diff --git a/tests/test_utils_pv.py b/tests/test_utils_pv.py index 38ba95ab0..2fbf937dc 100644 --- a/tests/test_utils_pv.py +++ b/tests/test_utils_pv.py @@ -1,38 +1,45 @@ -# test_utils_pv.py - +import pytest import asyncio import threading import time -import pytest -import colorama -from io import StringIO from contextlib import contextmanager +from io import StringIO -from caproto.server import pvproperty, PVGroup +import colorama +from slic.utils.pv import PV # The custom PV class to test -# --- IOC simulé avec caproto --- +# -------- caproto imports for the test IOC -------- +from caproto.server import pvproperty, PVGroup, run + +# Create a simple IOC with one PV (TEST:VAL) class TestIOC(PVGroup): - PV = pvproperty(value=0.0, name='TEST:PV', units='units') + val = pvproperty(value=0.0, units='units') + + def __init__(self, prefix, **kwargs): + super().__init__(prefix=prefix, **kwargs) -# --- Démarrage automatique de l'IOC --- +# Run the IOC in a background thread (shared across module) @pytest.fixture(scope="module", autouse=True) def run_test_ioc(): - ioc = TestIOC() + prefix = "TEST:" + ioc = TestIOC(prefix=prefix) - def run(): - asyncio.run(ioc.serve(async_lib='asyncio')) + # Define async loop to serve IOC + def start_ioc(): + asyncio.run(run(ioc.pvdb, startup_hook=None)) - thread = threading.Thread(target=run, daemon=True) + thread = threading.Thread(target=start_ioc, daemon=True) thread.start() + + # Wait a moment to ensure IOC is ready before test starts time.sleep(1.5) + + # Nothing to yield; test can now run yield -# --- Import de TA CLASSE PV depuis slic.utils.pv --- -from slic.utils.pv import PV - -# --- Capture stdout --- +# Helper to capture stdout output @pytest.fixture def capture_stdout(monkeypatch): buf = StringIO() @@ -40,52 +47,58 @@ def capture_stdout(monkeypatch): return buf -# --- PV instancié sur l’IOC Caproto en live --- -@pytest.fixture -def real_pv(): - pv = PV("TEST:PV") - assert pv.get() == 0.0 - return pv - - -@pytest.mark.parametrize("value, expected_bar, expected_color, expected_repr", [ - (25.0, "██▌ ", colorama.Fore.GREEN, 'PV "TEST:PV" at 25.0 units'), - (50.0, "█████ ", colorama.Fore.GREEN, 'PV "TEST:PV" at 50.0 units'), - (75.0, "███████▌ ", colorama.Fore.GREEN, 'PV "TEST:PV" at 75.0 units'), - (100.0, "██████████", colorama.Fore.GREEN, 'PV "TEST:PV" at 100.0 units'), - (150.0, ">>>>>>>>>>", colorama.Fore.RED, 'PV "TEST:PV" at 150.0 units'), - (-50.0, "<<<<<<<<<<", colorama.Fore.RED, 'PV "TEST:PV" at -50.0 units'), +@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(real_pv, capture_stdout, value, expected_bar, expected_color, expected_repr): - pv = real_pv +def test_put_with_progress_and_repr(capture_stdout, value, expected_bar, expected_color): + # Connect to the test PV using the custom slic PV class + pv = PV("TEST:VAL") + + # Set initial value to 0.0 + pv.put(0.0, wait=True) + assert pv.get() == 0.0 + + # Call put() with progress enabled pv.put(value, show_progress=True) + # Get printed output from RangeBar output = capture_stdout.getvalue() + + # Check progress bar content and color assert f"|{expected_bar}|" in output assert expected_color in output assert str(value) in output + # Verify PV value was correctly updated assert pv.get() == pytest.approx(value) + + # Check repr + expected_repr = f'PV "TEST:VAL" at {value} units' assert repr(pv) == expected_repr + assert pv.orig_repr().startswith('