126 lines
3.7 KiB
Python
126 lines
3.7 KiB
Python
import pytest
|
|
import time
|
|
import threading
|
|
import builtins
|
|
from io import StringIO
|
|
import colorama
|
|
from contextlib import contextmanager
|
|
|
|
from slic.utils.pv import PV
|
|
from morbidissimo import MorIOC
|
|
|
|
# === 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)
|
|
|
|
import re
|
|
|
|
def strip_ansi(text):
|
|
ansi_escape = re.compile(r'\x1b\[[0-9;]*m')
|
|
return ansi_escape.sub('', text)
|
|
|
|
# === Tests ===
|
|
@pytest.mark.parametrize("value, 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.GREEN),
|
|
(-50.0, colorama.Fore.GREEN)
|
|
])
|
|
|
|
def test_put_with_progress_and_repr(value, expected_color):
|
|
pv = PV("TEST:VAL", connection_timeout=2.0)
|
|
assert pv.wait_for_connection(timeout=2.0), "PV not connected"
|
|
pv.units = "//"
|
|
|
|
pv.put(0.0, wait=True)
|
|
assert pv.get() == pytest.approx(0.0)
|
|
|
|
# Capture tous les prints dans une liste
|
|
printed_lines = []
|
|
|
|
def fake_print(*args, **kwargs):
|
|
line = " ".join(str(a) for a in args)
|
|
printed_lines.append(line)
|
|
|
|
# Monkeypatch print uniquement dans ce contexte
|
|
original_print = builtins.print
|
|
builtins.print = fake_print
|
|
try:
|
|
pv.put(value, show_progress=True)
|
|
finally:
|
|
builtins.print = original_print
|
|
|
|
cleaned_lines = [strip_ansi(line) for line in printed_lines]
|
|
|
|
# Initialisation bar
|
|
matches = [line for line in cleaned_lines if f"| |" in line]
|
|
assert matches, f"Expected bar not found in:\n" + "\n".join(printed_lines)
|
|
|
|
# Vérifie que la bonne barre a été affichée au moins une fois
|
|
matches = [line for line in cleaned_lines if f"|██████████████████████████████|" in line]
|
|
assert matches, f"Expected bar not found in:\n" + "\n".join(printed_lines)
|
|
|
|
# Vérifie que la couleur est bien utilisée (au moins une fois dans les lignes printées)
|
|
color_matches = [line for line in printed_lines if expected_color in line]
|
|
assert color_matches, "Expected color code not found"
|
|
|
|
# Vérifie que la valeur finale est correcte
|
|
assert pv.get() == pytest.approx(value)
|
|
|
|
# Représentation
|
|
expected_repr = f'PV "TEST:VAL" at {value} //'
|
|
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 |