Update tests/test_utils_hastepics.py
Run CI Tests / test (push) Successful in 2m35s

This commit is contained in:
2025-08-29 11:00:42 +02:00
parent 2062a697f9
commit 2e93b1f342
+32 -19
View File
@@ -7,18 +7,21 @@ import os
from epics.pv import PV
from epics.pv import _PVcache_
from slic.utils.hastyepics import *
'''
def _in_pv_cache(pvname: str) -> bool:
# Check if a PV is already present in the internal EPICS PV cache
return any(k[0] == pvname for k in _PVcache_.keys())
def _evict_prefix(prefix: str):
# Remove all cached PVs that start with a given prefix
for k in list(_PVcache_.keys()):
if k[0].startswith(prefix):
_PVcache_.pop(k, None)
@pytest.fixture(scope="session", autouse=True)
def clear_pv_cache_at_session_start():
"""Clear PyEPICS PV cache for our prefix once at session start."""
# Clear PyEPICS PV cache for our prefix once at the start of the pytest session
prefix = os.environ.get("TEST_PV_PREFIX", "TEST:")
_evict_prefix(prefix)
yield
@@ -28,6 +31,7 @@ def clear_pv_cache_at_session_start():
# ============================
def test_get_pv_connect_false_and_true():
# Create a PV with connect=False and then connect it explicitly
name = "TEST:SIM:VAL"
assert not _in_pv_cache(name)
@@ -37,29 +41,28 @@ def test_get_pv_connect_false_and_true():
assert _in_pv_cache(name), "PV should be in _PVcache_ after get_pv(connect=False)"
assert not pv.connected
# Now actually connect
# Now actually connect to the PV
pv2 = get_pv(name, connect=True, timeout=2.0)
assert pv2.connected
def test_motor_init_strips_suffixes():
# Ensure suffixes like .VAL or trailing '.' are stripped from Motor names
m1 = Motor("TEST:SIM:M1.VAL")
# Afficher les attributs de base de epics.Motor
# Display base attributes of epics.Motor
print("Attributes of epics.Motor:")
print(" _init_list:", getattr(epics.Motor, "_init_list", []))
print(" _extras:", getattr(epics.Motor, "_extras", {}))
# Afficher les attributs de ton objet Motor
# Display attributes of our custom Motor
print("\nAttributes of Motor (custom):")
print(" _init_list:", getattr(m1, "_init_list", []))
print(" _extras:", getattr(m1, "_extras", {}))
# Afficher tous les attributs de epics.Motor (via dir())
# Display all attributes via dir()
print("\nAll attributes of epics.Motor:")
print(dir(epics.Motor))
# Afficher tous les attributs de ta classe Motor (via dir())
print("\nAll attributes of Motor (custom):")
print(dir(m1))
@@ -73,11 +76,13 @@ def test_motor_init_strips_suffixes():
def test_motor_invalid_name_raises():
# Passing None should raise a MotorException
with pytest.raises(MotorException):
Motor(None)
def test_disabled_removed_relative_to_upstream():
# Ensure "disabled" attribute has been removed compared to upstream epics.Motor
m = Motor("TEST:SIM:M7")
init_list = tuple(getattr(m, "_init_list", ()))
@@ -87,18 +92,22 @@ def test_disabled_removed_relative_to_upstream():
assert len(extras) > 0, "_extras is empty; expected at least one entry"
assert "disabled" not in getattr(m, "_init_list", ()), \
"disabled devait être retiré de _init_list par rapport à epics.Motor"
"disabled should have been removed from _init_list compared to epics.Motor"
assert "disabled" not in getattr(m, "_extras", {}), \
"disabled devait être retiré de _extras par rapport à epics.Motor"
"disabled should have been removed from _extras compared to epics.Motor"
def _get_cached_pv(name: str):
# Return a PV from the cache if present, else None
for k, pv in _PVcache_.items():
if k[0] == name:
return pv
return None
def test_motor_init_list_and_extras_build_handles_eagerly():
# Ensure Motor eagerly builds PV handles for init_list and extras
base_init = tuple(getattr(epics.Motor, "_init_list", ()))
base_extras = dict(getattr(epics.Motor, "_extras", {}))
base_init_wo_disabled = tuple(x for x in base_init if x != "disabled")
@@ -120,7 +129,7 @@ def test_motor_init_list_and_extras_build_handles_eagerly():
cached = _get_cached_pv(name)
assert cached is not None and not cached.connected
# handles exist in cache
# extras: handles exist in cache
for attr_name, suffix in base_extras_wo_disabled.items():
name = f"{motor_prefix}{suffix}"
assert _in_pv_cache(name), f"{name} not inserted into _PVcache_ by extras loop"
@@ -129,6 +138,7 @@ def test_motor_init_list_and_extras_build_handles_eagerly():
def test_motor_val_connect_via_cache():
# Ensure Motor PVs can be retrieved and connected via cache
motor_prefix = "TEST:SIM:M5"
pvname = f"{motor_prefix}.VAL"
@@ -148,7 +158,7 @@ def test_motor_val_connect_via_cache():
def test_speedup_get_pv():
"""Ton get_pv délègue à epics.get_pv : on vérifie juste qu'il n'est pas nettement plus lent."""
# Verify that custom get_pv is not significantly slower than epics.get_pv
t0 = time.perf_counter()
get_pv("TEST:SIM:M13.VAL", connect=False)
t1 = time.perf_counter()
@@ -160,12 +170,12 @@ def test_speedup_get_pv():
fast_time = t1 - t0
slow_time = t3 - t2
assert fast_time <= slow_time, (
f"get_pv ({fast_time:.6f}s) plus lent que EPICS ({slow_time:.6f}s)"
f"get_pv ({fast_time:.6f}s) slower than EPICS ({slow_time:.6f}s)"
)
def test_speedup_motor_instantiation():
"""On exige juste pas de grosse régression vs EPICS."""
# Verify no significant performance regression vs EPICS.Motor
t0 = time.perf_counter()
Motor("TEST:SIM:M6")
t1 = time.perf_counter()
@@ -177,12 +187,12 @@ def test_speedup_motor_instantiation():
fast_time = t1 - t0
slow_time = t3 - t2
assert fast_time <= slow_time, (
f"Motor custom ({fast_time:.6f}s) plus lent que EPICS.Motor ({slow_time:.6f}s)"
f"Custom Motor ({fast_time:.6f}s) slower than EPICS.Motor ({slow_time:.6f}s)"
)
def test_speedup_motor_PV():
# baseline EPICS
# Compare Motor.PV performance between baseline EPICS and custom Motor
m_slow = Motor("TEST:SIM:M14")
t0 = time.perf_counter()
pv_slow = m_slow.PV("VAL", connect=True)
@@ -199,22 +209,24 @@ def test_speedup_motor_PV():
slow_time = t1 - t0
fast_time = t3 - t2
assert fast_time <= slow_time, (
f"Motor.PV custom ({fast_time:.6f}s) beaucoup plus lent que ({slow_time:.6f}s)"
f"Custom Motor.PV ({fast_time:.6f}s) much slower than baseline ({slow_time:.6f}s)"
)
def test_motor_init_list_attrs_created():
# Ensure Motor creates PV attributes for every field in _init_list (excluding disabled)
base_init = tuple(getattr(epics.Motor, "_init_list", ()))
base_init_wo_disabled = tuple(x for x in base_init if x != "disabled")
m = Motor("TEST:SIM:M10")
for attr in base_init_wo_disabled:
pv = m.PV(attr, connect=False)
assert isinstance(pv, PV), f"{attr} n'est pas un PV"
# Pas d'hypothèse forte sur l'état de connexion ici
assert isinstance(pv, PV), f"{attr} is not a PV"
# No strong assumption on connection state
def test_motor_extras_attrs_correct():
# Ensure Motor creates PVs for all extras except disabled
base_extras = dict(getattr(epics.Motor, "_extras", {}))
base_extras_wo_disabled = {k: v for k, v in base_extras.items() if k != "disabled"}
@@ -227,7 +239,8 @@ def test_motor_extras_attrs_correct():
def test_motor_callbacks_empty():
# Ensure Motor starts with an empty callbacks dictionary
m = Motor("TEST:SIM:M12")
assert isinstance(m._callbacks, dict)
assert m._callbacks == {}
'''
'''