test_stage2_identity.py
Rama main · réplica de lectura del repositorio original.
"""Contrats de non-régression de l'étape 2 : identité et réparation.
Cette suite est entièrement locale : aucun compte, aucune socket et aucun média
réel. Les contrats absents à son introduction ont d'abord été protégés par
``expectedFailure`` ; leurs marqueurs ont été retirés avec le correctif qui les
satisfait, afin que toute régression rende désormais la suite rouge.
Exécution :
python -B -m unittest -v test_stage2_identity.py
"""
from __future__ import annotations
import argparse
import contextlib
import datetime as dt
import io
import json
import os
import tempfile
import unittest
from pathlib import Path
from unittest import mock
os.environ["BLINK_BOOTSTRAP"] = "none"
import blink2video as b2v # noqa: E402 - bootstrap neutralisé avant import
import blink_engine # noqa: E402
import blink_models # noqa: E402
import blink_registre # noqa: E402
INSTANT = dt.datetime(2026, 8, 13, 12, 0, tzinfo=dt.timezone.utc)
class Clip:
"""Surface commune aux objets USB et cloud dont l'identité est testée."""
def __init__(
self,
identifiant=42,
nom="jardin",
instant=INSTANT,
*,
device_id="camera-7",
network_id="reseau-1",
):
self.id = identifiant
self.name = nom
self.created_at = instant
self.device_id = device_id
self.network_id = network_id
self.size = 1
self.download_issue = None
class Sync:
def __init__(self, sync_id="hub-1", network_id="reseau-1"):
self.sync_id = sync_id
self.network_id = network_id
def _boite(nom: bytes, contenu: bytes = b"") -> bytes:
return (len(contenu) + 8).to_bytes(4, "big") + nom + contenu
# Réparation/reconciliation seulement : le contenu de mdat n'a pas besoin
# d'être un flux vidéo réel, valid_mp4_complet() est patché sur valid_mp4()
# (structurel) dans ces tests, comme test_blink2video_audit.py le fait déjà.
MP4_STRUCTUREL = (
_boite(b"ftyp", b"isom\x00\x00\x02\x00isomiso2")
+ _boite(b"moov")
+ _boite(b"mdat", b"paquet-video")
)
def arguments(output: Path, **changements):
valeurs = dict(
since=None,
camera=None,
command="download",
output=output,
hub=None,
overwrite=False,
source="cloud",
loop=None,
)
valeurs.update(changements)
return argparse.Namespace(**valeurs)
class BacASable(unittest.TestCase):
def setUp(self):
self.temporaire = tempfile.TemporaryDirectory(prefix="blink_stage2_")
self.racine = Path(self.temporaire.name)
self.home = self.racine / "home"
self.sortie = self.racine / "clips"
self.home.mkdir()
self.sortie.mkdir()
self.ancien_home = os.environ.get("BLINK_HOME")
os.environ["BLINK_HOME"] = str(self.home)
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 memoriser(
self,
etat: dict,
clip: Clip,
*,
sync=None,
source="usb",
contenu=b"video",
) -> Path:
cible = blink_models.target_path(
self.sortie, clip, sync=sync or Sync(), source=source,
)
cible.parent.mkdir(parents=True, exist_ok=True)
cible.write_bytes(contenu)
blink_registre.remember_download(
etat,
sync or Sync(),
"Maison",
clip,
self.sortie,
cible,
source=source,
)
return cible
class TestsIdentiteEtMatching(BacASable):
def test_T_B03_REINDEX_42_vers_99_garde_le_fichier_acquis(self):
"""Le chemin mémorisé est l'autorité après renumérotation distante."""
etat = {"version": 1, "clips": {}}
ancien = Clip(42)
nouveau = Clip(99)
ancienne_cible = self.memoriser(etat, ancien)
nouvelle_cible = blink_models.target_path(self.sortie, nouveau)
self.assertNotEqual(ancienne_cible, nouvelle_cible)
self.assertTrue(blink_registre.is_downloaded(etat, Sync(), nouveau, nouvelle_cible))
self.assertFalse(nouvelle_cible.exists())
def test_T_I04_USB_puis_cloud_ne_produit_qu_un_evenement(self):
usb = Clip(42, device_id="camera-stable")
cloud = Clip(900, device_id="camera-stable")
inedits, doublons = blink_models.rapprocher([usb], [cloud])
self.assertEqual(inedits, [])
self.assertEqual(doublons, [cloud])
def test_T_I04_cloud_puis_USB_ne_produit_qu_un_evenement(self):
"""L'identité du registre est indépendante de l'ordre des sources."""
etat = {"version": 1, "clips": {}}
cloud = Clip(900, device_id="camera-stable")
usb = Clip(42, device_id="camera-stable")
self.memoriser(etat, cloud, sync=Sync("cloud", "reseau-1"), source="cloud")
self.assertTrue(
blink_registre.is_downloaded(
etat,
Sync("hub-1", "reseau-1"),
usb,
blink_models.target_path(self.sortie, usb),
)
)
def test_T_I21_deux_evenements_legitimes_proches_restent_deux(self):
premier = Clip(1, instant=INSTANT, device_id="camera-stable")
second = Clip(
2,
instant=INSTANT + dt.timedelta(seconds=1),
device_id="camera-stable",
)
self.assertNotEqual(
blink_registre.state_key(Sync(), premier), blink_registre.state_key(Sync(), second)
)
self.assertNotEqual(
blink_models.target_path(self.sortie, premier),
blink_models.target_path(self.sortie, second),
)
def test_T_I21_un_match_local_n_est_consomme_qu_une_fois(self):
local = Clip(1, device_id="camera-stable")
cloud_exact = Clip(2, device_id="camera-stable")
cloud_proche = Clip(
3,
instant=INSTANT + dt.timedelta(seconds=1),
device_id="camera-stable",
)
inedits, doublons = blink_models.rapprocher(
[local], [cloud_proche, cloud_exact], tolerance=2
)
self.assertEqual(doublons, [cloud_exact])
self.assertEqual(inedits, [cloud_proche])
self.assertEqual(
blink_models._apparier_evenements([local], [cloud_proche, cloud_exact], 2),
[(0, 1)],
)
def test_T_I21_appariement_conserve_la_cardinalite_maximale(self):
locaux = [
Clip(1, instant=INSTANT, device_id="camera-stable"),
Clip(
2,
instant=INSTANT + dt.timedelta(seconds=3),
device_id="camera-stable",
),
]
cloud = [
Clip(
3,
instant=INSTANT + dt.timedelta(seconds=2),
device_id="camera-stable",
),
Clip(
4,
instant=INSTANT + dt.timedelta(seconds=4),
device_id="camera-stable",
),
]
inedits, doublons = blink_models.rapprocher(locaux, cloud, tolerance=2)
self.assertEqual(inedits, [])
self.assertEqual(doublons, cloud)
self.assertEqual(blink_models._apparier_evenements(locaux, cloud, 2), [(0, 0), (1, 1)])
def test_T_I16_camera_renommee_device_id_stable_ne_duplique_pas(self):
ancien_nom = Clip(1, nom="Jardin", device_id="camera-stable")
nouveau_nom = Clip(2, nom="Terrasse", device_id="camera-stable")
inedits, doublons = blink_models.rapprocher([ancien_nom], [nouveau_nom])
self.assertEqual(inedits, [])
self.assertEqual(doublons, [nouveau_nom])
def test_T_I16_A_slash_B_et_A_underscore_B_restent_distincts(self):
slash = Clip(1, nom="A/B", device_id="camera-slash")
underscore = Clip(2, nom="A_B", device_id="camera-underscore")
self.assertEqual(blink_models.safe_name(slash.name), blink_models.safe_name(underscore.name))
self.assertNotEqual(
blink_registre.state_key(Sync(), slash), blink_registre.state_key(Sync(), underscore)
)
inedits, doublons = blink_models.rapprocher([slash], [underscore])
self.assertEqual(inedits, [underscore])
self.assertEqual(doublons, [])
def test_T_I20_multi_reseaux_ne_collisionne_pas(self):
premier = Clip(1, network_id="reseau-1")
second = Clip(2, network_id="reseau-2")
self.assertNotEqual(
blink_registre.state_key(Sync("cloud", "reseau-1"), premier),
blink_registre.state_key(Sync("cloud", "reseau-2"), second),
)
inedits, doublons = blink_models.rapprocher([premier], [second])
self.assertEqual(inedits, [second])
self.assertEqual(doublons, [])
def test_T_I20_multi_hubs_ne_collisionne_pas(self):
clip = Clip(1, network_id="reseau-commun")
hub_a = Sync("hub-A", "reseau-commun")
hub_b = Sync("hub-B", "reseau-commun")
self.assertNotEqual(
blink_registre.state_key(hub_a, clip),
blink_registre.state_key(hub_b, clip),
)
self.assertNotEqual(
blink_models.target_path(self.sortie, clip, sync=hub_a, source="usb"),
blink_models.target_path(self.sortie, clip, sync=hub_b, source="usb"),
)
class TestsRegistreRobuste(BacASable):
@staticmethod
def _entry(clip, chemin, *, bytes_=1, excluded=False):
return {
"camera": clip.name,
"created_at": clip.created_at.isoformat(),
"path": chemin,
"bytes": bytes_,
"source": "usb",
"excluded": excluded,
}
def test_T_B03_entry_path_hostile_n_est_jamais_acquis(self):
clip = Clip()
cible = blink_models.target_path(self.sortie, clip)
hors_racine = self.racine / "hors-racine.mp4"
hors_racine.write_bytes(b"x")
(self.sortie / "un-dossier").mkdir()
chemins = {
"absolu": str(hors_racine.resolve()),
"traversal": "../hors-racine.mp4",
"repertoire": "un-dossier",
"vide": "",
}
for libelle, chemin in chemins.items():
with self.subTest(cas=libelle):
etat = {
"version": 2,
"clips": {
blink_registre.state_key(Sync(), clip): self._entry(clip, chemin),
},
}
self.assertFalse(blink_registre.is_downloaded(etat, Sync(), clip, cible))
def test_T_I21_fallback_registre_est_consomme_une_seule_fois(self):
archive = self.sortie / "archive.mp4"
archive.write_bytes(b"archive")
premier = Clip(101, instant=INSTANT)
second = Clip(102, instant=INSTANT + dt.timedelta(seconds=1))
etat = {
"version": 2,
"clips": {
"ancienne-cle": self._entry(
premier,
archive.relative_to(self.sortie).as_posix(),
bytes_=archive.stat().st_size,
),
},
}
consommees = set()
self.assertTrue(
blink_registre.is_downloaded(
etat,
Sync(),
premier,
blink_models.target_path(self.sortie, premier),
consommees,
)
)
self.assertFalse(
blink_registre.is_downloaded(
etat,
Sync(),
second,
blink_models.target_path(self.sortie, second),
consommees,
)
)
self.assertEqual(consommees, {"ancienne-cle"})
def test_T_B03_tombstone_ancienne_gagne_sur_cle_exacte_normale(self):
clip = Clip()
archive = self.sortie / "archive.mp4"
archive.write_bytes(b"archive")
etat = {
"version": 2,
"clips": {
blink_registre.state_key(Sync(), clip): self._entry(
clip,
archive.relative_to(self.sortie).as_posix(),
bytes_=archive.stat().st_size,
),
"ancienne-tombstone": self._entry(
clip, "absent.mp4", excluded=True,
),
},
}
correspondance = blink_registre._apparier_registre(etat, Sync(), [clip])[0]
self.assertEqual(correspondance[0], "ancienne-tombstone")
self.assertIs(correspondance[1]["excluded"], True)
def test_T_I20_lookup_cloud_utilise_source_et_remote_id(self):
etat = {"version": 2, "clips": {}}
sync = blink_engine._HubCloud("reseau-1")
premier = Clip(101)
second = Clip(202)
self.memoriser(etat, premier, sync=sync, source="cloud")
self.memoriser(etat, second, sync=sync, source="cloud")
cle, entree = blink_registre._trouver_entree(
etat, sync, second, source="cloud",
)
self.assertEqual(cle, blink_registre.state_key(sync, second, "cloud"))
self.assertEqual(entree["remote_id"], "202")
def test_T_B04_merge_v2_est_monotone_et_preserve_tombstone_disque(self):
clip = Clip()
registre = self.sortie / blink_registre.STATE_FILENAME
registre.write_text(
json.dumps(
{
"version": 2,
"clips": {
"commun": self._entry(
clip, "commun.mp4", excluded=True
),
"seulement-disque": self._entry(clip, "disque.mp4"),
},
}
),
encoding="utf-8",
)
etat_perime = {
"version": 1,
"clips": {
"commun": self._entry(clip, "commun.mp4", excluded=False),
"seulement-memoire": self._entry(clip, "memoire.mp4"),
},
}
blink_registre._ecrire_registre(registre, etat_perime)
fusion = json.loads(registre.read_text(encoding="utf-8"))
self.assertEqual(fusion["version"], 2)
self.assertEqual(
set(fusion["clips"]),
{"commun", "seulement-disque", "seulement-memoire"},
)
self.assertIs(fusion["clips"]["commun"]["excluded"], True)
def test_registre_survit_a_la_suppression_du_dossier_de_clips(self):
clip = Clip()
etat = {"version": 2, "clips": {
blink_registre.state_key(Sync(), clip): self._entry(clip, "ancien.mp4"),
}}
blink_registre.save_download_state(self.sortie, etat)
(self.sortie / blink_registre.STATE_FILENAME).unlink()
self.sortie.rmdir()
restaure = blink_registre.load_download_state(self.sortie)
self.assertEqual(set(restaure["clips"]), set(etat["clips"]))
blink_registre.save_download_state(self.sortie, restaure)
self.assertTrue((self.sortie / blink_registre.STATE_FILENAME).is_file())
class TestsReparationCloud(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self):
self.temporaire = tempfile.TemporaryDirectory(prefix="blink_stage2_async_")
self.racine = Path(self.temporaire.name)
self.home = self.racine / "home"
self.sortie = self.racine / "clips"
self.home.mkdir()
self.sortie.mkdir()
self.ancien_home = os.environ.get("BLINK_HOME")
os.environ["BLINK_HOME"] = str(self.home)
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()
@staticmethod
def _entry(clip, chemin="ancien.mp4", *, excluded=False):
return {
"camera": clip.name,
"created_at": clip.created_at.isoformat(),
"path": chemin,
"source": "cloud",
"excluded": excluded,
}
async def _executer(self, clip, entree, *, overwrite=False):
(self.sortie / blink_registre.STATE_FILENAME).write_text(
json.dumps({"version": 1, "clips": {"ancienne-cle": entree}}),
encoding="utf-8",
)
appels = []
async def telecharger(_blink, cible):
appels.append(cible)
cible.write_bytes(MP4_STRUCTUREL)
return True
clip.download_to = telecharger
with mock.patch.object(
blink_models, "read_cloud_manifest", new=mock.AsyncMock(return_value=[clip])
), mock.patch.object(
blink_engine.md, "valid_mp4_complet", side_effect=blink_engine.md.valid_mp4
), contextlib.redirect_stdout(io.StringIO()):
resultat = await blink_engine.traiter_cloud(
object(), arguments(self.sortie, overwrite=overwrite), []
)
return resultat, appels
async def test_T_B03_MISSING_fichier_purge_reste_deja_traite(self):
clip = Clip()
resultat, appels = await self._executer(clip, self._entry(clip))
self.assertEqual(resultat.downloaded, 0)
self.assertEqual(resultat.skipped, 1)
self.assertEqual(appels, [])
async def test_T_B03_MISSING_overwrite_retélécharge(self):
clip = Clip()
resultat, appels = await self._executer(
clip, self._entry(clip), overwrite=True,
)
self.assertEqual(resultat.downloaded, 1)
self.assertEqual(len(appels), 1)
async def test_T_B03_EXCLUDED_fichier_absent_exclu_est_ignore(self):
clip = Clip()
resultat, appels = await self._executer(
clip, self._entry(clip, excluded=True)
)
self.assertEqual(resultat.downloaded, 0)
self.assertEqual(resultat.skipped, 1)
self.assertEqual(appels, [])
self.assertFalse(blink_models.target_path(self.sortie, clip).exists())
async def test_T_B03_EXCLUDED_overwrite_cloud_reste_ignore(self):
clip = Clip()
async def ne_doit_pas_telecharger(_blink, cible):
cible.write_bytes(b"interdit")
return True
clip.download_to = ne_doit_pas_telecharger
(self.sortie / blink_registre.STATE_FILENAME).write_text(
json.dumps(
{
"version": 1,
"clips": {
"tombstone": self._entry(clip, excluded=True),
},
}
),
encoding="utf-8",
)
with mock.patch.object(
blink_models, "read_cloud_manifest", new=mock.AsyncMock(return_value=[clip])
), contextlib.redirect_stdout(io.StringIO()):
resultat = await blink_engine.traiter_cloud(
object(), arguments(self.sortie, overwrite=True), []
)
self.assertEqual(resultat.downloaded, 0)
self.assertEqual(resultat.skipped, 1)
self.assertFalse(blink_models.target_path(self.sortie, clip).exists())
async def test_T_I05_taille_divergente_reste_deja_traitee(self):
clip = Clip()
sync_cloud = Sync(clip.network_id, clip.network_id)
cible = blink_models.target_path(
self.sortie, clip, sync=sync_cloud, source="cloud"
)
cible.parent.mkdir(parents=True)
cible.write_bytes(b"tronque")
entree = self._entry(
clip,
cible.relative_to(self.sortie).as_posix(),
)
entree["bytes"] = 1000
(self.sortie / blink_registre.STATE_FILENAME).write_text(
json.dumps({"version": 2, "clips": {"incomplet": entree}}),
encoding="utf-8",
)
appels = []
async def reparer(_blink, partiel):
appels.append(partiel)
partiel.write_bytes(MP4_STRUCTUREL)
return True
clip.download_to = reparer
with mock.patch.object(
blink_models, "read_cloud_manifest", new=mock.AsyncMock(return_value=[clip])
), mock.patch.object(
blink_engine.md, "valid_mp4_complet", side_effect=blink_engine.md.valid_mp4
), contextlib.redirect_stdout(io.StringIO()):
resultat = await blink_engine.traiter_cloud(
object(), arguments(self.sortie), []
)
self.assertEqual(resultat.downloaded, 0)
self.assertEqual(resultat.skipped, 1)
self.assertEqual(resultat.adopted, 0)
self.assertEqual(appels, [])
self.assertEqual(cible.read_bytes(), b"tronque")
async def test_T_I16_filtre_cloud_A_slash_B_exclut_A_underscore_B(self):
slash = Clip(1, nom="A/B", device_id="camera-slash")
underscore = Clip(2, nom="A_B", device_id="camera-underscore")
appels = []
# rapprocher_paires porte tous les rapprochements du plan cloud
# (rapprocher() lui délègue aussi) depuis l'audit du 26/09/2026.
original = blink_models.rapprocher_paires
def rapprocher_trace(locaux, cloud, tolerance=2, compatibles=None):
appels.append(list(cloud))
return original(locaux, cloud, tolerance, compatibles)
with mock.patch.object(
blink_models,
"read_cloud_manifest",
new=mock.AsyncMock(return_value=[slash, underscore]),
), mock.patch.object(blink_models, "rapprocher_paires",
side_effect=rapprocher_trace), \
contextlib.redirect_stdout(io.StringIO()):
await blink_engine.traiter_cloud(
object(),
arguments(self.sortie, command="list", camera="A/B"),
[],
)
self.assertGreaterEqual(len(appels), 1)
self.assertTrue(all(clips == [slash] for clips in appels))
async def test_T_B03_EXCLUDED_overwrite_USB_reste_ignore(self):
clip = Clip()
(self.sortie / blink_registre.STATE_FILENAME).write_text(
json.dumps(
{
"version": 1,
"clips": {
"tombstone": self._entry(clip, excluded=True),
},
}
),
encoding="utf-8",
)
telecharger = mock.AsyncMock(return_value="failed")
with mock.patch.object(
blink_models, "read_local_manifest", new=mock.AsyncMock(return_value=[clip])
), mock.patch.object(blink_engine, "download_clip", new=telecharger), \
mock.patch.object(b2v.runtime, "travail"), \
mock.patch.object(b2v.runtime, "marquer"), \
mock.patch.object(b2v.runtime, "toast"), \
contextlib.redirect_stdout(io.StringIO()):
code = await blink_engine.un_passage(
object(),
arguments(self.sortie, source="usb", overwrite=True),
[("Maison", Sync())],
)
self.assertEqual(code, 0)
telecharger.assert_not_awaited()
async def test_T_I20_cloud_utilise_le_network_id_de_chaque_clip(self):
premier = Clip(1, instant=INSTANT, network_id="reseau-A")
second = Clip(
2,
instant=INSTANT + dt.timedelta(minutes=1),
network_id="reseau-B",
)
async def telecharger(_blink, cible):
cible.write_bytes(MP4_STRUCTUREL)
return True
premier.download_to = telecharger
second.download_to = telecharger
with mock.patch.object(
blink_models,
"read_cloud_manifest",
new=mock.AsyncMock(return_value=[premier, second]),
), mock.patch.object(
blink_engine.md, "valid_mp4_complet", side_effect=blink_engine.md.valid_mp4
), contextlib.redirect_stdout(io.StringIO()):
resultat = await blink_engine.traiter_cloud(
object(), arguments(self.sortie), [("Maison", Sync("hub-1", "X"))]
)
self.assertEqual(resultat.downloaded, 2)
registre = blink_registre.load_download_state(self.sortie)
reseaux = {entree.get("network_id") for entree in registre["clips"].values()}
self.assertEqual(reseaux, {"reseau-A", "reseau-B"})
class TestsEvenementsCloudProches(unittest.IsolatedAsyncioTestCase):
"""Audit du 26/09/2026, B01 : deux médias cloud aux identifiants distincts,
à une seconde d'écart, restent deux téléchargements. La tolérance de deux
secondes rapproche des sources différentes (USB et cloud) ou de vieilles
entrées sans identifiant ; elle ne fusionne jamais deux identifiants cloud,
et une entrée ne sert qu'une fois, comme dans le plan d'inventaire."""
async def asyncSetUp(self):
self.temporaire = tempfile.TemporaryDirectory(prefix="blink_b01_")
racine = Path(self.temporaire.name)
self.home = racine / "home"
self.sortie = racine / "clips"
self.home.mkdir()
self.sortie.mkdir()
self.ancien_home = os.environ.get("BLINK_HOME")
os.environ["BLINK_HOME"] = str(self.home)
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()
def _acquerir(self, clip, source):
"""Entrée déjà rapatriée, fichier présent, comme après un passage."""
etat = blink_registre.load_download_state(self.sortie)
sync = (blink_engine._HubCloud(clip.network_id) if source == "cloud"
else Sync())
cible = blink_models.target_path(self.sortie, clip, sync=sync, source=source)
cible.parent.mkdir(parents=True, exist_ok=True)
cible.write_bytes(MP4_STRUCTUREL)
blink_registre.remember_download(
etat, sync, "", clip, self.sortie, cible, source=source)
blink_registre.save_download_state(self.sortie, etat)
async def _executer(self, manifeste):
appels = []
def telechargeur(clip):
async def telecharger(_blink, cible):
appels.append(clip.id)
cible.write_bytes(MP4_STRUCTUREL)
return True
return telecharger
for clip in manifeste:
clip.download_to = telechargeur(clip)
with mock.patch.object(
blink_models, "read_cloud_manifest",
new=mock.AsyncMock(return_value=list(manifeste)),
), mock.patch.object(
blink_engine.md, "valid_mp4_complet", side_effect=blink_engine.md.valid_mp4
), contextlib.redirect_stdout(io.StringIO()):
resultat = await blink_engine.traiter_cloud(
object(), arguments(self.sortie), []
)
etat = blink_registre.load_download_state(self.sortie)
identifiants = sorted(
entree.get("remote_id") for entree in etat["clips"].values()
if entree.get("source") == "cloud"
)
return resultat, appels, identifiants
async def test_B01_deux_clips_proches_du_meme_passage(self):
premier = Clip(1, instant=INSTANT + dt.timedelta(seconds=1))
second = Clip(2, instant=INSTANT + dt.timedelta(seconds=2))
resultat, appels, identifiants = await self._executer([premier, second])
self.assertEqual((resultat.downloaded, resultat.skipped), (2, 0))
self.assertEqual(appels, [1, 2])
self.assertEqual(identifiants, ["1", "2"])
async def test_B01_clip_voisin_d_un_media_deja_rapatrie(self):
ancien = Clip(1, instant=INSTANT)
self._acquerir(ancien, "cloud")
nouveau = Clip(2, instant=INSTANT + dt.timedelta(seconds=1))
resultat, appels, identifiants = await self._executer(
[Clip(1, instant=INSTANT), nouveau])
self.assertEqual((resultat.downloaded, resultat.skipped), (1, 1))
self.assertEqual(appels, [2])
self.assertEqual(identifiants, ["1", "2"])
async def test_B01_une_entree_usb_n_absorbe_qu_un_clip_cloud(self):
# Même événement vu par la clé USB (numéro de manifeste 5) et par le
# cloud (identifiant 1) ; le clip cloud 2, une seconde plus tard, est
# un autre événement que le plan d'inventaire classe inédit.
self._acquerir(Clip(5, instant=INSTANT), "usb")
meme_evenement = Clip(1, instant=INSTANT)
autre = Clip(2, instant=INSTANT + dt.timedelta(seconds=1))
resultat, appels, identifiants = await self._executer(
[meme_evenement, autre])
self.assertEqual((resultat.downloaded, resultat.skipped), (1, 1))
self.assertEqual(appels, [2])
self.assertEqual(identifiants, ["2"])
async def test_B01_vieille_entree_sans_identifiant_garde_la_tolerance(self):
# Registre d'avant l'identité v2 : pas d'identifiant distant, la
# proximité dans le temps reste le seul signal, comme aujourd'hui.
ancien = Clip(1, instant=INSTANT)
self._acquerir(ancien, "cloud")
etat = blink_registre.load_download_state(self.sortie)
for entree in etat["clips"].values():
entree["remote_id"] = ""
blink_registre.save_download_state(self.sortie, etat)
resultat, appels, _ = await self._executer(
[Clip(9, instant=INSTANT + dt.timedelta(seconds=1))])
self.assertEqual((resultat.downloaded, resultat.skipped), (0, 1))
self.assertEqual(appels, [])
async def test_B01_passage_combine_l_entree_usb_ne_prend_qu_un_clip_cloud(self):
# Le module USB rapatrie l'événement ; le cloud le connaît aussi
# (identifiant 1) et en a un autre une seconde plus tard
# (identifiant 2). Le plan lie l'USB au cloud 1 : la nouvelle entrée
# USB ne doit pas absorber le cloud 2 à sa revalidation.
usb = Clip(5, instant=INSTANT)
cloud_meme = Clip(1, instant=INSTANT)
cloud_autre = Clip(2, instant=INSTANT + dt.timedelta(seconds=1))
appels_cloud = []
async def telecharger_usb(_blink, _clip, cible, _ecraser):
cible.parent.mkdir(parents=True, exist_ok=True)
cible.write_bytes(MP4_STRUCTUREL)
return "downloaded"
def telechargeur(clip):
async def telecharger(_blink, cible):
appels_cloud.append(clip.id)
cible.write_bytes(MP4_STRUCTUREL)
return True
return telecharger
for clip in (cloud_meme, cloud_autre):
clip.download_to = telechargeur(clip)
with mock.patch.object(
blink_models, "read_local_manifest",
new=mock.AsyncMock(return_value=[usb]),
), mock.patch.object(
blink_engine, "download_clip", side_effect=telecharger_usb,
), mock.patch.object(
blink_models, "read_cloud_manifest",
new=mock.AsyncMock(return_value=[cloud_meme, cloud_autre]),
), mock.patch.object(
blink_engine.md, "valid_mp4_complet", side_effect=blink_engine.md.valid_mp4
), mock.patch.object(
blink_engine.runtime, "toast"
) as toast, contextlib.redirect_stdout(io.StringIO()):
code = await blink_engine.un_passage(
object(), arguments(self.sortie, source="all"), [("Test", Sync())],
)
self.assertEqual(code, 0)
self.assertEqual(appels_cloud, [2])
# Deux clips rapatries : le passage annonce « 2 nouveaux clips ». Sans ce
# blocage, chaque execution locale de la suite posait un vrai toast Windows
# (les « clips fantomes » du 2026-10-05) et la CI Windows pouvait s'y bloquer.
toast.assert_called_once()
self.assertIn("2", toast.call_args.args[1])
class TestsMigrationV1(BacASable):
def _etat_v1(self, archive: Path):
clip = Clip(42, nom="Ancien nom", device_id="camera-stable")
cle = f"hub-1:{blink_models.safe_name(clip.name)}:{clip.created_at.isoformat()}"
return {
"version": 1,
"clips": {
cle: {
"hub": "Maison",
"camera": clip.name,
"created_at": clip.created_at.isoformat(),
"path": archive.relative_to(self.sortie).as_posix(),
"bytes": archive.stat().st_size,
"source": "usb",
}
},
}
def test_T_MIGRATION_V1_est_idempotente_et_archive_preservee(self):
archive = self.sortie / "Ancien nom" / "2026-08" / "archive-42.mp4"
archive.parent.mkdir(parents=True)
archive.write_bytes(b"archive-intacte")
etat_v1 = self._etat_v1(archive)
registre = self.sortie / blink_registre.STATE_FILENAME
original = json.dumps(etat_v1).encode("utf-8")
registre.write_bytes(original)
premier = blink_registre.load_download_state(self.sortie)
blink_registre.save_download_state(self.sortie, premier)
apres_premier = registre.read_bytes()
second = blink_registre.load_download_state(self.sortie)
blink_registre.save_download_state(self.sortie, second)
apres_second = registre.read_bytes()
self.assertEqual(premier["version"], 2)
self.assertEqual(second, premier)
self.assertEqual(apres_second, apres_premier)
self.assertEqual(archive.read_bytes(), b"archive-intacte")
sauvegarde = self.sortie / blink_registre.STATE_V1_BACKUP_FILENAME
self.assertEqual(sauvegarde.read_bytes(), original)
registre.write_bytes(sauvegarde.read_bytes())
restaure = blink_registre.load_download_state(self.sortie)
self.assertEqual(restaure["version"], 2)
self.assertEqual(set(restaure["clips"]), set(etat_v1["clips"]))
self.assertEqual(
next(iter(second["clips"].values()))["path"],
archive.relative_to(self.sortie).as_posix(),
)
def test_T_MIGRATION_V1_echec_backup_annule_ecriture_v2(self):
archive = self.sortie / "archive.mp4"
archive.write_bytes(b"archive")
registre = self.sortie / blink_registre.STATE_FILENAME
original = json.dumps(self._etat_v1(archive)).encode("utf-8")
registre.write_bytes(original)
etat = blink_registre.load_download_state(self.sortie)
with mock.patch.object(Path, "write_bytes", side_effect=OSError("disque")):
with self.assertRaisesRegex(OSError, "migration v2 annulée"):
blink_registre.save_download_state(self.sortie, etat)
self.assertEqual(registre.read_bytes(), original)
self.assertFalse(
(self.sortie / blink_registre.STATE_V1_BACKUP_FILENAME).exists()
)
class TestsNomsDeFichiers(BacASable):
def test_T_I16_noms_reserves_windows_sont_neutralises(self):
reserves = ["CON", "PRN", "AUX", "NUL", "COM1", "LPT9", "con.txt"]
for nom in reserves:
with self.subTest(nom=nom):
composant = blink_models.safe_name(nom)
racine = composant.split(".", 1)[0].casefold()
self.assertNotIn(
racine,
{"con", "prn", "aux", "nul", "com1", "lpt9"},
)
def test_T_I16_separateurs_et_path_traversal_restent_dans_output(self):
noms = ["../secret", r"..\secret", "dossier/fichier", r"dossier\fichier"]
for position, nom in enumerate(noms):
with self.subTest(nom=nom):
clip = Clip(position, nom=nom, device_id=f"camera-{position}")
cible = blink_models.target_path(self.sortie, clip)
cible.resolve().relative_to(self.sortie.resolve())
relatif = cible.relative_to(self.sortie)
self.assertEqual(len(relatif.parts), 3)
self.assertNotIn("..", relatif.parts)
self.assertNotIn("/", blink_models.safe_name(nom))
self.assertNotIn("\\", blink_models.safe_name(nom))
def test_T_I16_composants_et_chemin_total_sont_bornes(self):
clip = Clip(
"id-" + "9" * 500,
nom="caméra-" + "é" * 500,
device_id="device-" + "x" * 500,
)
cible = blink_models.target_path(self.sortie, clip)
self.assertTrue(all(len(part.encode("utf-8")) <= 255 for part in cible.parts))
self.assertLessEqual(len(str(cible)), 240)
if __name__ == "__main__":
unittest.main(verbosity=2)