""" ============================================================================= THẬP NGŨ NIÊN (THE FIFTEEN SPRINGS) - CHALLENGER 1 EMPIRICAL GATE VERIFICATION ============================================================================= Module: tests/test_m3_challenger1_probe.py Milestone: M3 Gate Verification (Challenger 1 Empirical Challenge) Author: Challenger 1 (Adversarial Critic) Mandatory Verification: 1. Empirically probe Start Frame resolution across all 140 shots of Ep01 Scenes 01 to 10: - Every shot must resolve to a physical file existing on disk. - Every file must be a valid readable image >= 720p. - Character safeguards must strictly hold across all 140 shots. 2. Empirically challenge the Shot Gate retake loop with real synthetic files: - Score < 0.8 retakes with incremented versions (_v1, _v2, _v3). - Terminates with True on score >= 0.8. - Terminates with False on max_retakes exhausted. - suggested_action == 'RETAKE_SHOT' overrides score >= 0.8. ============================================================================= """ import os import sys import json import shutil import cv2 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" KEYFRAMES_DIR = PROJECT_ROOT / "04_Assets" / "keyframes" CHARACTERS_DIR = PROJECT_ROOT / "04_Assets" / "characters" 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_140_shots(): """Load all shots for Ep01 Scenes 01 to 10.""" shots = po.get_all_shots("ep01") scenes = [f"ep01_scene{i:02d}" for i in range(1, 11)] filtered = {sid: sdata for sid, sdata in shots.items() if sid.split("_shot")[0] in scenes} assert len(filtered) == 140, f"Expected exactly 140 shots across Scenes 01 to 10, found {len(filtered)}" return filtered # ============================================================================= # CHALLENGE 1: 140-SHOT START FRAME RESOLUTION & INTEGRITY PROBE # ============================================================================= class TestAll140ShotsStartFrameResolution: """Rigorous empirical probe of start frame resolution for all 140 shots.""" def test_probe_140_shots_resolve_to_existing_physical_files(self, ep01_140_shots): """Every shot must resolve to a physical file that exists on disk.""" unresolved = [] non_existent = [] zero_byte = [] for sid, sdata in sorted(ep01_140_shots.items()): resolved_frame, take_type = po.resolve_start_frame_v2(sid, sdata) if not resolved_frame: unresolved.append((sid, take_type)) continue p = Path(resolved_frame) if not p.exists() or not p.is_file(): non_existent.append((sid, resolved_frame)) continue if p.stat().st_size == 0: zero_byte.append((sid, resolved_frame)) assert len(unresolved) == 0, f"Unresolved shots: {unresolved}" assert len(non_existent) == 0, f"Non-existent files resolved: {non_existent}" assert len(zero_byte) == 0, f"Zero-byte files resolved: {zero_byte}" def test_probe_all_resolved_files_are_valid_720p_images(self, ep01_140_shots): """Every resolved file must be a valid readable image >= 720p.""" tested_files = {} for sid, sdata in ep01_140_shots.items(): rf = po.resolve_start_frame(sid, sdata) assert rf is not None, f"Shot {sid} returned None" if rf not in tested_files: img = cv2.imread(rf) assert img is not None, f"Image could not be decoded by OpenCV: {rf}" h, w = img.shape[:2] assert h >= 700 and w >= 1200, f"Image {rf} is smaller than 720p standard ({w}x{h})" tested_files[rf] = (w, h) assert len(tested_files) >= 20, f"Expected at least 20 unique high-quality assets, got {len(tested_files)}" def test_probe_scene_openers_are_always_cinematic_cuts(self, ep01_140_shots): """Shot 01 of every scene (01-10) must be CINEMATIC_CUT and never link to a prev_tail.""" for sc in range(1, 11): sid = f"ep01_scene{sc:02d}_shot01" sdata = ep01_140_shots.get(sid, {}) rf, take_type = po.resolve_start_frame_v2(sid, sdata) assert take_type == "CINEMATIC_CUT", f"Scene opener {sid} classified as {take_type}, expected CINEMATIC_CUT" assert "clean_frame_239" not in rf.lower(), f"Scene opener {sid} improperly links to prev_tail: {rf}" def test_probe_character_safeguards_canonical_character_shots(self, ep01_140_shots): """Ensure canonical character shots resolve strictly to their own portraits and never leak Thúy Kiều.""" # Canonical Kim Trọng solo shots 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_140_shots.get(sid, {}) rf = po.resolve_start_frame(sid, sdata) assert rf is not None, f"Shot {sid} returned None" rf_lower = rf.lower() assert "thuy_kieu" not in rf_lower, f"LEAK: Kim Trọng shot {sid} leaked Thúy Kiều: {rf}" assert "kim_trong" in rf_lower, f"Kim Trọng shot {sid} expected Kim Trọng portrait, got: {rf}" # Canonical Vương Ông solo shots vuong_ong_shots = [ "ep01_scene02_shot04", "ep01_scene02_shot07", "ep01_scene02_shot10", "ep01_scene02_shot13" ] for sid in vuong_ong_shots: sdata = ep01_140_shots.get(sid, {}) rf = po.resolve_start_frame(sid, sdata) assert rf is not None, f"Shot {sid} returned None" rf_lower = rf.lower() assert "thuy_kieu" not in rf_lower, f"LEAK: Vương Ông shot {sid} leaked Thúy Kiều: {rf}" assert "vuong_ong" in rf_lower, f"Vương Ông shot {sid} expected Vương Ông portrait, got: {rf}" # Canonical Vương Quan solo shots 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_140_shots.get(sid, {}) rf = po.resolve_start_frame(sid, sdata) assert rf is not None, f"Shot {sid} returned None" rf_lower = rf.lower() assert "thuy_kieu" not in rf_lower, f"LEAK: Vương Quan shot {sid} leaked Thúy Kiều: {rf}" assert "vuong_quan" in rf_lower, f"Vương Quan shot {sid} expected Vương Quan portrait, got: {rf}" # Canonical Thúy Vân solo shots (currently properly resolved) thuy_van_passing_shots = [ "ep01_scene03_shot02", "ep01_scene03_shot04", "ep01_scene04_shot09" ] for sid in thuy_van_passing_shots: sdata = ep01_140_shots.get(sid, {}) rf = po.resolve_start_frame(sid, sdata) assert rf is not None, f"Shot {sid} returned None" rf_lower = rf.lower() assert "thuy_kieu" not in rf_lower, f"LEAK: Thúy Vân shot {sid} leaked Thúy Kiều: {rf}" assert "thuy_van" in rf_lower, f"Thúy Vân shot {sid} expected Thúy Vân portrait, got: {rf}" # Canonical Crowd Qingming Festival shots crowd_shots = [ "ep01_scene04_shot01", "ep01_scene04_shot02", "ep01_scene04_shot03", "ep01_scene04_shot04", "ep01_scene04_shot12" ] for sid in crowd_shots: sdata = ep01_140_shots.get(sid, {}) rf = po.resolve_start_frame(sid, sdata) assert rf is not None, f"Shot {sid} returned None" rf_lower = rf.lower() assert "thuy_kieu" not in rf_lower, f"LEAK: Crowd shot {sid} leaked Thúy Kiều: {rf}" assert "crowd_qingming" in rf_lower, f"Crowd shot {sid} expected crowd asset, got: {rf}" # Canonical Scenery / Environment shots 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_140_shots.get(sid, {}) rf = po.resolve_start_frame(sid, sdata) assert rf is not None, f"Shot {sid} returned None" rf_lower = rf.lower() assert "thuy_kieu" not in rf_lower, f"LEAK: Scenery shot {sid} leaked Thúy Kiều: {rf}" assert expected_key in rf_lower, f"Scenery shot {sid} expected {expected_key}, got: {rf}" def test_empirical_defect_scene03_thuy_van_shots_leak_thuy_kieu(self, ep01_140_shots): """ EMPIRICAL BUG REPRODUCTION: In Scene 03 (NỘI. KHUÊ PHÒNG THÚY KIỀU & THÚY VÂN): - Shot 03 (Đặc tả vẻ đẹp Thúy Vân / Close-Up Portrait Thúy Vân) resolves to Thúy Kiều. - Shot 05 (Thúy Vân đứng dậy gọi chị / Medium Shot Thúy Vân) resolves to Thúy Kiều. Both shots must resolve to Thúy Vân portrait (thuy_van_maiden_16yo_720p.png). """ for sid in ["ep01_scene03_shot03", "ep01_scene03_shot05"]: sdata = ep01_140_shots.get(sid, {}) rf = po.resolve_start_frame(sid, sdata) assert rf is not None, f"Shot {sid} returned None" rf_lower = rf.lower() assert "thuy_kieu" not in rf_lower, f"DEFECT CONFIRMED: Shot {sid} leaked Thúy Kiều: {rf}" assert "thuy_van" in rf_lower, f"Shot {sid} expected Thúy Vân portrait, got: {rf}" def test_probe_adversarial_bilateral_safeguards_override(self): """Adversarial test: even if prompt data contains corrupted character_asset_ref pointing to Thúy Kiều, safeguard overrides it.""" corrupted_cases = [ ("ep01_scene07_shot01", {"character_anchor": "kim_trong", "character_asset_ref": "04_Assets/characters/01_Main_Protagonists/thuy_kieu_maiden_16yo_720p.png"}, "kim_trong"), ("ep01_scene02_shot04", {"character_anchor": "vuong_ong", "character_asset_ref": "04_Assets/characters/01_Main_Protagonists/thuy_kieu_maiden_16yo_720p.png"}, "vuong_ong"), ("ep01_scene02_shot08", {"character_anchor": "vuong_quan", "character_asset_ref": "04_Assets/characters/01_Main_Protagonists/thuy_kieu_maiden_16yo_720p.png"}, "vuong_quan"), ("ep01_scene03_shot05", {"character_anchor": "thuy_van", "character_asset_ref": "04_Assets/characters/01_Main_Protagonists/thuy_kieu_maiden_16yo_720p.png"}, "thuy_van"), ("ep01_scene02_shot01", {"character_anchor": "thuy_kieu", "character_asset_ref": "04_Assets/characters/01_Main_Protagonists/kim_trong_18yo_720p.png"}, "thuy_kieu"), ] for shot_id, corrupted_data, expected_char in corrupted_cases: resolved = po.resolve_start_frame(shot_id, corrupted_data) assert resolved is not None r_lower = resolved.lower() assert expected_char in r_lower, f"Adversarial override failed for {shot_id}: expected {expected_char}, got {resolved}" if expected_char != "thuy_kieu": assert "thuy_kieu" not in r_lower, f"Adversarial leak of Thúy Kiều into {shot_id}: {resolved}" def test_classify_shot_take_actor_switches_are_cinematic_cuts(self, ep01_140_shots): """ EMPIRICAL PROBE: Actor switches between distinct characters must strictly be classified as CINEMATIC_CUT. - Scene 03 Shot 05 (Thúy Vân) -> Shot 06 (Thúy Kiều) - Scene 02 Shot 10 (Vương Ông) -> Shot 11 (Vương Quan) - Scene 08 Shot 07 (Kim Trọng) -> Shot 08 (Thúy Kiều) - 140 shots transition matrix audit across Scenes 01 to 10 (0 actor-switch mismatches). """ # 1. Shot 06: Thúy Vân -> Thúy Kiều assert po.classify_shot_take("ep01_scene03_shot06", shot_data=ep01_140_shots.get("ep01_scene03_shot06")) == "CINEMATIC_CUT" assert po.classify_shot_take("ep01_scene03_shot06") == "CINEMATIC_CUT" # 2. Shot 11: Vương Ông -> Vương Quan assert po.classify_shot_take("ep01_scene02_shot11", shot_data=ep01_140_shots.get("ep01_scene02_shot11")) == "CINEMATIC_CUT" assert po.classify_shot_take("ep01_scene02_shot11") == "CINEMATIC_CUT" # 3. Shot 08: Kim Trọng -> Thúy Kiều assert po.classify_shot_take("ep01_scene08_shot08", shot_data=ep01_140_shots.get("ep01_scene08_shot08")) == "CINEMATIC_CUT" assert po.classify_shot_take("ep01_scene08_shot08") == "CINEMATIC_CUT" # 4. Full 140 shots transition matrix across all 10 Scenes scenes = [f"ep01_scene{i:02d}" for i in range(1, 11)] mismatches = [] for sc in scenes: sc_shots = sorted([s for s in ep01_140_shots if s.startswith(sc + "_")]) for i in range(1, len(sc_shots)): p_id, c_id = sc_shots[i - 1], sc_shots[i] p_ref = ep01_140_shots[p_id].get("character_asset_ref", "") c_ref = ep01_140_shots[c_id].get("character_asset_ref", "") tt = po.classify_shot_take(c_id, prev_shot_id=p_id, shot_data=ep01_140_shots[c_id]) if p_ref and c_ref: p_group = po.Path(p_ref).stem.lower() c_group = po.Path(c_ref).stem.lower() for k in ["thuy_van", "thuy_kieu", "kim_trong", "vuong_ong", "vuong_ba", "vuong_quan", "dam_tien", "tieu_dong"]: if k in p_group: p_group = k if k in c_group: c_group = k if p_group != c_group and tt == "CONTINUOUS_TAKE": mismatches.append(f"{p_id} -> {c_id} ({p_group} -> {c_group})") assert len(mismatches) == 0, f"Actor switch mismatches found: {mismatches}" def test_simulated_tail_frame_actor_switch_prevents_face_leak(self, ep01_140_shots): """ EMPIRICAL PROBE: Simulate a physical tail frame (clean_frame_239.jpg) on disk for ep01_scene03_shot05 (Thúy Vân). Verify that resolving ep01_scene03_shot06 (Thúy Kiều) does NOT consume Thúy Vân's tail frame, classifies as CINEMATIC_CUT, and strictly resolves to Thúy Kiều's master portrait. """ kf5 = po.KEYFRAMES_DIR / "ep01_scene03_shot05" kf5.mkdir(parents=True, exist_ok=True) tail5 = kf5 / "clean_frame_239.jpg" tail5.write_bytes(b"dummy_thuy_van_tail_frame") try: rf_06, take_type = po.resolve_start_frame_v2("ep01_scene03_shot06", ep01_140_shots["ep01_scene03_shot06"]) assert take_type == "CINEMATIC_CUT", f"Expected CINEMATIC_CUT, got {take_type}" assert rf_06 is not None, "Resolved start frame is None" assert "clean_frame_239" not in rf_06.lower(), f"Face leak detected: consumed tail frame {rf_06}" assert "thuy_van" not in rf_06.lower(), f"Face leak detected: consumed Thúy Vân asset {rf_06}" assert "thuy_kieu" in rf_06.lower(), f"Expected Thúy Kiều portrait, got {rf_06}" finally: shutil.rmtree(kf5, ignore_errors=True) # ============================================================================= # CHALLENGE 2: EMPIRICAL SHOT GATE RETAKE LOOP STRESS HARNESS # ============================================================================= class TestEmpiricalShotGateRetakeLoopStressHarness: """ Stress-test the Shot Gate retake loop using actual video generation (dry_run=True) and empirical verification of versioned video files on disk. """ def test_retake_loop_eventual_approval_with_incremented_versions(self): """ Scenario: - Attempt 1: Critic gives score 0.60 (RETAKE_SHOT) -> generates _v1.mp4. - Attempt 2: Critic gives score 0.72 (RETAKE_SHOT) -> generates _v2.mp4 with updated critique prompt. - Attempt 3: Critic gives score 0.88 (APPROVE) -> generates _v3.mp4 and succeeds. Empirical check: - render_single_shot returns True. - _v1.mp4, _v2.mp4, and _v3.mp4 physically exist in VIDEOS_DIR. - clean_frame_239.jpg physically exists in KEYFRAMES_DIR. - Critic was called exactly 3 times. """ test_shot_id = "test_retake_eventual_pass" try: # Clean before test for f in VIDEOS_DIR.glob(f"{test_shot_id}*"): f.unlink(missing_ok=True) kf_dir = KEYFRAMES_DIR / test_shot_id if kf_dir.exists(): shutil.rmtree(kf_dir, ignore_errors=True) verdict_1 = VideoCriticVerdict( overall_score=0.60, approved=False, shot_eval=ShotEvaluation(character_match=False, visual_defects=["blur"], score=0.60), suggested_action="RETAKE_SHOT", critique_notes="Cần tăng độ nét khuôn mặt nhân vật." ) verdict_2 = VideoCriticVerdict( overall_score=0.72, approved=False, shot_eval=ShotEvaluation(character_match=True, visual_defects=["jitter"], score=0.72), suggested_action="RETAKE_SHOT", critique_notes="Chuyển động còn hơi giật ở biên khung hình." ) verdict_3 = VideoCriticVerdict( overall_score=0.88, approved=True, shot_eval=ShotEvaluation(character_match=True, visual_defects=[], score=0.90), suggested_action="APPROVE", critique_notes="Đạt chuẩn điện ảnh." ) prompts_received = [] def mock_gate(shot_id, video_path, expected_character, prompt): prompts_received.append(prompt) if len(prompts_received) == 1: return verdict_1 elif len(prompts_received) == 2: return verdict_2 else: return verdict_3 # Mock shot data in orchestrator mock_shots = { test_shot_id: { "scene_title": "Test Retake Scene", "motion_prompt": "Kiều đứng dưới bóng liễu", "character_anchor": "thuy_kieu", "character_asset_ref": "04_Assets/characters/01_Main_Protagonists/thuy_kieu_maiden_16yo_720p.png" } } with patch("production_orchestrator.get_all_shots", return_value=mock_shots), \ patch("production_orchestrator.evaluate_shot_gate", side_effect=mock_gate) as mock_eval: success = po.render_single_shot( shot_id=test_shot_id, dry_run=True, enable_critic=True, max_retakes=2 ) assert success is True, "render_single_shot should have succeeded on attempt 3" assert mock_eval.call_count == 3, f"Expected 3 gate calls, got {mock_eval.call_count}" # Verify version files on disk v1 = VIDEOS_DIR / f"{test_shot_id}_10s_v1.mp4" v2 = VIDEOS_DIR / f"{test_shot_id}_10s_v2.mp4" v3 = VIDEOS_DIR / f"{test_shot_id}_10s_v3.mp4" assert v1.exists(), f"Expected {v1} to exist on disk" assert v2.exists(), f"Expected {v2} to exist on disk" assert v3.exists(), f"Expected {v3} to exist on disk" assert v1.stat().st_size > 1000, "v1 video file corrupted or empty" assert v2.stat().st_size > 1000, "v2 video file corrupted or empty" assert v3.stat().st_size > 1000, "v3 video file corrupted or empty" # Verify critique notes were appended on retakes assert "[Critique Fix: Cần tăng độ nét khuôn mặt nhân vật.]" in prompts_received[1] assert "[Critique Fix: Chuyển động còn hơi giật ở biên khung hình.]" in prompts_received[2] # Verify tail frame extracted tail_frame = kf_dir / "clean_frame_239.jpg" assert tail_frame.exists(), f"Expected {tail_frame} to exist after approval" finally: # Teardown for f in VIDEOS_DIR.glob(f"{test_shot_id}*"): try: f.unlink() except Exception: pass kf_dir = KEYFRAMES_DIR / test_shot_id if kf_dir.exists(): shutil.rmtree(kf_dir, ignore_errors=True) def test_retake_loop_persistent_failure_terminates_on_max_retakes(self): """ Scenario: - Critic persistently rejects with score 0.55 (< 0.8). - max_retakes = 2. Empirical check: - render_single_shot terminates and returns False. - Critic was called exactly 3 times (1 initial + 2 retakes). - Exactly 3 video versions (_v1, _v2, _v3) were created on disk. """ test_shot_id = "test_retake_persistent_fail" try: for f in VIDEOS_DIR.glob(f"{test_shot_id}*"): f.unlink(missing_ok=True) kf_dir = KEYFRAMES_DIR / test_shot_id if kf_dir.exists(): shutil.rmtree(kf_dir, ignore_errors=True) failing_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="Nhân vật bị biến dạng mặt." ) mock_shots = { test_shot_id: { "scene_title": "Test Fail Scene", "motion_prompt": "Kiều buồn rầu ngắm hoa rơi", "character_anchor": "thuy_kieu", "character_asset_ref": "04_Assets/characters/01_Main_Protagonists/thuy_kieu_maiden_16yo_720p.png" } } with patch("production_orchestrator.get_all_shots", return_value=mock_shots), \ patch("production_orchestrator.evaluate_shot_gate", return_value=failing_verdict) as mock_eval: success = po.render_single_shot( shot_id=test_shot_id, dry_run=True, enable_critic=True, max_retakes=2 ) assert success is False, "render_single_shot should return False when max_retakes exceeded" assert mock_eval.call_count == 3, f"Expected 3 gate calls, got {mock_eval.call_count}" v1 = VIDEOS_DIR / f"{test_shot_id}_10s_v1.mp4" v2 = VIDEOS_DIR / f"{test_shot_id}_10s_v2.mp4" v3 = VIDEOS_DIR / f"{test_shot_id}_10s_v3.mp4" v4 = VIDEOS_DIR / f"{test_shot_id}_10s_v4.mp4" assert v1.exists() assert v2.exists() assert v3.exists() assert not v4.exists(), "v4 should NOT exist since max_retakes=2 limits to 3 total attempts" finally: for f in VIDEOS_DIR.glob(f"{test_shot_id}*"): try: f.unlink() except Exception: pass kf_dir = KEYFRAMES_DIR / test_shot_id if kf_dir.exists(): shutil.rmtree(kf_dir, ignore_errors=True) def test_retake_loop_immediate_pass_does_not_retake(self): """ Scenario: - Attempt 1 returns score 0.92 (APPROVE). Empirical check: - render_single_shot returns True immediately. - Critic called exactly 1 time. - Only _v1.mp4 exists on disk. """ test_shot_id = "test_retake_immediate_pass" try: for f in VIDEOS_DIR.glob(f"{test_shot_id}*"): f.unlink(missing_ok=True) kf_dir = KEYFRAMES_DIR / test_shot_id if kf_dir.exists(): shutil.rmtree(kf_dir, ignore_errors=True) pass_verdict = VideoCriticVerdict( overall_score=0.92, approved=True, shot_eval=ShotEvaluation(character_match=True, visual_defects=[], score=0.95), suggested_action="APPROVE", critique_notes="Tuyệt đối đạt chuẩn." ) mock_shots = { test_shot_id: { "scene_title": "Test Pass Scene", "motion_prompt": "Kim Trọng nhìn sang vườn Thúy", "character_anchor": "kim_trong", "character_asset_ref": "04_Assets/characters/01_Main_Protagonists/kim_trong_18yo_720p.png" } } with patch("production_orchestrator.get_all_shots", return_value=mock_shots), \ patch("production_orchestrator.evaluate_shot_gate", return_value=pass_verdict) as mock_eval: success = po.render_single_shot( shot_id=test_shot_id, dry_run=True, enable_critic=True, max_retakes=2 ) assert success is True assert mock_eval.call_count == 1 v1 = VIDEOS_DIR / f"{test_shot_id}_10s_v1.mp4" v2 = VIDEOS_DIR / f"{test_shot_id}_10s_v2.mp4" assert v1.exists() assert not v2.exists(), "v2 should NOT exist when shot passes on first attempt" finally: for f in VIDEOS_DIR.glob(f"{test_shot_id}*"): try: f.unlink() except Exception: pass kf_dir = KEYFRAMES_DIR / test_shot_id if kf_dir.exists(): shutil.rmtree(kf_dir, ignore_errors=True) def test_retake_loop_action_retake_shot_overrides_high_score(self): """ Scenario: - Attempt 1: score = 0.82 (>= 0.8) BUT suggested_action == "RETAKE_SHOT" (e.g. subtle flaw). - Attempt 2: score = 0.90 and suggested_action == "APPROVE". Empirical check: - The loop MUST retake on attempt 1 because suggested_action == "RETAKE_SHOT". - Successfully passes on attempt 2. - Exactly 2 versions (_v1, _v2) created. """ test_shot_id = "test_retake_action_override" try: for f in VIDEOS_DIR.glob(f"{test_shot_id}*"): f.unlink(missing_ok=True) kf_dir = KEYFRAMES_DIR / test_shot_id if kf_dir.exists(): shutil.rmtree(kf_dir, ignore_errors=True) verdict_override = VideoCriticVerdict( overall_score=0.82, approved=False, shot_eval=ShotEvaluation(character_match=True, visual_defects=["micro_glitch"], score=0.82), suggested_action="RETAKE_SHOT", critique_notes="Điểm khá nhưng đạo diễn yêu cầu retake do góc sáng lệch." ) verdict_approved = VideoCriticVerdict( overall_score=0.91, approved=True, shot_eval=ShotEvaluation(character_match=True, visual_defects=[], score=0.92), suggested_action="APPROVE", critique_notes="Hoàn hảo." ) mock_shots = { test_shot_id: { "scene_title": "Test Override Scene", "motion_prompt": "Vương Quan bước ra cửa", "character_anchor": "vuong_quan", "character_asset_ref": "04_Assets/characters/02_Vuong_Family_And_Fate/vuong_quan_16yo_720p.png" } } with patch("production_orchestrator.get_all_shots", return_value=mock_shots), \ patch("production_orchestrator.evaluate_shot_gate", side_effect=[verdict_override, verdict_approved]) as mock_eval: success = po.render_single_shot( shot_id=test_shot_id, dry_run=True, enable_critic=True, max_retakes=2 ) assert success is True assert mock_eval.call_count == 2, "Should have retaken despite score >= 0.8 due to suggested_action == 'RETAKE_SHOT'" v1 = VIDEOS_DIR / f"{test_shot_id}_10s_v1.mp4" v2 = VIDEOS_DIR / f"{test_shot_id}_10s_v2.mp4" v3 = VIDEOS_DIR / f"{test_shot_id}_10s_v3.mp4" assert v1.exists() assert v2.exists() assert not v3.exists() finally: for f in VIDEOS_DIR.glob(f"{test_shot_id}*"): try: f.unlink() except Exception: pass kf_dir = KEYFRAMES_DIR / test_shot_id if kf_dir.exists(): shutil.rmtree(kf_dir, ignore_errors=True)