"""Régressions de supervision et d'onboarding de ``blink_cli``. Ces tests ne lancent aucun vrai processus et n'ouvrent aucun navigateur. """ from __future__ import annotations import contextlib import io import os import unittest from types import SimpleNamespace from unittest import mock import blink_cli import tray class _ProcessusTermine: def __init__(self, pid: int, code: int): self.pid = pid self.returncode = code def poll(self): return self.returncode class TestsSupervision(unittest.TestCase): def test_le_code_du_dernier_processus_est_collecte(self): processus = [_ProcessusTermine(101, 3), _ProcessusTermine(102, 7)] with mock.patch.object( blink_cli.runtime, "verrou_controle", return_value=contextlib.nullcontext(), ), mock.patch.object( blink_cli.runtime, "inscrire_instance", ), mock.patch.object( blink_cli.runtime, "demarrer", side_effect=processus, ), mock.patch.object( blink_cli.runtime, "self_command", return_value=["faux"], ), mock.patch.object( blink_cli.runtime, "flags_enfant", return_value=0, ), mock.patch.object( blink_cli.runtime, "lire_reglages", return_value={"port": 8765}, ), mock.patch.object( blink_cli.runtime, "demander_arret", ), mock.patch.object( blink_cli.runtime, "effacer_arret_demande", ), mock.patch.object( tray, "disponible", return_value=False, ), mock.patch.object(blink_cli.time, "sleep"): code = blink_cli.executer([ ["serve"], ["download", "--loop", "1"], ]) self.assertEqual(code, 7) def test_download_boucle_seul_est_inscrit_pour_stop(self): arguments = SimpleNamespace() with mock.patch.object( blink_cli.runtime, "verrou_controle", return_value=contextlib.nullcontext(), ), mock.patch.object( blink_cli.runtime, "inscrire_instance", ) as inscrire, mock.patch.object( blink_cli, "parse_args", return_value=arguments, ), mock.patch.object( blink_cli, "main", new=mock.AsyncMock(return_value=0), ): code = blink_cli.executer([["download", "--loop=1"]]) self.assertEqual(code, 0) inscrire.assert_called_once_with([["download", "--loop=1"]]) class TestsOnboardingPort(unittest.TestCase): def test_port_forme_egale_est_sonde(self): processus = SimpleNamespace(pid=4242, poll=mock.Mock(return_value=None)) with mock.patch.object( blink_cli.runtime, "demarrer", return_value=processus, ), mock.patch.object( blink_cli.runtime, "self_command", side_effect=lambda *arguments: list(arguments), ), mock.patch.object( blink_cli.runtime, "arreter_processus", ), mock.patch.object( blink_cli.runtime, "bootstrap", ), mock.patch.object( blink_cli, "_port_ouvert", return_value=True, ) as port_ouvert, mock.patch.dict( os.environ, {"BLINK_NO_BROWSER": "1"}, clear=False, ): code = blink_cli.accueillir( {"authenticated": True, "error": None}, ["--port=55432"], delai=5, ) self.assertEqual(code, 0) port_ouvert.assert_called_once_with(55432) def _accueillir_avec_reglages(self, reglages: dict, supplement: list): processus = SimpleNamespace(pid=4242, poll=mock.Mock(return_value=None)) with mock.patch.object( blink_cli.runtime, "lire_reglages", return_value=reglages, ), mock.patch.object( blink_cli.runtime, "demarrer", return_value=processus, ) as demarrer, mock.patch.object( blink_cli.runtime, "self_command", side_effect=lambda *arguments: list(arguments), ), mock.patch.object( blink_cli.runtime, "arreter_processus", ), mock.patch.object( blink_cli.runtime, "bootstrap", ), mock.patch.object( blink_cli, "_port_ouvert", return_value=True, ) as port_ouvert, mock.patch.dict( os.environ, {"BLINK_NO_BROWSER": "1"}, clear=False, ), contextlib.redirect_stdout(io.StringIO()) as sortie: code = blink_cli.accueillir( {"authenticated": True, "error": None}, supplement, delai=5) return code, demarrer.call_args[0][0], port_ouvert, sortie.getvalue() def test_port_des_reglages_utilise_sans_port_explicite(self): """Session expirée + port personnalisé : le serveur temporaire ne doit pas retomber sur le 8765 codé en dur de serve.py.""" reglages = dict(blink_cli.runtime.REGLAGES_DEFAUT, port=9123, timezone="America/New_York", trusted_host="100.64.0.7") code, commande, port_ouvert, sortie = self._accueillir_avec_reglages( reglages, []) self.assertEqual(code, 0) self.assertEqual(commande, [ "serve", "--port", "9123", "--timezone", "America/New_York", "--trusted-host", "100.64.0.7", ]) port_ouvert.assert_called_once_with(9123) self.assertIn("http://127.0.0.1:9123/", sortie) def test_port_explicite_l_emporte_sur_les_reglages(self): reglages = dict(blink_cli.runtime.REGLAGES_DEFAUT, port=9123) code, commande, port_ouvert, _ = self._accueillir_avec_reglages( reglages, ["--port", "9555"]) self.assertEqual(code, 0) # Le supplément suit le bloc fixe : argparse retient sa valeur. self.assertEqual(commande[-2:], ["--port", "9555"]) port_ouvert.assert_called_once_with(9555) class _ProcessusVivant: """Enfant qui tourne jusqu'à l'arrêt coopératif (demander_arret).""" arret_demande = False def __init__(self, pid: int): self.pid = pid def poll(self): return 0 if _ProcessusVivant.arret_demande else None @contextlib.contextmanager def _superviseur_simule(*, processus, icone=None, lancer=None): """executer() sans vrai processus : verrous, fiches, drapeau d'arrêt et icône simulés. ``demander_arret`` arrête les enfants simulés.""" _ProcessusVivant.arret_demande = False def demander_arret(): _ProcessusVivant.arret_demande = True with mock.patch.object( blink_cli.runtime, "verrou_controle", return_value=contextlib.nullcontext(), ), mock.patch.object( blink_cli.runtime, "inscrire_instance", ), mock.patch.object( blink_cli.runtime, "demarrer", side_effect=processus, ), mock.patch.object( blink_cli.runtime, "self_command", return_value=["faux"], ), mock.patch.object( blink_cli.runtime, "flags_enfant", return_value=0, ), mock.patch.object( blink_cli.runtime, "lire_reglages", return_value={"port": 8765}, ), mock.patch.object( blink_cli.runtime, "demander_arret", side_effect=demander_arret, ) as arret, mock.patch.object( blink_cli.runtime, "effacer_arret_demande", ), mock.patch.object( blink_cli.runtime, "arreter_processus", ), mock.patch.object( blink_cli.runtime, "lancer", side_effect=lancer, ), mock.patch.object( tray, "disponible", return_value=icone is not None, ), mock.patch.object( tray, "executer", side_effect=icone, ), mock.patch.object(blink_cli.time, "sleep"), \ contextlib.redirect_stdout(io.StringIO()) as sortie: yield arret, sortie class TestsPortIcone(unittest.TestCase): """Audit du 26/09/2026, B05 : start place les options enregistrées avant les explicites. Le serveur retient la dernière occurrence de --port ; l'icône doit recevoir ce même port, forme --port=N comprise.""" def _port_recu(self, arguments_serve: list) -> int: recus = [] def icone(port, fin, nettoyer): recus.append(port) _ProcessusVivant.arret_demande = True with _superviseur_simule(processus=[_ProcessusVivant(101), _ProcessusVivant(102)], icone=icone): blink_cli.executer([["serve", *arguments_serve], ["watch", "--loop", "1"]]) return recus[0] def test_la_derniere_occurrence_l_emporte(self): self.assertEqual(self._port_recu(["--port", "8765", "--port", "9555"]), 9555) def test_la_forme_egale_est_reconnue(self): self.assertEqual(self._port_recu(["--port", "8765", "--port=9555"]), 9555) def test_sans_port_explicite_le_port_enregistre(self): self.assertEqual(self._port_recu([]), 8765) class TestsNettoyageEtapesPonctuelles(unittest.TestCase): """Audit du 26/09/2026, B06 : une interruption ou un échec de lancement pendant une étape ponctuelle arrête aussi les processus persistants déjà démarrés, au lieu de les laisser tourner sans superviseur.""" def test_echec_de_lancement_arrete_les_persistants(self): with _superviseur_simule(processus=[_ProcessusVivant(101)], lancer=OSError("introuvable")) as (arret, _): with self.assertRaises(OSError): blink_cli.executer([["serve"], ["download"]]) arret.assert_called_once() def test_ctrl_c_pendant_une_etape_arrete_les_persistants(self): with _superviseur_simule(processus=[_ProcessusVivant(101)], lancer=KeyboardInterrupt()) as (arret, sortie): try: blink_cli.executer([["serve"], ["download"]]) except KeyboardInterrupt: # Capturée ici pour échouer proprement : laissée libre, elle # interromprait toute la suite de tests. self.fail("Ctrl+C a échappé au superviseur, serve laissé actif") arret.assert_called_once() self.assertIn(blink_cli.msg("interruption_clavier"), sortie.getvalue()) def test_sans_persistant_aucun_drapeau_d_arret(self): # Le drapeau d'arrêt est commun à tous les processus blink2video : # une composition sans enfant ne doit jamais le poser. with _superviseur_simule(processus=[], lancer=KeyboardInterrupt()) as (arret, _): with self.assertRaises(KeyboardInterrupt): blink_cli.executer([["download"], ["merge"]]) arret.assert_not_called() if __name__ == "__main__": unittest.main()