test_merge_daily_source_partielle.py
Rama main · réplica de lectura del repositorio original.
"""Une source indisponible n'autorise pas à raccourcir les vidéos existantes."""
import contextlib
import datetime as dt
import io
import json
import unittest
from unittest import mock
import merge_daily as md
import runtime
import test_merge_daily_sauvegarde_incrementale as fixtures
class TestsSourcePartielle(unittest.TestCase):
def verifier(self, *, invalide=False, exclu=False, quotidienne_presente=True):
fixture = fixtures.TestsSauvegardeIncrementale()
fixture.setUp()
self.addCleanup(fixture.doCleanups)
args = fixture._args()
args.no_weekly = args.no_monthly = False
source = args.input / 'a.mp4'
source.write_bytes(fixtures.mp4_structurel())
if invalide:
(args.input / 'b.mp4').write_bytes(b'invalide')
clips = {nom: {'path': nom+'.mp4', 'camera': 'Salon',
'created_at': '2026-09-07T10:00:00+00:00',
'excluded': exclu and nom == 'b'} for nom in ('a', 'b')}
(args.input / md.DOWNLOAD_STATE).write_text(json.dumps({'clips': clips}), encoding='utf-8')
daily = args.output / 'Salon/2026-09-07_Salon.mp4'
daily.parent.mkdir()
if quotidienne_presente:
daily.write_bytes(b'quotidienne-complete')
state_path = args.output / md.MERGE_STATE
state_path.write_text(json.dumps({'groups': {'Salon|2026-09-07': {'fingerprint': 'ancien'}}}), encoding='utf-8')
state_original = state_path.read_bytes()
info = md.ClipInfo(created=dt.datetime(2026,9,7,10,tzinfo=dt.timezone.utc), source=source,
duration=5, width=1280, height=720, fps=15, has_audio=False)
with mock.patch.object(md, 'find_ffmpeg', return_value='ffmpeg-simule'), \
mock.patch.object(md, 'clip_info', return_value=info), \
mock.patch.object(md, 'camera_target', return_value=(1280,720,15)), \
mock.patch.object(md, 'normalize_clip', return_value=(True,'',True)), \
mock.patch.object(md, 'merge_group', return_value=(True,'')) as fusion, \
mock.patch.object(md, 'build_periods', return_value=(0,0,0)) as periodes, \
mock.patch.object(runtime, 'travail'), contextlib.redirect_stdout(io.StringIO()):
code = md._executer(args)
if exclu:
self.assertEqual(code, 0)
fusion.assert_called_once()
self.assertEqual(periodes.call_count, 2)
else:
# Jamais de journalière refaite avec les seuls clips restants.
fusion.assert_not_called()
self.assertEqual(state_path.read_bytes(), state_original)
if quotidienne_presente:
self.assertEqual(daily.read_bytes(), b'quotidienne-complete')
else:
self.assertFalse(daily.exists())
if invalide:
# Fichier là mais illisible : source indisponible, échec, agrégats reportés.
self.assertEqual(code, 1)
periodes.assert_not_called()
else:
# Fichier absent : effacé à la main (#118), noté comme purgé.
# Pas un échec, et les agrégats continuent d'être faits.
self.assertEqual(code, 0)
self.assertEqual(periodes.call_count, 2)
registre = json.loads((args.input / md.DOWNLOAD_STATE).read_text(encoding='utf-8'))
self.assertTrue(registre['clips']['b'].get('purge_manuelle'))
self.assertNotIn('purged_at', registre['clips']['a'])
def test_un_brut_efface_a_la_main_revient_s_il_est_remis(self):
fixture = fixtures.TestsSauvegardeIncrementale()
fixture.setUp()
self.addCleanup(fixture.doCleanups)
racine = fixture._args().input
etat = racine / md.DOWNLOAD_STATE
etat.write_text(json.dumps({'clips': {'b': {
'path': 'b.mp4', 'camera': 'Salon', 'created_at': '2026-09-07T10:00:00+00:00'}}}),
encoding='utf-8')
with contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(md.suivre_purges_manuelles(racine), 1)
self.assertTrue(json.loads(etat.read_text(encoding='utf-8'))['clips']['b']['purge_manuelle'])
(racine / 'b.mp4').write_bytes(fixtures.mp4_structurel())
self.assertEqual(md.suivre_purges_manuelles(racine), 1)
self.assertEqual(md.suivre_purges_manuelles(racine), 0)
entree = json.loads(etat.read_text(encoding='utf-8'))['clips']['b']
self.assertNotIn('purged_at', entree)
self.assertNotIn('purge_manuelle', entree)
def test_un_brut_efface_a_la_main_preserve_la_journaliere_sans_echec(self):
self.verifier()
def test_un_brut_invalide_preserve_la_journaliere(self):
self.verifier(invalide=True)
def test_aucune_journaliere_partielle_ne_peut_etre_creee(self):
self.verifier(quotidienne_presente=False)
def test_exclusion_expresse_autorise_la_reconstruction(self):
self.verifier(exclu=True)
class TestsAgregatsProteges(unittest.TestCase):
"""Audit du 26/09/2026, B02 : une journée qui a perdu à la fois ses clips
bruts et sa journalière protège les agrégats de sa semaine et de son
mois, même quand la fusion est filtrée sur un autre jour (--date)."""
LUNDI, MARDI = dt.date(2026, 9, 7), dt.date(2026, 9, 8)
def setUp(self):
fixture = fixtures.TestsSauvegardeIncrementale()
fixture.setUp()
self.addCleanup(fixture.doCleanups)
self.args = fixture._args()
self.args.no_weekly = self.args.no_monthly = False
# Même semaine ISO et même mois : la précondition du scénario.
self.assertEqual(md.period_label(self.LUNDI, 'weekly'),
md.period_label(self.MARDI, 'weekly'))
def _agregat(self, dossier, period, contenu):
label = md.period_label(self.LUNDI, period)
chemin = dossier / 'Salon' / f'{label}_Salon.mp4'
chemin.parent.mkdir(parents=True, exist_ok=True)
chemin.write_bytes(contenu)
etat = {'groups': {f'Salon|{label}': {
'fingerprint': 'ancien', 'path': f'Salon/{label}_Salon.mp4', 'days': 2}}}
(dossier / md.MERGE_STATE).write_text(json.dumps(etat), encoding='utf-8')
return chemin, (dossier / md.MERGE_STATE).read_bytes()
def test_filtre_sur_mardi_ne_raccourcit_pas_la_semaine_ni_le_mois(self):
args = self.args
args.date = self.MARDI.isoformat()
# Lundi : brut disparu et journalière absente. Mardi : tout est là.
(args.input / 'mardi.mp4').write_bytes(fixtures.mp4_structurel())
clips = {
'lundi': {'path': 'lundi.mp4', 'camera': 'Salon',
'created_at': '2026-09-07T10:00:00+00:00'},
'mardi': {'path': 'mardi.mp4', 'camera': 'Salon',
'created_at': '2026-09-08T10:00:00+00:00'},
}
(args.input / md.DOWNLOAD_STATE).write_text(
json.dumps({'clips': clips}), encoding='utf-8')
journaliere = args.output / 'Salon' / '2026-09-08_Salon.mp4'
journaliere.parent.mkdir()
journaliere.write_bytes(fixtures.mp4_structurel())
hebdo, etat_hebdo = self._agregat(args.weekly_output, 'weekly', b'semaine-complete')
mensuel, etat_mensuel = self._agregat(args.monthly_output, 'monthly', b'mois-complet')
info = md.ClipInfo(created=dt.datetime(2026, 9, 8, 10, tzinfo=dt.timezone.utc),
source=args.input / 'mardi.mp4', duration=5, width=1280,
height=720, fps=15, has_audio=False)
with mock.patch.object(md, 'find_ffmpeg', return_value='ffmpeg-simule'), \
mock.patch.object(md, 'clip_info', return_value=info), \
mock.patch.object(md, 'camera_target', return_value=(1280, 720, 15)), \
mock.patch.object(md, 'normalize_clip', return_value=(True, '', True)), \
mock.patch.object(md, 'merge_group', return_value=(True, '')), \
mock.patch.object(md, 'concat_videos', return_value=(True, '')) as concat, \
mock.patch.object(runtime, 'travail'), \
contextlib.redirect_stdout(io.StringIO()) as sortie:
md._executer(args)
concat.assert_not_called()
self.assertEqual(hebdo.read_bytes(), b'semaine-complete')
self.assertEqual(mensuel.read_bytes(), b'mois-complet')
self.assertEqual((args.weekly_output / md.MERGE_STATE).read_bytes(), etat_hebdo)
self.assertEqual((args.monthly_output / md.MERGE_STATE).read_bytes(), etat_mensuel)
self.assertIn(md.msg('periode_protegee', periode='Salon|2026-W37'), sortie.getvalue())
def test_une_periode_protegee_sans_journaliere_n_est_pas_supprimee(self):
hebdo, etat = self._agregat(self.args.weekly_output, 'weekly', b'semaine-complete')
cle = f"Salon|{md.period_label(self.LUNDI, 'weekly')}"
with contextlib.redirect_stdout(io.StringIO()):
md.build_periods('ffmpeg-simule', md.ZoneInfo('UTC'), self.args.output,
self.args.weekly_output, 'weekly', False, 'veryfast', 21,
proteges={cle})
self.assertEqual(hebdo.read_bytes(), b'semaine-complete')
self.assertEqual((self.args.weekly_output / md.MERGE_STATE).read_bytes(), etat)
def test_une_periode_vide_non_protegee_est_supprimee_sans_planter(self):
# Trouvé avec B02 : le message de ce nettoyage passait cle=... à
# msg(cle, **valeurs) et levait TypeError, avant même d'enregistrer
# l'état ; chaque fusion suivante plantait au même endroit.
hebdo, _ = self._agregat(self.args.weekly_output, 'weekly', b'semaine-complete')
with contextlib.redirect_stdout(io.StringIO()):
md.build_periods('ffmpeg-simule', md.ZoneInfo('UTC'), self.args.output,
self.args.weekly_output, 'weekly', False, 'veryfast', 21)
self.assertFalse(hebdo.exists())
etat = json.loads((self.args.weekly_output / md.MERGE_STATE).read_text(encoding='utf-8'))
self.assertEqual(etat['groups'], {})
def test_une_journee_indisponible_avec_sa_journaliere_ne_protege_rien(self):
# Le brut a disparu mais la journalière est là : l'agrégat la reprend
# telle quelle, rien n'est perdu, la période se reconstruit.
journaliere = self.args.output / 'Salon' / '2026-09-07_Salon.mp4'
journaliere.parent.mkdir()
journaliere.write_bytes(fixtures.mp4_structurel())
indisponibles = {('Salon', self.LUNDI.isoformat())}
self.assertEqual(md.periodes_a_proteger(indisponibles, self.args.output, 'weekly'),
set())
journaliere.unlink()
self.assertEqual(md.periodes_a_proteger(indisponibles, self.args.output, 'weekly'),
{'Salon|2026-W37'})