This commit is contained in:
+12
-27
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user