"""Non-régression de la progression globale des nouveaux clips.""" import contextlib import datetime as dt import io import os import sys import tempfile import threading import unittest from pathlib import Path from types import SimpleNamespace from unittest import mock import blink_engine import blink_cli import blink_models import blink_registre import runtime import serve def boite_mp4(nom: bytes, contenu: bytes = b"") -> bytes: return (len(contenu) + 8).to_bytes(4, "big") + nom + contenu VIDEO_VALIDE = ( boite_mp4(b"ftyp", b"isom\x00\x00\x02\x00isomiso2") + boite_mp4(b"moov") + boite_mp4(b"mdat", b"paquet-video") ) class FauxSync: def __init__(self, identifiant, reseau): self.sync_id = identifiant self.network_id = reseau class FauxClip: def __init__(self, identifiant, camera, instant, reseau, appareil): self.id = identifiant self.name = camera self.created_at = instant self.network_id = reseau self.device_id = appareil self.size = 1 self.download_issue = None self.telechargements = 0 async def download_to(self, _blink, cible): self.telechargements += 1 cible.parent.mkdir(parents=True, exist_ok=True) cible.write_bytes(VIDEO_VALIDE) return True async def delete_video(self, _blink): return True class ProgressionGlobaleTests(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.temporaire = tempfile.TemporaryDirectory(prefix="blink_progress_") self.home = Path(self.temporaire.name) self.ancien_home = os.environ.get("BLINK_HOME") os.environ["BLINK_HOME"] = str(self.home) # Les tests de cette classe portent sur l'ordonnancement ; leurs MP4 # synthétiques passent le contrôle structurel sans lancer ffmpeg. self.sonde_media = mock.patch.object( blink_engine.md, "valid_mp4_complet", side_effect=blink_engine.md.valid_mp4, ) self.sonde_media.start() self.addCleanup(self.sonde_media.stop) async def asyncTearDown(self): if self.ancien_home is None: os.environ.pop("BLINK_HOME", None) else: os.environ["BLINK_HOME"] = self.ancien_home self.temporaire.cleanup() async def test_multi_modules_cloud_et_repli_gardent_un_total_unique(self): """Deux modules + cloud donnent un seul 0/N…N/N sans doublon. Le premier clip USB échoue mais existe aussi dans le cloud : ce dernier est son repli et conserve le même cran, au lieu d'agrandir ou remettre à zéro la barre. """ base = dt.datetime(2026, 8, 28, 10, tzinfo=dt.timezone.utc) sync_a = FauxSync(10, 7) sync_b = FauxSync(20, 8) usb_repli = FauxClip("usb-a", "Entrée", base, 7, "cam-a") usb_ok = FauxClip("usb-b", "Jardin", base + dt.timedelta(minutes=1), 8, "cam-b") cloud_repli = FauxClip( "cloud-a", "Entrée", base + dt.timedelta(seconds=1), 7, "cam-a", ) cloud_c = FauxClip( "cloud-c", "Garage", base + dt.timedelta(minutes=2), 7, "cam-c", ) cloud_d = FauxClip( "cloud-d", "Salon", base + dt.timedelta(minutes=3), 7, "cam-d", ) locaux = {sync_a: [usb_repli], sync_b: [usb_ok]} appels_travail = [] async def lire_local(sync): return locaux[sync] async def telecharger_usb(_blink, clip, cible, _overwrite): if clip is usb_repli: return "failed" cible.parent.mkdir(parents=True, exist_ok=True) cible.write_bytes(VIDEO_VALIDE) return "downloaded" def publier(quoi, fait=0, total=0, cle=None): appels_travail.append((quoi, fait, total, cle)) arguments = SimpleNamespace( since=None, camera=None, command="download", output=self.home / "clips", hub=None, overwrite=False, source="all", loop=None, ) with mock.patch.object( blink_models, "read_local_manifest", side_effect=lire_local, ), mock.patch.object( blink_models, "read_cloud_manifest", new=mock.AsyncMock(return_value=[cloud_repli, cloud_c, cloud_d]), ), mock.patch.object( blink_engine, "download_clip", side_effect=telecharger_usb, ), mock.patch.object( runtime, "travail", side_effect=publier, ), mock.patch.object( runtime, "lire_suppression_auto", return_value=set(), ), mock.patch.object(runtime, "marquer"), \ mock.patch.object(runtime, "toast"), \ mock.patch.object(runtime, "lire_langue", return_value="fr"), \ mock.patch.object(runtime, "lire_reglages", return_value={"port": 8765}), \ contextlib.redirect_stdout(io.StringIO()) as sortie: code = await blink_engine.un_passage( object(), arguments, [("Maison", sync_a), ("Annexe", sync_b)], ) progression = [ (fait, total) for _quoi, fait, total, cle in appels_travail if cle == "phase.download_clips" ] self.assertTrue(progression) self.assertEqual(progression[0], (0, 4)) self.assertEqual(progression[-1], (4, 4)) self.assertTrue(all(total == 4 for _fait, total in progression)) self.assertEqual( [fait for fait, _total in progression], sorted(fait for fait, _total in progression), ) self.assertIn("[1/4] 0%", sortie.getvalue()) self.assertIn("[4/4] 100%", sortie.getvalue()) self.assertEqual(cloud_repli.telechargements, 1) self.assertEqual(cloud_c.telechargements, 1) self.assertEqual(cloud_d.telechargements, 1) # Le repli a sauvé le média : l'échec de sa première source reste dans # le journal, mais ne transforme pas le passage global en échec. self.assertEqual(code, 0) async def test_un_echec_definitif_termine_quand_meme_la_barre(self): """Un clip traité mais refusé compte comme fini, sans rester à 0/N.""" sync = FauxSync(10, 7) clip = FauxClip( "usb-echec", "Entrée", dt.datetime(2026, 8, 28, 10, tzinfo=dt.timezone.utc), 7, "cam-a", ) appels_travail = [] arguments = SimpleNamespace( since=None, camera=None, command="download", output=self.home / "clips-echec", hub=None, overwrite=False, source="usb", loop=None, ) def publier(quoi, fait=0, total=0, cle=None): appels_travail.append((fait, total, cle)) with mock.patch.object( blink_models, "read_local_manifest", new=mock.AsyncMock(return_value=[clip]), ), mock.patch.object( blink_engine, "download_clip", new=mock.AsyncMock(return_value="failed"), ), mock.patch.object(runtime, "travail", side_effect=publier), \ mock.patch.object(runtime, "lire_suppression_auto", return_value=set()), \ mock.patch.object(runtime, "marquer"), \ contextlib.redirect_stdout(io.StringIO()) as sortie: code = await blink_engine.un_passage( object(), arguments, [("Maison", sync)], ) progression = [ (fait, total) for fait, total, cle in appels_travail if cle == "phase.download_clips" ] self.assertEqual(progression[0], (0, 1)) self.assertEqual(progression[-1], (1, 1)) self.assertIn("[1/1] 100%", sortie.getvalue()) self.assertEqual(code, 1) async def test_doublon_cloud_deja_reussi_en_usb_ne_relance_pas_un_spinner(self): instant = dt.datetime(2026, 8, 28, 11, tzinfo=dt.timezone.utc) sync = FauxSync(10, 7) usb = FauxClip("usb", "Entrée", instant, 7, "cam-a") cloud = FauxClip( "cloud", "Entrée", instant + dt.timedelta(seconds=1), 7, "cam-a", ) arguments = SimpleNamespace( since=None, camera=None, command="download", output=self.home / "clips-doublon", hub=None, overwrite=False, source="all", loop=None, ) async def telecharger_usb(_blink, _clip, cible, _overwrite): cible.parent.mkdir(parents=True, exist_ok=True) cible.write_bytes(VIDEO_VALIDE) return "downloaded" with mock.patch.object( blink_models, "read_local_manifest", new=mock.AsyncMock(return_value=[usb]), ), mock.patch.object( blink_models, "read_cloud_manifest", new=mock.AsyncMock(return_value=[cloud]), ), mock.patch.object( blink_engine, "download_clip", side_effect=telecharger_usb, ), mock.patch.object(runtime, "travail"), \ mock.patch.object(runtime, "lire_suppression_auto", return_value=set()), \ mock.patch.object(runtime, "marquer"), \ mock.patch.object(runtime, "toast"), \ mock.patch.object(runtime, "lire_langue", return_value="fr"), \ mock.patch.object(runtime, "lire_reglages", return_value={"port": 8765}), \ contextlib.redirect_stdout(io.StringIO()) as sortie: code = await blink_engine.un_passage( object(), arguments, [("Maison", sync)], ) self.assertEqual(code, 0) self.assertEqual(cloud.telechargements, 0) self.assertNotIn("=== CLOUD DE L'ABONNEMENT ===", sortie.getvalue()) self.assertIn("[1/1] 100%", sortie.getvalue()) async def test_exception_de_suppression_usb_n_interrompt_pas_le_lot(self): sync = FauxSync(10, 7) instant = dt.datetime(2026, 8, 28, 12, tzinfo=dt.timezone.utc) premier = FauxClip("usb-1", "Entrée", instant, 7, "cam-a") second = FauxClip( "usb-2", "Jardin", instant + dt.timedelta(minutes=1), 7, "cam-b", ) premier.delete_video = mock.AsyncMock(side_effect=RuntimeError("API indisponible")) cles_suppression_auto = { blink_registre.camera_setting_key(sync, premier), blink_registre.camera_setting_key(sync, second), } appels_travail = [] arguments = SimpleNamespace( since=None, camera=None, command="download", output=self.home / "clips-suppression", hub=None, overwrite=False, source="usb", loop=None, ) async def telecharger(_blink, _clip, cible, _overwrite): cible.parent.mkdir(parents=True, exist_ok=True) cible.write_bytes(VIDEO_VALIDE) return "downloaded" with mock.patch.object( blink_models, "read_local_manifest", new=mock.AsyncMock(return_value=[premier, second]), ), mock.patch.object( blink_engine, "download_clip", side_effect=telecharger, ), mock.patch.object( runtime, "travail", side_effect=lambda _q, fait=0, total=0, cle=None: appels_travail.append((fait, total, cle)), ), mock.patch.object( runtime, "lire_suppression_auto", return_value=cles_suppression_auto, ), mock.patch.object(runtime, "marquer"), \ mock.patch.object(runtime, "toast"), \ mock.patch.object(runtime, "lire_langue", return_value="fr"), \ mock.patch.object(runtime, "lire_reglages", return_value={"port": 8765}), \ contextlib.redirect_stdout(io.StringIO()) as sortie: code = await blink_engine.un_passage( object(), arguments, [("Maison", sync)], ) progression = [ (fait, total) for fait, total, cle in appels_travail if cle == "phase.download_clips" ] self.assertEqual(code, 0) self.assertEqual(progression[-1], (2, 2)) self.assertIn("[2/2] 100%", sortie.getvalue()) self.assertIn("Suppression impossible", sortie.getvalue()) async def test_planificateur_commence_global_puis_respecte_les_deux_cadences(self): horloge = {"t": 0.0} sources = [] async def faire(_blink, arguments, _modules, repetition): self.assertTrue(repetition) sources.append(arguments.source) return blink_engine._ResultatPassage(0, True) async def attendre(echeance): horloge["t"] = echeance return True arguments = SimpleNamespace(source="usb", usb_loop=2, cloud_loop=1, loop=None) with mock.patch.object(blink_engine, "_faire_passage", side_effect=faire), \ mock.patch.object(blink_engine, "_attendre_echeance", side_effect=attendre), \ mock.patch.object(blink_engine.time, "monotonic", side_effect=lambda: horloge["t"]), \ mock.patch.object(runtime, "arret_demande", side_effect=lambda: len(sources) >= 3): code = await blink_engine._boucler_sources( object(), arguments, [("Maison", FauxSync(10, 7))], 2, 1, ) self.assertEqual(code, 0) self.assertEqual(sources, ["all", "cloud", "all"]) self.assertEqual(arguments.source, "usb", "les copies ne mutent pas la CLI") async def test_bootstrap_global_est_reessaye_apres_contention(self): sources = [] arrete = {"oui": False} async def faire(_blink, arguments, _modules, repetition): sources.append(arguments.source) if len(sources) == 1: return blink_engine._ResultatPassage(0, False) arrete["oui"] = True return blink_engine._ResultatPassage(0, True) with mock.patch.object(blink_engine, "_faire_passage", side_effect=faire), \ mock.patch.object( blink_engine, "_attendre_echeance", new=mock.AsyncMock(return_value=True), ), mock.patch.object(runtime, "arret_demande", side_effect=lambda: arrete["oui"]): await blink_engine._boucler_sources( object(), SimpleNamespace(source="all"), [], 10, 1, ) self.assertEqual(sources, ["all", "all"]) async def test_contention_d_un_clic_manuel_n_est_pas_un_faux_succes(self): class VerrouOccupe: def __enter__(self): raise runtime.BusyError("worker automatique") def __exit__(self, *_args): return False with mock.patch.object(runtime, "verrou", return_value=VerrouOccupe()), \ contextlib.redirect_stdout(io.StringIO()): resultat = await blink_engine._faire_passage( object(), SimpleNamespace(), [], repetition=False, ) self.assertEqual(resultat, blink_engine._ResultatPassage(1, False)) class ProgressionRuntimeTests(unittest.TestCase): def setUp(self): self.temporaire = tempfile.TemporaryDirectory(prefix="blink_work_state_") self.ancien_home = os.environ.get("BLINK_HOME") os.environ["BLINK_HOME"] = self.temporaire.name def tearDown(self): if self.ancien_home is None: os.environ.pop("BLINK_HOME", None) else: os.environ["BLINK_HOME"] = self.ancien_home self.temporaire.cleanup() def test_fin_n_n_reste_affichable_sans_rester_active(self): runtime.travail( "Téléchargement des clips", 1, 1, cle="phase.download_clips", ) runtime.fin_travail(conserver=10) self.assertEqual(runtime.travail_en_cours(), {}) visible = runtime.travail_affichable() self.assertEqual((visible["fait"], visible["total"]), (1, 1)) self.assertTrue(visible["termine"]) self.assertFalse(visible["actif"]) def test_download_termine_et_merge_actif_ne_s_ecrasent_pas(self): with mock.patch.object(runtime, "processus_vivant", return_value=True): with mock.patch.object(runtime.os, "getpid", return_value=101): runtime.travail( "Téléchargement des clips", 4, 4, cle="phase.download_clips", ) runtime.fin_travail(conserver=10) with mock.patch.object(runtime.os, "getpid", return_value=202): runtime.travail( "Assemblage des vidéos", 1, 3, cle="phase.merge_videos", ) visible = runtime.travail_affichable() self.assertEqual(visible["pid"], 101) self.assertTrue(visible["termine"]) self.assertTrue(visible["actif"], "le merge actif doit garder le bouton bloqué") self.assertEqual(runtime.travail_en_cours()["pid"], 202) with mock.patch.object(runtime.os, "getpid", return_value=202): runtime.fin_travail() self.assertEqual(runtime.travail_en_cours(), {}) self.assertEqual(runtime.travail_affichable()["pid"], 101) def test_conclusion_mise_a_jour_prioritaire_sur_telechargement(self): for termine in (False, True): with self.subTest(termine=termine), \ mock.patch.object(runtime, "processus_vivant", return_value=True): with mock.patch.object(runtime.os, "getpid", return_value=101): runtime.travail("Téléchargement", 1, 1, cle="phase.download_clips") if termine: runtime.fin_travail(conserver=10) with mock.patch.object(runtime.os, "getpid", return_value=202): runtime.travail("Arrêt échoué", 1, 1, cle="phase.update_noop") runtime.fin_travail(conserver=10) visible = runtime.travail_affichable() self.assertEqual(visible["cle"], "phase.update_noop") self.assertTrue(visible["termine"]) self.assertEqual(visible["actif"], not termine) self.assertEqual(bool(runtime.travail_en_cours()), not termine) with mock.patch.object(runtime.os, "getpid", return_value=202): runtime.fin_travail() self.assertEqual(runtime.travail_affichable()["pid"], 101) def test_reessai_retire_seulement_la_conclusion_de_mise_a_jour(self): with mock.patch.object(runtime, "processus_vivant", return_value=True): with mock.patch.object(runtime.os, "getpid", return_value=101): runtime.travail("Téléchargement", 1, 1, cle="phase.download_clips") runtime.fin_travail(conserver=10) with mock.patch.object(runtime.os, "getpid", return_value=202): runtime.travail("Arrêt échoué", 1, 1, cle="phase.update_noop") runtime.fin_travail(conserver=10) with mock.patch.object(runtime.os, "getpid", return_value=303): runtime.travail("Assemblage", 0, 1, cle="phase.merge") self.assertEqual(runtime.travail_affichable()["cle"], "phase.update_noop") self.assertTrue(runtime.effacer_conclusion_travail("phase.update_noop")) self.assertFalse(runtime._fichier_travail(202).exists()) self.assertTrue(runtime._fichier_travail(101).exists()) self.assertTrue(runtime._fichier_travail(303).exists()) visible = runtime.travail_affichable() self.assertEqual(visible["cle"], "phase.download_clips") self.assertTrue(visible["actif"]) self.assertEqual(runtime.travail_en_cours()["pid"], 303) def test_reessai_refuse_si_la_conclusion_ne_peut_pas_etre_retiree(self): runtime.travail("Arrêt échoué", 1, 1, cle="phase.update_noop") runtime.fin_travail(conserver=10) with mock.patch.object(runtime, "_supprimer_fiche_travail", return_value=False): self.assertFalse(runtime.effacer_conclusion_travail("phase.update_noop")) self.assertEqual(runtime.travail_affichable()["cle"], "phase.update_noop") def test_purge_ordinaire_reste_sans_attente_pendant_un_reessai(self): with mock.patch.object(runtime.os, "getpid", return_value=101): runtime.travail("Assemblage", 0, 1, cle="phase.merge") with mock.patch.object(runtime.os, "getpid", return_value=202): runtime.travail("Arrêt échoué", 1, 1, cle="phase.update_noop") runtime.fin_travail(conserver=10) with mock.patch.object(runtime, "processus_vivant", return_value=False), \ mock.patch.object(runtime, "_supprimer_fiche_travail", return_value=True) as retirer: self.assertTrue(runtime.effacer_conclusion_travail("phase.update_noop")) attentes = {appel.args[1]["pid"]: appel.kwargs["attente"] for appel in retirer.call_args_list} self.assertEqual(attentes, {101: 0, 202: 0.25}) def test_echec_du_snapshot_terminal_ne_laisse_pas_un_faux_actif(self): runtime.travail( "Téléchargement des clips", 1, 1, cle="phase.download_clips", ) with mock.patch.object(runtime, "_ecrire_fiche_travail", return_value=False): runtime.fin_travail(conserver=10) self.assertEqual(runtime.travail_en_cours(), {}) class ProgressionCliTests(unittest.TestCase): def test_cadences_internes_acceptent_un_download_global(self): argv = [ "blink2video", "download", "--from", "all", "--usb-loop", "10", "--cloud-loop", "1", ] with mock.patch.object(sys, "argv", argv): arguments = blink_cli.parse_args() self.assertEqual((arguments.usb_loop, arguments.cloud_loop), (10, 1)) def test_une_seule_cadence_interne_est_refusee(self): argv = ["blink2video", "download", "--from", "all", "--usb-loop", "10"] with mock.patch.object(sys, "argv", argv), \ contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): blink_cli.parse_args() class ProgressionInterfaceTests(unittest.TestCase): def _commandes_actualisation(self, hub, reglages=None, authentifie=True, evenements=None): with tempfile.TemporaryDirectory(prefix="blink_refresh_hub_") as dossier: racine = Path(dossier) if authentifie: (racine / "blink_auth.json").write_text("{}", encoding="utf-8") commandes = [] faux = SimpleNamespace( hub=hub, send_event=lambda evenement: (evenements.append(evenement) if evenements is not None else None) or True, suivre=lambda commande, _env, _phase: commandes.append(commande) or True, ) with mock.patch.object(serve, "BASE_DIR", racine), \ mock.patch.object( runtime, "self_command", side_effect=lambda *args: list(args), ), mock.patch.object( runtime, "lire_reglages", return_value=reglages or dict(runtime.REGLAGES_DEFAUT), ): serve.Handler.run_refresh(faux) return commandes def test_actualiser_fusion_desactivee_ne_lance_que_download(self): # Les sous-options mémorisées ne doivent pas rallumer l'interrupteur # général, même si elles sont restées cochées avant sa désactivation. for semaine in (False, True): for mois in (False, True): with self.subTest(semaine=semaine, mois=mois): evenements = [] reglages = dict(runtime.REGLAGES_DEFAUT, merge_jour=False, merge_semaine=semaine, merge_mois=mois) self.assertEqual(self._commandes_actualisation( "Jardin", reglages, evenements=evenements), [["download", "--hub", "Jardin"]]) self.assertNotIn("phase.step_merge", [e.get("phase_key") for e in evenements]) self.assertEqual(evenements[-1], {"done": True, "ok": True}) def test_actualiser_sans_session_et_sans_fusion_ne_lance_rien(self): evenements = [] self.assertEqual(self._commandes_actualisation( None, dict(runtime.REGLAGES_DEFAUT, merge_jour=False), authentifie=False, evenements=evenements), []) self.assertIn("fusion est désactivée", evenements[0]["line"]) self.assertEqual(evenements[-1], {"done": True, "ok": True}) def test_actualiser_sans_session_conserve_la_fusion_si_activee(self): self.assertEqual(self._commandes_actualisation(None, authentifie=False), [["merge", "--timezone", "Europe/Paris", "--no-timestamp", "--no-weekly", "--no-monthly"]]) def test_actualiser_sans_hub_laisse_download_choisir_tous_les_modules(self): # Réglages usine (REGLAGES_DEFAUT) : Incrustation, Hebdomadaire et # Mensuelle sont décochées par défaut, donc présentes dès le premier # lancement, pas seulement si l'utilisateur les décoche à la main. self.assertEqual( self._commandes_actualisation(None), [["download"], ["merge", "--timezone", "Europe/Paris", "--no-timestamp", "--no-weekly", "--no-monthly"]], ) def test_actualiser_conserve_un_hub_explicitement_demande(self): self.assertEqual( self._commandes_actualisation("Jardin"), [["download", "--hub", "Jardin"], ["merge", "--timezone", "Europe/Paris", "--no-timestamp", "--no-weekly", "--no-monthly"]], ) def test_actualiser_respecte_incrustation_et_agregats_actives(self): """Non-régression du bug du 2026-09-05 : Actualiser reconstruisait toujours avec horodatage et agrégats hebdo/mensuels, quels que soient les réglages de la page.""" reglages_tout_actif = { **runtime.REGLAGES_DEFAUT, "timestamp": True, "merge_semaine": True, "merge_mois": True, } self.assertEqual( self._commandes_actualisation(None, reglages_tout_actif), [["download"], ["merge", "--timezone", "Europe/Paris"]], ) def test_api_travail_ne_relit_pas_le_registre_de_clips(self): reponses = [] faux = SimpleNamespace( path="/api/travail", hote_autorise=lambda: True, jeton_valide=lambda: True, send_json=lambda corps, *_args: reponses.append(corps), ) travail = {"quoi": "Téléchargement", "fait": 2, "total": 8} with mock.patch.object(runtime, "travail_affichable", return_value=travail), \ mock.patch.object( serve, "read_entries", side_effect=AssertionError("registre relu par la sonde rapide"), ): serve.Handler.do_GET(faux) self.assertEqual(reponses, [{"travail": travail, "dossier_absent": ""}]) def test_actualiser_conserve_le_fuseau_configure(self): reglages = dict(runtime.REGLAGES_DEFAUT, timezone="America/New_York") fusion = self._commandes_actualisation(None, reglages)[1] self.assertEqual(fusion[fusion.index("--timezone") + 1], "America/New_York") def test_suivre_transmet_explicitement_zero_et_cent_pour_cent(self): class Processus: def __init__(self): self.stdout = io.StringIO( " [1/2] 0%\n" " [1/2] clip-a.mp4\n" " [1/2] 100%\n" " [2/2] clip-b.mp4\n" " [2/2] 100%\n" ) self.returncode = 0 def wait(self): return self.returncode def terminate(self): self.returncode = 1 evenements = [] faux = SimpleNamespace( send_event=lambda evenement: evenements.append(evenement) or True, ) with mock.patch.object(runtime, "demarrer", return_value=Processus()): resultat = serve.Handler.suivre(faux, ["download"], {}, "Téléchargement") progressions = [e["progress"] for e in evenements if "progress" in e] self.assertTrue(resultat) self.assertEqual(progressions[0], {"done": 0.0, "total": 2}) self.assertEqual(progressions[-1], {"done": 2.0, "total": 2}) def test_actualiser_ne_preverrouille_pas_le_hub_du_downloader(self): """Le faux enfant doit pouvoir devenir l'unique propriétaire du hub.""" with tempfile.TemporaryDirectory(prefix="blink_refresh_lock_") as dossier: ancien_home = os.environ.get("BLINK_HOME") os.environ["BLINK_HOME"] = dossier execute = [] evenements = [] def run_refresh(): with blink_engine.hub_lock("downloader-enfant"): execute.append(True) faux = SimpleNamespace( lock=threading.Lock(), send_response=lambda *_args: None, send_header=lambda *_args: None, end_headers=lambda: None, send_event=lambda evenement: evenements.append(evenement) or True, run_refresh=run_refresh, ) try: with mock.patch.object(runtime, "travail_en_cours", return_value={}), \ mock.patch.object(serve, "MODULE_SLOT", threading.Semaphore(1)): serve.Handler.stream_refresh(faux) finally: if ancien_home is None: os.environ.pop("BLINK_HOME", None) else: os.environ["BLINK_HOME"] = ancien_home self.assertEqual(execute, [True]) self.assertFalse(any(e.get("ok") is False for e in evenements)) def test_polling_reste_rapide_pour_un_worker_demarre_plus_tard(self): self.assertIn('fetch("/api/travail", { cache: "no-store" })', serve.PAGE) self.assertIn("etatDuTravail();\n(function veillerTravail()", serve.PAGE) self.assertIn("function veillerTravail()", serve.PAGE) self.assertIn("veillerTravail();\n }, 3000);", serve.PAGE) self.assertIn("veillerPassages();\n }, 60000);", serve.PAGE) self.assertIn("let travailVisible = false;", serve.PAGE) if __name__ == "__main__": unittest.main()