Update tests/test_utils_pv.py

tests/test_utils_pv.py

@@ -1,51 +1,22 @@
 import pytest
-import asyncio
-import threading
 import time
-from contextlib import contextmanager
 from io import StringIO
-
 import colorama
-from slic.utils.pv import PV
-# Caproto is used to simulate a real EPICS IOC for testing
-from caproto.server import pvproperty, PVGroup, ioc_arg_parser, run
-import caproto.server as cs
 
-# IOC definition: a single PV TEST:VAL with units
-class TestIOC(PVGroup):
-    val = pvproperty(value=0.0, units='units')
-
-    def __init__(self, prefix, **kwargs):
-        super().__init__(prefix=prefix, **kwargs)
+from epicman import EpicManServer
+from slic.utils.pv import PV  # Ton wrapper autour de epics.PV
 
 
-# Fixture to launch IOC in a background thread before tests
 @pytest.fixture(scope="module", autouse=True)
-def run_test_ioc():
-    prefix = "TEST:"
-    ioc = TestIOC(prefix=prefix)
-
-    # Manually construct run_options to avoid argparse issues
-    run_options = cs.RunOptions(
-        prefix=prefix,
-        interfaces=["127.0.0.1"],
-        log_pv_names=True,
-        startup_hook=None,
-        async_lib='asyncio',
-        simulate=False,
-        macros={}
-    )
-
-    print("✅ Starting IOC with prefix", prefix)
-    ioc.start(run_options)
-    import time
-    time.sleep(1)  # give some time to start up
-
-    yield
+def start_epicman():
+    # Lancer un PV simulé avec une valeur initiale
+    server = EpicManServer(pvs={"TEST:VAL": 0.0})
+    server.start()
+    time.sleep(1)  # attendre que le serveur soit prêt
+    yield server
+    server.stop()
 
 
-
-# Fixture to capture stdout (for tqdm / progress bar)
 @pytest.fixture
 def capture_stdout(monkeypatch):
     buf = StringIO()
@@ -62,48 +33,34 @@ def capture_stdout(monkeypatch):
     (-50.0, "<<<<<<<<<<", colorama.Fore.RED)
 ])
 def test_put_with_progress_and_repr(capture_stdout, value, expected_bar, expected_color):
-    # Use the custom PV class
-    pv = PV("TEST:VAL")
+    pv = PV("TEST:VAL", connection_timeout=2.0)
 
     pv.wait_for_connection(timeout=2.0)
     assert pv.connected
 
-    # Set to initial value
+    # Mise à 0 initiale
     pv.put(0.0, wait=True)
+    assert pv.get() == pytest.approx(0.0)
 
-    # Wait for update to propagate
-    for _ in range(20):
-        val = pv.get()
-        if val is not None:
-            break
-        time.sleep(0.1)
-
-    assert val is not None
-    assert val == 0.0
-
-    # Put new value with progress bar
+    # Put avec la barre de progression
     pv.put(value, show_progress=True)
 
     output = capture_stdout.getvalue()
-
-    # Verify visual bar and color
     assert f"|{expected_bar}|" in output
     assert expected_color in output
     assert str(value) in output
 
-    # Verify value updated
     assert pv.get() == pytest.approx(value)
 
-    # Check custom and original 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():
-    pv = PV("TEST:VAL")
+def test_use_callback_context_manager(start_epicman):
+    pv = PV("TEST:VAL", connection_timeout=2.0)
     pv.put(0.0, wait=True)
- 
+
     seen_values = []
 
     def callback(value=None, **kwargs):
@@ -111,12 +68,11 @@ def test_use_callback_context_manager():
 
     initial_count = len(pv._callbacks)
 
-    # This is the function under test
-    with use_callback(pv, callback):
+    # Gestion du contexte de callback
+    with pv.use_callback(callback):
         assert len(pv._callbacks) == initial_count + 1
         pv.put(42.0, wait=True)
-        time.sleep(0.2)  # Ensure callback has time to trigger
+        time.sleep(0.2)
         assert 42.0 in seen_values
 
-    # After context manager exit, callback should be removed
     assert len(pv._callbacks) == initial_count