test_serve_appliquer_selection_lock.py
Rama main · réplica de lectura del repositorio original.
"""Non-régression du blocage de /api/appliquer-selection par le verrou
inter-processus « registre » (AUDIT-2026-08-13.md, section 28.33, puis
28.75 : /api/toggle a été remplacé par /api/appliquer-selection pour traiter
un lot de clips en un seul appel) : la réponse HTTP ne doit toujours pas
attendre ce verrou, seulement le travail qu'il protège en profite."""
from __future__ import annotations
import io
import json
import os
import tempfile
import threading
import time
import unittest
from pathlib import Path
from unittest import mock
try:
from zoneinfo import ZoneInfo
except ImportError: # Python 3.8, édition Windows 7
from backports.zoneinfo import ZoneInfo
os.environ["BLINK_BOOTSTRAP"] = "none"
_TEST_HOME = tempfile.TemporaryDirectory(prefix="blink-toggle-lock-")
os.environ["BLINK_HOME"] = _TEST_HOME.name
import runtime # noqa: E402 - environnement de test posé avant import
import serve # noqa: E402 - bootstrap neutralisé avant import
class TestAppliquerSelectionNAttendPasLeVerrouRegistre(unittest.TestCase):
def setUp(self) -> None:
self.temporaire = tempfile.TemporaryDirectory(prefix="blink_toggle_lock_test_")
self.racine = Path(self.temporaire.name)
self.paths = {
"input": self.racine / "clips",
"normalized": self.racine / "normalized",
"excluded": self.racine / "excluded",
"thumbs": self.racine / "thumbs",
}
for chemin in self.paths.values():
chemin.mkdir(parents=True, exist_ok=True)
# runtime.verrou() écrit son fichier sous app_dir() : redirigé vers
# le dossier du test pour ne jamais toucher une vraie instance.
self.patch_app_dir = mock.patch.object(runtime, "app_dir", return_value=self.racine)
self.patch_app_dir.start()
# Hors sujet ici (28.32/28.9) : ce test porte sur le verrou registre,
# pas sur la reconstruction. Sans ce mock, le thread de fond lance un
# vrai sous-processus « merge » qui peut encore tenir un descripteur
# sur le dossier temporaire au moment du nettoyage (PermissionError
# Windows, tempfile._rmtree en boucle).
# identite_processus() utilise aussi ce lanceur sous POSIX. Une sortie
# vide garde l'identité sérialisable tout en neutralisant le vrai merge.
self.patch_lancer = mock.patch.object(
runtime, "lancer", return_value=mock.Mock(stdout=""),
)
self.patch_lancer.start()
self.identity = "jardin/test.mp4"
(self.paths["input"] / "jardin").mkdir(parents=True, exist_ok=True)
(self.paths["input"] / self.identity).write_bytes(b"\x00" * 16)
registre = {
"version": 2,
"clips": {
"cle-1": {
"path": self.identity,
"camera": "jardin",
"created_at": "2026-08-18T12:00:00+00:00",
"hub": "Maison",
"source": "usb",
}
},
}
(self.paths["input"] / ".blink_download_state.json").write_text(
json.dumps(registre), encoding="utf-8")
def tearDown(self) -> None:
self.patch_lancer.stop()
self.patch_app_dir.stop()
self.temporaire.cleanup()
def construire_handler(self):
# Un vrai Handler, sans passer par __init__ (qui attend une vraie
# socket) : do_POST ne lit que self.path/.headers/.rfile, faciles à
# simuler sans ouvrir de connexion réseau.
handler = serve.Handler.__new__(serve.Handler)
handler.paths = self.paths
handler.timezone = ZoneInfo("UTC")
return handler
def appeler(self, handler, corps_dict: dict):
corps = json.dumps(corps_dict).encode("utf-8")
handler.path = "/api/appliquer-selection"
# Host + jeton requis depuis 28.60 (protection CSRF/Origin) : sans
# eux, ce handler construit à la main est rejeté en 403.
handler.headers = {"Content-Length": str(len(corps)), "Host": "127.0.0.1",
"X-Blink-Token": serve.TOKEN}
handler.rfile = io.BytesIO(corps)
reponses = []
handler.send_json = lambda payload, code=200: reponses.append((code, payload))
handler.do_POST()
return reponses
def test_la_reponse_ne_bloque_plus_sur_le_verrou_tenu_ailleurs(self):
# Simule une écriture de téléchargement en cours : le même verrou
# que set_excluded() prend en interne, tenu ici depuis un autre fil,
# comme le ferait un vrai processus séparé.
pret = threading.Event()
relacher = threading.Event()
def tenir_le_verrou():
with runtime.verrou("registre", "test-telechargement", stale_after=60, attente=0):
pret.set()
relacher.wait(timeout=5)
fil_verrou = threading.Thread(target=tenir_le_verrou, daemon=True)
fil_verrou.start()
self.assertTrue(pret.wait(timeout=2), "le verrou de test n'a pas été pris à temps")
try:
handler = self.construire_handler()
debut = time.monotonic()
reponses = self.appeler(handler, {"exclure": [self.identity]})
duree = time.monotonic() - debut
finally:
relacher.set()
fil_verrou.join(timeout=5)
self.assertLess(duree, 1.0,
f"la réponse a attendu {duree:.2f}s le verrou tenu ailleurs")
self.assertEqual(reponses, [(200, {"ok": True, "resultats": {}})])
# Le travail réel (l'exclusion) doit tout de même aboutir une fois le
# verrou libre, même si la réponse HTTP n'a pas attendu dessus.
# load_download_state() plutôt qu'une lecture JSON directe : un
# remplacement atomique en cours peut transitoirement refuser la
# lecture côté Windows (PermissionError), déjà toléré par ce helper.
# Dix secondes au plus, l'exclusion aboutissant en quelques dixièmes.
# Les échecs des CI Windows des 25 et 26/09/2026, pris d'abord pour
# de la lenteur, venaient d'un refus de Windows à la recréation du
# verrou, qui tuait le fil de l'exclusion (voir runtime.verrou()).
limite = time.monotonic() + 10
while True:
etat = serve.blink_registre.load_download_state(self.paths["input"])
if etat["clips"].get("cle-1", {}).get("excluded"):
break
if time.monotonic() > limite:
self.fail("l'exclusion n'a jamais été appliquée en arrière-plan")
time.sleep(0.1)
self.assertTrue((self.paths["excluded"] / self.identity).is_file(),
"le brut aurait dû être déplacé vers Blink_Excluded")
def test_identifiant_invalide_est_refuse_sans_verrou(self):
# Un lot malformé est rejeté avant toute mutation, pas partiellement
# appliqué avec un faux succès.
handler = self.construire_handler()
reponses = self.appeler(handler, {"exclure": ["../../etc/passwd"]})
self.assertEqual(reponses[0][0], 400)
self.assertIn("error", reponses[0][1])
etat_fichier = self.paths["input"] / ".blink_download_state.json"
etat = json.loads(etat_fichier.read_text(encoding="utf-8"))
self.assertNotIn("excluded", etat["clips"]["cle-1"],
"un identifiant invalide n'aurait jamais dû être appliqué")
if __name__ == "__main__":
unittest.main()