Update tests/test_utils_opmsg.py

2025-08-09 00:45:51 +02:00
parent eb61546b79
commit 5eb56c4301
1 changed files with 49 additions and 54 deletions
@@ -4,68 +4,63 @@ import threading
 import pytest
 from morbidissimo import MorIOC

-from slic.utils.opmsg import *
+from slic.utils.opmsg import (
+    IDS, BEAMLINES, N_MSG_HISTORY,
+    OperationMessageStatus, OperationMessageEntry,
+    OperationMessage, OperationMessages, MachineStatus,
+    clean_name, IDS_INVERSE,
+)

-def ioc():
+def start_single_ioc():
    """
-    IOC de test pour:
-      - STATUS / STATUS-DATE
-      - OP-DATEi / OP-MSGi
-      - OP-MSG-tmp / OP-MSG-TMP
-      - CATEGORY / DOWNTIME
+    One MorIOC instance that serves ALL PVs to avoid port collisions.
    """
-    def run_op_prefix(prefix: str):
-        with MorIOC(prefix) as mor:
-            mor.host(
-                STATUS={"type": "enum", "enums": ["OFFLINE", "PREPARATION", "REMOTE", "ATTENDED"]},
-                **{"STATUS-DATE": str},
-                **{f"OP-DATE{i}": str for i in range(N_MSG_HISTORY)},
-                **{f"OP-MSG{i}": str for i in range(N_MSG_HISTORY)},
-                **{"OP-MSG-tmp": str},
-                **{"OP-MSG-TMP": str},
-            )
-            mor.serve(
-                STATUS="OFFLINE",
-                **{"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)},
-                **{f"OP-MSG{i}": f"Initial message {i}" for i in range(N_MSG_HISTORY)},
-                **{"OP-MSG-tmp": ""},
-                **{"OP-MSG-TMP": ""},
-            )
+    def run():
+        with MorIOC("") as mor:  # single IOC, no fixed prefix
+            # --- OPMSG tree ---
+            for ID in IDS.values():
+                base = f"SF-OP:{ID}-MSG"
+                # declare PVs
+                mor.host(
+                    **{f"{base}:STATUS": {"type": "enum", "enums": ["OFFLINE", "PREPARATION", "REMOTE", "ATTENDED"]}},
+                    **{f"{base}:STATUS-DATE": str},
+                    **{f"{base}:OP-MSG-tmp": str},
+                    **{f"{base}:OP-MSG-TMP": str},
+                    **{f"{base}:OP-DATE{i}": str for i in range(N_MSG_HISTORY)},
+                    **{f"{base}:OP-MSG{i}":  str for i in range(N_MSG_HISTORY)},
+                )
+                # initial values
+                mor.serve(
+                    **{f"{base}:STATUS": "OFFLINE"},
+                    **{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.1)
+                time.sleep(0.05)

-    def run_status_prefix(beamline: str):
-        # IMPORTANT: le code utilise SF-STATUS-{beamline}
-        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
+    t = threading.Thread(target=run, daemon=True)
+    t.start()
+    return t

@pytest.fixture(scope="module", autouse=True)
-def run_all_iocs():
-    threads = ioc()
-    time.sleep(2.0)  # laisser le temps aux serveurs de démarrer
+def ioc_up():
+    start_single_ioc()
+    time.sleep(1.0)  # give CA some time to announce PVs
    yield