diff --git a/tests/test_utils_hastepics.py b/tests/test_utils_hastepics.py index b1f124e56..ea46f89f9 100644 --- a/tests/test_utils_hastepics.py +++ b/tests/test_utils_hastepics.py @@ -9,6 +9,18 @@ from slic.utils.hastyepics import * def _in_pv_cache(pvname: str) -> bool: return any(k[0] == pvname for k in _PVcache_.keys()) +def _evict_prefix(prefix: str): + 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.""" + prefix = os.environ.get("TEST_PV_PREFIX", "TEST:") + _evict_prefix(prefix) + yield + # ============================ # Tests # ============================ @@ -49,53 +61,70 @@ def test_disabled_removed_relative_to_upstream(): base_init = tuple(getattr(epics.Motor, "_init_list", ())) base_extras = dict(getattr(epics.Motor, "_extras", {})) - had_in_init = "disabled" in base_init - had_in_extras = "disabled" in base_extras + assert "disabled" in base_init + assert "disabled" in base_extras m = Motor("TEST:SIM:M7") - if had_in_init: - assert "disabled" not in getattr(m, "_init_list", ()), \ + assert "disabled" not in getattr(m, "_init_list", ()), \ "disabled devait être retiré de _init_list par rapport à epics.Motor" - if had_in_extras: - assert "disabled" not in getattr(m, "_extras", {}), \ + assert "disabled" not in getattr(m, "_extras", {}), \ "disabled devait être retiré de _extras par rapport à epics.Motor" +def _get_cached(name: str): + for k, pv in _PVcache_.items(): + if k[0] == name: + return pv + return None -def test_motor_init_list_and_extras(): - """ - Vérifie la cohérence de _init_list/_extras vs l'upstream : - - tout ce que l'upstream expose (hors 'disabled') peut être résolu via PV()/attribut. - Sans forcer que _extras soit non vide si l'upstream ne l'est pas. - """ +def test_motor_init_list_and_extras_build_handles_eagerly(): 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") base_extras_wo_disabled = {k: v for k, v in base_extras.items() if k != "disabled"} - m = Motor("TEST:SIM:M4") + motor_prefix = "TEST:SIM:M4" - # Pour tous les attributs "init_list" upstream (sans 'disabled'), on doit avoir un PV accessible - for attr in base_init_wo_disabled: - pv = m.PV(attr, connect=False) - assert isinstance(pv, PV), f"{attr} n'est pas un PV" + _evict_prefix(motor_prefix) + init_pvnames = [f"{motor_prefix}.{attr}" for attr in base_init_wo_disabled] + extra_pvnames = [f"{motor_prefix}{suffix}" for suffix in base_extras_wo_disabled.values()] + for name in init_pvnames + extra_pvnames: + assert not _in_pv_cache(name), f"{name} should not be in cache before Motor()" - # Pour tous les extras upstream (sans 'disabled'), si l'attribut existe sur m, le pvname doit matcher + m = Motor(motor_prefix) + + # init_list: handles exist in cache and are disconnected + for name in init_pvnames: + assert _in_pv_cache(name), f"{name} not inserted into _PVcache_ by __init__" + cached = _get_cached_pv(name) + assert cached is not None and not cached.connected + + # handles exist in cache for attr_name, suffix in base_extras_wo_disabled.items(): - if hasattr(m, attr_name): - pv = getattr(m, attr_name) - assert isinstance(pv, PV), f"{attr_name} n'est pas un PV" - assert pv.pvname == f"TEST:SIM:M4{suffix}" + name = f"{motor_prefix}{suffix}" + assert _in_pv_cache(name), f"{name} not inserted into _PVcache_ by extras loop" + cached = _get_cached_pv(name) + assert cached is not None and not cached.connected -def test_motor_pv_method_connects_and_reads(): - m = Motor("TEST:SIM:M5") - pv_val = m.PV("VAL", connect=True, timeout=3.0) - assert pv_val.connected - pv_val.put(123.0) - time.sleep(0.1) - assert pv_val.get() == 123.0 +def test_motor_val_connect_via_cache(): + motor_prefix = "TEST:SIM:M5" + pvname = f"{motor_prefix}.VAL" + + assert not _in_pv_cache(pvname) + + m = Motor(motor_prefix) + + assert _in_pv_cache(pvname), f"{pvname} should be cached after Motor()" + + pv = _get_cached_pv(pvname) + assert pv is not None + assert not pv.connected + + assert pv.connect(timeout=2.0), "VAL did not connect" + pv.put(123.0, wait=True) + assert pv.get(timeout=1.0) == 123.0 def test_speedup_get_pv(): @@ -140,15 +169,14 @@ def test_speedup_motor_instantiation(): def test_speedup_motor_PV(): # baseline EPICS - m_epics = epics.Motor("TEST:SIM:M4") + m_epics = epics.Motor("TEST:SIM:M14") t0 = time.perf_counter() pv_epics = m_epics.PV("VAL", connect=True, timeout=2.0) t1 = time.perf_counter() - # version "fast" (ton Motor) - m_fast = Motor("TEST:SIM:M5") + m_fast = Motor("TEST:SIM:M15") t2 = time.perf_counter() - pv_fast = m_fast.PV("VAL", connect=False) # volontairement non connecté + pv_fast = m_fast.PV("VAL", connect=False) t3 = time.perf_counter() assert pv_epics.connected