72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
import os
|
|
import time
|
|
import threading
|
|
import pytest
|
|
from morbidissimo import MorIOC
|
|
|
|
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 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)
|
|
os.environ.setdefault(PREFIX_ENV, prefix)
|
|
|
|
stop = threading.Event()
|
|
|
|
def run():
|
|
with MorIOC(prefix) as mor:
|
|
# ---- Host PVs once
|
|
# Legacy/top-level PVs used by other tests
|
|
mor.host(
|
|
VAL=float, # TEST:VAL
|
|
PV1=float, PV2=float, PV3=float,
|
|
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
|
|
)
|
|
|
|
# ---- 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
|
|
while not stop.is_set():
|
|
# Accept writes
|
|
for name in state.keys():
|
|
new_val = mor.get(name)
|
|
if new_val is not None:
|
|
state[name] = new_val
|
|
|
|
# Publish current values; PV1..PV6 are static constants
|
|
mor.serve(
|
|
**state,
|
|
PV1=1.0, PV2=2.0, PV3=3.0,
|
|
PV4=4.0, PV5=5.0, PV6=6.0,
|
|
)
|
|
time.sleep(0.05) # avoid busy loop
|
|
|
|
t = threading.Thread(target=run, daemon=True)
|
|
t.start()
|
|
time.sleep(0.8) # let CA announce PVs
|
|
|
|
yield
|
|
|
|
stop.set()
|
|
t.join(timeout=2.0)
|
|
time.sleep(0.2)
|