"""Contrats de planification observables depuis l'orchestration de fusion.""" from __future__ import annotations import contextlib import datetime as dt import io import json import unittest from pathlib import Path from types import SimpleNamespace from unittest import mock import merge_daily as md import test_merge_daily_sauvegarde_incrementale as fixtures class TestsPlanNormalisation(unittest.TestCase): def setUp(self): self.fixture = fixtures.TestsSauvegardeIncrementale() self.fixture.setUp() self.addCleanup(self.fixture.doCleanups) self.args = self.fixture._args() self.moment = dt.datetime(2026, 9, 8, 10, tzinfo=dt.timezone.utc) def _clip(self, nom, jour=8): return self.moment.replace(day=jour), self.args.input / nom def _executer(self, groupes, *, connus=None, invalides=(), dimensions=None, identites_indisponibles=()): """Les seuls effets réels sont les registres dans le dossier temporaire.""" registre = {"version": 1, "cameras": {}, "clips": connus or {}} self.fixture.registry_path.write_text(json.dumps(registre), encoding="utf-8") (self.args.output / md.MERGE_STATE).write_text( json.dumps({"version": 1, "groups": {}}), encoding="utf-8") dimensions = dimensions or {} invalides = set(invalides) sortie = io.StringIO() appels = SimpleNamespace() def information(ffmpeg, registry, identity, created, source): largeur, hauteur, cadence = dimensions.get(identity, (1280, 720, 15.0)) return md.ClipInfo(created=created, source=source, duration=5.0, width=largeur, height=hauteur, fps=cadence, has_audio=False) def normalisation(ffmpeg, timezone, registry, normalized_dir, identity, info, target, key, font_path, preset, crf, force, on_progress=None, style=None): registry["clips"][identity] = {"key": key} return True, "", on_progress is not None with contextlib.ExitStack() as pile: def remplacer(nom, **options): appel = pile.enter_context(mock.patch.object(md, nom, **options)) setattr(appels, nom, appel) return appel remplacer("find_ffmpeg", return_value="ffmpeg-simule") remplacer("check_drawtext_available") remplacer("find_font", return_value=self.args.input / "police.ttf") remplacer("check_timestamp_rendering") remplacer("load_groups", return_value=groupes) remplacer("journees_a_source_indisponible", return_value=(set(), set(identites_indisponibles))) remplacer("clip_info", side_effect=information) remplacer("camera_target", wraps=md.camera_target) remplacer("render_key", side_effect=lambda identity, *args, **kw: "rendu/" + identity) remplacer("valid_mp4", side_effect=lambda chemin: chemin.name not in invalides) remplacer("normalize_clip", side_effect=normalisation) remplacer("progress_printer", side_effect=lambda libelle: mock.Mock(name=libelle)) remplacer("merge_group", return_value=(True, "")) remplacer("build_periods", return_value=(0, 0, 0)) remplacer("prune_normalized", return_value=0) appels.travail = pile.enter_context(mock.patch.object(md.runtime, "travail")) # Une extraction ne doit pas introduire d'appel réel à ffmpeg. pile.enter_context(mock.patch.object(md.runtime, "demarrer", side_effect=AssertionError("processus inattendu"))) pile.enter_context(contextlib.redirect_stdout(sortie)) appels.code = md._executer(self.args) appels.sortie = sortie.getvalue() appels.registre = json.loads(self.fixture.registry_path.read_text(encoding="utf-8")) return appels def test_les_cibles_utilisent_tous_les_clips_meme_avec_deux_filtres(self): self.args.camera = "sAlOn" self.args.date = "2026-09-08" groupes = { ("Terrasse", "2026-09-08"): [self._clip("terrasse.mp4")], ("Salon", "2026-09-08"): [self._clip("salon.mp4")], ("Salon", "2026-09-07"): [self._clip("salon-hd.mp4", 7)], } resultat = self._executer(groupes, dimensions={"salon-hd.mp4": (1920, 1080, 30.0)}) self.assertEqual(resultat.code, 0) self.assertEqual([appel.args[1] for appel in resultat.camera_target.call_args_list], ["Salon", "Terrasse"]) self.assertEqual([info.source.name for info in resultat.camera_target.call_args_list[0].args[2]], ["salon.mp4", "salon-hd.mp4"]) resultat.normalize_clip.assert_called_once() self.assertEqual(resultat.normalize_clip.call_args.args[4], "salon.mp4") self.assertEqual(resultat.normalize_clip.call_args.args[6], (1920, 1080, 30.0)) self.assertEqual(set(resultat.registre["cameras"]), {"Salon", "Terrasse"}) resultat.prune_normalized.assert_not_called() def test_le_plan_trie_les_groupes_et_conserve_l_ordre_interne_des_clips(self): groupes = { ("Terrasse", "2026-09-08"): [self._clip("t.mp4")], ("Salon", "2026-09-08"): [self._clip("b.mp4"), self._clip("a.mp4")], ("Salon", "2026-09-07"): [self._clip("ancien.mp4", 7)], } resultat = self._executer(groupes) attendus = ["ancien.mp4", "b.mp4", "a.mp4", "t.mp4"] self.assertEqual([appel.args[0] for appel in resultat.render_key.call_args_list], attendus) self.assertEqual([appel.args[4] for appel in resultat.normalize_clip.call_args_list], attendus) self.assertEqual([[segment.name for segment in appel.args[1]] for appel in resultat.merge_group.call_args_list], [["ancien.mp4"], ["b.mp4", "a.mp4"], ["t.mp4"]]) def test_les_cles_sont_calculees_aussi_hors_selection(self): self.args.camera = "salon" self.args.date = "2026-09-08" self.args.preset = "fast" self.args.crf = 27 groupes = { ("Salon", "2026-09-08"): [self._clip("retenu.mp4")], ("Salon", "2026-09-07"): [self._clip("ancien.mp4", 7)], ("Jardin", "2026-09-08"): [self._clip("ailleurs.mp4")], } resultat = self._executer(groupes) self.assertEqual([appel.args[0] for appel in resultat.render_key.call_args_list], ["ailleurs.mp4", "ancien.mp4", "retenu.mp4"]) for appel in resultat.render_key.call_args_list: self.assertEqual(appel.args[3], int(appel.args[1].created.timestamp())) self.assertEqual(appel.args[4:], (None, "fast", 27)) resultat.normalize_clip.assert_called_once() self.assertEqual(resultat.normalize_clip.call_args.args[4], "retenu.mp4") def test_un_filtre_sans_correspondance_calcule_encore_cibles_et_cles(self): self.args.camera = "absente" resultat = self._executer({("Salon", "2026-09-08"): [self._clip("a.mp4")]}) self.assertEqual(resultat.code, 0) self.assertIn("Aucun groupe de clips à fusionner.", resultat.sortie) resultat.camera_target.assert_called_once() resultat.render_key.assert_called_once() resultat.normalize_clip.assert_not_called() resultat.valid_mp4.assert_not_called() resultat.prune_normalized.assert_not_called() def test_la_date_filtre_exactement_et_la_camera_ignore_la_casse_unicode(self): self.args.camera = "STRASSE" self.args.date = "2026-09-08" resultat = self._executer({ ("Straße", "2026-09-07"): [self._clip("avant.mp4", 7)], ("Straße", "2026-09-08"): [self._clip("retenu.mp4")], ("Strasse annexe", "2026-09-08"): [self._clip("autre.mp4")], }) resultat.normalize_clip.assert_called_once() self.assertEqual(resultat.normalize_clip.call_args.args[4], "retenu.mp4") def test_force_ou_cle_perimee_evitent_la_sonde_mp4(self): cas = [(True, "rendu/a.mp4"), (True, "ancienne"), (False, "ancienne")] for force, cle in cas: with self.subTest(force=force, cle=cle): self.args.force = force resultat = self._executer( {("Salon", "2026-09-08"): [self._clip("a.mp4")]}, connus={"a.mp4": {"key": cle}}, ) resultat.valid_mp4.assert_not_called() resultat.progress_printer.assert_called_once_with("[1/1]") self.assertIsNotNone(resultat.normalize_clip.call_args.args[-1]) def test_un_segment_a_jour_est_verifie_et_reutilise_sans_progression(self): resultat = self._executer( {("Salon", "2026-09-08"): [self._clip("a.mp4")]}, connus={"a.mp4": {"key": "rendu/a.mp4"}}, ) resultat.valid_mp4.assert_called_once_with(self.args.normalized_output.resolve() / "a.mp4") resultat.normalize_clip.assert_called_once() self.assertIsNone(resultat.normalize_clip.call_args.args[-1]) resultat.progress_printer.assert_not_called() self.assertIn("0 clip(s) encodé(s), 1 réutilisé(s)", resultat.sortie) def test_un_segment_invalide_est_reencode_meme_si_sa_cle_est_a_jour(self): resultat = self._executer( {("Salon", "2026-09-08"): [self._clip("a.mp4")]}, connus={"a.mp4": {"key": "rendu/a.mp4"}}, invalides={"a.mp4"}, ) resultat.valid_mp4.assert_called_once_with(self.args.normalized_output.resolve() / "a.mp4") resultat.progress_printer.assert_called_once_with("[1/1]") self.assertIn("1 clip(s) encodé(s), 0 réutilisé(s)", resultat.sortie) def test_le_total_ne_compte_que_les_identites_a_encoder(self): resultat = self._executer({ ("Salon", "2026-09-08"): [self._clip("a.mp4"), self._clip("reutilise.mp4")], ("Terrasse", "2026-09-08"): [self._clip("b.mp4")], }, connus={"reutilise.mp4": {"key": "rendu/reutilise.mp4"}}) self.assertEqual(resultat.progress_printer.call_args_list, [mock.call("[1/2]"), mock.call("[2/2]")]) preparation = [appel for appel in resultat.travail.call_args_list if appel.kwargs.get("cle") == "phase.prepare_clips"] self.assertEqual(preparation, [ mock.call("Préparation des clips", 0, 2, cle="phase.prepare_clips"), mock.call("Préparation des clips", 1, 2, cle="phase.prepare_clips"), ]) def test_le_nettoyage_conserve_les_segments_utilises_et_sources_indisponibles(self): resultat = self._executer({ ("Salon", "2026-09-08"): [self._clip("a.mp4")], ("Terrasse", "2026-09-08"): [self._clip("b.mp4")], }, identites_indisponibles={"c.mp4"}) resultat.prune_normalized.assert_called_once() self.assertEqual(resultat.prune_normalized.call_args.args[2], {self.args.normalized_output.resolve() / nom for nom in ("a.mp4", "b.mp4", "c.mp4")}) def test_un_alias_de_stockage_utilise_le_chemin_resolu_a_toutes_les_etapes(self): dossier_reel = self.args.normalized_output.resolve() alias = dossier_reel.with_name("alias-normalized") self.args.normalized_output = alias resoudre = Path.resolve def resoudre_alias(chemin, *args, **kwargs): # Simule un nom court Windows ou /var -> /private/var sur macOS, # sans créer de lien symbolique ni demander de privilèges. return dossier_reel if chemin == alias else resoudre(chemin, *args, **kwargs) with mock.patch.object(Path, "resolve", autospec=True, side_effect=resoudre_alias): resultat = self._executer( {("Salon", "2026-09-08"): [self._clip("a.mp4")]}, connus={"a.mp4": {"key": "rendu/a.mp4"}}, identites_indisponibles={"b.mp4"}, ) self.assertEqual(resultat.code, 0) resultat.valid_mp4.assert_called_once_with(dossier_reel / "a.mp4") self.assertEqual(resultat.normalize_clip.call_args.args[3], dossier_reel) self.assertEqual(resultat.merge_group.call_args.args[1], [dossier_reel / "a.mp4"]) self.assertEqual(resultat.prune_normalized.call_args.args[0], dossier_reel) self.assertEqual(resultat.prune_normalized.call_args.args[2], {dossier_reel / "a.mp4", dossier_reel / "b.mp4"}) def test_un_identifiant_repete_ne_gonfle_pas_le_total_annonce(self): resultat = self._executer({ ("Salon", "2026-09-08"): [self._clip("a.mp4"), self._clip("a.mp4")], }) self.assertIn("Normalisation : 1 clip(s) à encoder", resultat.sortie) preparation = [appel for appel in resultat.travail.call_args_list if appel.kwargs.get("cle") == "phase.prepare_clips"] self.assertTrue(preparation) self.assertTrue(all(appel.args[2] == 1 for appel in preparation)) def test_sans_horodatage_aucune_police_ni_sonde_drawtext(self): resultat = self._executer({("Salon", "2026-09-08"): [self._clip("a.mp4")]}) resultat.check_drawtext_available.assert_not_called() resultat.find_font.assert_not_called() resultat.check_timestamp_rendering.assert_not_called() self.assertIsNone(resultat.render_key.call_args.args[4]) self.assertIsNone(resultat.normalize_clip.call_args.args[8]) def test_avec_horodatage_la_police_est_validee_et_transmise(self): self.args.no_timestamp = False self.args.font = self.args.input / "choix.ttf" resultat = self._executer({("Salon", "2026-09-08"): [self._clip("a.mp4")]}) police = self.args.input / "police.ttf" style_attendu = md.TimestampStyle(None, "white", 0.55) resultat.check_drawtext_available.assert_called_once_with("ffmpeg-simule") resultat.find_font.assert_called_once_with(self.args.font) resultat.check_timestamp_rendering.assert_called_once_with( "ffmpeg-simule", police, style_attendu) self.assertEqual(resultat.render_key.call_args.args[4], police) self.assertEqual(resultat.render_key.call_args.kwargs["style"], style_attendu) self.assertEqual(resultat.normalize_clip.call_args.args[8], police) self.assertEqual(resultat.normalize_clip.call_args.kwargs["style"], style_attendu) if __name__ == "__main__": unittest.main()