89 lines
2.4 KiB
Python
89 lines
2.4 KiB
Python
'''
|
|
|
|
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
|
|
|
|
# simulate un PV en background
|
|
@pytest.fixture(scope="module", autouse=True)
|
|
def softioc_server():
|
|
def run_ioc():
|
|
builder.SetDeviceName("TEST")
|
|
pv = builder.aOut('VAL', initial_value=0.0)
|
|
builder.LoadDatabase()
|
|
softioc.iocInit()
|
|
softioc.run()
|
|
|
|
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
|
|
|
|
@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)
|
|
|
|
@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.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)
|
|
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
|
|
''' |