Update tests/test_utils_pvpreload.py

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