Update tests/test_utils_opmsg.py

tests/test_utils_opmsg.py

@@ -1,4 +1,3 @@
-import patch_put
 import time
 import threading
 import pytest
@@ -8,59 +7,69 @@ from slic.utils.opmsg import (
     IDS, BEAMLINES, N_MSG_HISTORY,
     OperationMessageStatus, OperationMessageEntry,
     OperationMessage, OperationMessages, MachineStatus,
-    clean_name, IDS_INVERSE,
+    clean_name, IDS_INVERSE
 )
 
-def start_single_ioc():
+def ioc():
     """
-    One MorIOC instance that serves ALL PVs to avoid port collisions.
+    IOC de test pour:
+      - STATUS / STATUS-DATE
+      - OP-DATEi / OP-MSGi
+      - OP-MSG-tmp / OP-MSG-TMP
+      - CATEGORY / DOWNTIME
     """
-    def run():
+    def run_op_prefix(prefix: str):
-        with MorIOC("") as mor:  # single IOC, no fixed prefix
+        with MorIOC(prefix) as mor:
-            # --- OPMSG tree ---
+            mor.host(
-            for ID in IDS.values():
+                STATUS={"type": "enum", "enums": ["OFFLINE", "PREPARATION", "REMOTE", "ATTENDED"]},
-                base = f"SF-OP:{ID}-MSG"
+                **{"STATUS-DATE": str},
-                # declare PVs
+                **{f"OP-DATE{i}": str for i in range(N_MSG_HISTORY)},
-                mor.host(
+                **{f"OP-MSG{i}": str for i in range(N_MSG_HISTORY)},
-                    **{f"{base}:STATUS": {"type": "enum", "enums": ["OFFLINE", "PREPARATION", "REMOTE", "ATTENDED"]}},
+                **{"OP-MSG-tmp": str},
-                    **{f"{base}:STATUS-DATE": str},
+                **{"OP-MSG-TMP": str},
-                    **{f"{base}:OP-MSG-tmp": str},
+            )
-                    **{f"{base}:OP-MSG-TMP": str},
+            mor.serve(
-                    **{f"{base}:OP-DATE{i}": str for i in range(N_MSG_HISTORY)},
+                STATUS="OFFLINE",
-                    **{f"{base}:OP-MSG{i}":  str for i in range(N_MSG_HISTORY)},
+                **{"STATUS-DATE": "2024-01-01 00:00:00"},
-                )
+                **{f"OP-DATE{i}": f"2024-01-01 00:00:0{i}" for i in range(N_MSG_HISTORY)},
-                # initial values
+                **{f"OP-MSG{i}": f"Initial message {i}" for i in range(N_MSG_HISTORY)},
-                mor.serve(
+                **{"OP-MSG-tmp": ""},
-                    **{f"{base}:STATUS": "OFFLINE"},
+                **{"OP-MSG-TMP": ""},
-                    **{f"{base}:STATUS-DATE": "2024-01-01 00:00:00"},
+            )
-                    **{f"{base}:OP-MSG-tmp": ""},
-                    **{f"{base}:OP-MSG-TMP": ""},
-                    **{f"{base}:OP-DATE{i}": f"2024-01-01 00:00:0{i}" for i in range(N_MSG_HISTORY)},
-                    **{f"{base}:OP-MSG{i}":  f"Initial message {i}" for i in range(N_MSG_HISTORY)},
-                )
-
-            # --- STATUS tree ---
-            for bl in BEAMLINES:
-                base = f"SF-STATUS-{bl}"
-                mor.host(
-                    **{f"{base}:CATEGORY": {"type": "enum", "enums": ["USER", "MD", "SD", "ACCESS", "DOWN"]}},
-                    **{f"{base}:DOWNTIME": str},
-                )
-                mor.serve(**{f"{base}:CATEGORY": "USER", f"{base}:DOWNTIME": "00:00:00"})
-
-            # serve loop
             while True:
                 mor.serve()
-                time.sleep(0.05)
+                time.sleep(0.1)
 
-    t = threading.Thread(target=run, daemon=True)
+    def run_status_prefix(beamline: str):
-    t.start()
+        # IMPORTANT: le code utilise SF-STATUS-{beamline}
-    return t
+        with MorIOC(f"SF-STATUS-{beamline}") as mor:
+            mor.host(
+                CATEGORY={"type": "enum", "enums": ["USER", "MD", "SD", "ACCESS", "DOWN"]},
+                DOWNTIME=str,
+            )
+            mor.serve(CATEGORY="USER", DOWNTIME="00:00:00")
+            while True:
+                mor.serve()
+                time.sleep(0.1)
+
+    threads = []
+    # IOCs pour opmsg (un par ID)
+    for ID in IDS.values():
+        prefix = f"SF-OP:{ID}-MSG"
+        t = threading.Thread(target=run_op_prefix, args=(prefix,), daemon=True)
+        t.start()
+        threads.append(t)
+    # IOCs pour machinestatus
+    for bl in BEAMLINES:
+        t = threading.Thread(target=run_status_prefix, args=(bl,), daemon=True)
+        t.start()
+        threads.append(t)
+    return threads
 
 @pytest.fixture(scope="module", autouse=True)
-def ioc_up():
+def run_all_iocs():
-    start_single_ioc()
+    threads = ioc()
-    time.sleep(1.0)  # give CA some time to announce PVs
+    time.sleep(2.0)  # laisser le temps aux serveurs de démarrer
     yield
 
 
@@ -80,14 +89,14 @@ class TestOperationMessageStatus:
         assert set(["OFFLINE", "PREPARATION", "REMOTE", "ATTENDED"]).issubset(set(allowed))
 
     def test_properties_date_status(self, status):
-        status.pv_date.put("2025-08-08 10:00:00")
+        status.pv_date.put("2025-08-08 10:00:00", wait=True)
-        status.pv_status.put("REMOTE")
+        status.pv_status.put("REMOTE", wait=True)
         assert status.date == "2025-08-08 10:00:00"
         assert status.status == "REMOTE"
 
     def test_repr_uses_properties(self, status):
-        status.pv_date.put("2025-08-08 10:01:02")
+        status.pv_date.put("2025-08-08 10:01:02", wait=True)
-        status.pv_status.put("ATTENDED")
+        status.pv_status.put("ATTENDED", wait=True)
         r = repr(status)
         assert "2025-08-08 10:01:02 ATTENDED" in r
 
@@ -127,19 +136,19 @@ class TestOperationMessageEntry:
 
     def test_date_property(self, entry):
         test_val = "2025-08-08 12:34:56"
-        entry.pv_date.put(test_val)
+        entry.pv_date.put(test_val, wait=True)
         assert entry.date == test_val
 
     def test_msg_property(self, entry):
         test_val = "Hello from test"
-        entry.pv_msg.put(test_val)
+        entry.pv_msg.put(test_val, wait=True)
         assert entry.msg == test_val
 
     def test_repr(self, entry):
         date_val = "2025-08-08 13:00:00"
         msg_val  = "System maintenance"
-        entry.pv_date.put(date_val)
+        entry.pv_date.put(date_val, wait=True)
-        entry.pv_msg.put(msg_val)
+        entry.pv_msg.put(msg_val, wait=True)
         # petite margede temps pour la lecture
         time.sleep(0.05)
         assert repr(entry) == f"{date_val} {msg_val}"