Update tests/ioc_plugin.py
Run CI Tests / test (push) Successful in 2m21s

This commit is contained in:
2025-08-11 01:37:40 +02:00
parent 9142d6e96c
commit 4a3e10db06
+33 -22
View File
@@ -4,57 +4,68 @@ import threading
import pytest
from morbidissimo import MorIOC
# Environment variable to share the PV prefix with tests
PREFIX_ENV = "TEST_PV_PREFIX"
DEFAULT_PREFIX = "TEST:"
# Motors you reference in tests: M1..M13
MOTOR_IDS = [f"M{i}" for i in range(1, 14)]
@pytest.fixture(scope="session", autouse=True)
def ioc_session():
"""
Start exactly ONE IOC for the entire pytest session.
All tests can reuse it to avoid port/prefix conflicts.
Add all PVs needed by your test suite here.
Start exactly ONE IOC for the whole test session.
Hosts both legacy PVs (TEST:VAL, TEST:PV1..PV6) and SIM PVs
(TEST:SIM:VAL, TEST:SIM:M*.VAL) used by the motor tests.
"""
prefix = os.environ.get(PREFIX_ENV, DEFAULT_PREFIX)
# Ensure tests can read the chosen prefix
os.environ.setdefault(PREFIX_ENV, prefix)
stop = threading.Event()
def run():
with MorIOC(prefix) as mor:
# Host PVs once for the whole session
# ---- Host PVs once
# Legacy/top-level PVs used by other tests
mor.host(
VAL=float, # writable PV used in progress-bar tests
VAL=float, # TEST:VAL
PV1=float, PV2=float, PV3=float,
PV4=float, PV5=float, PV6=float, # static PVs
PV4=float, PV5=float, PV6=float,
)
# SIM namespace PVs
mor.host(
**{"SIM:VAL": float}, # TEST:SIM:VAL
**{f"SIM:{mid}.VAL": float for mid in MOTOR_IDS}, # TEST:SIM:M*.VAL
)
current_val = 0.0 # backing state for VAL
# ---- Backing state (writable PVs keep last written value)
state = {
"VAL": 0.0,
"SIM:VAL": 0.0,
**{f"SIM:{mid}.VAL": 0.0 for mid in MOTOR_IDS},
}
# Serve loop; keep running until the session ends
# ---- Serve loop
while not stop.is_set():
# Make VAL writable: read incoming value and persist it
new_val = mor.get("VAL")
if new_val is not None:
current_val = new_val
# Accept writes
for name in state.keys():
new_val = mor.get(name)
if new_val is not None:
state[name] = new_val
# Serve current values (static PVs as constants)
# Publish current values; PV1..PV6 are static constants
mor.serve(
VAL=current_val,
**state,
PV1=1.0, PV2=2.0, PV3=3.0,
PV4=4.0, PV5=5.0, PV6=6.0,
)
time.sleep(0.05) # small delay to avoid a busy loop
time.sleep(0.05) # avoid busy loop
# Start IOC in a background thread
t = threading.Thread(target=run, daemon=True)
t.start()
time.sleep(0.8) # give Channel Access time to announce PVs
time.sleep(0.8) # let CA announce PVs
yield # <<< tests execute during this period >>>
yield
# Graceful teardown
stop.set()
t.join(timeout=2.0)
time.sleep(0.2) # allow OS to release sockets
time.sleep(0.2)