158 lines
5.2 KiB
Python
158 lines
5.2 KiB
Python
import pytest
|
|
import time
|
|
import threading
|
|
from morbidissimo import MorIOC
|
|
from slic.utils.hastyepics import get_pv as PV
|
|
from slic.utils.pyepics import *
|
|
'''
|
|
# ------------------------------------------------------------------------------
|
|
# Fixture IOC
|
|
# ------------------------------------------------------------------------------
|
|
@pytest.fixture(scope="module", autouse=True)
|
|
def epics_ioc():
|
|
stop = threading.Event()
|
|
|
|
def ioc():
|
|
with MorIOC("TEST") as mor:
|
|
mor.serve(
|
|
TEST_ENUM=["OFF", "ON", "STANDBY"],
|
|
TEST_ANALOG=0.0,
|
|
TEST_STRING="default",
|
|
TEST_SPECIAL="a.b c@d"
|
|
)
|
|
while not stop.is_set():
|
|
time.sleep(0.1)
|
|
|
|
t = threading.Thread(target=ioc)
|
|
t.start()
|
|
time.sleep(2) # Attente connexion
|
|
yield
|
|
stop.set()
|
|
t.join()
|
|
|
|
# ------------------------------------------------------------------------------
|
|
# Tests EnumWrapper
|
|
# ------------------------------------------------------------------------------
|
|
def test_enum_wrapper_initial_state():
|
|
enum = EnumWrapper("TEST_ENUM")
|
|
assert enum._pv.wait_for_connection(timeout=2.0), "PV non connecté"
|
|
assert enum.get() == 0
|
|
assert enum.get_name() == "OFF"
|
|
|
|
def test_enum_wrapper_set_valid_string():
|
|
enum = EnumWrapper("TEST_ENUM")
|
|
assert enum._pv.wait_for_connection(timeout=2.0), "PV non connecté"
|
|
enum.set("ON")
|
|
assert enum.get() == 1
|
|
assert enum.get_name() == "ON"
|
|
|
|
def test_enum_wrapper_set_valid_index():
|
|
enum = EnumWrapper("TEST_ENUM")
|
|
assert enum._pv.wait_for_connection(timeout=2.0), "PV non connecté"
|
|
enum.set(2)
|
|
assert enum.get_name() == "STANDBY"
|
|
|
|
def test_enum_wrapper_set_invalid_string_raises():
|
|
enum = EnumWrapper("TEST_ENUM")
|
|
assert enum._pv.wait_for_connection(timeout=2.0), "PV non connecté"
|
|
with pytest.raises(AssertionError):
|
|
enum.set("INVALID")
|
|
|
|
def test_enum_wrapper_set_invalid_index_raises():
|
|
enum = EnumWrapper("TEST_ENUM")
|
|
assert enum._pv.wait_for_connection(timeout=2.0), "PV non connecté"
|
|
with pytest.raises(AssertionError):
|
|
enum.set(10)
|
|
|
|
def test_enum_wrapper_set_negative_index_raises():
|
|
enum = EnumWrapper("TEST_ENUM")
|
|
assert enum._pv.wait_for_connection(timeout=2.0), "PV non connecté"
|
|
with pytest.raises(AssertionError):
|
|
enum.set(-1)
|
|
|
|
def test_enum_wrapper_setters_property():
|
|
enum = EnumWrapper("TEST_ENUM")
|
|
assert enum._pv.wait_for_connection(timeout=2.0), "PV non connecté"
|
|
setters = enum.setters
|
|
setters.ON()
|
|
assert enum.get_name() == "ON"
|
|
|
|
# ------------------------------------------------------------------------------
|
|
# Tests MonitorAccumulator
|
|
# ------------------------------------------------------------------------------
|
|
def test_monitor_accumulator_initial_state():
|
|
pv = PV("TEST_ANALOG")
|
|
assert pv._pv.wait_for_connection(timeout=2.0), "PV non connecté"
|
|
monitor = MonitorAccumulator(pv)
|
|
assert len(monitor.values) == 0
|
|
|
|
def test_monitor_accumulator_captures_values():
|
|
pv = PV("TEST_ANALOG")
|
|
assert pv._pv.wait_for_connection(timeout=2.0), "PV non connecté"
|
|
monitor = MonitorAccumulator(pv)
|
|
monitor.accumulate()
|
|
|
|
pv.put(42.0)
|
|
time.sleep(0.5)
|
|
|
|
assert len(monitor.values) > 0
|
|
assert monitor.values[-1][0] == pytest.approx(42.0)
|
|
|
|
def test_monitor_accumulator_cycle_clears_values():
|
|
pv = PV("TEST_ANALOG")
|
|
assert pv._pv.wait_for_connection(timeout=2.0), "PV non connecté"
|
|
monitor = MonitorAccumulator(pv)
|
|
monitor.accumulate()
|
|
|
|
pv.put(42.0)
|
|
time.sleep(0.5)
|
|
|
|
data = monitor.cycle()
|
|
assert len(data) > 0
|
|
assert len(monitor.values) == 0
|
|
|
|
# ------------------------------------------------------------------------------
|
|
# Tests Positioner
|
|
# ------------------------------------------------------------------------------
|
|
def test_positioner_creates_methods():
|
|
def mock_action():
|
|
return "executed"
|
|
|
|
pos = Positioner([("TEST", mock_action)])
|
|
assert pos.TEST() == "executed"
|
|
|
|
def test_positioner_handles_special_chars():
|
|
def mock_action():
|
|
return "ok"
|
|
|
|
pos = Positioner([("1.2", mock_action), ("a b", mock_action)])
|
|
assert pos.v1p2() == "ok"
|
|
assert pos.a_b() == "ok"
|
|
|
|
# ------------------------------------------------------------------------------
|
|
# Tests EpicsString
|
|
# ------------------------------------------------------------------------------
|
|
def test_epics_string_initial_value():
|
|
estr = EpicsString("TEST_STRING")
|
|
assert estr._pv.wait_for_connection(timeout=2.0), "PV non connecté"
|
|
assert estr.get() == "default"
|
|
|
|
def test_epics_string_set_value():
|
|
estr = EpicsString("TEST_STRING")
|
|
assert estr._pv.wait_for_connection(timeout=2.0), "PV non connecté"
|
|
estr.set("new_value")
|
|
assert estr.get() == "new_value"
|
|
|
|
def test_epics_string_call_syntax():
|
|
estr = EpicsString("TEST_STRING")
|
|
assert estr._pv.wait_for_connection(timeout=2.0), "PV non connecté"
|
|
estr("updated")
|
|
assert estr.get() == "updated"
|
|
|
|
def test_epics_string_special_chars():
|
|
estr = EpicsString("TEST_SPECIAL")
|
|
assert estr._pv.wait_for_connection(timeout=2.0), "PV non connecté"
|
|
estr.set("test@value")
|
|
assert estr.get() == "test@value"
|
|
|
|
''' |