test_blink_registre_lecture_concurrente.py
Rama main · réplica de lectura del repositorio original.
"""Un registre momentanément inaccessible ne doit jamais passer pour vide.
Incident réel du 2026-10-17 : la passe USB a lu le registre pendant que la
boucle cloud le remplaçait (os.replace). Sous Windows, cette lecture échoue en
PermissionError, et load_download_state() la traitait comme un registre
corrompu, donc vide : six clips déjà acquis ont été retéléchargés sous leur
nouveau numéro de manifeste, puis annoncés comme nouveaux.
Les deux premiers tests font tourner une vraie concurrence, entre deux
processus, sur un vrai fichier : ils échouaient sous Windows avant le
correctif, et passent sans effort ailleurs, où un remplacement ne bloque jamais
un lecteur. Les suivants simulent le refus pour fixer le contrat sur tous les
systèmes."""
from __future__ import annotations # Python 3.8 (build Windows 7)
import json
import multiprocessing
import os
import tempfile
import time
import unittest
from pathlib import Path
from unittest import mock
from nico579_commons import atomique
import blink_registre
import runtime
DUREE_S = 2.0
def _etat(nombre: int) -> dict:
return {"version": 2, "clips": {
f"v2:{indice:05d}": {
"hub": "Maison",
"camera": "jardin",
"created_at": (f"2026-10-{1 + indice // 1440 % 28:02d}"
f"T{indice // 60 % 24:02d}:{indice % 60:02d}:00+00:00"),
"path": f"jardin/2026-10/clip-{indice:05d}.mp4",
"bytes": 1000 + indice,
"source": "usb",
}
for indice in range(nombre)
}}
def _remplacer_sans_relache(registre: str, contenu: str, pret, arret) -> None:
"""La boucle cloud, dans son propre processus : réécrit le registre d'un bloc."""
cible = Path(registre)
temporaire = cible.with_name("autre-boucle.tmp")
pret.set()
while not arret.is_set():
temporaire.write_text(contenu, encoding="utf-8")
try:
os.replace(temporaire, cible)
except PermissionError:
pass # refusé pendant une lecture : retenté au tour suivant
def _lire_par_moments(registre: str, pret, arret) -> None:
"""La page ou l'autre boucle : lisent le registre par moments, pas en continu."""
cible = Path(registre)
pret.set()
while not arret.is_set():
try:
cible.read_bytes()
except OSError:
pass # croisé un remplacement : sans objet pour ce test
time.sleep(0.002)
class _Registre(unittest.TestCase):
nombre = 3
def setUp(self):
dossier = tempfile.TemporaryDirectory()
self.addCleanup(dossier.cleanup)
self.sortie = Path(dossier.name)
self.registre = self.sortie / blink_registre.STATE_FILENAME
self.contenu = json.dumps(_etat(self.nombre), indent=2, ensure_ascii=False)
self.registre.write_text(self.contenu, encoding="utf-8")
# Le verrou « registre » de save_download_state() reste dans le bac à sable.
verrous = mock.patch.object(runtime, "app_dir", return_value=self.sortie)
verrous.start()
self.addCleanup(verrous.stop)
class TestsConcurrenceReelle(_Registre):
# Un petit registre se relit vite : autant de lectures de plus, donc
# d'occasions de croiser un remplacement pendant DUREE_S.
nombre = 200
def _en_parallele(self, travail, args: tuple, action) -> None:
"""Répète ``action`` pendant DUREE_S, ``travail`` tournant dans un
second processus. Deux fils d'un même processus, sérialisés par le
GIL, ne se croisaient presque jamais : le test passait sur le code
fautif (essai du 2026-10-17)."""
contexte = multiprocessing.get_context("spawn")
pret, arret = contexte.Event(), contexte.Event()
processus = contexte.Process(target=travail, args=(*args, pret, arret))
processus.start()
try:
self.assertTrue(pret.wait(60), "processus concurrent jamais démarré")
fin = time.monotonic() + DUREE_S
while time.monotonic() < fin:
action()
finally:
arret.set()
processus.join(timeout=10)
if processus.is_alive():
processus.terminate()
processus.join(timeout=5)
processus.close()
def test_lecture_pendant_des_remplacements_ne_rend_jamais_un_registre_vide(self):
tailles = []
self._en_parallele(
_remplacer_sans_relache, (str(self.registre), self.contenu),
lambda: tailles.append(
len(blink_registre.load_download_state(self.sortie)["clips"])))
incompletes = [taille for taille in tailles if taille != self.nombre]
self.assertEqual(incompletes, [],
f"{len(incompletes)} lectures incomplètes sur {len(tailles)}")
def test_ecriture_pendant_des_lectures_aboutit(self):
etat = blink_registre.load_download_state(self.sortie)
self._en_parallele(
_lire_par_moments, (str(self.registre),),
lambda: blink_registre.save_download_state(self.sortie, etat))
self.assertEqual(
len(blink_registre.load_download_state(self.sortie)["clips"]), self.nombre)
class TestsAbsencePassagere(_Registre):
"""Sous Windows, os.replace laisse un instant la cible absente : la sonde
du 2026-11-06 n'a trouvé que le temporaire dans le dossier, jusqu'à 125 ms."""
def setUp(self):
super().setUp()
for correctif in (mock.patch.object(blink_registre.time, "sleep"),
mock.patch.object(blink_registre, "REMPLACEMENT_NON_ATOMIQUE", True)):
correctif.start()
self.addCleanup(correctif.stop)
def _absent(self, combien: int):
"""Fait passer le registre pour absent ``combien`` lectures durant."""
lire = atomique.lire_json
restants = [combien]
def lecture(chemin, defaut, **options):
if restants[0] != 0:
restants[0] -= 1
return defaut
return lire(chemin, defaut, **options)
return mock.patch.object(atomique, "lire_json", side_effect=lecture)
def test_absence_passagere_ne_rend_pas_un_registre_vide(self):
# Quatre lectures : registre et copie absents, deux tours de suite.
with self._absent(4):
etat = blink_registre.load_download_state(self.sortie)
self.assertEqual(len(etat["clips"]), self.nombre)
def test_absence_avere_reste_vide_et_bornee(self):
self.registre.unlink()
etat = blink_registre.load_download_state(self.sortie)
self.assertEqual(etat["clips"], {})
self.assertEqual(blink_registre.time.sleep.call_count,
blink_registre.TENTATIVES_ABSENCE)
def test_installation_neuve_n_attend_pas(self):
etat = blink_registre.load_download_state(self.sortie / "Blink_Clips")
self.assertEqual(etat["clips"], {})
blink_registre.time.sleep.assert_not_called()
class TestsRefusSimule(_Registre):
def setUp(self):
super().setUp()
# Les nouvelles tentatives sans leurs pauses de 50 ms.
pauses = mock.patch.object(atomique.time, "sleep")
pauses.start()
self.addCleanup(pauses.stop)
def _refuser(self, combien: int):
"""Refuse ``combien`` lectures du registre (toutes si négatif)."""
lire = Path.read_text
restants = [combien]
def lecture(chemin, *args, **kwargs):
if chemin == self.registre and restants[0] != 0:
restants[0] -= 1
raise PermissionError(13, "Accès refusé", str(chemin))
return lire(chemin, *args, **kwargs)
return mock.patch.object(Path, "read_text", autospec=True, side_effect=lecture)
def test_refus_passager_est_reessaye(self):
with self._refuser(2):
etat = blink_registre.load_download_state(self.sortie)
self.assertEqual(len(etat["clips"]), self.nombre)
def test_refus_persistant_leve_au_lieu_de_rendre_un_registre_vide(self):
with self._refuser(-1), self.assertRaises(PermissionError):
blink_registre.load_download_state(self.sortie)
def test_registre_illisible_n_est_jamais_ecrase(self):
# Avant le correctif, la relecture de fusion rendait un registre vide
# et l'écriture ne gardait que les entrées de l'appelant : tout
# l'historique, exclusions comprises, disparaissait.
nouveau = {"version": 2, "clips": {
"v2:nouveau": _etat(1)["clips"]["v2:00000"]}}
avant = self.registre.read_bytes()
with self._refuser(-1), self.assertRaises(PermissionError):
blink_registre.save_download_state(self.sortie, nouveau)
self.assertEqual(self.registre.read_bytes(), avant)
if __name__ == "__main__":
unittest.main()