diff --git a/tests/test_utils_preload.py b/tests/test_utils_preload.py new file mode 100644 index 000000000..da9dd8ef8 --- /dev/null +++ b/tests/test_utils_preload.py @@ -0,0 +1,157 @@ +import pytest +import time +import threading +import pickle +from pathlib import Path + +from slic.utils.picklio import unpickle +from freezegun import freeze_time +from datetime import datetime, timedelta +from slic.utils.hastyepics import get_pv +from epics.pv import _PVcache_ + +import sys +import os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) +from slic.utils.pvpreload import * + +from morbidissimo import MorIOC + + +# IOC simulation +@pytest.fixture(scope="module") +def epics_ioc(): + from epics import ca + pvname = "TEST:PV1" + + def ioc_thread(): + value = 0.0 + while True: + ca.put(pvname, value) + value += 1.0 + time.sleep(0.1) + + thread = threading.Thread(target=ioc_thread, daemon=True) + thread.start() + time.sleep(1) + +# Tests + +# file_age() +@pytest.mark.parametrize("age_seconds, expected", [ + (30, timedelta(seconds=30)), # 0:00:30 + (300, timedelta(minutes=5)), # 0:05:00 + (3600, timedelta(hours=1)), # 1:00:00 + (86400, timedelta(days=1)), # 1 day + (1209600, timedelta(days=14)), # 2 weeks +]) +@freeze_time("2025-08-07 12:00:00") +def test_file_age(tmp_path, age_seconds, expected): + test_file = tmp_path / "testfile" + test_file.touch() + + past_timestamp = time.time() - age_seconds + os.utime(test_file, (past_timestamp, past_timestamp)) + + result = file_age(test_file) + assert result == expected + +# preload() + +def test_preload_fichier_valide(tmp_path, caplog): + f = tmp_path / "valide.pkl" + pvname_1 = "TEST:PV1" + pvname_2 = "TEST:PV2" + + # Crée un fichier pickle avec les noms de PV + with open(f, "wb") as pkl: + pickle.dump([pvname_1, pvname_2], pkl) + + # Vérifie que les PV ne sont pas encore dans le cache EPICS + assert pvname_1 not in _PVcache_ + assert pvname_2 not in _PVcache_ + + # Patch le chemin utilisé dans preload() + with pytest.MonkeyPatch().context() as mp: + mp.setattr("slic.utils.pvpreload.fn", f) + + with caplog.at_level("DEBUG"): + preload() + + # Vérifie que les PV ont bien été créés et sont dans le cache + assert pvname_1 in _PVcache_ + assert pvname_2 in _PVcache_ + + pv_1 = _PVcache_[pvname_1] + pv_2 = _PVcache_[pvname_2] + + # Vérifie qu'ils sont bien connectés + assert pv_1.wait_for_connection(timeout=2) + assert pv_2.wait_for_connection(timeout=2) + + # Vérifie le log final + logs = "\n".join(caplog.messages) + assert "PV preload done" in logs + +import os +import pickle +from datetime import datetime, timedelta +import pytest +from slic.utils.pvpreload import preload, fn as original_fn, lifetime + +def test_preload_fichier_trop_vieux(tmp_path, caplog): + f = tmp_path / "trop_vieux.pkl" + pvname = "TEST:PV_X" + + # Crée un fichier pickle valide avec un nom de PV + with open(f, "wb") as pkl: + pickle.dump([pvname], pkl) + + # Vieillit artificiellement le fichier (par exemple 2h d'ancienneté) + old_time = datetime.timestamp(datetime.now() - 2 * lifetime) + os.utime(f, (old_time, old_time)) + + # Patch le chemin du fichier dans preload + with pytest.MonkeyPatch().context() as mp: + mp.setattr("slic.utils.pvpreload.fn", f) + + with caplog.at_level("INFO"): + preload() + + # Vérifie que le fichier a été détecté comme trop vieux + logs = "\n".join(caplog.messages) + assert "PV preload file too old" in logs + + # Vérifie que le PV n’a pas été créé (pas dans le cache) + assert pvname not in _PVcache_ + +# offload() +def test_offload(tmp_path, caplog): + fake_file = tmp_path / "offload_test.pkl" + pvname_3 = "TEST:PV3" + pvname_4 = "TEST:PV4" + + # Connecte 2 PVs + pv3 = get_pv(pvname_3) + pv4 = get_pv(pvname_4) + assert pv3.wait_for_connection(timeout=2) + assert pv4.wait_for_connection(timeout=2) + + # Patch le fichier et le delay + with pytest.MonkeyPatch().context() as mp: + mp.setattr("slic.utils.pvpreload.fn", fake_file) + mp.setattr("slic.utils.pvpreload.delay", 0.01) + + with caplog.at_level("DEBUG"): + offload() + + # Vérifie que le fichier contient bien les bons noms + names = pickle.load(open(fake_file, "rb")) + assert pvname_3 in names + assert pvname_4 in names + + # Vérifie les logs + logs = "\n".join(caplog.messages) + assert "PV offload start" in logs + assert "PV offload done" in logs +