From 5a1f37e209f3a40717e8051efff0896c36f6686e Mon Sep 17 00:00:00 2001 From: tligui_y Date: Fri, 8 Aug 2025 16:00:33 +0200 Subject: [PATCH] Update tests/test_utils_opmsg.py --- tests/test_utils_opmsg.py | 76 +++++++++++++++------------------------ 1 file changed, 29 insertions(+), 47 deletions(-) diff --git a/tests/test_utils_opmsg.py b/tests/test_utils_opmsg.py index d552db5b1..64a6d5ffa 100644 --- a/tests/test_utils_opmsg.py +++ b/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"