""" ============================================================================= CHALLENGER 1 EMPIRICAL ADVERSARIAL STRESS HARNESS - M1 ITERATION 2 ============================================================================= Authoritative independent stress-test suite executed by Challenger 1. Validates: 1. evaluate_scene_gate: - Non-existent files (alone and mixed with valid files) -> score=0.0, approved=False, action='RETAKE_SHOT' - Empty 0-byte files -> score=0.0, approved=False, action='RETAKE_SHOT' - Corrupted/garbage files -> score=0.0, approved=False, action='RETAKE_SHOT' - Truncated headers -> score=0.0, approved=False, action='RETAKE_SHOT' - Mixed valid and invalid files -> score=0.0, approved=False, action='RETAKE_SHOT' 2. evaluate_shot_gate on frozen video: - 48, 120, 240 frames frozen -> docked >= 0.35, capping score <= 0.65, approved=False, action='RETAKE_SHOT' 3. evaluate_shot_gate on Thúy Kiều video evaluated as Vương Ông / Vương Quan / Kim Trọng: - Static portrait video -> char_match=False, score <= 0.50, approved=False, action='RETAKE_SHOT' - Moving Thúy Kiều video (non-frozen motion) -> char_match=False, score <= 0.50, approved=False, action='RETAKE_SHOT' - Case-insensitive / whitespace-tolerant expected_character names ("Vương Ông", " vuong_quan ") 4. Engine dispatch (both force_engine='heuristic' and default force_engine=None): - Consistent behavior across invocation modes ============================================================================= """ import os import sys import tempfile import shutil from pathlib import Path import pytest import numpy as np import cv2 PROJECT_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(PROJECT_ROOT / "05_Production_Pipeline")) from antigravity_critic_gate import ( VideoCriticVerdict, ShotEvaluation, SceneEvaluation, evaluate_shot_gate, evaluate_scene_gate, OfflineHeuristicEngine, ) @pytest.fixture(scope="session") def harness_sandbox(tmp_path_factory): temp_dir = tmp_path_factory.mktemp("challenger1_stress") yield temp_dir shutil.rmtree(str(temp_dir), ignore_errors=True) def create_test_video(filepath: Path, num_frames=72, fps=24, width=1280, height=720, mode="motion", color=(120, 160, 200)): fourcc = cv2.VideoWriter_fourcc(*"mp4v") out = cv2.VideoWriter(str(filepath), fourcc, fps, (width, height)) for i in range(num_frames): if mode == "frozen": frame = np.full((height, width, 3), color, dtype=np.uint8) elif mode == "motion": frame = np.full((height, width, 3), color, dtype=np.uint8) cx = int(width / 2 + np.sin(i / 10.0) * 150) cy = int(height / 2 + np.cos(i / 10.0) * 80) cv2.circle(frame, (cx, cy), 50, (255, 255, 255), -1) elif mode == "black": frame = np.zeros((height, width, 3), dtype=np.uint8) else: frame = np.random.randint(0, 255, (height, width, 3), dtype=np.uint8) out.write(frame) out.release() return filepath # ============================================================================= # 1. EVALUATE_SCENE_GATE ADVERSARIAL STRESS TESTS # ============================================================================= class TestChallengerSceneGateStress: def test_scene_gate_pure_nonexistent_files(self): """Non-existent files must yield score=0.0, approved=False, action='RETAKE_SHOT'.""" for engine in [None, "heuristic"]: v = evaluate_scene_gate("sc_test_missing", ["non_existent_1.mp4", "non_existent_2.mp4"], force_engine=engine) assert v.overall_score == 0.0, f"Expected 0.0, got {v.overall_score}" assert v.approved is False assert v.suggested_action == "RETAKE_SHOT" assert v.scene_eval is not None assert v.scene_eval.score == 0.0 def test_scene_gate_mixed_valid_and_nonexistent(self, harness_sandbox): """1 valid file + 1 non-existent file must yield score=0.0, approved=False, action='RETAKE_SHOT'.""" valid_p = harness_sandbox / "valid_for_mix.mp4" create_test_video(valid_p, num_frames=48, mode="motion") missing_p = str(harness_sandbox / "phantom_file.mp4") for engine in [None, "heuristic"]: v = evaluate_scene_gate("sc_mixed_missing", [str(valid_p), missing_p], force_engine=engine) assert v.overall_score == 0.0 assert v.approved is False assert v.suggested_action == "RETAKE_SHOT" assert "Missing file" in v.critique_notes def test_scene_gate_empty_zero_byte_files(self, harness_sandbox): """Empty 0-byte files must yield score=0.0, approved=False, action='RETAKE_SHOT'.""" zero_1 = harness_sandbox / "zero_1.mp4" zero_2 = harness_sandbox / "zero_2.mp4" zero_1.write_bytes(b"") zero_2.write_bytes(b"") for engine in [None, "heuristic"]: v = evaluate_scene_gate("sc_zero_bytes", [str(zero_1), str(zero_2)], force_engine=engine) assert v.overall_score == 0.0 assert v.approved is False assert v.suggested_action == "RETAKE_SHOT" assert "Zero-byte file" in v.critique_notes def test_scene_gate_corrupted_garbage_files(self, harness_sandbox): """Corrupted/garbage header files must yield score=0.0, approved=False, action='RETAKE_SHOT'.""" garb_1 = harness_sandbox / "garbage_1.mp4" garb_2 = harness_sandbox / "garbage_2.mp4" garb_1.write_bytes(b"RANDOM_GARBAGE_HEADER_DATA_1234567890") garb_2.write_bytes(b"\x00\x00\x00\x18ftypmp42corrupted_tail") for engine in [None, "heuristic"]: v = evaluate_scene_gate("sc_garbage", [str(garb_1), str(garb_2)], force_engine=engine) assert v.overall_score == 0.0 assert v.approved is False assert v.suggested_action == "RETAKE_SHOT" assert ("Unreadable file" in v.critique_notes or "Zero frames" in v.critique_notes) def test_scene_gate_mixed_valid_and_corrupted(self, harness_sandbox): """1 valid file + 1 corrupted 0-byte file must fail immediately with score=0.0.""" valid_p = harness_sandbox / "valid_mix_corrupt.mp4" create_test_video(valid_p, num_frames=48, mode="motion") corrupt_p = harness_sandbox / "corrupt_sub.mp4" corrupt_p.write_bytes(b"") v = evaluate_scene_gate("sc_mix_corrupt", [str(valid_p), str(corrupt_p)]) assert v.overall_score == 0.0 assert v.approved is False assert v.suggested_action == "RETAKE_SHOT" # ============================================================================= # 2. EVALUATE_SHOT_GATE ON FROZEN VIDEO ADVERSARIAL STRESS TESTS # ============================================================================= class TestChallengerShotGateFrozenStress: @pytest.mark.parametrize("frame_count", [48, 120, 240]) def test_shot_gate_frozen_penalty_and_cap(self, harness_sandbox, frame_count): """ Frozen video across multiple durations (2s, 5s, 10s): MUST dock >= 0.35, capping score <= 0.65, approved=False, action='RETAKE_SHOT'. """ frozen_file = harness_sandbox / f"frozen_{frame_count}f.mp4" create_test_video(frozen_file, num_frames=frame_count, mode="frozen", color=(140, 150, 160)) for engine in [None, "heuristic"]: v = evaluate_shot_gate( shot_id=f"shot_frozen_{frame_count}", video_path=str(frozen_file), expected_character="none", force_engine=engine ) # Dock >= 0.35 -> from 1.0 baseline, score must be <= 0.65 assert v.overall_score <= 0.65, f"Score {v.overall_score} exceeds 0.65 cap for frozen video!" assert v.approved is False assert v.suggested_action == "RETAKE_SHOT" assert v.shot_eval is not None assert any("Frozen" in d for d in v.shot_eval.visual_defects), f"No Frozen defect in {v.shot_eval.visual_defects}" def test_shot_gate_frozen_bright_and_dark(self, harness_sandbox): """Frozen videos with various colors (not pure black) must all be penalized <= 0.65.""" for idx, col in enumerate([(220, 220, 220), (50, 60, 70), (100, 180, 120)]): p = harness_sandbox / f"frozen_color_{idx}.mp4" create_test_video(p, num_frames=60, mode="frozen", color=col) v = evaluate_shot_gate(f"shot_fcol_{idx}", str(p)) assert v.overall_score <= 0.65 assert v.approved is False assert v.suggested_action == "RETAKE_SHOT" # ============================================================================= # 3. EVALUATE_SHOT_GATE CHARACTER CONTAMINATION ADVERSARIAL STRESS TESTS # ============================================================================= class TestChallengerCharacterContaminationStress: @classmethod def setup_class(cls): cls.kieu_portrait = PROJECT_ROOT / "04_Assets" / "characters" / "01_Main_Protagonists" / "thuy_kieu_maiden_16yo_720p.png" assert cls.kieu_portrait.exists(), "Thuy Kieu master portrait must exist!" cls.kieu_img = cv2.imread(str(cls.kieu_portrait)) assert cls.kieu_img is not None, "Failed to decode Thuy Kieu master portrait" def test_contamination_static_portrait_evaluated_as_vuong_ong(self, harness_sandbox): """Thúy Kiều video evaluated as Vương Ông: MUST detect cross-contamination, char_match=False, score <= 0.50, RETAKE_SHOT.""" vid_p = harness_sandbox / "tk_static_as_vo.mp4" h, w = self.kieu_img.shape[:2] out = cv2.VideoWriter(str(vid_p), cv2.VideoWriter_fourcc(*"mp4v"), 24, (w, h)) for _ in range(48): out.write(self.kieu_img) out.release() v = evaluate_shot_gate("shot_test_vo", str(vid_p), expected_character="vuong_ong") assert v.overall_score <= 0.50, f"Score {v.overall_score} exceeds 0.50 for cross-contaminated shot" assert v.shot_eval.character_match is False assert v.approved is False assert v.suggested_action == "RETAKE_SHOT" assert any("cross-contamination" in d.lower() for d in v.shot_eval.visual_defects) def test_contamination_static_portrait_evaluated_as_vuong_quan(self, harness_sandbox): """Thúy Kiều video evaluated as Vương Quan: MUST detect cross-contamination, char_match=False, score <= 0.50, RETAKE_SHOT.""" vid_p = harness_sandbox / "tk_static_as_vq.mp4" h, w = self.kieu_img.shape[:2] out = cv2.VideoWriter(str(vid_p), cv2.VideoWriter_fourcc(*"mp4v"), 24, (w, h)) for _ in range(48): out.write(self.kieu_img) out.release() v = evaluate_shot_gate("shot_test_vq", str(vid_p), expected_character="vuong_quan") assert v.overall_score <= 0.50, f"Score {v.overall_score} exceeds 0.50" assert v.shot_eval.character_match is False assert v.approved is False assert v.suggested_action == "RETAKE_SHOT" assert any("cross-contamination" in d.lower() for d in v.shot_eval.visual_defects) def test_contamination_moving_kieu_video_evaluated_as_vuong_ong(self, harness_sandbox): """ Adversarial: Thúy Kiều video WITH MICRO-MOTION (not frozen), evaluated as Vương Ông. MUST still detect cross-contamination, char_match=False, score <= 0.50, RETAKE_SHOT. """ vid_p = harness_sandbox / "tk_moving_as_vo.mp4" h, w = self.kieu_img.shape[:2] out = cv2.VideoWriter(str(vid_p), cv2.VideoWriter_fourcc(*"mp4v"), 24, (w, h)) for i in range(72): # Apply slight translation/panning (micro-motion) to avoid frozen detection dx = int(np.sin(i / 8.0) * 8) dy = int(np.cos(i / 8.0) * 5) M = np.float32([[1, 0, dx], [0, 1, dy]]) shifted = cv2.warpAffine(self.kieu_img, M, (w, h), borderMode=cv2.BORDER_REFLECT) out.write(shifted) out.release() v = evaluate_shot_gate("shot_moving_vo", str(vid_p), expected_character="vuong_ong") # Ensure it was NOT flagged as frozen assert not any("Frozen" in d for d in v.shot_eval.visual_defects), "Should have micro-motion" # But MUST be flagged as cross-contamination assert any("cross-contamination" in d.lower() for d in v.shot_eval.visual_defects) assert v.shot_eval.character_match is False assert v.overall_score <= 0.50 assert v.approved is False assert v.suggested_action == "RETAKE_SHOT" def test_contamination_moving_kieu_video_evaluated_as_vuong_quan(self, harness_sandbox): """ Adversarial: Thúy Kiều video WITH MICRO-MOTION (not frozen), evaluated as Vương Quan. MUST detect cross-contamination, char_match=False, score <= 0.50, RETAKE_SHOT. """ vid_p = harness_sandbox / "tk_moving_as_vq.mp4" h, w = self.kieu_img.shape[:2] out = cv2.VideoWriter(str(vid_p), cv2.VideoWriter_fourcc(*"mp4v"), 24, (w, h)) for i in range(72): dx = int(np.sin(i / 8.0) * 8) dy = int(np.cos(i / 8.0) * 5) M = np.float32([[1, 0, dx], [0, 1, dy]]) shifted = cv2.warpAffine(self.kieu_img, M, (w, h), borderMode=cv2.BORDER_REFLECT) out.write(shifted) out.release() v = evaluate_shot_gate("shot_moving_vq", str(vid_p), expected_character="vuong_quan") assert not any("Frozen" in d for d in v.shot_eval.visual_defects) assert any("cross-contamination" in d.lower() for d in v.shot_eval.visual_defects) assert v.shot_eval.character_match is False assert v.overall_score <= 0.50 assert v.approved is False assert v.suggested_action == "RETAKE_SHOT" def test_contamination_with_whitespace_and_case_variants(self, harness_sandbox): """Robustness: expected_character with spaces or capital letters (' Vương Ông ', 'VUONG_QUAN').""" vid_p = harness_sandbox / "tk_static_as_vo_case.mp4" h, w = self.kieu_img.shape[:2] out = cv2.VideoWriter(str(vid_p), cv2.VideoWriter_fourcc(*"mp4v"), 24, (w, h)) for _ in range(48): out.write(self.kieu_img) out.release() # Test " vuong_ong " v1 = evaluate_shot_gate("shot_vo_ws", str(vid_p), expected_character=" vuong_ong ") assert v1.shot_eval.character_match is False assert v1.overall_score <= 0.50 # Test "VUONG_QUAN" v2 = evaluate_shot_gate("shot_vq_upper", str(vid_p), expected_character="VUONG_QUAN") assert v2.shot_eval.character_match is False assert v2.overall_score <= 0.50 # ============================================================================= # 4. BONA FIDE CHARACTER PASSING VERIFICATION (ORACLE BASELINE) # ============================================================================= class TestChallengerLegitimateCharacterPass: def test_thuy_kieu_video_evaluated_as_thuy_kieu_passes(self, harness_sandbox): """Oracle check: Bona fide Thúy Kiều video with motion evaluated as Thúy Kiều MUST pass without contamination.""" kieu_portrait = PROJECT_ROOT / "04_Assets" / "characters" / "01_Main_Protagonists" / "thuy_kieu_maiden_16yo_720p.png" kieu_img = cv2.imread(str(kieu_portrait)) h, w = kieu_img.shape[:2] vid_p = harness_sandbox / "genuine_kieu_pass.mp4" out = cv2.VideoWriter(str(vid_p), cv2.VideoWriter_fourcc(*"mp4v"), 24, (w, h)) for i in range(72): dx = int(np.sin(i / 8.0) * 10) dy = int(np.cos(i / 8.0) * 6) M = np.float32([[1, 0, dx], [0, 1, dy]]) shifted = cv2.warpAffine(kieu_img, M, (w, h), borderMode=cv2.BORDER_REFLECT) out.write(shifted) out.release() v = evaluate_shot_gate("shot_genuine_kieu", str(vid_p), expected_character="thuy_kieu") assert v.shot_eval.character_match is True assert not any("cross-contamination" in d.lower() for d in v.shot_eval.visual_defects) assert v.overall_score >= 0.80 assert v.approved is True assert v.suggested_action == "APPROVE" if __name__ == "__main__": pytest.main([__file__, "-v"])