"""Non-régression des validateurs d'arguments partagés.""" import argparse import contextlib import io import json import os import sys import tempfile import threading import time import unittest from pathlib import Path from unittest import mock _TEST_HOME = tempfile.TemporaryDirectory(prefix="blink-validation-") os.environ["BLINK_BOOTSTRAP"] = "none" os.environ["BLINK_HOME"] = _TEST_HOME.name import runtime # noqa: E402 def analyser(parse_args, programme: str, *arguments: str): """Appelle un parseur de module sans toucher aux arguments du test runner.""" with mock.patch.object(sys, "argv", [programme, *arguments]): return parse_args() class ValidationBoucleTests(unittest.TestCase): def setUp(self) -> None: self.parser = argparse.ArgumentParser(prog="test-loop") runtime.ajouter_boucle(self.parser) def test_absence_et_valeur_implicite_conservees(self) -> None: self.assertIsNone(self.parser.parse_args([]).loop) self.assertEqual(self.parser.parse_args(["--loop"]).loop, 10) def test_cadence_positive_acceptee(self) -> None: self.assertEqual(self.parser.parse_args(["--loop", "1"]).loop, 1) def test_zero_negatif_et_texte_refuses_proprement(self) -> None: for valeur in ("0", "-1", "texte"): with self.subTest(valeur=valeur): with contextlib.redirect_stderr(io.StringIO()): with self.assertRaises(SystemExit) as erreur: self.parser.parse_args(["--loop", valeur]) self.assertEqual(erreur.exception.code, 2) class ValidationPortTests(unittest.TestCase): def test_bornes_valides_acceptees(self) -> None: self.assertEqual(runtime.port_valide("1"), 1) self.assertEqual(runtime.port_valide("65535"), 65535) def test_ports_hors_plage_et_texte_refuses(self) -> None: for valeur in ("0", "-1", "65536", "texte"): with self.subTest(valeur=valeur): with self.assertRaises(argparse.ArgumentTypeError): runtime.port_valide(valeur) def test_serve_utilise_le_validateur_partage(self) -> None: import serve self.assertEqual(analyser(serve.parse_args, "serve", "--port", "1").port, 1) self.assertIsNone(analyser(serve.parse_args, "serve").hub) self.assertEqual( analyser(serve.parse_args, "serve", "--hub", "Jardin").hub, "Jardin", ) with contextlib.redirect_stderr(io.StringIO()): with self.assertRaises(SystemExit) as erreur: analyser(serve.parse_args, "serve", "--port", "65536") self.assertEqual(erreur.exception.code, 2) def test_watch_utilise_le_validateur_partage(self) -> None: import watch self.assertEqual( analyser(watch.parse_args, "watch", "--port", "65535").port, 65535, ) with contextlib.redirect_stderr(io.StringIO()): with self.assertRaises(SystemExit) as erreur: analyser(watch.parse_args, "watch", "--port", "0") self.assertEqual(erreur.exception.code, 2) class ValidationJoursTests(unittest.TestCase): def test_zero_et_valeur_positive_sont_acceptes(self) -> None: self.assertEqual(runtime.jours_non_negatifs("0"), 0) self.assertEqual(runtime.jours_non_negatifs("30"), 30) def test_valeur_negative_et_texte_sont_refuses(self) -> None: for valeur in ("-1", "texte"): with self.subTest(valeur=valeur): with self.assertRaises(argparse.ArgumentTypeError): runtime.jours_non_negatifs(valeur) class ProcessusSansPsTests(unittest.TestCase): """AUDIT-2026-08-13, 28.85 : `ps` absent (python:3.12-slim, sans procps) plantait toute l'appli des le premier "start" en conteneur - constate en reel sur l'image Docker Hub. Reproduit ici l'environnement POSIX sans dependre d'une vraie machine sans `ps`.""" def test_identite_processus_sans_ps_degrade_sans_planter(self) -> None: with mock.patch.object(runtime.os, "name", "posix"), \ mock.patch.object(runtime, "lancer", side_effect=FileNotFoundError(2, "No such file or directory", "ps")): self.assertIsNone(runtime.identite_processus(os.getpid())) def test_processus_vivant_sans_ps_reste_vivant(self) -> None: with mock.patch.object(runtime.os, "name", "posix"), \ mock.patch.object(runtime.os, "kill", return_value=None), \ mock.patch.object(runtime, "lancer", side_effect=FileNotFoundError(2, "No such file or directory", "ps")): self.assertTrue(runtime.processus_vivant(os.getpid())) @unittest.skipUnless(os.name == "nt", "nécessite l'API Win32") class ProcessusVivantWindowsTests(unittest.TestCase): """Revue du 27/08 : sous Windows, processus_vivant() interroge l'OS directement via OpenProcess/GetExitCodeProcess (API Win32), sans passer par un sous-processus tasklist - à la fois pour la sûreté (bug 1 : un échec d'ouverture ne prouve rien sur le pid, sauf s'il est absent) et pour la vitesse/déterminisme (constaté en réel : arreter() vérifie plusieurs membres chaque seconde pendant l'attente, un tasklist par vérification rend le délai dépendant de la charge machine).""" @staticmethod def _get_exit_code(code: int): def cb(handle, ref): ref._obj.value = code return 1 return cb def test_pid_vivant_est_detecte(self) -> None: with mock.patch.object(runtime.os, "name", "nt"), \ mock.patch("ctypes.windll.kernel32.OpenProcess", return_value=12345), \ mock.patch("ctypes.windll.kernel32.GetExitCodeProcess", side_effect=self._get_exit_code(259)), \ mock.patch("ctypes.windll.kernel32.CloseHandle", return_value=1): self.assertTrue(runtime.processus_vivant(os.getpid())) def test_pid_termine_est_bien_mort(self) -> None: with mock.patch.object(runtime.os, "name", "nt"), \ mock.patch("ctypes.windll.kernel32.OpenProcess", return_value=12345), \ mock.patch("ctypes.windll.kernel32.GetExitCodeProcess", side_effect=self._get_exit_code(0)), \ mock.patch("ctypes.windll.kernel32.CloseHandle", return_value=1): self.assertFalse(runtime.processus_vivant(999_999_999)) def test_ouverture_refusee_reste_vivant(self) -> None: with mock.patch.object(runtime.os, "name", "nt"), \ mock.patch("ctypes.windll.kernel32.OpenProcess", return_value=0), \ mock.patch("ctypes.windll.kernel32.GetLastError", return_value=5): # ERROR_ACCESS_DENIED self.assertTrue(runtime.processus_vivant(os.getpid())) def test_ouverture_echouee_pid_absent_est_bien_mort(self) -> None: with mock.patch.object(runtime.os, "name", "nt"), \ mock.patch("ctypes.windll.kernel32.OpenProcess", return_value=0), \ mock.patch("ctypes.windll.kernel32.GetLastError", return_value=87): # ERROR_INVALID_PARAMETER self.assertFalse(runtime.processus_vivant(999_999_999)) def test_get_exit_code_en_echec_reste_vivant(self) -> None: with mock.patch.object(runtime.os, "name", "nt"), \ mock.patch("ctypes.windll.kernel32.OpenProcess", return_value=12345), \ mock.patch("ctypes.windll.kernel32.GetExitCodeProcess", return_value=0), \ mock.patch("ctypes.windll.kernel32.CloseHandle", return_value=1): self.assertTrue(runtime.processus_vivant(os.getpid())) class VerrouTests(unittest.TestCase): """B-05 : acquisition atomique, jamais de vol d'un propriétaire vivant, récupération d'un verrou abandonné par un processus mort.""" def fichier(self, nom: str) -> "object": return runtime.app_dir() / f".blink_{nom}.lock" def test_B05_verrou_d_un_processus_mort_est_recupere(self) -> None: cible = self.fichier("crash-test") cible.write_text(json.dumps( {"owner": "victime", "pid": 999_999, "jeton": "perime", "at": 0} ), encoding="utf-8") with mock.patch.object(runtime, "processus_vivant", return_value=False): with runtime.verrou("crash-test", "sauveteur"): self.assertTrue(cible.exists()) contenu = json.loads(cible.read_text(encoding="utf-8")) self.assertEqual(contenu["owner"], "sauveteur") self.assertFalse(cible.exists()) def test_verrou_pid_recycle_identite_differente_est_recupere(self) -> None: """AUDIT-2026-08-13, 28.82/28.84 : constaté en réel, la boucle merge est restée bloquée plus de 15h après un redémarrage Windows. Un pid vivant (recyclé par un autre processus après la mort du vrai propriétaire) ne doit pas passer pour lui : seule l'identité (date de démarrage réelle, jamais recyclée) fait foi.""" cible = self.fichier("pid-recycle") cible.write_text(json.dumps( {"owner": "fantome", "pid": os.getpid(), "jeton": "perime", "at": time.time(), "identite": "identite-qui-ne-correspond-a-rien"} ), encoding="utf-8") with runtime.verrou("pid-recycle", "sauveteur"): self.assertTrue(cible.exists()) contenu = json.loads(cible.read_text(encoding="utf-8")) self.assertEqual(contenu["owner"], "sauveteur") self.assertFalse(cible.exists()) def test_verrou_meme_identite_reste_protege(self) -> None: """Contrepoint du precedent (B-05) : un pid vivant dont l'identite correspond vraiment ne doit toujours pas etre vole.""" cible = self.fichier("meme-identite") identite = runtime.identite_processus(os.getpid()) cible.write_text(json.dumps( {"owner": "legitime", "pid": os.getpid(), "jeton": "valide", "at": time.time(), "identite": identite} ), encoding="utf-8") with self.assertRaises(runtime.BusyError): with runtime.verrou("meme-identite", "voleur", attente=0): pass cible.unlink() def test_verrou_ancien_format_sans_identite_reste_protege(self) -> None: """Une marque ecrite avant ce correctif n'a pas de champ "identite" : rien a comparer, donc pas de purge a tort - ancien comportement conserve pour ce cas.""" cible = self.fichier("ancien-format") cible.write_text(json.dumps( {"owner": "legitime", "pid": os.getpid(), "jeton": "valide", "at": time.time()} ), encoding="utf-8") with self.assertRaises(runtime.BusyError): with runtime.verrou("ancien-format", "voleur", attente=0): pass cible.unlink() def test_verrou_identite_actuelle_illisible_est_un_pid_recycle(self) -> None: """Règle de la revue du 27/08 (bug 1) renversée le 29/09/2026, avec l'accord de Nico : elle tenait un None pour « on ne sait pas » et gardait le verrou. En réel, après un redémarrage, le pid du verrou est repris par un service protégé (accès refusé) et le démarrage automatique restait bloqué. Un propriétaire de notre compte est toujours lisible : illisible, avec une identité enregistrée, veut dire recyclé.""" cible = self.fichier("identite-inconnue") cible.write_text(json.dumps( {"owner": "ancien", "pid": os.getpid(), "jeton": "valide", "at": time.time(), "identite": "identite-enregistree-a-la-creation"} ), encoding="utf-8") with mock.patch.object(runtime, "identite_processus", return_value=None): with runtime.verrou("identite-inconnue", "nouveau", attente=0): self.assertEqual(json.loads(cible.read_text(encoding="utf-8"))["owner"], "nouveau") def test_verrou_corrompu_leve_busyerror_au_lieu_de_boucler(self) -> None: """Bug #3, revue de code du 0eab463 : un fichier de verrou présent mais illisible (JSON corrompu) bouclait indéfiniment en ignorant `attente`, jamais de BusyError, jamais de main rendue. Lancé sur un thread à part avec join(timeout) : si le correctif régresse, ce test échoue par timeout plutôt que de pendre toute la suite.""" cible = self.fichier("corrompu") cible.write_text("pas du json valide", encoding="utf-8") resultat = {} def tenter(): debut = time.monotonic() try: with runtime.verrou("corrompu", "moi", attente=0.2): pass except runtime.BusyError: resultat["busy"] = True except Exception as erreur: # pragma: no cover - diagnostic seulement resultat["erreur"] = erreur resultat["duree"] = time.monotonic() - debut fil = threading.Thread(target=tenter, daemon=True) fil.start() fil.join(timeout=5) self.assertFalse(fil.is_alive(), "verrou() sur fichier corrompu ne rend jamais la main") self.assertTrue(resultat.get("busy"), resultat.get("erreur")) self.assertLess(resultat["duree"], 3) cible.unlink(missing_ok=True) def test_verrou_double_purge_ne_supprime_pas_un_verrou_frais(self) -> None: """Bug #3, revue de code du 0eab463 : un second processus qui conclut aussi « propriétaire mort » ne doit pas supprimer, entre-temps, le verrou qu'un premier vient de recréer sous un jeton différent - il doit constater le nouveau jeton et renoncer, pas le voler.""" cible = self.fichier("double-purge") perime = {"owner": "victime", "pid": 999_999, "jeton": "perime", "at": 0} frais = {"owner": "rescape", "pid": os.getpid(), "jeton": "frais", "at": time.time()} cible.write_text(json.dumps(perime), encoding="utf-8") lectures = [perime, frais, frais, frais, frais] def vivant(pid): return pid == os.getpid() with mock.patch.object(runtime, "_lire_verrou", side_effect=lectures), \ mock.patch.object(runtime, "processus_vivant", side_effect=vivant): with self.assertRaises(runtime.BusyError): with runtime.verrou("double-purge", "voleur", attente=0): pass # Le fichier réel n'a jamais bougé : la lecture simulée « frais » ne # provient que du mock, mais la garantie testée est que rien n'a # tenté de le supprimer entre les deux lectures. self.assertTrue(cible.exists()) cible.unlink() def test_B05_liberation_ne_retire_que_son_propre_jeton(self) -> None: cible = self.fichier("jeton-etranger") with runtime.verrou("jeton-etranger", "moi"): # Un autre propriétaire a repris ce fichier entre-temps (cas # limite) : notre sortie ne doit pas effacer sa marque. cible.write_text(json.dumps( {"owner": "autrui", "pid": os.getpid(), "jeton": "pas-le-mien"} ), encoding="utf-8") self.assertTrue(cible.exists()) self.assertEqual( json.loads(cible.read_text(encoding="utf-8"))["owner"], "autrui") cible.unlink() def test_B05_relache_normalement_a_la_sortie(self) -> None: cible = self.fichier("cycle-normal") with runtime.verrou("cycle-normal", "moi"): self.assertTrue(cible.exists()) self.assertFalse(cible.exists()) @unittest.skipUnless(os.name == "nt", "état propre au système de fichiers de Windows") class VerrouEnAttenteDeSuppressionTests(unittest.TestCase): """CI Windows des 25 et 26/09/2026 : un verrou que son propriétaire vient de supprimer, mais qu'un antivirus ou l'indexeur tient encore ouvert, reste « en attente de suppression ». Le recréer lève alors PermissionError, ni succès ni FileExistsError, et verrou() laissait passer l'erreur : l'exclusion lancée en arrière-plan mourait sans rien appliquer. On reproduit ici ce vrai état du système de fichiers, par un lecteur qui partage tout et une suppression à la fermeture, la seule que connaisse Windows 7. Sous Windows 10 22H2, os.remove() efface le nom tout de suite, même lecteur ouvert ; les runners Windows de la CI ont pourtant bien connu cet état.""" GENERIC_READ = 0x80000000 DELETE = 0x00010000 PARTAGE_TOTAL = 0x1 | 0x2 | 0x4 # lecture, écriture, suppression OPEN_EXISTING = 3 FILE_FLAG_DELETE_ON_CLOSE = 0x04000000 def setUp(self) -> None: import ctypes from ctypes import wintypes self.ctypes = ctypes self.invalide = wintypes.HANDLE(-1).value # Instance propre à ce test : ses argtypes ne touchent pas # ctypes.windll.kernel32, partagé avec le code testé. self.k32 = ctypes.WinDLL("kernel32", use_last_error=True) self.k32.CreateFileW.restype = wintypes.HANDLE self.k32.CreateFileW.argtypes = [ wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE] self.k32.CloseHandle.argtypes = [wintypes.HANDLE] temporaire = tempfile.TemporaryDirectory(prefix="blink-verrou-suppression-") self.addCleanup(temporaire.cleanup) self.racine = Path(temporaire.name) self.fichier = self.racine / ".blink_suppression.lock" self.lecteurs = [] self.addCleanup(self.liberer) def ouvrir(self, chemin: Path, acces: int, drapeaux: int = 0): poignee = self.k32.CreateFileW(str(chemin), acces, self.PARTAGE_TOTAL, None, self.OPEN_EXISTING, drapeaux, None) if poignee in (None, self.invalide): raise self.ctypes.WinError(self.ctypes.get_last_error()) return poignee def mettre_en_suppression(self, chemin: Path) -> None: """Laisse `chemin` en attente de suppression, retenu par un lecteur que liberer() fermera.""" chemin.write_bytes(b"{}") self.lecteurs.append(self.ouvrir(chemin, self.GENERIC_READ)) self.k32.CloseHandle(self.ouvrir(chemin, self.DELETE, self.FILE_FLAG_DELETE_ON_CLOSE)) # L'état est bien reproduit : la création exclusive est refusée. with self.assertRaises(PermissionError): os.close(os.open(chemin, os.O_CREAT | os.O_EXCL | os.O_WRONLY)) def liberer(self) -> None: """Ferme le lecteur, ce qui achève la suppression. Sans effet la seconde fois : une poignée fermée deux fois pourrait en viser une autre, réattribuée entre-temps.""" try: poignee = self.lecteurs.pop() except IndexError: return self.k32.CloseHandle(poignee) def prendre(self, attente: float) -> dict: """Tente le verrou sur un fil à part : une régression qui bouclerait sans fin échoue ici au lieu de pendre toute la suite.""" resultat = {} def tenter() -> None: try: with runtime.verrou("suppression", "moi", attente=attente, racine=self.racine): resultat["owner"] = json.loads( self.fichier.read_text(encoding="utf-8"))["owner"] except Exception as erreur: # rapportée par l'assertion du test resultat["erreur"] = erreur fil = threading.Thread(target=tenter, daemon=True) fil.start() fil.join(timeout=10) self.assertFalse(fil.is_alive(), "verrou() ne rend jamais la main") return resultat def test_verrou_obtenu_des_que_la_suppression_s_acheve(self) -> None: self.mettre_en_suppression(self.fichier) threading.Timer(0.3, self.liberer).start() self.assertEqual(self.prendre(attente=5), {"owner": "moi"}) self.assertFalse(self.fichier.exists()) def test_refus_qui_dure_leve_busyerror_a_l_echeance(self) -> None: self.mettre_en_suppression(self.fichier) erreur = self.prendre(attente=0.3).get("erreur") self.assertIsInstance(erreur, runtime.BusyError, repr(erreur)) def test_marque_de_purge_en_attente_de_suppression(self) -> None: # Verrou abandonné par un processus mort : sa purge passe par la # marque .purge, elle aussi créée puis supprimée à chaque passage. self.fichier.write_text(json.dumps( {"owner": "victime", "pid": 999_999, "jeton": "perime", "at": 0}), encoding="utf-8") self.mettre_en_suppression(self.fichier.with_name(self.fichier.name + ".purge")) threading.Timer(0.3, self.liberer).start() with mock.patch.object(runtime, "processus_vivant", return_value=False): self.assertEqual(self.prendre(attente=5), {"owner": "moi"}) class RepeterTests(unittest.TestCase): """I-17 et O-05 : une erreur de tour ne doit pas tuer la boucle, et la prochaine échéance doit se calculer depuis le début du tour courant.""" def test_I17_une_erreur_de_tour_n_arrete_pas_la_repetition(self) -> None: appels = [] def travail(): appels.append(len(appels)) if len(appels) == 1: raise RuntimeError("panne transitoire") if len(appels) >= 3: raise KeyboardInterrupt return 0 # Horloge et sommeil liés : depuis l'arrêt coopératif (revue du # 27/08), le sommeil entre deux tours est scindé en tranches d'une # seconde qui relisent time.monotonic() à chaque tranche - un # sommeil muet avec une horloge réelle non mockée boucle alors à # vide pendant toute la période réelle (~5 min ici) avant de s'en # apercevoir. Ce que ce test vérifie (une erreur n'arrête pas la # répétition) ne dépend pas du nombre exact de tranches. horloge = {"t": 0.0} def faux_sommeil(duree): horloge["t"] += duree with mock.patch("time.monotonic", side_effect=lambda: horloge["t"]), \ mock.patch("time.sleep", side_effect=faux_sommeil) as sommeil, \ contextlib.redirect_stdout(io.StringIO()): code = runtime.repeter(travail, 5) self.assertEqual(code, 0) self.assertEqual(len(appels), 3) self.assertGreater(sommeil.call_count, 0) def test_O05_echeance_calculee_depuis_le_debut_du_tour(self) -> None: """Un tour de 15 s à cadence 1 min doit dormir ~45 s au total, pas 60 s.""" horloge = {"t": 1_000.0} def maintenant(): return horloge["t"] appels = [] def travail(): appels.append(None) horloge["t"] += 15.0 # le tour "dure" 15 s if len(appels) >= 2: raise KeyboardInterrupt durees = [] def faux_sommeil(duree): durees.append(duree) horloge["t"] += duree with mock.patch("time.monotonic", side_effect=maintenant), \ mock.patch("time.sleep", side_effect=faux_sommeil), \ contextlib.redirect_stdout(io.StringIO()): runtime.repeter(travail, 1) # Le sommeil est désormais scindé en tranches courtes (arrêt # coopératif, revue du 27/08) plutôt qu'un seul gros sommeil : c'est # leur somme qui doit valoir 45 s, pas dépasser jusqu'à 60 s. self.assertEqual(sum(durees), 45.0) class ExtraireModeBootstrapTests(unittest.TestCase): """Revue du 27/08, bug 4 : reproductible avec "download --bootstrap= none", dont les arguments n'étaient nettoyés par personne avant d'atteindre argparse (aucun parseur de verbe ne déclare --bootstrap).""" def setUp(self) -> None: self.ancien = os.environ.pop("BLINK_BOOTSTRAP", None) def tearDown(self) -> None: if self.ancien is None: os.environ.pop("BLINK_BOOTSTRAP", None) else: os.environ["BLINK_BOOTSTRAP"] = self.ancien def test_retire_loption_et_pose_la_variable(self) -> None: reste = runtime.extraire_mode_bootstrap( ["download", "--bootstrap=none", "--from", "usb"]) self.assertEqual(reste, ["download", "--from", "usb"]) self.assertEqual(os.environ.get("BLINK_BOOTSTRAP"), "none") def test_argv_sans_loption_reste_inchange(self) -> None: argv = ["download", "--from", "usb"] self.assertEqual(runtime.extraire_mode_bootstrap(argv), argv) self.assertNotIn("BLINK_BOOTSTRAP", os.environ) class DependancesTests(unittest.TestCase): def test_imageio_ffmpeg_fait_partie_des_dependances(self) -> None: """find_ffmpeg() (merge_daily.py) en dépend par défaut : absent du venv auto-créé, un compte sans ffmpeg système échouait en silence jusqu'au premier assemblage.""" racine = Path(__file__).resolve().parent self.assertIn("imageio-ffmpeg", runtime._amorcage.dependances_directes(racine / "requirements.in")) if __name__ == "__main__": unittest.main()