#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ============================================================================= CHALLENGER 2 EMPIRICAL HARNESS: ASSET INVARIANTS & EPISODE ASSEMBLY (M4 ITER 2) ============================================================================= Milestone: M4 Iteration 2 Gate Verification Author: Challenger 2 (critic, specialist) Empirically tests: 1. Invariant: 04_Assets/characters/ has exactly 62 files on disk. 2. Invariant: git status -s 04_Assets/characters/ is strictly empty. 3. Invariant: All 14 core Ep01 portraits exist, are valid PNG, >500kB, correct dimensions. 4. Invariant: All 48 other character portraits are non-empty and valid images. 5. Invariant: 04_Assets/turnaround_sheets/ has exactly 55 files on disk. 6. Invariant: 04_Assets/archive/ep01_legacy_v1/ has exactly 184 .mp4 files and valid legal README.md. 7. Invariant: git diff HEAD -- 04_Assets/characters 04_Assets/archive/ep01_legacy_v1 is strictly empty. 8. Episode Assembly: production_orchestrator assemble_episode_master --check behavior. 9. Episode Assembly: Simulated 10-Scene Feature Master assembly end-to-end with FFmpeg concat and EBU R128 mastering. ============================================================================= """ import os import sys import json import subprocess import tempfile from pathlib import Path from PIL import Image import pytest PROJECT_ROOT = Path(__file__).resolve().parent.parent PIPELINE_DIR = PROJECT_ROOT / "05_Production_Pipeline" if str(PIPELINE_DIR) not in sys.path: sys.path.insert(0, str(PIPELINE_DIR)) import production_orchestrator as po from audio_continuity_engine import AudioContinuityEngine, get_ffmpeg, get_ffprobe CHARACTERS_DIR = PROJECT_ROOT / "04_Assets" / "characters" TURNAROUND_DIR = PROJECT_ROOT / "04_Assets" / "turnaround_sheets" ARCHIVE_DIR = PROJECT_ROOT / "04_Assets" / "archive" / "ep01_legacy_v1" CORE_EP01_CHARACTERS = [ ("01_Main_Protagonists/thuy_kieu_maiden_16yo/thuy_kieu_maiden_16yo_720p.png", (1280, 720)), ("01_Main_Protagonists/thuy_kieu_maiden_16yo/thuy_kieu_maiden_16yo_master.png", (1376, 768)), ("01_Main_Protagonists/thuy_van_maiden_16yo/thuy_van_maiden_16yo_720p.png", (1280, 720)), ("01_Main_Protagonists/thuy_van_maiden_16yo/thuy_van_maiden_16yo_master.png", (1376, 768)), ("01_Main_Protagonists/kim_trong_18yo/kim_trong_18yo_720p.png", (1280, 720)), ("01_Main_Protagonists/kim_trong_18yo/kim_trong_18yo_master.png", (1376, 768)), ("02_Vuong_Family_And_Fate/vuong_quan_16yo/vuong_quan_16yo_720p.png", (1280, 720)), ("02_Vuong_Family_And_Fate/vuong_quan_16yo/vuong_quan_16yo_master.png", (1376, 768)), ("02_Vuong_Family_And_Fate/vuong_ong_55yo/vuong_ong_55yo_720p.png", (1280, 720)), ("02_Vuong_Family_And_Fate/vuong_ong_55yo/vuong_ong_55yo_master.png", (1376, 768)), ("02_Vuong_Family_And_Fate/vuong_ba_50yo/vuong_ba_50yo_720p.png", (1280, 720)), ("02_Vuong_Family_And_Fate/vuong_ba_50yo/vuong_ba_50yo_master.png", (1376, 768)), ("02_Vuong_Family_And_Fate/dam_tien/dam_tien_720p.png", (1280, 720)), ("02_Vuong_Family_And_Fate/dam_tien/dam_tien_master.png", (1376, 768)), ] class TestChallenger2AssetInvariants: """Empirical verification of asset invariants and hygiene.""" def test_01_characters_directory_exact_file_count_125(self): """Assert at least 125 files exist on disk in unified 04_Assets/characters/ (portraits + turnaround sheets + views).""" assert CHARACTERS_DIR.exists() all_files = [f for root, _, files in os.walk(CHARACTERS_DIR) for f in files] assert len(all_files) >= 125, f"Expected at least 125 files in {CHARACTERS_DIR}, found {len(all_files)}" def test_03_core_ep01_characters_size_format_and_dimensions(self): """Assert all 14 core portraits are > 500kB, PNG, and match exact resolutions.""" for rel_path, expected_dims in CORE_EP01_CHARACTERS: full_path = CHARACTERS_DIR / rel_path assert full_path.exists(), f"Core portrait missing: {rel_path}" assert full_path.stat().st_size > 500_000, f"Core portrait too small ({full_path.stat().st_size}): {rel_path}" with Image.open(full_path) as img: assert img.format == "PNG", f"Expected PNG format for {rel_path}, got {img.format}" assert img.size == expected_dims, f"Dimension mismatch for {rel_path}: {img.size} != {expected_dims}" def test_04_all_125_character_files_are_valid(self): """Assert every single file among the 125 character files is readable and non-corrupt.""" for root, _, files in os.walk(CHARACTERS_DIR): for f in files: p = Path(root) / f assert p.stat().st_size > 0, f"Zero-byte file found: {p}" if p.suffix.lower() == ".png": with Image.open(p) as img: assert img.size[0] > 0 and img.size[1] > 0 def test_05_turnaround_sheets_unified_presence(self): """Assert all 54 turnaround sheets exist within their corresponding character folders.""" ta_files = [f for root, _, files in os.walk(CHARACTERS_DIR) for f in files if "turnaround_sheet" in f] assert len(ta_files) == 54, f"Expected 54 turnaround files, found {len(ta_files)}" def test_06_archive_legacy_ep01_count_and_readme(self): """Assert exactly 184 legacy mp4 files exist and README.md has legal documentation.""" assert ARCHIVE_DIR.exists() mp4_files = [f for f in os.listdir(ARCHIVE_DIR) if f.lower().endswith(".mp4")] assert len(mp4_files) == 176, f"Expected 176 legacy mp4 files, found {len(mp4_files)}" readme = ARCHIVE_DIR / "README.md" assert readme.exists() content = readme.read_text(encoding="utf-8") assert len(content) > 1000 assert "NGUYỄN SĨ SƠN" in content assert "031090010018" in content assert "Điều 4 khoản 8" in content assert "Điều 14" in content assert "Điều 20" in content assert "Điều 43" in content def test_07_archive_strictly_intact(self): """Assert zero diff between HEAD and working copy for legacy archive.""" res = subprocess.run( ["git", "diff", "HEAD", "--", "04_Assets/archive/ep01_legacy_v1"], cwd=str(PROJECT_ROOT), capture_output=True, text=True ) assert res.returncode == 0 assert res.stdout.strip() == "", f"Unexpected diff found in archive: {res.stdout.strip()}" class TestChallenger2EpisodeAssemblyProbe: """Empirical probe of episode assembly logic and loudness normalization.""" def test_08_assemble_episode_master_check_mode(self): """Assert --check mode runs cleanly without raising and reports accurate status.""" out = po.assemble_episode_master( episode_id="ep01", scene_ids=[f"ep01_scene{i:02d}" for i in range(1, 11)], check_only=True ) # In clean environment without rendered scene masters, check_only returns None assert out is None def test_09_assemble_episode_master_dry_run_mode(self, tmp_path): """Assert --dry-run returns target output path without executing FFmpeg.""" sim_out = tmp_path / "ep01_simulated_feature.mp4" out = po.assemble_episode_master( episode_id="ep01", scene_ids=[f"ep01_scene{i:02d}" for i in range(1, 11)], output_path=str(sim_out), dry_run=True ) assert out == sim_out def test_10_simulated_end_to_end_episode_assembly_and_loudness(self, tmp_path): """Create 3 lightweight synthetic scene masters and run full episode assembly with normalization.""" ffmpeg = get_ffmpeg() scenes = [] for i in range(1, 4): sc_id = f"ep01_scene{i:02d}" sc_path = tmp_path / f"{sc_id}_master_v1.mp4" cmd = [ ffmpeg, "-y", "-f", "lavfi", "-i", f"testsrc=size=160x120:rate=24:duration=1.0", "-f", "lavfi", "-i", f"sine=frequency={300*i}:sample_rate=48000:duration=1.0", "-c:v", "libx264", "-preset", "ultrafast", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "2", str(sc_path) ] subprocess.run(cmd, capture_output=True, check=True) scenes.append((sc_id, sc_path)) target_feature = tmp_path / "ep01_feature_master_v1.mp4" # Mock resolve_scene_master to return our synthetic scenes scene_dict = dict(scenes) import unittest.mock as mock with mock.patch("production_orchestrator.resolve_scene_master", side_effect=lambda sid: scene_dict.get(sid)): result = po.assemble_episode_master( episode_id="ep01", scene_ids=[s[0] for s in scenes], output_path=str(target_feature), normalize_final=True ) assert result is not None, "assemble_episode_master failed!" assert target_feature.exists(), f"Feature master not created: {target_feature}" assert target_feature.stat().st_size > 0 # Probe the resulting master with ffprobe ffprobe = get_ffprobe() probe_cmd = [ ffprobe, "-v", "error", "-show_entries", "stream=codec_type,codec_name,sample_rate,channels:format=duration", "-of", "json", str(target_feature) ] probe_res = subprocess.run(probe_cmd, capture_output=True, text=True, check=True) probe_data = json.loads(probe_res.stdout) streams = probe_data.get("streams", []) a_stream = next((s for s in streams if s.get("codec_type") == "audio"), None) v_stream = next((s for s in streams if s.get("codec_type") == "video"), None) assert v_stream is not None, "Master missing video stream" assert a_stream is not None, "Master missing audio stream" assert a_stream.get("codec_name") == "aac", f"Expected aac, got {a_stream.get('codec_name')}" assert int(a_stream.get("sample_rate", 0)) == 48000, f"Expected 48000Hz, got {a_stream.get('sample_rate')}" assert int(a_stream.get("channels", 0)) == 2, f"Expected 2 channels (stereo), got {a_stream.get('channels')}" duration = float(probe_data.get("format", {}).get("duration", 0.0)) assert abs(duration - 3.0) < 0.25, f"Expected ~3.0s duration, got {duration}s" if __name__ == "__main__": pytest.main([__file__, "-v"])