Update tests/test_utils_pv.py

tests/test_utils_pv.py

@@ -1,103 +1,91 @@
-# Fichier test_utils_pv.py corrigé
+# test_utils_pv.py
 
+import asyncio
+import threading
+import time
 import pytest
-from io import StringIO
 import colorama
+from io import StringIO
+from contextlib import contextmanager
+
+from caproto.server import pvproperty, PVGroup
+
+# --- IOC simulé avec caproto ---
+class TestIOC(PVGroup):
+    PV = pvproperty(value=0.0, name='TEST:PV', units='units')
+
+
+# --- Démarrage automatique de l'IOC ---
+@pytest.fixture(scope="module", autouse=True)
+def run_test_ioc():
+    ioc = TestIOC()
+
+    def run():
+        asyncio.run(ioc.serve(async_lib='asyncio'))
+
+    thread = threading.Thread(target=run, daemon=True)
+    thread.start()
+    time.sleep(1.5)
+    yield
+
+
+# --- Import de TA CLASSE PV depuis slic.utils.pv ---
 from slic.utils.pv import PV
-from epics import PV as EpicsPV
-
-class FakeEPICS(EpicsPV):
-    def __init__(self, pvname, **kwargs):
-        super().__init__(pvname, **kwargs)
-        self._value = 0.0  # Initial value set to 0.0
-        self.units = "units"
-        self._callbacks = {}
-        self.pvname = pvname  # Important pour le repr
-
-    def get(self, *args, **kwargs):
-        return self._value
-
-    def put(self, value, *args, **kwargs):
-        self._value = value
-        # Trigger all callbacks
-        for callback in self._callbacks.values():
-            callback(value=value, **kwargs)
-        return value
-
-    def add_callback(self, callback, **kwargs):
-        cb_index = len(self._callbacks) + 1
-        self._callbacks[cb_index] = callback
-        return cb_index
-
-    def remove_callback(self, cb_index):
-        self._callbacks.pop(cb_index, None)
-
-    def __repr__(self):
-        return f"PV {self.pvname} at {self._value} {self.units}"
-
-@pytest.fixture
-def fake_epics_pv(monkeypatch):
-    monkeypatch.setattr("epics.PV", FakeEPICS)
-    pv = PV("TEST:PV")
-    # Force initial value
-    pv.put(0.0, wait=True)
-    return pv
 
+# --- Capture stdout ---
 @pytest.fixture
 def capture_stdout(monkeypatch):
     buf = StringIO()
     monkeypatch.setattr("sys.stdout", buf)
     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')
+    (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_progress_and_repr(fake_epics_pv, capture_stdout, value, expected_bar, expected_color, expected_repr):
-    pv = fake_epics_pv
-    
-    # Verify initial value is properly set to 0.0
-    assert pv.get() == 0.0
-    
-    # Test put() with progress
+def test_put_with_progress(real_pv, capture_stdout, value, expected_bar, expected_color, expected_repr):
+    pv = real_pv
     pv.put(value, show_progress=True)
+
     output = capture_stdout.getvalue()
-    
-    # Verify progress bar display
     assert f"|{expected_bar}|" in output
     assert expected_color in output
     assert str(value) in output
-    
-    # Verify value was updated
+
     assert pv.get() == pytest.approx(value)
-    
-    # Test string representation
     assert repr(pv) == expected_repr
-    assert pv.orig_repr() == f"PV TEST:PV at {value} units"
-    
-    capture_stdout.truncate(0)
 
-def test_use_callback_context_manager(fake_epics_pv):
-    pv = fake_epics_pv
-    callback_log = []
-    
-    def test_callback(value, **kwargs):
-        callback_log.append(value)
 
-    # Verify no callbacks initially
-    assert len(pv._pv._callbacks) == 0
+def test_use_callback_context_manager(real_pv):
+    pv = real_pv
+    callback_values = []
 
-    with pv.use_callback(test_callback):
-        # Callback should be added
-        assert len(pv._pv._callbacks) == 1
-        
-        # Trigger callback
-        pv.put(10.0)
-        assert callback_log == [10.0]
+    def cb(value=None, **kwargs):
+        callback_values.append(value)
 
-    # Callback should be removed
-    assert len(pv._pv._callbacks) == 0
+    # Le callback est ajouté temporairement
+    with pv.use_callback(cb):
+        pv.put(42.0, wait=True)
+        time.sleep(0.1)
+
+    # Il a bien été appelé
+    assert any(val == 42.0 for val in callback_values)
+
+    # Hors du contexte : pas de rappel
+    callback_values.clear()
+    pv.put(50.0, wait=True)
+    time.sleep(0.1)
+    assert callback_values == []