""" ============================================================================= THẬP NGŨ NIÊN (THE FIFTEEN SPRINGS) - PRODUCTION PIPELINE TEST SUITE (M3) ============================================================================= Module: tests/test_production_pipeline_m3.py Milestone: M3 (Ep01 10-Scene Production Pipeline Implementation) Standard: AGENTS.md, GEMINI.md, ORIGINAL_REQUEST.md §R1-R3, PROJECT.md Covers: 1. 100% Start Frame resolution across all 140 shots (Ep01 Scenes 01 to 10). 2. Character invariant check (0 character misidentifications). 3. Shot Gate integration and automated retake loop (threshold 0.8, max_retakes=2). 4. Scene Gate integration and junction validation. 5. Audio Guard directive enforcement (100% No-BGM compliance). 6. Dry-run pipeline end-to-end execution with Head-Tail chaining and critic gate. ============================================================================= """ import os import sys import json import shutil from pathlib import Path from unittest.mock import patch, MagicMock import pytest import cv2 PROJECT_ROOT = Path(__file__).resolve().parent.parent PIPELINE_DIR = PROJECT_ROOT / "05_Production_Pipeline" VIDEOS_DIR = PROJECT_ROOT / "04_Assets" / "videos" KEYFRAMES_DIR = PROJECT_ROOT / "04_Assets" / "keyframes" CHARACTERS_DIR = PROJECT_ROOT / "04_Assets" / "characters" BACKGROUNDS_DIR = PROJECT_ROOT / "04_Assets" / "backgrounds" EP01_PROMPTS_FILE = PROJECT_ROOT / "episodes" / "ep01" / "prompts" / "muse_prompts.json" if str(PIPELINE_DIR) not in sys.path: sys.path.insert(0, str(PIPELINE_DIR)) import production_orchestrator as po from run_shot import run_shot_pipeline, ensure_audio_guard, AUDIO_GUARD_CANONICAL from antigravity_critic_gate import VideoCriticVerdict, ShotEvaluation, SceneEvaluation @pytest.fixture(scope="module") def ep01_shots_data(): """Tải toàn bộ dữ liệu shot của Ep01.""" shots = po.get_all_shots("ep01") scenes = [f"ep01_scene{i:02d}" for i in range(1, 11)] filtered = {} for sid, sdata in shots.items(): sc = sid.split("_shot")[0] if sc in scenes: filtered[sid] = sdata return filtered # ============================================================================= # TEST 1: 100% START FRAME COVERAGE ACROSS 140 SHOTS # ============================================================================= class TestStartFrameCoverage: """Kiểm tra độ bao phủ 100% của Start Frame cho 140 shots thuộc Scenes 01 đến 10.""" def test_01_all_140_shots_exist_in_manifest(self, ep01_shots_data): """Xác nhận có đúng 140 shots trải dài từ Scene 01 đến Scene 10.""" assert len(ep01_shots_data) == 140, f"Kỳ vọng 140 shots, tìm thấy: {len(ep01_shots_data)}" def test_01_all_140_shots_start_frame_resolution(self, ep01_shots_data): """Xác nhận 100% trong số 140 shots đều resolve thành công ra file ảnh tồn tại trên đĩa.""" unresolved = [] invalid_files = [] for sid, sdata in ep01_shots_data.items(): resolved_frame, take_type = po.resolve_start_frame_v2(sid, sdata) if not resolved_frame: unresolved.append((sid, take_type, "RETURNED_NONE")) continue p = Path(resolved_frame) if not p.exists(): unresolved.append((sid, take_type, f"FILE_NOT_FOUND: {resolved_frame}")) continue if not p.is_file() or p.stat().st_size < 100: invalid_files.append((sid, resolved_frame, f"SIZE_TOO_SMALL: {p.stat().st_size}")) assert len(unresolved) == 0, f"Có {len(unresolved)} shots không resolve được Start Frame: {unresolved}" assert len(invalid_files) == 0, f"Có {len(invalid_files)} file ảnh không hợp lệ: {invalid_files}" def test_01_all_resolved_frames_are_valid_images(self, ep01_shots_data): """Kiểm tra toàn bộ các start frames resolve được đều là ảnh đọc được với kích thước chuẩn.""" checked_paths = set() for sid, sdata in ep01_shots_data.items(): resolved_frame = po.resolve_start_frame(sid, sdata) assert resolved_frame is not None if resolved_frame in checked_paths: continue checked_paths.add(resolved_frame) img = cv2.imread(resolved_frame) assert img is not None, f"Không thể đọc file ảnh: {resolved_frame}" h, w = img.shape[:2] assert h >= 700 and w >= 1200, f"Kích thước ảnh không đạt chuẩn 720p: ({w}x{h}) tại {resolved_frame}" # ============================================================================= # TEST 2: CHARACTER INVARIANT CHECK (0 CHARACTER MISIDENTIFICATIONS) # ============================================================================= class TestCharacterInvariantSafeguards: """Kiểm tra bất biến nhân vật: tuyệt đối không nhầm lẫn chân dung Thúy Kiều cho vai khác.""" def test_02_kim_trong_canonical_shots_safeguard(self, ep01_shots_data): """Kiểm tra toàn bộ các shot Kim Trọng không bao giờ bị trả về ảnh Thúy Kiều.""" kim_trong_shots = [ "ep01_scene07_shot01", "ep01_scene07_shot03", "ep01_scene07_shot04", "ep01_scene07_shot05", "ep01_scene07_shot06", "ep01_scene07_shot08", "ep01_scene08_shot04", "ep01_scene08_shot05", "ep01_scene08_shot07", "ep01_scene09_shot01", "ep01_scene09_shot04", "ep01_scene10_shot04", "ep01_scene10_shot05", "ep01_scene10_shot13" ] for sid in kim_trong_shots: sdata = ep01_shots_data.get(sid, {}) resolved = po.resolve_start_frame(sid, sdata) assert resolved is not None r_lower = resolved.lower() assert "thuy_kieu" not in r_lower, f"LEAK: Kim Trọng shot {sid} bị gán Thúy Kiều: {resolved}" assert "kim_trong" in r_lower, f"Kim Trọng shot {sid} kỳ vọng chân dung Kim Trọng, nhận: {resolved}" def test_02_vuong_family_shots_safeguard(self, ep01_shots_data): """Kiểm tra các shot Vương Ông, Vương Quan không bao giờ bị trả về ảnh Thúy Kiều.""" vuong_ong_shots = ["ep01_scene02_shot04", "ep01_scene02_shot07", "ep01_scene02_shot10", "ep01_scene02_shot13"] for sid in vuong_ong_shots: sdata = ep01_shots_data.get(sid, {}) resolved = po.resolve_start_frame(sid, sdata) assert resolved is not None assert "thuy_kieu" not in resolved.lower(), f"LEAK: Vương Ông shot {sid} bị gán Thúy Kiều: {resolved}" assert "vuong_ong" in resolved.lower() vuong_quan_shots = ["ep01_scene02_shot08", "ep01_scene02_shot09", "ep01_scene02_shot12", "ep01_scene02_shot15", "ep01_scene05_shot12"] for sid in vuong_quan_shots: sdata = ep01_shots_data.get(sid, {}) resolved = po.resolve_start_frame(sid, sdata) assert resolved is not None assert "thuy_kieu" not in resolved.lower(), f"LEAK: Vương Quan shot {sid} bị gán Thúy Kiều: {resolved}" assert "vuong_quan" in resolved.lower() def test_02_crowd_and_scenery_shots_safeguard(self, ep01_shots_data): """Kiểm tra cảnh đại chúng Hội Thanh Minh và các ngoại cảnh không bị gán Thúy Kiều.""" crowd_shots = ["ep01_scene04_shot01", "ep01_scene04_shot02", "ep01_scene04_shot03", "ep01_scene04_shot04", "ep01_scene04_shot12"] for sid in crowd_shots: sdata = ep01_shots_data.get(sid, {}) resolved = po.resolve_start_frame(sid, sdata) assert resolved is not None assert "crowd_qingming_festival" in resolved.lower(), f"Cảnh đại chúng {sid} kỳ vọng crowd keyframe, nhận: {resolved}" scenery_shots = { "ep01_scene05_shot07": "suoi_ngoc_rang_lieu", "ep01_scene05_shot10": "mo_dam_tien", "ep01_scene06_shot01": "hoang_hon_rang_lieu", "ep01_scene06_shot02": "duong_mon_hoang_hon", "ep01_scene08_shot01": "buc_tuong_hoa", } for sid, expected_key in scenery_shots.items(): sdata = ep01_shots_data.get(sid, {}) resolved = po.resolve_start_frame(sid, sdata) assert resolved is not None assert expected_key in resolved.lower(), f"Ngoại cảnh {sid} kỳ vọng {expected_key}, nhận: {resolved}" # ============================================================================= # TEST 3: SHOT GATE INTEGRATION & AUTOMATED RETAKE LOOP # ============================================================================= class TestShotGateAndRetakeLoop: """Kiểm tra tích hợp Shot Gate và vòng lặp retake tự động trong production_orchestrator.py.""" def test_03_shot_gate_approves_when_score_above_threshold(self, tmp_path): """Xác nhận Shot Gate phê duyệt khi overall_score >= 0.8.""" mock_verdict = VideoCriticVerdict( overall_score=0.92, approved=True, shot_eval=ShotEvaluation(character_match=True, visual_defects=[], score=0.95), suggested_action="APPROVE", critique_notes="Tạo hình chuẩn xác, không dị tật." ) with patch("production_orchestrator.run_shot_pipeline", return_value=(True, "tail.jpg")), \ patch("production_orchestrator.find_rendered_video", return_value=Path("mock_video.mp4")), \ patch("production_orchestrator._extract_tail_frame", return_value=Path("tail.jpg")), \ patch("production_orchestrator.evaluate_shot_gate", return_value=mock_verdict) as mock_gate: res = po.render_single_shot("ep01_scene02_shot01", enable_critic=True, max_retakes=2) assert res is True assert mock_gate.call_count == 1 def test_03_shot_gate_triggers_retake_loop_on_defect(self): """Xác nhận Shot Gate kích hoạt vòng lặp retake khi score < 0.8 hoặc suggested_action == RETAKE_SHOT.""" bad_verdict = VideoCriticVerdict( overall_score=0.65, approved=False, shot_eval=ShotEvaluation(character_match=False, visual_defects=["squint"], score=0.60), suggested_action="RETAKE_SHOT", critique_notes="Phát hiện mắt lé nhân vật." ) good_verdict = VideoCriticVerdict( overall_score=0.88, approved=True, shot_eval=ShotEvaluation(character_match=True, visual_defects=[], score=0.90), suggested_action="APPROVE", critique_notes="Retake hoàn hảo." ) with patch("production_orchestrator.run_shot_pipeline", return_value=(True, "tail.jpg")) as mock_pipeline, \ patch("production_orchestrator.find_rendered_video", return_value=Path("mock_video.mp4")), \ patch("production_orchestrator._extract_tail_frame", return_value=Path("tail.jpg")), \ patch("production_orchestrator.evaluate_shot_gate", side_effect=[bad_verdict, good_verdict]) as mock_gate: res = po.render_single_shot("ep01_scene02_shot01", enable_critic=True, max_retakes=2) assert res is True assert mock_gate.call_count == 2 assert mock_pipeline.call_count == 2 def test_03_shot_gate_fails_after_max_retakes_exceeded(self): """Xác nhận Shot Gate trả về False nếu vượt quá max_retakes mà vẫn không đạt điểm.""" bad_verdict = VideoCriticVerdict( overall_score=0.55, approved=False, shot_eval=ShotEvaluation(character_match=False, visual_defects=["severe_morphing"], score=0.50), suggested_action="RETAKE_SHOT", critique_notes="Biến dạng khuôn mặt nghiêm trọng." ) with patch("production_orchestrator.run_shot_pipeline", return_value=(True, "tail.jpg")) as mock_pipeline, \ patch("production_orchestrator.find_rendered_video", return_value=Path("mock_video.mp4")), \ patch("production_orchestrator.evaluate_shot_gate", return_value=bad_verdict) as mock_gate: res = po.render_single_shot("ep01_scene02_shot01", enable_critic=True, max_retakes=2) assert res is False # 1 initial try + 2 retakes = 3 total attempts assert mock_gate.call_count == 3 assert mock_pipeline.call_count == 3 # ============================================================================= # TEST 4: SCENE GATE INTEGRATION & JUNCTION CONTINUITY # ============================================================================= class TestSceneGateIntegration: """Kiểm tra tích hợp Scene Gate trong batch_render_scene và concat_scene_shots.""" def test_04_scene_gate_validates_continuity_before_concat(self, tmp_path): """Xác nhận evaluate_scene_gate được gọi thẩm định tiếp biên trước khi xuất Master.""" mock_scene_verdict = VideoCriticVerdict( overall_score=0.88, approved=True, scene_eval=SceneEvaluation(junction_smoothness=0.9, axis_180_ok=True, color_continuity=0.9, score=0.9), suggested_action="APPROVE", critique_notes="Tiếp biên mượt mà, trục 180 độ an toàn." ) out_file = tmp_path / "ep01_scene02_master_v1.mp4" def mock_stitch(paths, target, **kwargs): Path(target).write_bytes(b"dummy master video content") return True mock_videos = [Path(f"mock_s{i}.mp4") for i in range(1, 4)] with patch("production_orchestrator.get_shots_for_scene", return_value=[(f"s{i}", {}) for i in range(1, 4)]), \ patch("production_orchestrator.find_rendered_video", side_effect=mock_videos), \ patch("production_orchestrator.evaluate_scene_gate", return_value=mock_scene_verdict) as mock_gate, \ patch("production_orchestrator.AudioContinuityEngine") as mock_engine_cls: mock_engine = mock_engine_cls.return_value mock_engine.stitch_with_audio_crossfade.side_effect = mock_stitch res = po.concat_scene_shots("ep01_scene02", output_path=str(out_file), enable_critic=True) assert res is not None assert mock_gate.call_count == 1 assert out_file.exists() def test_04_scene_gate_rejects_concat_on_critical_junction_error(self): """Xác nhận concat_scene_shots bị hủy bỏ nếu Scene Gate từ chối (RETAKE_SHOT).""" reject_verdict = VideoCriticVerdict( overall_score=0.50, approved=False, scene_eval=SceneEvaluation(junction_smoothness=0.3, axis_180_ok=False, color_continuity=0.4, score=0.4), suggested_action="RETAKE_SHOT", critique_notes="Nhảy trục 180 độ nghiêm trọng tại shot 2 và shot 3." ) mock_videos = [Path(f"mock_s{i}.mp4") for i in range(1, 4)] with patch("production_orchestrator.get_shots_for_scene", return_value=[(f"s{i}", {}) for i in range(1, 4)]), \ patch("production_orchestrator.find_rendered_video", side_effect=mock_videos), \ patch("production_orchestrator.evaluate_scene_gate", return_value=reject_verdict) as mock_gate: res = po.concat_scene_shots("ep01_scene02", enable_critic=True) assert res is None assert mock_gate.call_count == 1 # ============================================================================= # TEST 5: AUDIO GUARD DIRECTIVE ENFORCEMENT # ============================================================================= class TestAudioGuardEnforcement: """Kiểm tra tuân thủ 100% chỉ thị Audio Guard cấm nhạc nền (No-BGM) theo AGENTS.md §3.""" def test_05_audio_guard_formatting_compliance(self): """Xác nhận Audio Guard chứa đúng tiền tố và nội dung quy chuẩn AGENTS.md §3.""" assert "Quy tắc âm thanh:" in AUDIO_GUARD_CANONICAL assert "Tuyệt đối KHÔNG sinh nhạc nền (no music/BGM)" in AUDIO_GUARD_CANONICAL assert "không âm thanh điện tử, không tạp âm rè nhiễu" in AUDIO_GUARD_CANONICAL assert "Chỉ sinh âm thanh môi trường tự nhiên (foley, ambience) và thoại nhân vật chân thực" in AUDIO_GUARD_CANONICAL def test_05_all_ep01_prompts_strictly_enforce_no_bgm(self): """Xác nhận 100% prompt trong episodes/ep01/prompts/muse_prompts.json có Audio Guard.""" with open(EP01_PROMPTS_FILE, "r", encoding="utf-8") as f: data = json.load(f) motion_prompts = data.get("motion_prompts", {}) missing_guard = [] for sid, sdata in motion_prompts.items(): prompt = sdata.get("motion_prompt", "") if not po.validate_audio_guard(prompt): missing_guard.append(sid) assert len(missing_guard) == 0, f"Có {len(missing_guard)} prompts thiếu Audio Guard: {missing_guard}" def test_05_ensure_audio_guard_helper_appends_canonical_guard(self): """Kiểm tra helper ensure_audio_guard tự động gắn Audio Guard vào prompt trần.""" raw_prompt = "Camera lia chậm cận cảnh nàng Kiều gảy đàn tỳ bà." guarded = po.ensure_audio_guard(raw_prompt) assert po.validate_audio_guard(guarded) assert guarded.endswith(AUDIO_GUARD_CANONICAL) # Idempotency check: Gắn 2 lần không bị lặp double_guarded = po.ensure_audio_guard(guarded) assert double_guarded == guarded # ============================================================================= # TEST 6: DRY-RUN PIPELINE EXECUTION & HEAD-TAIL CHAINING # ============================================================================= class TestDryRunPipelineEndToEnd: """Kiểm thử chu trình render giả lập (Dry-Run) không treo browser trong < 15 giây.""" def test_06_dry_run_generates_valid_720p_video_and_tail_frame(self, tmp_path): """Xác nhận chế độ MUSE_DRY_RUN=1 sinh video 720p có audio AAC và trích Tail Frame thành công.""" test_shot_id = "test_dry_run_shot01" try: ok, tail_path = run_shot_pipeline( shot_id=test_shot_id, prompt="Dry run test shot", dry_run=True ) assert ok is True assert tail_path is not None assert Path(tail_path).exists() # Kiểm tra video sinh ra target_video = po.find_rendered_video(test_shot_id) assert target_video is not None assert target_video.exists() cap = cv2.VideoCapture(str(target_video)) assert cap.isOpened() w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) cap.release() assert w == 1280 and h == 720, f"Kích thước video không đúng 720p: {w}x{h}" # Kiểm tra tail frame tail_img = cv2.imread(str(tail_path)) assert tail_img is not None assert tail_img.shape[:2] == (720, 1280) finally: # Teardown để giữ workspace sạch sẽ cho test_m2_hygiene for f in VIDEOS_DIR.glob(f"{test_shot_id}*"): try: f.unlink() except Exception: pass kf_test = KEYFRAMES_DIR / test_shot_id if kf_test.exists(): shutil.rmtree(kf_test, ignore_errors=True) def test_06_dry_run_head_tail_chaining_and_critic_approval(self): """Xác nhận chuỗi Head-Tail liên hoàn giữa 2 shot: Shot 1 tạo frame 239 -> Shot 2 nhận làm start frame.""" shot1_id = "test_chain_shot01" shot2_id = "test_chain_shot02" try: # Render Shot 1 qua dry-run ok1, tail1 = run_shot_pipeline(shot1_id, prompt="Shot 1 take", dry_run=True) assert ok1 is True assert Path(tail1).exists() # Shot 2 là CONTINUOUS_TAKE kế thừa tail1 shot2_data = { "character_anchor": "thuy_kieu", "motion_prompt": "Shot 2 continuation." } resolved_frame, take_type = po.resolve_start_frame( shot2_id, shot_data=shot2_data, return_classification=True ) # Find tail of shot 1 tail_shot1 = po.find_tail_frame(shot1_id) assert tail_shot1 is not None assert str(tail_shot1.resolve()) == str(Path(tail1).resolve()) finally: # Cleanup test artifacts for sid in (shot1_id, shot2_id): for f in VIDEOS_DIR.glob(f"{sid}*"): try: f.unlink() except Exception: pass kf_test = KEYFRAMES_DIR / sid if kf_test.exists(): shutil.rmtree(kf_test, ignore_errors=True)