test_blink_cli_redemarrer.py
Rama main · réplica de lectura del repositorio original.
"""Non-régression de blink_cli.redemarrer (AUDIT-2026-08-13.md, section
28.32) : verbe « restart » qui sert le panneau de réglages de la page web
(bouton Appliquer et bouton Stop), sans jamais toucher à un vrai processus.
Le premier temps (sans --finaliser) doit rendre la main tout de suite après
avoir lancé le second : c'est cette sortie rapide qui le détache de l'arbre
que « stop » va abattre. Le second temps (--finaliser) doit arrêter puis,
sauf --sans-relance, relancer « start »."""
from __future__ import annotations
import contextlib
import os
import tempfile
import threading
import time
import unittest
from unittest import mock
import blink_cli
class TestsRedemarrerPremierTemps(unittest.TestCase):
"""Sans --finaliser : uniquement lancer le second temps, détaché."""
def test_lance_le_second_temps_et_rend_la_main(self):
with mock.patch.object(blink_cli.runtime, "demarrer") as demarrer, \
mock.patch.object(blink_cli.runtime, "lancer") as lancer:
code = blink_cli.redemarrer([])
self.assertEqual(code, 0)
lancer.assert_not_called()
demarrer.assert_called_once()
commande = demarrer.call_args[0][0]
self.assertIn("restart", commande)
self.assertIn("--finaliser", commande)
self.assertNotIn("--sans-relance", commande)
def test_relaie_sans_relance_au_second_temps(self):
with mock.patch.object(blink_cli.runtime, "demarrer") as demarrer:
blink_cli.redemarrer(["--sans-relance"])
commande = demarrer.call_args[0][0]
self.assertIn("--finaliser", commande)
self.assertIn("--sans-relance", commande)
class TestsRedemarrerSecondTemps(unittest.TestCase):
"""Avec --finaliser : arrêter, attendre, puis relancer sauf --sans-relance."""
def test_arrete_attend_puis_relance(self):
with mock.patch.object(blink_cli, "_arreter_instances", return_value=0) as arreter, \
mock.patch.object(blink_cli.runtime, "lire_instances",
return_value=[]) as lire_instances, \
mock.patch.object(blink_cli.runtime, "verrou_controle",
return_value=contextlib.nullcontext()), \
mock.patch.object(blink_cli.runtime, "demarrer") as demarrer, \
mock.patch.object(blink_cli.time, "sleep") as sommeil:
code = blink_cli.redemarrer(["--finaliser"])
self.assertEqual(code, 0)
arreter.assert_called_once()
# Plus rien à attendre dès le premier passage : aucune pause.
lire_instances.assert_called_once()
sommeil.assert_not_called()
# Puis relancé, avec « start ».
demarrer.assert_called_once()
self.assertIn("start", demarrer.call_args[0][0])
def test_sans_relance_n_appelle_pas_start(self):
with mock.patch.object(blink_cli, "_arreter_instances", return_value=0), \
mock.patch.object(blink_cli.runtime, "lire_instances", return_value=[]), \
mock.patch.object(blink_cli.runtime, "verrou_controle",
return_value=contextlib.nullcontext()), \
mock.patch.object(blink_cli.runtime, "demarrer") as demarrer:
code = blink_cli.redemarrer(["--finaliser", "--sans-relance"])
self.assertEqual(code, 0)
demarrer.assert_not_called()
def test_attend_que_les_instances_disparaissent(self):
# Encore là deux fois, plus là ensuite : deux pauses, pas vingt.
etats = [["encore"], ["encore"], []]
with mock.patch.object(blink_cli, "_arreter_instances", return_value=0), \
mock.patch.object(blink_cli.runtime, "lire_instances",
side_effect=etats) as lire_instances, \
mock.patch.object(blink_cli.runtime, "verrou_controle",
return_value=contextlib.nullcontext()), \
mock.patch.object(blink_cli.runtime, "demarrer"), \
mock.patch.object(blink_cli.time, "sleep") as sommeil:
blink_cli.redemarrer(["--finaliser"])
self.assertEqual(lire_instances.call_count, 3)
self.assertEqual(sommeil.call_count, 2)
def test_renonce_apres_vingt_passages_sans_empiler_une_instance(self):
with mock.patch.object(blink_cli, "_arreter_instances", return_value=0), \
mock.patch.object(blink_cli.runtime, "lire_instances",
return_value=["toujours_la"]) as lire_instances, \
mock.patch.object(blink_cli.runtime, "verrou_controle",
return_value=contextlib.nullcontext()), \
mock.patch.object(blink_cli.runtime, "demarrer") as demarrer, \
mock.patch.object(blink_cli.time, "sleep"):
code = blink_cli.redemarrer(["--finaliser"])
self.assertEqual(code, 1)
self.assertEqual(lire_instances.call_count, 20)
demarrer.assert_not_called()
def test_echec_de_stop_est_retourne_sans_relance(self):
with mock.patch.object(blink_cli, "_arreter_instances", return_value=3), \
mock.patch.object(blink_cli.runtime, "lire_instances") as lire_instances, \
mock.patch.object(blink_cli.runtime, "verrou_controle",
return_value=contextlib.nullcontext()), \
mock.patch.object(blink_cli.runtime, "demarrer") as demarrer:
code = blink_cli.redemarrer(["--finaliser"])
self.assertEqual(code, 3)
lire_instances.assert_not_called()
demarrer.assert_not_called()
class TestsArretIncomplet(unittest.TestCase):
def test_conserve_la_demande_arret_si_un_processus_survit(self):
instance = {
"pid": 123, "depuis": "maintenant", "verbes": [["serve"]],
"enfants": [], "identites": {"123": "creation-attendue"},
"fiche": "inutilisee.json",
}
horloge = {"t": 0.0}
with mock.patch.object(blink_cli.runtime, "lire_instances",
return_value=[instance]), \
mock.patch.object(blink_cli.runtime, "demander_arret") as demander, \
mock.patch.object(blink_cli.runtime, "effacer_arret_demande") as effacer, \
mock.patch.object(blink_cli.runtime, "processus_vivant", return_value=True), \
mock.patch.object(blink_cli.runtime, "identite_processus",
return_value="creation-attendue"), \
mock.patch.object(blink_cli.runtime, "arreter_processus"), \
mock.patch.object(blink_cli.time, "time", side_effect=lambda: horloge["t"]), \
mock.patch.object(blink_cli.time, "sleep",
side_effect=lambda d: horloge.__setitem__("t", horloge["t"] + d)):
code = blink_cli._arreter_instances()
self.assertEqual(code, 1)
demander.assert_called_once()
effacer.assert_not_called()
def test_pid_illisible_est_un_pid_recycle_pas_un_survivant(self):
"""Règle inversée le 29/09/2026, avec l'accord de Nico : un PID dont
l'identité est illisible comptait comme survivant. Après un
redémarrage, la fiche périmée citait un pid repris par
AggregatorHost.exe (accès refusé) : `stop` échouait, et la mise à jour
qui l'appelle, ayant déjà tué le reste, s'abandonnait sans rien
relancer. Un processus de notre compte est toujours lisible."""
instance = {
"pid": 123, "depuis": "maintenant", "verbes": [["serve"]],
"enfants": [], "identites": {"123": "creation-attendue"},
"fiche": "inutilisee.json",
}
horloge = {"t": 0.0}
with mock.patch.object(blink_cli.runtime, "lire_instances",
return_value=[instance]), \
mock.patch.object(blink_cli.runtime, "demander_arret"), \
mock.patch.object(blink_cli.runtime, "effacer_arret_demande") as effacer, \
mock.patch.object(blink_cli.runtime, "processus_vivant", return_value=True), \
mock.patch.object(blink_cli.runtime, "identite_processus", return_value=None), \
mock.patch.object(blink_cli.runtime, "arreter_processus") as arreter, \
mock.patch.object(blink_cli.time, "time", side_effect=lambda: horloge["t"]), \
mock.patch.object(blink_cli.time, "sleep",
side_effect=lambda d: horloge.__setitem__("t", horloge["t"] + d)):
code = blink_cli._arreter_instances()
self.assertEqual(code, 0)
arreter.assert_not_called()
effacer.assert_called_once()
class TestsSerialisationArret(unittest.TestCase):
def test_stop_execute_son_corps_sous_verrou(self):
with mock.patch.object(blink_cli.runtime, "verrou_controle",
return_value=contextlib.nullcontext()) as verrou, \
mock.patch.object(blink_cli, "_arreter_instances", return_value=0) as corps:
self.assertEqual(blink_cli.arreter([]), 0)
verrou.assert_called_once_with("stop", attente=10)
corps.assert_called_once()
def test_stop_concurrent_echoue_sans_toucher_aux_instances(self):
with mock.patch.object(blink_cli.runtime, "verrou_controle",
side_effect=blink_cli.runtime.BusyError("occupé")), \
mock.patch.object(blink_cli, "_arreter_instances") as corps:
self.assertEqual(blink_cli.arreter([]), 1)
corps.assert_not_called()
def test_stop_attend_la_fin_d_un_lancement(self):
# Vrai verrou sur disque, dans un dossier de contrôle temporaire : un
# « launch » le tient une seconde, comme un enfant qui inscrit sa
# fiche juste après « start ». Le stop doit attendre puis s'exécuter,
# au lieu d'être refusé.
tenu = threading.Event()
def lancement():
with blink_cli.runtime.verrou_controle("launch"):
tenu.set()
time.sleep(1)
with tempfile.TemporaryDirectory(prefix="blink-controle-") as controle, \
mock.patch.dict(os.environ, {"BLINK_CONTROL_HOME": controle}), \
mock.patch.object(blink_cli, "_arreter_instances", return_value=0) as corps:
fil = threading.Thread(target=lancement)
fil.start()
self.assertTrue(tenu.wait(5))
debut = time.monotonic()
code = blink_cli.arreter([])
duree = time.monotonic() - debut
fil.join()
self.assertEqual(code, 0)
corps.assert_called_once()
self.assertGreaterEqual(duree, 0.5, "stop n'a pas attendu le verrou")
class TestsRedemarrerDansVerbes(unittest.TestCase):
def test_restart_route_vers_un_fichier_existant(self):
commande = blink_cli.runtime.self_command("restart")
from pathlib import Path
self.assertTrue(Path(commande[2]).is_file())
if __name__ == "__main__":
unittest.main()