"""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)