Update tests/test_utils_opmsg.py

tests/test_utils_opmsg.py

@@ -3,36 +3,31 @@ import threading
 import pytest
 from morbidissimo import MorIOC
 
-from slic.utils.opmsg import *
-
-
-import time
-import threading
-import pytest
-from morbidissimo import MorIOC
-from slic.utils.opmsg import IDS, BEAMLINES, N_MSG_HISTORY
+from slic.utils.opmsg import (
+    IDS, BEAMLINES, N_MSG_HISTORY,
+    OperationMessageStatus, OperationMessageEntry,
+    OperationMessage, OperationMessages, MachineStatus,
+    clean_name, IDS_INVERSE
+)
 
 def ioc():
     """
     IOC de test pour:
       - STATUS / STATUS-DATE
       - OP-DATEi / OP-MSGi
-      - OP-MSG-tmp (exactement ce que ton code utilise)
+      - OP-MSG-tmp / OP-MSG-TMP
       - CATEGORY / DOWNTIME
     """
     def run_op_prefix(prefix: str):
         with MorIOC(prefix) as mor:
-            # Crée tous les PVs nécessaires
             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},  # Les deux cas pour couvrir toutes les possibilités
+                **{"OP-MSG-TMP": str},
             )
-
-            # Initialise les valeurs
             mor.serve(
                 STATUS="OFFLINE",
                 **{"STATUS-DATE": "2024-01-01 00:00:00"},
@@ -41,13 +36,13 @@ def ioc():
                 **{"OP-MSG-tmp": ""},
                 **{"OP-MSG-TMP": ""},
             )
-
             while True:
                 mor.serve()
                 time.sleep(0.1)
 
     def run_status_prefix(beamline: str):
-        with MorIOC(f"STATUS-{beamline}") as mor:
+        # 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,
@@ -58,30 +53,27 @@ def ioc():
                 time.sleep(0.1)
 
     threads = []
-    
-    # Crée les IOCs pour tous les IDs
+    # 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)
-    
-    # Crée les IOCs pour les beamlines
+    # 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 run_all_iocs():
     threads = ioc()
-    time.sleep(1.0)  
+    time.sleep(1.0)  # laisser le temps aux serveurs de démarrer
     yield
 
 
-# Tests: OperationMessageStatus
+# -------- OperationMessageStatus --------
 class TestOperationMessageStatus:
     @pytest.fixture
     def status(self):
@@ -91,24 +83,18 @@ class TestOperationMessageStatus:
         return s
 
     def test_initialization_and_enum(self, status):
-        # Ensure PVs are created and connected, and allowed values match expectations
-        assert status.pv_date is not None
-        assert status.pv_status is not None
         assert status.pv_date.connected
         assert status.pv_status.connected
-
         allowed = status.get_allowed()
         assert set(["OFFLINE", "PREPARATION", "REMOTE", "ATTENDED"]).issubset(set(allowed))
 
     def test_properties_date_status(self, status):
-        # Verify the .date and .status properties return correct values
         status.pv_date.put("2025-08-08 10:00:00", wait=True)
         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):
-        # Check that __repr__ includes both date and status
         status.pv_date.put("2025-08-08 10:01:02", wait=True)
         status.pv_status.put("ATTENDED", wait=True)
         r = repr(status)
@@ -134,7 +120,7 @@ class TestOperationMessageStatus:
         assert "not from allowed values" in str(exc.value)
 
 
-# Tests: OperationMessageEntry
+# -------- OperationMessageEntry --------
 class TestOperationMessageEntry:
     @pytest.fixture(params=range(N_MSG_HISTORY))
     def entry(self, request):
@@ -145,8 +131,6 @@ class TestOperationMessageEntry:
         return e
 
     def test_initialization(self, entry):
-        assert entry.pv_date is not None
-        assert entry.pv_msg is not None
         assert entry.pv_date.connected
         assert entry.pv_msg.connected
 
@@ -162,14 +146,15 @@ class TestOperationMessageEntry:
 
     def test_repr(self, entry):
         date_val = "2025-08-08 13:00:00"
-        msg_val = "System maintenance"
+        msg_val  = "System maintenance"
         entry.pv_date.put(date_val, wait=True)
         entry.pv_msg.put(msg_val, wait=True)
-        expected_repr = f"{date_val} {msg_val}"
-        assert repr(entry) == expected_repr
+        # petite margede temps pour la lecture
+        time.sleep(0.05)
+        assert repr(entry) == f"{date_val} {msg_val}"
 
 
-# Tests: OperationMessage
+# -------- OperationMessage --------
 class TestOperationMessage:
     def test_init_with_name(self):
         om = OperationMessage(name="Control Room")
@@ -187,7 +172,7 @@ class TestOperationMessage:
     def test_init_with_id(self):
         om = OperationMessage(ID="cr")  # case-insensitive
         assert om.ID == "CR"
-        assert om.name == IDS_INVERSE["CR"]  # "Control Room"
+        assert om.name == IDS_INVERSE["CR"]
         assert om.prefix == "SF-OP:CR-MSG"
 
     def test_init_error_when_missing_both(self):
@@ -205,6 +190,8 @@ class TestOperationMessage:
         assert om.pv_send.wait_for_connection(timeout=2.0)
         msg = "Beam down at 14:00"
         om.update(msg)
+        # laisser l’IOC cycler (serve() toutes 100 ms)
+        time.sleep(0.2)
         got = om.pv_send.get(as_string=True)
         if isinstance(got, bytes):
             got = got.decode("utf-8", errors="ignore")
@@ -212,7 +199,6 @@ class TestOperationMessage:
 
     def test_status_object_is_wired(self):
         om = OperationMessage(name="Control Room")
-        assert isinstance(om.status, OperationMessageStatus)
         om.status.update("REMOTE")
         assert om.status.status == "REMOTE"
 
@@ -228,15 +214,13 @@ class TestOperationMessage:
         assert "Slot A ok" in r
 
 
-# Tests: OperationMessages 
+# -------- OperationMessages --------
 class TestOperationMessages:
     def test_initialization_populates_entries_and_items(self):
         oms = OperationMessages()
 
-        # One entry per human-readable name
         assert set(oms.entries.keys()) == set(IDS.keys())
 
-        # Each value is a wired/connected OperationMessage
         for name, om in oms.entries.items():
             assert isinstance(om, OperationMessage)
             assert om.pv_send.wait_for_connection(timeout=2.0)
@@ -246,7 +230,6 @@ class TestOperationMessages:
                 assert entry.pv_date.wait_for_connection(timeout=2.0), f"{om.prefix}:OP-DATE{i} not connected"
                 assert entry.pv_msg.wait_for_connection(timeout=2.0), f"{om.prefix}:OP-MSG{i} not connected"
 
-        # Dynamic attributes exist (cleaned names → attributes)
         for name in IDS.keys():
             attr = clean_name(name)
             assert hasattr(oms, attr), f"Missing attribute: {attr}"
@@ -268,7 +251,7 @@ class TestOperationMessages:
         assert isinstance(om_by_id, OperationMessage)
         assert om_by_id is om_by_name
 
-        om_by_original_name = oms["Control Room"] 
+        om_by_original_name = oms["Control Room"]
         assert om_by_original_name is om_by_name
 
         with pytest.raises(KeyError):
@@ -280,8 +263,8 @@ class TestOperationMessages:
         assert len(vals) == len(IDS)
         assert all(isinstance(v, OperationMessage) for v in vals)
 
-        # All prefixes serve identical initial data; verify contents quickly
         for om in vals:
+            # état initial injecté par l’IOC de test
             assert om.status.status == "OFFLINE"
             assert om.status.date == "2024-01-01 00:00:00"
             for i, entry in enumerate(om.entries):
@@ -299,15 +282,15 @@ class TestOperationMessages:
     def test_repr_includes_each_submessage_repr(self):
         oms = OperationMessages()
         r = repr(oms)
-        assert "Control Room (CR)" in r  # header from one submessage
+        assert "Control Room (CR)" in r
 
         om_cr = oms["control_room"]
-        # Modify two consecutive entries
         om_cr.entries[0].pv_date.put("2025-08-08 12:00:00", wait=True)
         om_cr.entries[0].pv_msg.put("Test message 1", wait=True)
         om_cr.entries[1].pv_date.put("2025-08-08 12:05:00", wait=True)
         om_cr.entries[1].pv_msg.put("Test message 2", wait=True)
 
+        time.sleep(0.05)
         r2 = repr(oms)
         assert "2025-08-08 12:00:00" in r2
         assert "Test message 1" in r2
@@ -315,7 +298,7 @@ class TestOperationMessages:
         assert "Test message 2" in r2
 
 
-# Tests: MachineStatus
+# -------- MachineStatus --------
 class TestMachineStatus:
     @pytest.mark.parametrize("beamline", BEAMLINES)
     def test_init_and_connections(self, beamline):
@@ -336,7 +319,6 @@ class TestMachineStatus:
         ms.pv_downtime.put("01:23:45", wait=True)
         assert ms.category == "MD"
         assert ms.downtime == "01:23:45"
-
         ms.pv_category.put("SD", wait=True)
         ms.pv_downtime.put("00:10:00", wait=True)
         assert ms.category == "SD"