92 lines
2.7 KiB
Python
92 lines
2.7 KiB
Python
import pytest
|
|
import time
|
|
import threading
|
|
from io import StringIO
|
|
import colorama
|
|
from contextlib import contextmanager
|
|
|
|
from slic.utils.pv import PV
|
|
from morbidissimo import MorIOC # alias 'cs' dans votre repo
|
|
|
|
# === IOC simulé avec MorIOC ===
|
|
@pytest.fixture(scope="module", autouse=True)
|
|
def morioc_server():
|
|
def run_ioc():
|
|
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 serveur soit prêt
|
|
|
|
yield # le thread tourne en fond
|
|
|
|
# === Utils ===
|
|
@pytest.fixture
|
|
def capture_stdout(monkeypatch):
|
|
buf = StringIO()
|
|
monkeypatch.setattr("sys.stdout", buf)
|
|
return buf
|
|
|
|
@contextmanager
|
|
def use_callback(pv, callback):
|
|
cbid = pv.add_callback(callback)
|
|
try:
|
|
yield
|
|
finally:
|
|
pv.remove_callback(cbid)
|
|
|
|
# === Tests ===
|
|
@pytest.mark.parametrize("value, expected_bar, expected_color", [
|
|
(25.0, "██▌ ", colorama.Fore.GREEN),
|
|
(50.0, "█████ ", colorama.Fore.GREEN),
|
|
(75.0, "███████▌ ", colorama.Fore.GREEN),
|
|
(100.0, "██████████", colorama.Fore.GREEN),
|
|
(150.0, ">>>>>>>>>>", colorama.Fore.RED),
|
|
(-50.0, "<<<<<<<<<<", colorama.Fore.RED)
|
|
])
|
|
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), "PV not connected"
|
|
|
|
pv.put(0.0, wait=True)
|
|
assert pv.get() == pytest.approx(0.0)
|
|
|
|
pv.put(value, show_progress=True)
|
|
output = capture_stdout.getvalue()
|
|
|
|
assert f"|{expected_bar}|" in output
|
|
assert expected_color in output
|
|
assert str(value) in output
|
|
assert pv.get() == pytest.approx(value)
|
|
|
|
expected_repr = f'PV "TEST:VAL" at {value} units'
|
|
assert repr(pv) == expected_repr
|
|
assert pv.orig_repr().startswith('<epics.pv.PV')
|
|
|
|
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 = []
|
|
|
|
def callback(value=None, **kwargs):
|
|
seen_values.append(value)
|
|
|
|
initial_count = len(pv._callbacks)
|
|
|
|
with use_callback(pv, callback):
|
|
assert len(pv._callbacks) == initial_count + 1
|
|
pv.put(42.0, wait=True)
|
|
time.sleep(0.2)
|
|
assert 42.0 in seen_values
|
|
|
|
assert len(pv._callbacks) == initial_count |