diff --git a/tests/test_utils_pvpreload.py b/tests/test_utils_pvpreload.py index bc136366..a840697a 100644 --- a/tests/test_utils_pvpreload.py +++ b/tests/test_utils_pvpreload.py @@ -3,24 +3,19 @@ import time import threading import pickle as std_pickle from pathlib import Path - import logging -from logzero import logger, setup_default_logger -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_ -from slic.utils.pv import PV -import epics - - -import sys import os -sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) -from slic.utils.pvpreload import * +import sys +from datetime import datetime, timedelta -from morbidissimo import MorIOC +from logzero import logger +from freezegun import freeze_time +from epics.pv import _PVcache_ +import epics +from slic.utils.pv import PV +from slic.utils.hastyepics import get_pv +from morbidissimo import MorIOC +from slic.utils.pvpreload import * # IOC simulation @@ -46,9 +41,8 @@ def run_test_ioc(): def configure_logzero_for_pytest(caplog): logger.handlers.clear() - logger.propagate = True # Propager vers le root logger capturé par caplog - caplog.set_level(level = logging.DEBUG) - + logger.propagate = True # Propagate to root logger captured by caplog + caplog.set_level(level=logging.DEBUG) def is_pv_in_cache(name): return any(k[0] == name for k in epics.pv._PVcache_) @@ -59,15 +53,14 @@ def get_pv_from_cache(pvname): return pv return None -# Tests -# file_age() +# Tests for 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 + (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): @@ -80,32 +73,31 @@ def test_file_age(tmp_path, age_seconds, expected): result = file_age(test_file) assert result == expected -# preload() - -def test_preload_fichier_valide(tmp_path, caplog): +# Tests for preload() +def test_preload_valid_file(tmp_path, caplog): configure_logzero_for_pytest(caplog) - f = tmp_path / "valide.pkl" + f = tmp_path / "valid.pkl" pvname_1 = "TEST:PV1" pvname_2 = "TEST:PV2" - # Crée un fichier pickle avec les noms de PV + # Create pickle file with PV names with open(f, "wb") as pkl: std_pickle.dump([pvname_1, pvname_2], pkl) - # Vérifie que les PV ne sont pas encore dans le cache EPICS + # Verify PVs are not yet in EPICS cache assert not is_pv_in_cache(pvname_1) assert not is_pv_in_cache(pvname_2) - # Patch le chemin utilisé dans preload() + # Patch the path used in 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 + # Verify PVs were created and are in cache assert is_pv_in_cache(pvname_1) assert is_pv_in_cache(pvname_2) @@ -115,50 +107,48 @@ def test_preload_fichier_valide(tmp_path, caplog): assert not pv_1.connected assert not pv_2.connected - # Vérifie le log final + # Verify final log logs = "\n".join(caplog.messages) assert "PV preload done" in logs - def test_preload_old_file(tmp_path, caplog): - configure_logzero_for_pytest(caplog) f = tmp_path / "too_old.pkl" pvname = "TEST:PV_X" - # Crée un fichier pickle valide avec un nom de PV + # Create valid pickle file with PV name with open(f, "wb") as pkl: std_pickle.dump([pvname], pkl) - # Vieillit artificiellement le fichier (par exemple 2h d'ancienneté) + # Artificially age the file (e.g., 2 hours old) old_time = datetime.timestamp(datetime.now() - 2 * lifetime) os.utime(f, (old_time, old_time)) - # Patch le chemin du fichier dans preload + # Patch the file path in 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 + # Verify file was detected as too old 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) + # Verify PV was not created (not in cache) assert not is_pv_in_cache(pvname) -def test_preload_fichier_corrompu(tmp_path, caplog): +def test_preload_corrupt_file(tmp_path, caplog): configure_logzero_for_pytest(caplog) f = tmp_path / "corrupt.pkl" - # Écrit un fichier non valide (pas du pickle) + # Write invalid file (not pickle) with open(f, "w") as fcorrupt: fcorrupt.write("not a pickle at all") - # Patch le chemin du fichier dans preload + # Patch the file path in preload with pytest.MonkeyPatch().context() as mp: mp.setattr("slic.utils.pvpreload.fn", f) @@ -169,9 +159,8 @@ def test_preload_fichier_corrompu(tmp_path, caplog): assert "PV preload file not loaded due to: UnpicklingError:" in logs -# offload() +# Tests for offload() def test_offload(tmp_path, caplog): - configure_logzero_for_pytest(caplog) fake_file = tmp_path / "offload_test.pkl" @@ -179,23 +168,20 @@ def test_offload(tmp_path, caplog): pvname_4 = "TEST:PV4" # Create 2 PVs - pv_3 = get_pv(pvname_3, connect = True) - pv_4 = get_pv(pvname_4, connect = True) + pv_3 = get_pv(pvname_3, connect=True) + pv_4 = get_pv(pvname_4, connect=True) - # 3. Attente active de connexion - assert pv_3.wait_for_connection(timeout=2), "PV3 non connecté" - assert pv_4.wait_for_connection(timeout=2), "PV4 non connecté" + # Wait for connection + assert pv_3.wait_for_connection(timeout=2), "PV3 not connected" + assert pv_4.wait_for_connection(timeout=2), "PV4 not connected" - - assert pv_3.connected, "PV3 marqué comme non connecté" - assert pv_4.connected, "PV4 marqué comme non connecté" - - print(epics.pv._PVcache_.values()) + assert pv_3.connected, "PV3 marked as not connected" + assert pv_4.connected, "PV4 marked as not connected" assert is_pv_in_cache(pvname_3) assert is_pv_in_cache(pvname_4) - # Patch le fichier et le delay + # Patch file and delay with pytest.MonkeyPatch().context() as mp: mp.setattr("slic.utils.pvpreload.fn", fake_file) mp.setattr("slic.utils.pvpreload.delay", 0.01) @@ -204,29 +190,22 @@ def test_offload(tmp_path, caplog): offload() assert fake_file.exists(), "Offload file not created" - - assert fake_file.exists(), "Le fichier n'existe pas" file_size = fake_file.stat().st_size - assert file_size > 0, f"Le fichier est vide (taille: {file_size} bytes)" + assert file_size > 0, f"File is empty (size: {file_size} bytes)" - # Lisez le contenu brut pour inspection - file_content = fake_file.read_bytes() - print(f"Contenu brut du fichier (hex): {file_content.hex()}") - print(f"Contenu brut (str): {file_content!r}") - - # Lire le contenu du fichier + # Read file content with open(fake_file, "rb") as f: try: names = std_pickle.load(f) except Exception as e: pytest.fail(f"Failed to unpickle: {e}") - # Vérifications finales + # Final checks assert isinstance(names, set), "Saved data is not a set" assert pvname_3 in names, f"PV3 missing in {names}" assert pvname_4 in names, f"PV4 missing in {names}" - # Vérifie les logs + # Verify logs logs = "\n".join(caplog.messages) assert "PV offload start" in logs assert "PV offload done" in logs @@ -234,18 +213,18 @@ def test_offload(tmp_path, caplog): def test_offload_exception(tmp_path, caplog): configure_logzero_for_pytest(caplog) - # Fichier invalide : un répertoire à la place d'un fichier + # Invalid file: directory instead of file fake_file = tmp_path / "invalid_dir" - fake_file.mkdir() # <-- on crée un répertoire avec ce nom, pas un fichier + fake_file.mkdir() pvname = "TEST:PV1" - # Créer et connecter un PV pour qu'il soit dans le cache + # Create and connect PV to be in cache pv = get_pv(pvname, connect=True) - assert pv.wait_for_connection(timeout=2), "PV non connecté" + assert pv.wait_for_connection(timeout=2), "PV not connected" assert is_pv_in_cache(pvname) - # Patch la variable fn pour qu'elle pointe vers le mauvais "fichier" + # Patch fn variable to point to wrong "file" with pytest.MonkeyPatch().context() as mp: mp.setattr("slic.utils.pvpreload.fn", fake_file) mp.setattr("slic.utils.pvpreload.delay", 0.01) @@ -257,45 +236,44 @@ def test_offload_exception(tmp_path, caplog): assert "PV preload file not saved due to: IsADirectoryError:" in logs def test_pvpreload_end_to_end(tmp_path, caplog): - configure_logzero_for_pytest(caplog) preload_file = tmp_path / "end2end.pkl" - # Patch preload/offload pour utiliser le bon fichier et un petit délai + # Patch preload/offload to use correct file and small delay with pytest.MonkeyPatch().context() as mp: mp.setattr("slic.utils.pvpreload.fn", preload_file) mp.setattr("slic.utils.pvpreload.delay", 0.05) - # Lancer le thread une bonne fois pour toutes + # Start thread once and for all with caplog.at_level("DEBUG"): pvpreload() - # Étape 1 : Crée PV5 et attend offload + # Step 1: Create PV5 and wait for offload pvname_5 = "TEST:PV5" pv_5 = get_pv(pvname_5, connect=True) assert pv_5.wait_for_connection(timeout=2) - time.sleep(0.2) # Laisser le offload le capturer + time.sleep(0.2) # Let offload capture it - assert preload_file.exists(), "Fichier non créé après offload" + assert preload_file.exists(), "File not created after offload" names = std_pickle.load(open(preload_file, "rb")) - assert pvname_5 in names, f"{pvname_5} absent du fichier" + assert pvname_5 in names, f"{pvname_5} missing from file" - # Étape 2 : Injecte PV6 dans le fichier pour le prochain preload + # Step 2: Inject PV6 into file for next preload pvname_6 = "TEST:PV6" with open(preload_file, "wb") as f: std_pickle.dump([pvname_6], f) assert not is_pv_in_cache(pvname_6) - # Attendre que le prochain preload recharge PV6 + # Wait for next preload to reload PV6 with caplog.at_level("DEBUG"): pvpreload() time.sleep(0.2) - assert is_pv_in_cache(pvname_6), "PV6 non rechargé par preload" + assert is_pv_in_cache(pvname_6), "PV6 not reloaded by preload" - # Vérifie les logs + # Verify logs logs = "\n".join(caplog.messages) assert "PV preload done" in logs - assert "PV offload done" in logs + assert "PV offload done" in logs \ No newline at end of file