This commit is contained in:
+21
-18
@@ -1,32 +1,34 @@
|
||||
'''
|
||||
|
||||
import pytest
|
||||
import time
|
||||
import threading
|
||||
from io import StringIO
|
||||
import colorama
|
||||
|
||||
from slic.utils.pv import PV
|
||||
from contextlib import contextmanager
|
||||
|
||||
# softioc
|
||||
from softioc import softioc, builder
|
||||
from slic.utils.pv import PV
|
||||
from morbidissimo import MorIOC # alias 'cs' dans votre repo
|
||||
|
||||
# simulate un PV en background
|
||||
# === IOC simulé avec MorIOC ===
|
||||
@pytest.fixture(scope="module", autouse=True)
|
||||
def softioc_server():
|
||||
def morioc_server():
|
||||
def run_ioc():
|
||||
builder.SetDeviceName("TEST")
|
||||
pv = builder.aOut('VAL', initial_value=0.0)
|
||||
builder.LoadDatabase()
|
||||
softioc.iocInit()
|
||||
softioc.run()
|
||||
with MorIOC("TEST") as mor:
|
||||
val = 0.0
|
||||
while True:
|
||||
mor.host(VAL=float)
|
||||
input_val = mor.get("VAL")
|
||||
if input_val is not None:
|
||||
val = input_val
|
||||
mor.serve(VAL=val)
|
||||
time.sleep(0.1)
|
||||
|
||||
thread = threading.Thread(target=run_ioc, daemon=True)
|
||||
thread.start()
|
||||
time.sleep(1) # attendre que le PV soit en ligne
|
||||
yield # rien à retourner, il tourne en fond
|
||||
time.sleep(1) # attendre que le serveur soit prêt
|
||||
|
||||
yield # le thread tourne en fond
|
||||
|
||||
# === Utils ===
|
||||
@pytest.fixture
|
||||
def capture_stdout(monkeypatch):
|
||||
buf = StringIO()
|
||||
@@ -41,6 +43,7 @@ def use_callback(pv, callback):
|
||||
finally:
|
||||
pv.remove_callback(cbid)
|
||||
|
||||
# === Tests ===
|
||||
@pytest.mark.parametrize("value, expected_bar, expected_color", [
|
||||
(25.0, "██▌ ", colorama.Fore.GREEN),
|
||||
(50.0, "█████ ", colorama.Fore.GREEN),
|
||||
@@ -51,7 +54,7 @@ def use_callback(pv, callback):
|
||||
])
|
||||
def test_put_with_progress_and_repr(capture_stdout, value, expected_bar, expected_color):
|
||||
pv = PV("TEST:VAL", connection_timeout=2.0)
|
||||
assert pv.wait_for_connection(timeout=2.0)
|
||||
assert pv.wait_for_connection(timeout=2.0), "PV not connected"
|
||||
|
||||
pv.put(0.0, wait=True)
|
||||
assert pv.get() == pytest.approx(0.0)
|
||||
@@ -70,6 +73,7 @@ def test_put_with_progress_and_repr(capture_stdout, value, expected_bar, expecte
|
||||
|
||||
def test_use_callback_context_manager():
|
||||
pv = PV("TEST:VAL", connection_timeout=2.0)
|
||||
assert pv.wait_for_connection(timeout=2.0), "PV not connected"
|
||||
pv.put(0.0, wait=True)
|
||||
|
||||
seen_values = []
|
||||
@@ -85,5 +89,4 @@ def test_use_callback_context_manager():
|
||||
time.sleep(0.2)
|
||||
assert 42.0 in seen_values
|
||||
|
||||
assert len(pv._callbacks) == initial_count
|
||||
'''
|
||||
assert len(pv._callbacks) == initial_count
|
||||
Reference in New Issue
Block a user