diff --git a/tests/test_utils_opmsg.py b/tests/test_utils_opmsg.py index b5d95179e..d552db5b1 100644 --- a/tests/test_utils_opmsg.py +++ b/tests/test_utils_opmsg.py @@ -22,16 +22,17 @@ def ioc(): """ 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}, + **{"OP-MSG-TMP": str}, # Les deux cas pour couvrir toutes les possibilités ) - # Seed une seule fois + # Initialise les valeurs mor.serve( STATUS="OFFLINE", **{"STATUS-DATE": "2024-01-01 00:00:00"}, @@ -43,10 +44,10 @@ def ioc(): while True: mor.serve() - time.sleep(0.02) + time.sleep(0.1) def run_status_prefix(beamline: str): - with MorIOC(f"SF-STATUS-{beamline}") as mor: + with MorIOC(f"STATUS-{beamline}") as mor: mor.host( CATEGORY={"type": "enum", "enums": ["USER", "MD", "SD", "ACCESS", "DOWN"]}, DOWNTIME=str, @@ -54,48 +55,32 @@ def ioc(): mor.serve(CATEGORY="USER", DOWNTIME="00:00:00") while True: mor.serve() - time.sleep(0.02) + time.sleep(0.1) threads = [] + + # Crée les IOCs pour tous les IDs 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 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(0.5) + time.sleep(1.0) 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) - 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 - - # Tests: OperationMessageStatus class TestOperationMessageStatus: @pytest.fixture