Update tests/test_utils_pv.py

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():
+    # Define async loop to serve IOC
-        asyncio.run(ioc.serve(async_lib='asyncio'))
+    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 ---
+# Helper to capture stdout output
-from slic.utils.pv import PV
-
-# --- Capture stdout ---
 @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.mark.parametrize("value, expected_bar, expected_color", [
-@pytest.fixture
+    (25.0, "██▌       ", colorama.Fore.GREEN),
-def real_pv():
+    (50.0, "█████     ", colorama.Fore.GREEN),
-    pv = PV("TEST:PV")
+    (75.0, "███████▌  ", colorama.Fore.GREEN),
-    assert pv.get() == 0.0
+    (100.0, "██████████", colorama.Fore.GREEN),
-    return pv
+    (150.0, ">>>>>>>>>>", colorama.Fore.RED),
-
+    (-50.0, "<<<<<<<<<<", colorama.Fore.RED)
-
-@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'),
 ])
-def test_put_with_progress(real_pv, capture_stdout, value, expected_bar, expected_color, expected_repr):
+def test_put_with_progress_and_repr(capture_stdout, value, expected_bar, expected_color):
-    pv = real_pv
+    # 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('<epics.pv.PV')
 
 
-def test_use_callback_context_manager(real_pv):
+def test_use_callback_context_manager():
-    pv = real_pv
     callback_values = []
 
+    pv = PV("TEST:VAL")
+    pv.put(0.0, wait=True)
+
     def cb(value=None, **kwargs):
         callback_values.append(value)
 
-    # Le callback est ajouté temporairement
+    # Ensure initially no callback is set
+    initial_callbacks = len(pv._callbacks)
+
     with pv.use_callback(cb):
-        pv.put(42.0, wait=True)
+        assert len(pv._callbacks) == initial_callbacks + 1
-        time.sleep(0.1)
+        pv.put(10.0, wait=True)
+        assert callback_values[-1] == 10.0
 
-    # Il a bien été appelé
+    # Callback should be removed after context exit
-    assert any(val == 42.0 for val in callback_values)
+    assert len(pv._callbacks) == initial_callbacks
-
-    # Hors du contexte : pas de rappel
-    callback_values.clear()
-    pv.put(50.0, wait=True)
-    time.sleep(0.1)
-    assert callback_values == []