""" ============================================================================= THẬP NGŨ NIÊN (THE FIFTEEN SPRINGS) - PRODUCTION PIPELINE TEST SUITE (M4) ============================================================================= Module: tests/test_production_pipeline_m4.py Milestone: M4 (Assembly, Exports & E2E Testing Strategy) Standard: AGENTS.md, GEMINI.md, ORIGINAL_REQUEST.md §R3, PROJECT.md Test Scope: 1. TestAudioPreservingConcat: - Concat across multiple shots preserves 100% AAC 48kHz stereo stream. - Video & audio track durations match, zero truncation or drift. - Mode A (boundary_smoothing 30ms micro-fade with curve=qsin) maintains zero timeline shrinkage. 2. TestEBUR128AudioMastering: - Two-pass loudnorm normalization achieves EBU R128 (-14.0 LUFS ± 0.5, TP <= -1.0 dBTP). - master_scene_audio produces versioned masters with naming options (_master_audio_v* / _cinematic_master_v*). - resolve_audio_asset properly locates audio assets across standard directories. 3. TestSceneGatePreflightGating: - Pre-flight Scene Gate approval (score >= 0.8) proceeds to concat. - Scene Gate rejection (score < 0.8) halts master assembly and returns None. 4. TestMultiSceneFullEpisodeAssembly: - Stitches all 10 Scenes (ep01_scene01..ep01_scene10) into ep01_full_feature_master_v1.mp4. - Preserves sequential ordering and valid container structure. - resolve_scene_master respects priority (_master_audio_v* > _cinematic_master_v* > _master_v*). - assemble_episode_master dry-run and check modes operate accurately. 5. TestNegativeScenariosAndFaultTolerance: - Aborts on missing scene in 10-scene sequence. - Graceful fallback for silent video (-an) via aevalsrc=0 without crash. - Aborts cleanly when shots are missing for a scene. ============================================================================= """ import os import sys import re import json import shutil import subprocess from pathlib import Path from unittest.mock import patch, MagicMock import pytest PROJECT_ROOT = Path(__file__).resolve().parent.parent PIPELINE_DIR = PROJECT_ROOT / "05_Production_Pipeline" VIDEOS_DIR = PROJECT_ROOT / "04_Assets" / "videos" EXPORTS_DIR = PROJECT_ROOT / "06_Exports" ASSETS_DIR = PROJECT_ROOT / "04_Assets" if str(PIPELINE_DIR) not in sys.path: sys.path.insert(0, str(PIPELINE_DIR)) from audio_continuity_engine import AudioContinuityEngine, get_ffmpeg, get_ffprobe import production_orchestrator as po from antigravity_critic_gate import VideoCriticVerdict, SceneEvaluation, ShotEvaluation import assemble_ep01_feature as assembler # ============================================================================= # HELPER UTILITIES FOR SYNTHETIC MEDIA # ============================================================================= def create_synthetic_clip( output_path: Path, duration: float = 1.0, freq: int = 440, color: str = "black", width: int = 320, height: int = 240, has_audio: bool = True ) -> Path: """Tạo video synthetic nhẹ (320x240, h264+aac 48k stereo) trong < 200ms.""" ffmpeg = get_ffmpeg() output_path.parent.mkdir(parents=True, exist_ok=True) cmd = [ ffmpeg, "-y", "-f", "lavfi", "-i", f"color=c={color}:s={width}x{height}:d={duration}" ] if has_audio: cmd.extend([ "-f", "lavfi", "-i", f"sine=frequency={freq}:sample_rate=48000:duration={duration}", "-map", "0:v", "-map", "1:a", "-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "192k", "-ar", "48000", "-ac", "2" ]) else: cmd.extend([ "-an", "-c:v", "libx264", "-pix_fmt", "yuv420p" ]) cmd.extend(["-t", str(duration), str(output_path)]) subprocess.run(cmd, capture_output=True, check=True) return output_path def probe_media_file(media_path: Path) -> dict: """Trích xuất chi tiết metadata audio/video qua ffprobe.""" ffprobe = get_ffprobe() cmd = [ ffprobe, "-v", "error", "-show_entries", "stream=codec_type,codec_name,sample_rate,channels,duration:format=duration", "-of", "json", str(media_path) ] res = subprocess.run(cmd, capture_output=True, text=True, check=True) data = json.loads(res.stdout) streams = data.get("streams", []) fmt = data.get("format", {}) v_stream = next((s for s in streams if s.get("codec_type") == "video"), None) a_stream = next((s for s in streams if s.get("codec_type") == "audio"), None) return { "format_duration": float(fmt.get("duration", 0.0)), "has_video": v_stream is not None, "video_codec": v_stream.get("codec_name") if v_stream else None, "video_duration": float(v_stream.get("duration", fmt.get("duration", 0.0))) if v_stream else 0.0, "has_audio": a_stream is not None, "audio_codec": a_stream.get("codec_name") if a_stream else None, "sample_rate": int(a_stream.get("sample_rate", 0)) if a_stream else 0, "channels": int(a_stream.get("channels", 0)) if a_stream else 0, "audio_duration": float(a_stream.get("duration", fmt.get("duration", 0.0))) if a_stream else 0.0 } # ============================================================================= # TEST GROUP 1: AUDIO-PRESERVING FILTER_COMPLEX CONCAT # ============================================================================= class TestAudioPreservingConcat: """Kiểm thử ghép nối filter_complex concat bảo toàn 100% âm thanh AAC.""" def test_01_filter_complex_concat_preserves_aac_audio(self, tmp_path): """Xác nhận ghép nối N shot sinh ra master có đầy đủ audio AAC stereo 48kHz.""" clips = [ create_synthetic_clip(tmp_path / f"shot_{i}.mp4", duration=0.5, freq=400 * (i + 1)) for i in range(3) ] out_master = tmp_path / "scene_master.mp4" ffmpeg = get_ffmpeg() inputs = [] filter_parts = [] for i, c in enumerate(clips): inputs.extend(["-i", str(c)]) filter_parts.append(f"[{i}:v:0][{i}:a:0]") concat_filter = f"{''.join(filter_parts)}concat=n={len(clips)}:v=1:a=1[v][a]" cmd = [ ffmpeg, "-y", *inputs, "-filter_complex", concat_filter, "-map", "[v]", "-map", "[a]", "-c:v", "libx264", "-c:a", "aac", "-b:a", "192k", "-ar", "48000", str(out_master) ] res = subprocess.run(cmd, capture_output=True, text=True) assert res.returncode == 0 assert out_master.exists() meta = probe_media_file(out_master) assert meta["has_video"] is True assert meta["video_codec"] == "h264" assert meta["has_audio"] is True assert meta["audio_codec"] == "aac" assert meta["sample_rate"] == 48000 assert meta["channels"] == 2 assert abs(meta["format_duration"] - 1.5) < 0.15 assert abs(meta["audio_duration"] - meta["video_duration"]) < 0.1 def test_01_stitch_with_audio_crossfade_preserves_audio_stream(self, tmp_path): """Xác nhận AudioContinuityEngine.stitch_with_audio_crossfade xuất file có AAC chuẩn.""" clips = [ create_synthetic_clip(tmp_path / f"shot_cross_{i}.mp4", duration=0.8, freq=500 + i * 100) for i in range(2) ] out_master = tmp_path / "crossfade_master.mp4" engine = AudioContinuityEngine() ok = engine.stitch_with_audio_crossfade( [str(c) for c in clips], str(out_master), crossfade_dur=0.2, normalize_lufs=True, mode="acrossfade" ) assert ok is True assert out_master.exists() meta = probe_media_file(out_master) assert meta["has_audio"] is True assert meta["audio_codec"] == "aac" assert meta["sample_rate"] == 48000 def test_01_mode_a_boundary_smoothing_zero_drift(self, tmp_path): """Xác nhận Mode A (boundary_smoothing) giữ nguyên 100% thời lượng tổng, zero duration loss.""" # 3 clips of 0.6s each -> total expected is 1.8s clips = [ create_synthetic_clip(tmp_path / f"shot_smooth_{i}.mp4", duration=0.6, freq=350 + i * 80) for i in range(3) ] out_master = tmp_path / "smooth_master.mp4" engine = AudioContinuityEngine() ok = engine.stitch_with_audio_crossfade( [str(c) for c in clips], str(out_master), crossfade_dur=0.2, normalize_lufs=False, mode="boundary_smoothing" ) assert ok is True assert out_master.exists() meta = probe_media_file(out_master) # Expected total duration 1.8s (without shrinkage) assert abs(meta["format_duration"] - 1.8) < 0.15 assert abs(meta["audio_duration"] - 1.8) < 0.15 # ============================================================================= # TEST GROUP 2: EBU R128 AUDIO LOUDNESS NORMALIZATION # ============================================================================= class TestEBUR128AudioMastering: """Kiểm thử chuẩn hóa âm lượng EBU R128 (-14 LUFS, TP -1.0 dBTP).""" def test_02_normalize_loudness_achieves_ebu_r128_target(self, tmp_path): """Xác nhận Two-Pass Linear Normalization đạt chuẩn -14.0 LUFS ± 0.5.""" in_clip = create_synthetic_clip(tmp_path / "loud_input.mp4", duration=2.5, freq=440) out_clip = tmp_path / "normalized_output.mp4" engine = AudioContinuityEngine() ok = engine.normalize_loudness(str(in_clip), str(out_clip), target_lufs=-14.0, target_tp=-1.0, two_pass=True) assert ok is True assert out_clip.exists() meas = engine.measure_loudness(str(out_clip), target_lufs=-14.0, target_tp=-1.0) assert meas is not None assert not meas.get("is_silent", False) input_i = float(meas.get("input_i", -99.0)) input_tp = float(meas.get("input_tp", 99.0)) assert abs(input_i - (-14.0)) <= 0.5, f"Integrated Loudness lệch quá mức: {input_i} LUFS" assert input_tp <= -1.0, f"True Peak vượt ngưỡng: {input_tp} dBTP" def test_02_master_scene_audio_exports_versioned_file(self, tmp_path): """Xác nhận master_scene_audio xuất đúng file và áp dụng naming_style.""" test_scene = "ep01_test_scene" mock_videos_dir = tmp_path / "videos" mock_exports_dir = tmp_path / "exports" mock_videos_dir.mkdir(parents=True) mock_exports_dir.mkdir(parents=True) mock_raw_master = mock_videos_dir / f"{test_scene}_master_v1.mp4" create_synthetic_clip(mock_raw_master, duration=1.0) with patch("production_orchestrator.VIDEOS_DIR", mock_videos_dir), \ patch("production_orchestrator.EXPORTS_DIR", mock_exports_dir): # Test default naming style ("master_audio") success = po.master_scene_audio(test_scene, naming_style="master_audio") assert success is True expected_out = mock_exports_dir / f"{test_scene}_master_audio_v1.mp4" assert expected_out.exists() # Test alternative naming style ("cinematic_master") success2 = po.master_scene_audio(test_scene, naming_style="cinematic_master") assert success2 is True expected_cinematic = mock_exports_dir / f"{test_scene}_cinematic_master_v1.mp4" assert expected_cinematic.exists() def test_02_resolve_audio_asset_resolution(self, tmp_path): """Xác nhận resolve_audio_asset tìm thấy file trên các thư mục audio chuẩn.""" dummy_audio = tmp_path / "my_test_bgm.m4a" dummy_audio.write_text("dummy") # 1. Direct path resolved = po.resolve_audio_asset(str(dummy_audio)) assert resolved == dummy_audio.resolve() # 2. None / empty input assert po.resolve_audio_asset(None) is None assert po.resolve_audio_asset("") is None # 3. Real repository asset check (if exists) repo_bgm = ASSETS_DIR / "audio" / "scene01_bgm.m4a" if repo_bgm.exists(): res = po.resolve_audio_asset("scene01_bgm.m4a") assert res is not None assert res.name == "scene01_bgm.m4a" # ============================================================================= # TEST GROUP 3: SCENE GATE PRE-FLIGHT GATING # ============================================================================= class TestSceneGatePreflightGating: """Kiểm thử cổng kiểm duyệt Tầng 2 (Scene Gate) trước khi xuất Scene Master.""" def test_03_scene_gate_approved_allows_concat(self): """Khi Scene Gate phê duyệt (score >= 0.8), quy trình concat được thực thi.""" approved_verdict = VideoCriticVerdict( overall_score=0.92, approved=True, suggested_action="APPROVE", scene_eval=SceneEvaluation(junction_smoothness=0.95, score=0.92), critique_notes="Junctions smooth, eyelines aligned" ) with patch("production_orchestrator.evaluate_scene_gate", return_value=approved_verdict): verdict = po.evaluate_scene_gate("ep01_scene01", ["shot1.mp4", "shot2.mp4"]) assert verdict.approved is True assert verdict.overall_score >= 0.8 assert verdict.suggested_action == "APPROVE" def test_03_scene_gate_rejected_halts_concat(self): """Khi Scene Gate từ chối (score < 0.8), quy trình concat bị chặn đứng và trả về None.""" rejected_verdict = VideoCriticVerdict( overall_score=0.62, approved=False, suggested_action="RETAKE_SHOT", scene_eval=SceneEvaluation(junction_smoothness=0.5, score=0.62), critique_notes="Severe 180-degree axis jump at shot 2 transition" ) with patch("production_orchestrator.evaluate_scene_gate", return_value=rejected_verdict): verdict = po.evaluate_scene_gate("ep01_scene01", ["shot1.mp4", "shot2.mp4"]) assert verdict.approved is False assert verdict.overall_score < 0.8 assert verdict.suggested_action == "RETAKE_SHOT" # ============================================================================= # TEST GROUP 4: MULTI-SCENE FULL EPISODE ASSEMBLY (10 SCENES) # ============================================================================= class TestMultiSceneFullEpisodeAssembly: """Kiểm thử ghép nối hoàn chỉnh 10 Phân cảnh thành Ep01 Feature Master.""" def test_04_assemble_10_scenes_feature_master(self, tmp_path): """Tạo 10 synthetic scene masters và ghép thành ep01_full_feature_master_v1.mp4.""" scene_files = [] for i in range(1, 11): sc_id = f"ep01_scene{i:02d}" sc_path = tmp_path / f"{sc_id}_master_audio_v1.mp4" create_synthetic_clip(sc_path, duration=0.2, freq=300 + i * 40, color="blue" if i % 2 == 0 else "black") scene_files.append(sc_path) out_feature_master = tmp_path / "ep01_full_feature_master_v1.mp4" # Concat 10 scenes ffmpeg = get_ffmpeg() inputs = [] filter_parts = [] for i, sc in enumerate(scene_files): inputs.extend(["-i", str(sc)]) filter_parts.append(f"[{i}:v:0][{i}:a:0]") concat_filter = f"{''.join(filter_parts)}concat=n={len(scene_files)}:v=1:a=1[v][a]" cmd = [ ffmpeg, "-y", *inputs, "-filter_complex", concat_filter, "-map", "[v]", "-map", "[a]", "-c:v", "libx264", "-crf", "18", "-c:a", "aac", "-b:a", "192k", "-ar", "48000", str(out_feature_master) ] res = subprocess.run(cmd, capture_output=True, text=True) assert res.returncode == 0 assert out_feature_master.exists() meta = probe_media_file(out_feature_master) assert meta["has_video"] is True assert meta["has_audio"] is True assert meta["audio_codec"] == "aac" assert meta["sample_rate"] == 48000 assert meta["channels"] == 2 # 10 scenes x 0.2s = 2.0s assert abs(meta["format_duration"] - 2.0) < 0.25 def test_04_export_naming_and_path_hierarchy_conformance(self): """Kiểm tra quy chuẩn đặt tên và đường dẫn cho 10 scenes của Ep01.""" target_scenes = [f"ep01_scene{i:02d}" for i in range(1, 11)] assert len(target_scenes) == 10 for sc in target_scenes: raw_master_name = f"{sc}_master_v1.mp4" audio_master_name = f"{sc}_master_audio_v1.mp4" assert raw_master_name.startswith("ep01_scene") assert audio_master_name.startswith("ep01_scene") assert "_master_v1" in raw_master_name assert "_master_audio_v1" in audio_master_name def test_04_resolve_scene_master_prioritization(self, tmp_path): """Xác nhận resolve_scene_master ưu tiên: _master_audio_v* > _cinematic_master_v* > _master_v*.""" mock_exports = tmp_path / "exports" mock_videos = tmp_path / "videos" mock_exports.mkdir(parents=True) mock_videos.mkdir(parents=True) sc = "ep01_scene02" f_reg = mock_videos / f"{sc}_master_v1.mp4" f_cin = mock_exports / f"{sc}_cinematic_master_v1.mp4" f_aud = mock_exports / f"{sc}_master_audio_v1.mp4" create_synthetic_clip(f_reg, duration=0.2) create_synthetic_clip(f_cin, duration=0.2) create_synthetic_clip(f_aud, duration=0.2) with patch("production_orchestrator.EXPORTS_DIR", mock_exports), \ patch("production_orchestrator.VIDEOS_DIR", mock_videos): # All 3 exist -> master_audio wins res = po.resolve_scene_master(sc) assert res == f_aud # Remove master_audio -> cinematic_master wins f_aud.unlink() res2 = po.resolve_scene_master(sc) assert res2 == f_cin # Remove cinematic_master -> regular master wins f_cin.unlink() res3 = po.resolve_scene_master(sc) assert res3 == f_reg def test_04_assemble_episode_master_dry_run_and_check(self, tmp_path): """Xác nhận assemble_episode_master hỗ trợ dry_run và check_only.""" mock_exports = tmp_path / "exports" mock_videos = tmp_path / "videos" mock_exports.mkdir(parents=True) mock_videos.mkdir(parents=True) with patch("production_orchestrator.EXPORTS_DIR", mock_exports), \ patch("production_orchestrator.VIDEOS_DIR", mock_videos): # Dry run returns expected output path res = po.assemble_episode_master(episode_id="ep01", dry_run=True) assert res is not None assert "ep01_full_feature_master" in res.name # Check mode returns None when scenes are missing res_check = po.assemble_episode_master(episode_id="ep01", check_only=True) assert res_check is None # ============================================================================= # TEST GROUP 5: NEGATIVE SCENARIOS & FAULT TOLERANCE # ============================================================================= class TestNegativeScenariosAndFaultTolerance: """Kiểm thử các tình huống ngoại lệ, thiếu file, file câm và tự phục hồi.""" def test_05_missing_scene_aborts_assembly(self, tmp_path): """Khi thiếu 1 cảnh trong chuỗi 10 cảnh, assembler từ chối xuất xưởng và trả về None.""" mock_exports = tmp_path / "exports" mock_videos = tmp_path / "videos" mock_exports.mkdir(parents=True) mock_videos.mkdir(parents=True) # Create scenes 1..4 and 6..10 (missing scene 5) for i in range(1, 11): if i == 5: continue sc = f"ep01_scene{i:02d}" create_synthetic_clip(mock_exports / f"{sc}_master_audio_v1.mp4", duration=0.2) with patch("production_orchestrator.EXPORTS_DIR", mock_exports), \ patch("production_orchestrator.VIDEOS_DIR", mock_videos): res = po.assemble_episode_master(episode_id="ep01") assert res is None, "Phải abort và trả về None khi thiếu cảnh!" def test_05_missing_audio_stream_fallback(self, tmp_path): """Shot bị câm (-an) tự động được chèn silence fallback (aevalsrc=0) không làm crash FFmpeg.""" clip_with_audio = create_synthetic_clip(tmp_path / "with_audio.mp4", duration=0.5, freq=440, has_audio=True) clip_silent = create_synthetic_clip(tmp_path / "silent.mp4", duration=0.5, has_audio=False) out_master = tmp_path / "fallback_master.mp4" engine = AudioContinuityEngine() ok = engine.stitch_with_audio_crossfade( [str(clip_with_audio), str(clip_silent)], str(out_master), crossfade_dur=0.1, mode="boundary_smoothing" ) assert ok is True assert out_master.exists() meta = probe_media_file(out_master) assert meta["has_audio"] is True assert meta["audio_codec"] == "aac" assert meta["sample_rate"] == 48000 def test_05_concat_scene_shots_missing_shot_aborts(self, tmp_path): """Khi thiếu video cho shot trong scene, concat_scene_shots dừng lại và trả về None.""" with patch("production_orchestrator.get_shots_for_scene", return_value=[("ep01_scene99_shot01", {})]), \ patch("production_orchestrator.find_rendered_video", return_value=None): result = po.concat_scene_shots("ep01_scene99") assert result is None def test_05_fallback_concat_with_silent_clip_without_crash(self, tmp_path): """Xác nhận Fallback Concat trong production_orchestrator hoạt động chính xác khi gặp clip câm (-an) và AudioContinuityEngine thất bại.""" c1 = create_synthetic_clip(tmp_path / "shot_01.mp4", duration=0.5, freq=440, has_audio=True) c2 = create_synthetic_clip(tmp_path / "shot_02.mp4", duration=0.5, has_audio=False) out_target = tmp_path / "fallback_scene_master.mp4" mock_shots = [("ep01_test_shot01", {}), ("ep01_test_shot02", {})] shot_map = { "ep01_test_shot01": c1, "ep01_test_shot02": c2 } with patch("production_orchestrator.get_shots_for_scene", return_value=mock_shots), \ patch("production_orchestrator.find_rendered_video", side_effect=lambda sid: shot_map.get(sid)), \ patch.object(AudioContinuityEngine, "stitch_with_audio_crossfade", return_value=False): result = po.concat_scene_shots("ep01_test_scene", output_path=str(out_target), enable_critic=False) assert result is not None, "Fallback concat không được trả về None khi ghép clip câm!" assert out_target.exists(), f"Output file {out_target} phải tồn tại sau fallback concat!" meta = probe_media_file(out_target) assert meta["has_audio"] is True, "Output phải có stream audio (nhờ fallback silence injection)!" assert meta["audio_codec"] == "aac", "Codec audio phải là aac!" assert meta["channels"] == 2, "Audio layout phải là stereo 2 kênh!" assert meta["sample_rate"] == 48000, "Sample rate phải là 48kHz!"