diff --git a/tests/test_utils_opmsg.py b/tests/test_utils_opmsg.py index b5d95179e..90fe145a8 100644 --- a/tests/test_utils_opmsg.py +++ b/tests/test_utils_opmsg.py @@ -1,47 +1,64 @@ import time import threading import pytest -from morbidissimo import MorIOC -from slic.utils.opmsg import * +from epics import PV +from slic.utils.opmsg * -import time -import threading -import pytest -from morbidissimo import MorIOC -from slic.utils.opmsg import IDS, BEAMLINES, N_MSG_HISTORY +# ----------------------------- +# IOC: start all prefixes used by the SUT +# ----------------------------- def ioc(): """ - IOC de test pour: - - STATUS / STATUS-DATE - - OP-DATEi / OP-MSGi - - OP-MSG-tmp (exactement ce que ton code utilise) - - CATEGORY / DOWNTIME + Start MorIOC servers for ALL PVs used by: + - OperationMessageStatus (STATUS, STATUS-DATE) + - OperationMessageEntry (OP-DATEi, OP-MSGi) + - OperationMessage (OP-MSG-tmp) + - MachineStatus (CATEGORY, DOWNTIME) + + One MorIOC per prefix: + - For each ID in IDS: 'SF-OP:{ID}-MSG' + - For each beamline in BEAMLINES:'SF-STATUS-{beamline}' + + Important decisions: + - Enums are declared with pcaspy-style dicts so enum_strs are present. + - OP-MSG-tmp is hosted EXACTLY as the production code expects (lowercase), + and aliased to OP-MSG-TMP to be robust. + - We seed values ONCE; inside the loop we only call mor.serve() to + process CA traffic, without clobbering test writes. """ + from morbidissimo import MorIOC # imported here to avoid test env surprises + def run_op_prefix(prefix: str): with MorIOC(prefix) as mor: - mor.host( - STATUS={"type": "enum", "enums": ["OFFLINE", "PREPARATION", "REMOTE", "ATTENDED"]}, - **{"STATUS-DATE": str}, + pvdb = { + # true enum (pcaspy dict) so enum_strs is populated + "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}, - ) - # Seed une seule fois + # Host both variants; production uses the lowercase one. + # Canonical storage: + "OP-MSG-TMP": str, + # Alias with the exact lowercase name used in production code: + "OP-MSG-tmp": {"alias": "OP-MSG-TMP"}, + } + mor.host(**pvdb) + + # Seed ONCE 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": ""}, ) while True: + # process CA traffic without overwriting user/test writes mor.serve() time.sleep(0.02) @@ -51,12 +68,15 @@ def ioc(): CATEGORY={"type": "enum", "enums": ["USER", "MD", "SD", "ACCESS", "DOWN"]}, DOWNTIME=str, ) + # Seed ONCE mor.serve(CATEGORY="USER", DOWNTIME="00:00:00") + while True: mor.serve() time.sleep(0.02) threads = [] + for ID in IDS.values(): prefix = f"SF-OP:{ID}-MSG" t = threading.Thread(target=run_op_prefix, args=(prefix,), daemon=True) @@ -70,62 +90,83 @@ def ioc(): return threads -@pytest.fixture(scope="module", autouse=True) -def run_all_iocs(): - threads = ioc() - time.sleep(0.5) - yield - @pytest.fixture(scope="module", autouse=True) def run_all_iocs(): threads = ioc() - # Short boot delay to ensure IOCs are listening - time.sleep(0.3) + # Slightly longer boot time to ensure all servers are listening + time.sleep(1.0) yield # Daemon threads exit with pytest -# Autouse fixture to start IOC once per module -@pytest.fixture(scope="module", autouse=True) -def run_all_iocs(): - threads = ioc() - # Small boot delay so all IOCs are ready before the tests run - time.sleep(0.3) - yield - # Daemon threads end with pytest +# ----------------------------- +# Helpers for tests +# ----------------------------- + +def _ensure_enum_metadata(pv: PV): + """ + Ensure pyepics has fetched enum metadata; if not, force a read once. + """ + allowed = getattr(pv, "enum_strs", None) + if not allowed: + _ = pv.get(as_string=True) + allowed = getattr(pv, "enum_strs", None) + return allowed +def _reset_operation_message_to_defaults(om: OperationMessage): + """ + Reset a full OperationMessage object to the default IOC seeded values. + """ + om.status.pv_status.put("OFFLINE", wait=True) + om.status.pv_date.put("2024-01-01 00:00:00", wait=True) + for i, e in enumerate(om.entries): + e.pv_date.put(f"2024-01-01 00:00:0{i}", wait=True) + e.pv_msg.put(f"Initial message {i}", wait=True) + # tmp channel + # (the production object points to OP-MSG-tmp which is an alias to OP-MSG-TMP) + om.pv_send.put("", wait=True) + + +# ----------------------------- # Tests: OperationMessageStatus +# ----------------------------- + class TestOperationMessageStatus: @pytest.fixture def status(self): s = OperationMessageStatus("SF-OP:CR-MSG") assert s.pv_date.wait_for_connection(timeout=2.0), "PV STATUS-DATE not connected" assert s.pv_status.wait_for_connection(timeout=2.0), "PV STATUS not connected" + # ensure enum metadata present + _ensure_enum_metadata(s.pv_status) 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() + if not allowed: + _ = status.pv_status.get(as_string=True) + 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) assert "2025-08-08 10:01:02 ATTENDED" in r @@ -136,20 +177,28 @@ class TestOperationMessageStatus: ("set_attended", "ATTENDED"), ]) def test_set_methods_change_status(self, status, method, expected): + # Make sure enum metadata is ready + _ensure_enum_metadata(status.pv_status) + getattr(status, method)() assert status.status == expected def test_update_direct_valid(self, status): + _ensure_enum_metadata(status.pv_status) status.update("REMOTE") assert status.status == "REMOTE" def test_update_invalid_raises_valueerror(self, status): + _ensure_enum_metadata(status.pv_status) with pytest.raises(ValueError) as exc: status.update("TOTALLY_INVALID_STATUS") assert "not from allowed values" in str(exc.value) +# ----------------------------- # Tests: OperationMessageEntry +# ----------------------------- + class TestOperationMessageEntry: @pytest.fixture(params=range(N_MSG_HISTORY)) def entry(self, request): @@ -180,29 +229,30 @@ class TestOperationMessageEntry: 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 +# ----------------------------- # Tests: OperationMessage +# ----------------------------- + class TestOperationMessage: def test_init_with_name(self): om = OperationMessage(name="Control Room") - assert om.name == "Control Room" - assert om.ID == "CR" - assert om.prefix == "SF-OP:CR-MSG" - + # connections assert om.pv_send.wait_for_connection(timeout=2.0) assert om.status.pv_status.wait_for_connection(timeout=2.0) - assert len(om.entries) == N_MSG_HISTORY for i, entry in enumerate(om.entries): assert entry.pv_date.wait_for_connection(timeout=2.0), f"OP-DATE{i} not connected" assert entry.pv_msg.wait_for_connection(timeout=2.0), f"OP-MSG{i} not connected" + assert len(om.entries) == N_MSG_HISTORY def test_init_with_id(self): - om = OperationMessage(ID="cr") # case-insensitive + om = OperationMessage(ID="cr") assert om.ID == "CR" - assert om.name == IDS_INVERSE["CR"] # "Control Room" + assert om.name is not None # from IDS_INVERSE assert om.prefix == "SF-OP:CR-MSG" def test_init_error_when_missing_both(self): @@ -211,15 +261,16 @@ class TestOperationMessage: def test_getitem_returns_entry(self): om = OperationMessage(name="Control Room") - for i in range(N_MSG_HISTORY): - e = om[i] - assert isinstance(e, OperationMessageEntry) + for idx in range(N_MSG_HISTORY): + assert isinstance(om[idx], OperationMessageEntry) def test_update_puts_to_tmp_channel(self): om = OperationMessage(name="Control Room") assert om.pv_send.wait_for_connection(timeout=2.0) + msg = "Beam down at 14:00" om.update(msg) + got = om.pv_send.get(as_string=True) if isinstance(got, bytes): got = got.decode("utf-8", errors="ignore") @@ -227,15 +278,18 @@ class TestOperationMessage: def test_status_object_is_wired(self): om = OperationMessage(name="Control Room") - assert isinstance(om.status, OperationMessageStatus) + _ensure_enum_metadata(om.status.pv_status) om.status.update("REMOTE") assert om.status.status == "REMOTE" def test_repr_contains_header_and_entries(self): om = OperationMessage(name="Control Room") + _reset_operation_message_to_defaults(om) + om.status.pv_status.put("ATTENDED", wait=True) om.entries[0].pv_date.put("2025-08-08 12:00:00", wait=True) om.entries[0].pv_msg.put("Slot A ok", wait=True) + r = repr(om) assert "Control Room (CR)" in r assert "ATTENDED" in r @@ -243,15 +297,15 @@ class TestOperationMessage: assert "Slot A ok" in r -# Tests: OperationMessages +# ----------------------------- +# Tests: 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) @@ -261,9 +315,8 @@ 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) + attr = name.lower().replace(" ", "_") assert hasattr(oms, attr), f"Missing attribute: {attr}" assert getattr(oms, attr) is oms.entries[name] @@ -272,7 +325,7 @@ class TestOperationMessages: name = "Control Room" ID = IDS[name] - cleaned_name = "control_room" + cleaned_name = "control_room" cleaned_id = "cr" om_by_name = oms[cleaned_name] @@ -283,7 +336,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): @@ -291,11 +344,16 @@ class TestOperationMessages: def test_iter_yields_operationmessage_instances(self): oms = OperationMessages() + + # Reset everything to defaults to avoid cross-test leakage + for om in oms: + _reset_operation_message_to_defaults(om) + vals = list(iter(oms)) assert len(vals) == len(IDS) assert all(isinstance(v, OperationMessage) for v in vals) - # All prefixes serve identical initial data; verify contents quickly + # Check default contents for om in vals: assert om.status.status == "OFFLINE" assert om.status.date == "2024-01-01 00:00:00" @@ -306,47 +364,61 @@ class TestOperationMessages: def test_key_completions_exposes_all_cleaned_keys(self): oms = OperationMessages() keys = set(oms._ipython_key_completions_()) - expected_names = {clean_name(n) for n in IDS.keys()} - expected_ids = {clean_name(ID) for ID in IDS.values()} + + expected_names = {n.lower().replace(" ", "_") for n in IDS.keys()} + expected_ids = {id_.lower() for id_ in IDS.values()} + assert expected_names.issubset(keys) assert expected_ids.issubset(keys) def test_repr_includes_each_submessage_repr(self): oms = OperationMessages() - r = repr(oms) - assert "Control Room (CR)" in r # header from one submessage - om_cr = oms["control_room"] - # Modify two consecutive entries + + # ensure baseline then set two lines + _reset_operation_message_to_defaults(om_cr) + 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) r2 = repr(oms) + assert "Control Room (CR)" in r2 assert "2025-08-08 12:00:00" in r2 assert "Test message 1" in r2 assert "2025-08-08 12:05:00" in r2 assert "Test message 2" in r2 +# ----------------------------- # Tests: MachineStatus +# ----------------------------- + class TestMachineStatus: @pytest.mark.parametrize("beamline", BEAMLINES) def test_init_and_connections(self, beamline): + from slic.utils.opmsg import MachineStatus ms = MachineStatus(beamline) assert ms.beamline == beamline.upper() assert ms.pv_category.wait_for_connection(timeout=2.0), f"{beamline}: CATEGORY not connected" assert ms.pv_downtime.wait_for_connection(timeout=2.0), f"{beamline}: DOWNTIME not connected" def test_init_invalid_beamline_raises(self): + from slic.utils.opmsg import MachineStatus with pytest.raises(ValueError) as exc: MachineStatus("NOT_A_BEAMLINE") - assert 'beamline "NOT_A_BEAMLINE" must be from:' in str(exc.value) + assert 'must be from:' in str(exc.value) @pytest.mark.parametrize("beamline", BEAMLINES) def test_properties_category_and_downtime(self, beamline): + from slic.utils.opmsg import MachineStatus ms = MachineStatus(beamline) + + # ensure enum metadata + _ensure_enum_metadata(ms.pv_category) + ms.pv_category.put("MD", wait=True) ms.pv_downtime.put("01:23:45", wait=True) assert ms.category == "MD" @@ -359,9 +431,13 @@ class TestMachineStatus: @pytest.mark.parametrize("beamline", BEAMLINES) def test_repr_contains_header_and_values(self, beamline): + from slic.utils.opmsg import MachineStatus ms = MachineStatus(beamline) + + _ensure_enum_metadata(ms.pv_category) ms.pv_category.put("ACCESS", wait=True) ms.pv_downtime.put("00:05:00", wait=True) + r = repr(ms) assert beamline in r assert "ACCESS" in r