Add tests/test_utils_preload.py

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
+