79 lines
2.2 KiB
Python
79 lines
2.2 KiB
Python
import pytest
|
|
import time
|
|
from io import StringIO
|
|
import colorama
|
|
|
|
from epicman import EpicManServer
|
|
from slic.utils.pv import PV # Ton wrapper autour de epics.PV
|
|
|
|
|
|
@pytest.fixture(scope="module", autouse=True)
|
|
def start_epicman():
|
|
# Lancer un PV simulé avec une valeur initiale
|
|
server = EpicManServer(pvs={"TEST:VAL": 0.0})
|
|
server.start()
|
|
time.sleep(1) # attendre que le serveur soit prêt
|
|
yield server
|
|
server.stop()
|
|
|
|
|
|
@pytest.fixture
|
|
def capture_stdout(monkeypatch):
|
|
buf = StringIO()
|
|
monkeypatch.setattr("sys.stdout", buf)
|
|
return buf
|
|
|
|
|
|
@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)
|
|
|
|
pv.wait_for_connection(timeout=2.0)
|
|
assert pv.connected
|
|
|
|
# Mise à 0 initiale
|
|
pv.put(0.0, wait=True)
|
|
assert pv.get() == pytest.approx(0.0)
|
|
|
|
# Put avec la barre de progression
|
|
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(start_epicman):
|
|
pv = PV("TEST:VAL", connection_timeout=2.0)
|
|
pv.put(0.0, wait=True)
|
|
|
|
seen_values = []
|
|
|
|
def callback(value=None, **kwargs):
|
|
seen_values.append(value)
|
|
|
|
initial_count = len(pv._callbacks)
|
|
|
|
# Gestion du contexte de callback
|
|
with pv.use_callback(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
|