""" Tier 1: Feature Coverage Test Suite (F1 - F13) Contains >= 5 distinct, authoritative test cases per feature (Total: 65 test cases). """ import os import sys import re import json import unittest from pathlib import Path # Add tests directory and base directory to sys.path TESTS_DIR = Path(__file__).resolve().parent BASE_DIR = TESTS_DIR.parent if str(BASE_DIR) not in sys.path: sys.path.insert(0, str(BASE_DIR)) if str(TESTS_DIR) not in sys.path: sys.path.insert(0, str(TESTS_DIR)) from tests import ( FILMMAKER_DIR, PROMPTS_DIR, ASSETS_DIR, BIBLE_DIR, PIPELINE_DIR, EXPORTS_DIR, WEB_REVIEW_DIR, read_ep01_screenplay, get_screenplay_scenes, find_male_crying_violations, load_muse_prompts_data, load_banana_prompts_data ) class TestTier1Features(unittest.TestCase): """ Tier 1 tests verify the primary requirement contracts for F1 through F13. """ # ------------------------------------------------------------------------- # F1: EP01 Screenplay & Pacing Expansion (>= 25m / 1500s, >= 150 shots, 3 Acts) # ------------------------------------------------------------------------- def test_f1_01_ep01_shot_count_minimum(self): """F1.1: Verify EP01 screenplay defines >= 150 shots (target: 188 shots).""" screenplay = read_ep01_screenplay() self.assertTrue(len(screenplay) > 0, "TAP_01 screenplay file must not be empty") shot_matches = re.findall(r"(?:shot\s*(\d+)|cảnh\s*\d+[^:\n]*\bshot\b|\bSH_?\d+)", screenplay, re.IGNORECASE) # Also check explicit shot listing in FilmMaker/INDEX_VA_DANH_MUC_CANH_QUAY.md or TAP_01 muse_data = load_muse_prompts_data() ep01_prompts = [k for k in muse_data.get("motion_prompts", {}) if "ep01" in k.lower()] effective_count = max(len(shot_matches), len(ep01_prompts)) self.assertGreaterEqual( effective_count, 150, f"EP01 total shot count ({effective_count}) must be >= 150 for 25-30m feature length." ) def test_f1_02_ep01_total_duration(self): """F1.2: Verify calculated total runtime of EP01 is >= 1500 seconds (25 minutes).""" muse_data = load_muse_prompts_data() ep01_shots = [k for k in muse_data.get("motion_prompts", {}) if "ep01" in k.lower()] total_duration_sec = len(ep01_shots) * 10 self.assertGreaterEqual( total_duration_sec, 1500, f"EP01 total runtime ({total_duration_sec}s) must be >= 1500s (25 minutes)." ) def test_f1_03_ep01_three_act_structure_presence(self): """F1.3: Verify clear 3-Act classical cinema structure in EP01 screenplay.""" screenplay = read_ep01_screenplay() has_act_1 = bool(re.search(r"HỒI\s*1\b", screenplay, re.IGNORECASE)) has_act_2 = bool(re.search(r"HỒI\s*2\b", screenplay, re.IGNORECASE)) has_act_3 = bool(re.search(r"HỒI\s*3\b", screenplay, re.IGNORECASE)) self.assertTrue(has_act_1, "EP01 screenplay must contain Act 1 (HỒI 1)") self.assertTrue(has_act_2, "EP01 screenplay must contain Act 2 (HỒI 2)") self.assertTrue(has_act_3, "EP01 screenplay must contain Act 3 (HỒI 3)") def test_f1_04_ep01_scene_continuity_indexing(self): """F1.4: Verify sequential scene indexing from Scene 01 up through Scene 15.""" screenplay = read_ep01_screenplay() scenes = get_screenplay_scenes(screenplay) self.assertGreaterEqual(len(scenes), 12, "EP01 must have at least 12 distinct scenes") scene_indices = [] for header, _ in scenes: m = re.search(r"CẢNH\s*0?(\d+)", header, re.IGNORECASE) if m: scene_indices.append(int(m.group(1))) for expected in range(1, min(len(scene_indices) + 1, 14)): self.assertIn(expected, scene_indices, f"Missing Scene {expected:02d} in EP01 screenplay") def test_f1_05_ep01_pacing_act_duration_distribution(self): """F1.5: Verify balanced pacing distribution across acts (Act 1 >= 500s, Act 2 >= 500s, Act 3 >= 350s).""" screenplay = read_ep01_screenplay() act1_match = re.search(r"HỒI\s*1.*?(?=HỒI\s*2|\Z)", screenplay, re.DOTALL | re.IGNORECASE) act2_match = re.search(r"HỒI\s*2.*?(?=HỒI\s*3|\Z)", screenplay, re.DOTALL | re.IGNORECASE) act3_match = re.search(r"HỒI\s*3.*?(?=\Z)", screenplay, re.DOTALL | re.IGNORECASE) self.assertIsNotNone(act1_match, "Act 1 must be delineable") self.assertIsNotNone(act2_match, "Act 2 must be delineable") self.assertIsNotNone(act3_match, "Act 3 must be delineable") # Each act should contain substantive content (> 1500 chars) self.assertGreater(len(act1_match.group(0)), 1500, "Act 1 content must be comprehensive") self.assertGreater(len(act2_match.group(0)), 1500, "Act 2 content must be comprehensive") self.assertGreater(len(act3_match.group(0)), 1500, "Act 3 content must be comprehensive") # ------------------------------------------------------------------------- # F2: Đạm Tiên Grave 3-Beat Sequence (C05-A, C05-B, C05-C) # ------------------------------------------------------------------------- def test_f2_01_c05_a_stroll_along_stream(self): """F2.1: Verify Scene 05 contains Beat A (Stroll along 'tiểu khê' / winding stream).""" screenplay = read_ep01_screenplay() scenes = get_screenplay_scenes(screenplay) scene05_text = "" for h, c in scenes: if "CẢNH 05" in h or "CẢNH 5" in h: scene05_text = c break self.assertTrue(len(scene05_text) > 0, "Scene 05 must exist in screenplay") has_tieu_khe = bool(re.search(r"(tiểu khê|suối nhỏ|bước dần theo ngọn|lần theo ngọn)", scene05_text, re.IGNORECASE)) self.assertTrue(has_tieu_khe, "Scene 05 Beat A must feature stroll along 'tiểu khê' / winding stream") def test_f2_02_c05_b_inquiry_at_mound(self): """F2.2: Verify Scene 05 contains Beat B (Inquiry & wonder at deserted mound 'sè sè').""" screenplay = read_ep01_screenplay() scenes = get_screenplay_scenes(screenplay) scene05_text = "".join(c for h, c in scenes if "CẢNH 05" in h or "CẢNH 5" in h) has_se_se = bool(re.search(r"(sè sè|nấm mồ|nấm đất)", scene05_text, re.IGNORECASE)) has_inquiry = bool(re.search(r"(hương khói vắng tanh|sao trong tiết thanh minh|vắng tanh)", scene05_text, re.IGNORECASE)) self.assertTrue(has_se_se, "Scene 05 Beat B must feature the deserted mound 'sè sè'") self.assertTrue(has_inquiry, "Scene 05 Beat B must feature Kiều's inquiry at deserted mound") def test_f2_03_c05_c_grief_and_tears(self): """F2.3: Verify Scene 05 contains Beat C (Emotional grief for women's fate 'bạc mệnh' and tears).""" screenplay = read_ep01_screenplay() scenes = get_screenplay_scenes(screenplay) scene05_text = "".join(c for h, c in scenes if "CẢNH 05" in h or "CẢNH 5" in h) has_grief = bool(re.search(r"(đau đớn thay phận đàn bà|bạc mệnh cũng là lời chung)", scene05_text, re.IGNORECASE)) has_kieu_tears = bool(re.search(r"(nước mắt|rơi lệ|châu sa|khóc)", scene05_text, re.IGNORECASE)) self.assertTrue(has_grief, "Scene 05 Beat C must contain iconic grief for women's fate ('bạc mệnh')") self.assertTrue(has_kieu_tears, "Scene 05 Beat C must contain Thúy Kiều weeping for Đạm Tiên") def test_f2_04_c05_c_supernatural_footprints(self): """F2.4: Verify Scene 05 contains Beat C supernatural footprints in moss ('dấu hài' / 'dấu giày' 'in rêu').""" screenplay = read_ep01_screenplay() scenes = get_screenplay_scenes(screenplay) scene05_text = "".join(c for h, c in scenes if "CẢNH 05" in h or "CẢNH 5" in h) has_footprints = bool(re.search(r"(dấu hài|dấu giày|in rêu|dấu chân.*rêu)", scene05_text, re.IGNORECASE)) self.assertTrue(has_footprints, "Scene 05 must contain supernatural footprints in moss ('dấu hài in rêu')") def test_f2_05_scene05_sub_beat_ordering(self): """F2.5: Verify strict temporal order: Stroll (A) -> Wonder/Inquiry (B) -> Grief & Footprints (C).""" screenplay = read_ep01_screenplay() scenes = get_screenplay_scenes(screenplay) scene05_text = "".join(c for h, c in scenes if "CẢNH 05" in h or "CẢNH 5" in h) pos_a = re.search(r"(tiểu khê|bước dần|suối)", scene05_text, re.IGNORECASE) pos_b = re.search(r"(sè sè|nấm mồ|vắng tanh)", scene05_text, re.IGNORECASE) pos_c = re.search(r"(đau đớn thay|bạc mệnh|dấu hài|dấu giày)", scene05_text, re.IGNORECASE) self.assertIsNotNone(pos_a, "Beat A keyword must exist in Scene 05") self.assertIsNotNone(pos_b, "Beat B keyword must exist in Scene 05") self.assertIsNotNone(pos_c, "Beat C keyword must exist in Scene 05") self.assertLess(pos_a.start(), pos_b.start(), "Beat A (stroll) must precede Beat B (inquiry)") self.assertLess(pos_b.start(), pos_c.start(), "Beat B (inquiry) must precede Beat C (grief)") # ------------------------------------------------------------------------- # F3: Anti-Male-Tears Script Purge (0% Male Tears in EP01) # ------------------------------------------------------------------------- def test_f3_01_no_male_crying_in_ep01_screenplay(self): """F3.1: Zero male character crying cues across all action beats in EP01 screenplay.""" screenplay = read_ep01_screenplay() violations = find_male_crying_violations(screenplay) violation_messages = [f"Line {v['line_num']}: {v['character']} -> '{v['text']}'" for v in violations] self.assertEqual( len(violations), 0, f"Detected {len(violations)} Anti-Male-Tears violations in EP01:\n" + "\n".join(violation_messages) ) def test_f3_02_kim_trong_parting_stoicism(self): """F3.2: Specifically verify Kim Trọng does NOT weep during parting scene.""" screenplay = read_ep01_screenplay() lines = screenplay.splitlines() for idx, line in enumerate(lines, 1): if "KIM TRỌNG" in line and any(w in line.lower() for w in ["khóc", "rơi lệ", "rơi nước mắt", "nghẹn ngào rơi"]): self.fail(f"Line {idx} violates Anti-Male-Tears: Kim Trọng weeps: {line}") def test_f3_03_vuong_quan_stoic_delivery(self): """F3.3: Verify Vương Quan maintains stoic composure during Đạm Tiên tale and family crisis.""" screenplay = read_ep01_screenplay() lines = screenplay.splitlines() for idx, line in enumerate(lines, 1): if "VƯƠNG QUAN" in line and any(w in line.lower() for w in ["khóc", "rơi lệ", "rơi nước mắt"]): self.fail(f"Line {idx} violates Anti-Male-Tears: Vương Quan weeps: {line}") def test_f3_04_vuong_ong_dignity_during_arrest(self): """F3.4: Verify Vương Ông preserves paternal dignity during arrest without hysterical sobbing.""" screenplay = read_ep01_screenplay() lines = screenplay.splitlines() for idx, line in enumerate(lines, 1): if "VƯƠNG ÔNG" in line and any(w in line.lower() for w in ["khóc ngất", "khóc rống", "gào khóc"]): self.fail(f"Line {idx} violates Anti-Male-Tears: Vương Ông breaks stoic dignity: {line}") def test_f3_05_feudal_stoicism_expression_presence(self): """F3.5: Verify presence of codified feudal stoic expressions (đanh lại, trĩu nặng, nén, hàm bạnh).""" screenplay = read_ep01_screenplay() has_stoic_expressions = bool(re.search(r"(đanh lại|trĩu nặng|nén thở|hàm bạnh|ánh nhìn xa xăm|trầm mặc)", screenplay, re.IGNORECASE)) self.assertTrue( has_stoic_expressions, "Screenplay must substitute tears with feudal stoic micro-expressions (đanh lại, trĩu nặng, nén)." ) # ------------------------------------------------------------------------- # F4: Universal Acting Directives Codified in Project Bible # ------------------------------------------------------------------------- def test_f4_01_anti_male_tears_directive_in_bible(self): """F4.1: Verify explicit Anti-Male-Tears section codified in Project Bible.""" char_bible_path = BIBLE_DIR / "CHARACTER_BIBLE.md" master_bible_path = BIBLE_DIR / "MASTER_BIBLE.md" content = "" if char_bible_path.exists(): content += char_bible_path.read_text(encoding="utf-8") if master_bible_path.exists(): content += master_bible_path.read_text(encoding="utf-8") has_directive = bool(re.search(r"(Anti-Male-Tears|Nam nhân tuyệt đối không rơi lệ|quy chuẩn diễn xuất nam nhân)", content, re.IGNORECASE)) self.assertTrue(has_directive, "Project Bible must codify Anti-Male-Tears directive") def test_f4_02_male_stoic_micro_expressions_codified(self): """F4.2: Verify Project Bible specifies male micro-expressions (jaw clench, heavy gaze, compressed breath).""" char_bible_path = BIBLE_DIR / "CHARACTER_BIBLE.md" content = char_bible_path.read_text(encoding="utf-8") if char_bible_path.exists() else "" has_cues = bool(re.search(r"(cơ mặt đanh|hàm bạnh|ánh mắt trĩu nặng|hơi thở nén|jaw clench|heavy gaze)", content, re.IGNORECASE)) self.assertTrue(has_cues, "Project Bible must detail male stoic micro-expressions") def test_f4_03_female_nuanced_acting_standards(self): """F4.3: Verify Project Bible codifies nuanced female acting standards avoiding melodrama.""" char_bible_path = BIBLE_DIR / "CHARACTER_BIBLE.md" content = char_bible_path.read_text(encoding="utf-8") if char_bible_path.exists() else "" has_female_standards = bool(re.search(r"(micro-expression|diễn xuất.*nữ|làn thu thủy|đoan trang)", content, re.IGNORECASE)) self.assertTrue(has_female_standards, "Project Bible must codify nuanced female acting standards") def test_f4_04_crowd_and_extras_acting_rules(self): """F4.4: Verify Project Bible codifies authentic crowd / extras motion rules.""" bible_path = BIBLE_DIR / "MASTER_BIBLE.md" content = bible_path.read_text(encoding="utf-8") if bible_path.exists() else "" has_crowd = bool(re.search(r"(quần chúng|extras|đám đông|du xuân)", content, re.IGNORECASE)) self.assertTrue(has_crowd, "Project Bible must specify crowd & extras behavior standards") def test_f4_05_historical_feudal_decorum_mandate(self): """F4.5: Verify Project Bible mandates East Asian feudal decorum (Lễ giáo, phong thái nho nhã).""" bible_path = BIBLE_DIR / "MASTER_BIBLE.md" content = bible_path.read_text(encoding="utf-8") if bible_path.exists() else "" has_decorum = bool(re.search(r"(phong kiến|nho gia|lễ giáo|cổ phong|đoan trang)", content, re.IGNORECASE)) self.assertTrue(has_decorum, "Project Bible must mandate feudal decorum and historical composure") # ------------------------------------------------------------------------- # F5: Sweet-Spot Start Frame Protocol (Gemini Banana 720p, zero reverse) # ------------------------------------------------------------------------- def test_f5_01_gemini_banana_resolution_standard(self): """F5.1: Verify start frame prompt registry specifies 1280x720 resolution.""" banana_data = load_banana_prompts_data() self.assertTrue(len(banana_data) > 0, "gemini_banana_prompts.json must not be empty") # Check standard resolution key or top-level entries res_found = False raw_str = json.dumps(banana_data) if "1280x720" in raw_str or "720p" in raw_str: res_found = True self.assertTrue(res_found, "Gemini Banana prompt registry must enforce 720p (1280x720) sweet-spot standard") def test_f5_02_contextual_environment_blending(self): """F5.2: Verify start frame prompts combine character anchor with specific environmental anchors.""" banana_data = load_banana_prompts_data() raw_str = json.dumps(banana_data) has_env = any(env in raw_str.lower() for env in ["suối", "cỏ", "mộ", "bắc kinh", "vườn", "thanh minh"]) self.assertTrue(has_env, "Start frame prompts must combine character with environmental anchors") def test_f5_03_elimination_of_raw_studio_portraits(self): """F5.3: Verify outdoor scene prompts do not directly feed raw studio portraits to Muse.ai.""" muse_data = load_muse_prompts_data() prompts = muse_data.get("motion_prompts", {}) # Outdoor scenes: Scene 04 (Festival), Scene 05 (Dam Tien) outdoor_raw_portraits = [] for k, v in prompts.items(): if any(sc in k for sc in ["scene04", "scene05", "scene06"]): ref = v.get("character_asset_ref", "") if isinstance(v, dict) else "" if "characters" in ref and "backgrounds" not in ref and "keyframes" not in ref: outdoor_raw_portraits.append(k) self.assertEqual( len(outdoor_raw_portraits), 0, f"Outdoor prompts directly referencing raw studio portraits: {outdoor_raw_portraits}" ) def test_f5_04_directional_motion_vector_guard(self): """F5.4: Verify motion prompts specify clear forward motion vectors (tiến tới, bay về phía trước).""" muse_data = load_muse_prompts_data() prompts = muse_data.get("motion_prompts", {}) sample_prompts = list(prompts.values())[:30] prompts_with_motion = 0 for p in sample_prompts: text = p.get("motion_prompt", "") if isinstance(p, dict) else str(p) if any(w in text.lower() for w in ["camera", "chuyển động", "bước", "bay", "tiến", "pan", "dolly"]): prompts_with_motion += 1 self.assertGreaterEqual(prompts_with_motion, 15, "Motion prompts must specify active directional motion vectors") def test_f5_05_reverse_physics_prohibition(self): """F5.5: Verify prompt instructions forbid reverse physics (no backward bird flight, no moonwalking).""" muse_data = load_muse_prompts_data() raw_str = json.dumps(muse_data, ensure_ascii=False) has_guard = bool(re.search(r"(bay về phía trước|tuyệt đối không|chuyển động tự nhiên|thuận chiều)", raw_str, re.IGNORECASE)) self.assertTrue(has_guard, "Motion prompts must guard against reverse optical motion") # ------------------------------------------------------------------------- # F6: Lip-Sync & Off-Screen Dialogue Guard (Closed Lips) # ------------------------------------------------------------------------- def test_f6_01_lip_sync_guard_present_in_offscreen_dialogue(self): """F6.1: Verify off-screen / distant dialogue prompts contain mandatory closed-lips clause.""" muse_data = load_muse_prompts_data() prompts = muse_data.get("motion_prompts", {}) for shot_id, shot_info in prompts.items(): audio_text = shot_info.get("audio_prompt", "") if isinstance(shot_info, dict) else "" motion_text = shot_info.get("motion_prompt", "") if isinstance(shot_info, dict) else str(shot_info) if "từ xa" in audio_text.lower() or "ngoài khung" in audio_text.lower() or "v.o." in audio_text.lower(): has_closed_lips = any(w in motion_text.lower() for w in ["khép chặt", "khép môi", "lips closed", "không mấp máy"]) self.assertTrue(has_closed_lips, f"Shot {shot_id} has off-screen dialogue but missing closed-lips guard") def test_f6_02_scene08_kim_trong_hairpin_guard(self): """F6.2: Verify Scene 08 Shot 02 (Kim Trọng holding hairpin) enforces closed lips when Kiều speaks off-screen.""" muse_data = load_muse_prompts_data() prompts = muse_data.get("motion_prompts", {}) c08_s02 = prompts.get("ep01_scene08_shot02") if c08_s02: motion = c08_s02.get("motion_prompt", "") if isinstance(c08_s02, dict) else str(c08_s02) has_guard = any(w in motion.lower() for w in ["khép môi", "khép chặt", "lips closed", "không mấp máy"]) self.assertTrue(has_guard, "Scene 08 Shot 02 must guard Kim Trọng's lips when Kiều calls from afar") def test_f6_03_voiceover_prompts_lip_guard(self): """F6.3: Verify shots featuring on-screen characters with voiceover enforce closed lips.""" muse_data = load_muse_prompts_data() prompts = muse_data.get("motion_prompts", {}) vo_shots = [k for k, v in prompts.items() if "v.o." in json.dumps(v).lower() or "người dẫn chuyện" in json.dumps(v).lower()] for shot_id in vo_shots: shot_str = json.dumps(prompts[shot_id]).lower() if "nhân vật" in shot_str and ("khuôn mặt" in shot_str or "cận cảnh" in shot_str): has_guard = any(w in shot_str for w in ["khép môi", "khép chặt", "lips closed", "không mấp máy"]) self.assertTrue(has_guard, f"VO shot {shot_id} must have closed-lips guard") def test_f6_04_lip_sync_bilingual_directive(self): """F6.4: Verify closed-lips directive uses standardized terminology (Vietnamese and/or English).""" muse_data = load_muse_prompts_data() raw_str = json.dumps(muse_data, ensure_ascii=False) has_phrase = "khép chặt" in raw_str or "lips closed" in raw_str or "không mấp máy" in raw_str self.assertTrue(has_phrase, "Lip-sync guard must use standardized terminology (khép chặt/lips closed)") def test_f6_05_on_screen_vs_off_screen_classification(self): """F6.5: Verify prompts classify dialogue context as on-screen vs off-screen.""" muse_data = load_muse_prompts_data() prompts = muse_data.get("motion_prompts", {}) classified = 0 for p in prompts.values(): text = json.dumps(p) if "thoại" in text.lower() or "audio" in text.lower(): classified += 1 self.assertGreater(classified, 10, "Prompts must classify dialogue context") # ------------------------------------------------------------------------- # F7: EP01 188-Shot AI Prompt Registry # ------------------------------------------------------------------------- def test_f7_01_muse_prompt_registry_ep01_shot_count(self): """F7.1: Verify muse_ai_video_prompts.json defines >= 150 shots (target: 188 shots) for EP01.""" muse_data = load_muse_prompts_data() prompts = muse_data.get("motion_prompts", {}) ep01_shots = [k for k in prompts if "ep01" in k.lower()] self.assertGreaterEqual( len(ep01_shots), 150, f"EP01 motion prompts count ({len(ep01_shots)}) must be >= 150 (target: 188)." ) def test_f7_02_gemini_banana_prompt_registry_ep01_shots(self): """F7.2: Verify gemini_banana_prompts.json contains corresponding EP01 start frame prompts.""" banana_data = load_banana_prompts_data() self.assertTrue(len(banana_data) > 0, "gemini_banana_prompts.json must contain prompt entries") def test_f7_03_shot_id_format_compliance(self): """F7.3: Verify EP01 shot IDs adhere to canonical pattern ep01_sceneXX_shotYY.""" muse_data = load_muse_prompts_data() prompts = muse_data.get("motion_prompts", {}) ep01_shots = [k for k in prompts if "ep01" in k.lower()] pattern = re.compile(r"^ep01_(?:scene\d{2}|intro)_shot\d{2}$", re.IGNORECASE) invalid = [k for k in ep01_shots if not pattern.match(k)] self.assertEqual(len(invalid), 0, f"Non-conforming shot IDs: {invalid}") def test_f7_04_audio_guard_clause_in_all_prompts(self): """F7.4: Verify 100% of EP01 motion prompts include mandatory audio guard clause.""" muse_data = load_muse_prompts_data() prompts = muse_data.get("motion_prompts", {}) ep01_shots = {k: v for k, v in prompts.items() if "ep01" in k.lower()} unguarded = [] for k, v in ep01_shots.items(): text = json.dumps(v, ensure_ascii=False).lower() if "tuyệt đối không sinh nhạc nền" not in text and "no music" not in text: unguarded.append(k) self.assertEqual( len(unguarded), 0, f"{len(unguarded)} shots missing audio guard clause:\n{unguarded[:5]}" ) def test_f7_05_prompt_non_emptiness_and_length(self): """F7.5: Verify EP01 prompts contain non-empty, detailed descriptions (>= 50 chars).""" muse_data = load_muse_prompts_data() prompts = muse_data.get("motion_prompts", {}) ep01_shots = {k: v for k, v in prompts.items() if "ep01" in k.lower()} short_prompts = [] for k, v in ep01_shots.items(): m_text = v.get("motion_prompt", "") if isinstance(v, dict) else str(v) if len(m_text.strip()) < 50: short_prompts.append((k, len(m_text))) self.assertEqual(len(short_prompts), 0, f"Prompts too short (<50 chars): {short_prompts}") # ------------------------------------------------------------------------- # F8: Audio Continuity & Crossfade Concat # ------------------------------------------------------------------------- def test_f8_01_crossfade_method_availability(self): """F8.1: Verify AudioContinuityEngine implements stitch_with_audio_crossfade.""" from audio_continuity_engine import AudioContinuityEngine engine = AudioContinuityEngine() self.assertTrue(hasattr(engine, "stitch_with_audio_crossfade"), "Missing stitch_with_audio_crossfade method") self.assertTrue(callable(engine.stitch_with_audio_crossfade), "stitch_with_audio_crossfade must be callable") def test_f8_02_equal_power_curve_configuration(self): """F8.2: Verify equal-power crossfade curve (qsin or cbrt) is used in stitch_with_audio_crossfade.""" import inspect from audio_continuity_engine import AudioContinuityEngine src = inspect.getsource(AudioContinuityEngine.stitch_with_audio_crossfade) has_curve = "qsin" in src or "cbrt" in src self.assertTrue(has_curve, "Audio crossfade must use equal-power curve (qsin/cbrt)") def test_f8_03_default_crossfade_duration(self): """F8.3: Verify default crossfade duration is 1.0 second.""" import inspect from audio_continuity_engine import AudioContinuityEngine sig = inspect.signature(AudioContinuityEngine.stitch_with_audio_crossfade) param = sig.parameters.get("crossfade_dur") self.assertIsNotNone(param, "stitch_with_audio_crossfade must accept crossfade_dur") self.assertEqual(param.default, 1.0, "Default crossfade duration must be 1.0s") def test_f8_04_production_orchestrator_crossfade_default(self): """F8.4: Verify production_orchestrator integrates AudioContinuityEngine or seamless concat.""" from production_orchestrator import concat_scene_shots self.assertTrue(callable(concat_scene_shots), "concat_scene_shots must be callable") def test_f8_05_delta_volume_warning_threshold(self): """F8.5: Verify inspect_sequence warns on volume delta > 3.0 dB.""" import inspect from audio_continuity_engine import AudioContinuityEngine src = inspect.getsource(AudioContinuityEngine.inspect_sequence) self.assertIn("3.0", src, "inspect_sequence must threshold volume deltas at 3.0 dB") # ------------------------------------------------------------------------- # F9: Continuous Stem 2 Festival Ambience # ------------------------------------------------------------------------- def test_f9_01_festival_ambience_asset_presence(self): """F9.1: Verify festival crowd ambience audio asset exists in 04_Assets/audio/.""" audio_dir = ASSETS_DIR / "audio" sfx_dir = ASSETS_DIR / "audio_sfx" candidates = list(audio_dir.glob("*festival*")) + list(audio_dir.glob("*ambience*")) + \ list(sfx_dir.glob("*festival*")) + list(sfx_dir.glob("*ambience*")) # Also check bible definition if asset is being produced in M3 bible_path = BIBLE_DIR / "CINEMATIC_AUDIO_PIPELINE.md" has_spec = bible_path.exists() and "stem 2" in bible_path.read_text(encoding="utf-8").lower() self.assertTrue(len(candidates) > 0 or has_spec, "Festival ambience asset or specification must exist") def test_f9_02_ambience_audio_duration_coverage(self): """F9.2: Verify festival ambience specification or file covers festival scene sequence (>= 60s).""" bible_path = BIBLE_DIR / "CINEMATIC_AUDIO_PIPELINE.md" content = bible_path.read_text(encoding="utf-8") if bible_path.exists() else "" has_coverage = bool(re.search(r"(50s|60s|180s|liên tục)", content)) self.assertTrue(has_coverage, "Ambience bed track must cover multi-shot sequences continuously") def test_f9_03_ambience_format_and_sample_rate(self): """F9.3: Verify audio pipeline enforces 48000 Hz sample rate for stems.""" from audio_continuity_engine import AudioContinuityEngine import inspect src = inspect.getsource(AudioContinuityEngine) self.assertIn("48000", src, "Audio engine must enforce 48000 Hz sample rate") def test_f9_04_four_stem_architecture_compliance(self): """F9.4: Verify CINEMATIC_AUDIO_PIPELINE.md defines 4-stem architecture.""" bible_path = BIBLE_DIR / "CINEMATIC_AUDIO_PIPELINE.md" self.assertTrue(bible_path.exists(), "CINEMATIC_AUDIO_PIPELINE.md must exist") content = bible_path.read_text(encoding="utf-8") self.assertIn("Stem 1", content) self.assertIn("Stem 2", content) self.assertIn("Stem 3", content) self.assertIn("Stem 4", content) def test_f9_05_ambience_layering_integration(self): """F9.5: Verify AudioContinuityEngine supports dynamic ducking and layering.""" from audio_continuity_engine import AudioContinuityEngine engine = AudioContinuityEngine() self.assertTrue(hasattr(engine, "layer_bgm_with_ducking"), "Must implement layer_bgm_with_ducking") # ------------------------------------------------------------------------- # F10: EBU R128 -14 LUFS Two-Pass Normalization # ------------------------------------------------------------------------- def test_f10_01_target_lufs_configuration(self): """F10.1: Verify normalize_loudness targets -14.0 LUFS.""" import inspect from audio_continuity_engine import AudioContinuityEngine sig = inspect.signature(AudioContinuityEngine.normalize_loudness) self.assertEqual(sig.parameters["target_lufs"].default, -14.0) def test_f10_02_target_true_peak_configuration(self): """F10.2: Verify normalize_loudness targets True Peak <= -1.0 dBTP.""" import inspect from audio_continuity_engine import AudioContinuityEngine sig = inspect.signature(AudioContinuityEngine.normalize_loudness) self.assertEqual(sig.parameters["target_tp"].default, -1.0) def test_f10_03_target_loudness_range_configuration(self): """F10.3: Verify normalize_loudness targets LRA in range 9.0 to 11.0 LU.""" import inspect from audio_continuity_engine import AudioContinuityEngine sig = inspect.signature(AudioContinuityEngine.normalize_loudness) self.assertAlmostEqual(sig.parameters["target_lra"].default, 9.0, delta=2.0) def test_f10_04_loudnorm_filter_string_construction(self): """F10.4: Verify FFmpeg loudnorm filter string sets I=-14 and TP=-1.0.""" import inspect from audio_continuity_engine import AudioContinuityEngine src = inspect.getsource(AudioContinuityEngine.normalize_loudness) self.assertIn("loudnorm=", src) self.assertIn("target_lufs", src) def test_f10_05_measurement_tolerance_window(self): """F10.5: Verify YouTube Green Dollar compliance window is [-14.5, -13.5] LUFS.""" # Mathematical verification of tolerance window target = -14.0 tolerance = 0.5 min_lufs = target - tolerance max_lufs = target + tolerance self.assertEqual(min_lufs, -14.5) self.assertEqual(max_lufs, -13.5) # ------------------------------------------------------------------------- # F11: Web Review Studio Port 1515 Synchronization # ------------------------------------------------------------------------- def test_f11_01_fastapi_app_instance_valid(self): """F11.1: Verify web_review.server defines a valid FastAPI app instance.""" from server import app self.assertIsNotNone(app) self.assertEqual(app.title, "Thập Ngũ Niên - Cinema Review Studio") def test_f11_02_library_endpoint_structure(self): """F11.2: Verify /api/library returns expected dictionary structure.""" from server import app from starlette.testclient import TestClient client = TestClient(app) resp = client.get("/api/library") self.assertEqual(resp.status_code, 200) data = resp.json() self.assertIn("summary", data) self.assertIn("feature_masters", data) self.assertIn("raw_scenes", data) def test_f11_03_shots_endpoint_registered(self): """F11.3: Verify /api/shots endpoint is registered and responsive.""" from server import app from starlette.testclient import TestClient client = TestClient(app) resp = client.get("/api/shots") # In F11 contract, /api/shots must return 200 with shots catalog self.assertEqual( resp.status_code, 200, f"/api/shots endpoint returned status {resp.status_code} (expected 200)" ) def test_f11_04_scenes_endpoint_registered(self): """F11.4: Verify /api/scenes endpoint is registered and responsive.""" from server import app from starlette.testclient import TestClient client = TestClient(app) resp = client.get("/api/scenes") self.assertEqual( resp.status_code, 200, f"/api/scenes endpoint returned status {resp.status_code} (expected 200)" ) def test_f11_05_status_endpoint_contract(self): """F11.5: Verify /api/status endpoint returns gen_state structure.""" from server import app from starlette.testclient import TestClient client = TestClient(app) resp = client.get("/api/status") self.assertEqual(resp.status_code, 200) data = resp.json() self.assertIn("status", data) self.assertIn("logs", data) # ------------------------------------------------------------------------- # F12: Automated Production Runner CLI Integration # ------------------------------------------------------------------------- def test_f12_01_production_orchestrator_cli_flags(self): """F12.1: Verify production_orchestrator.py defines required CLI arguments.""" import subprocess proc = subprocess.run( [sys.executable, str(PIPELINE_DIR / "production_orchestrator.py"), "--help"], capture_output=True, text=True, encoding="utf-8", errors="replace" ) self.assertEqual(proc.returncode, 0) help_text = proc.stdout or "" self.assertIn("--status", help_text) self.assertIn("--list-shots", help_text) self.assertIn("--render-shot", help_text) self.assertIn("--concat-scene", help_text) self.assertIn("--master-audio", help_text) def test_f12_02_run_shot_pipeline_callable(self): """F12.2: Verify run_shot.py exports run_shot_pipeline function.""" import run_shot self.assertTrue(hasattr(run_shot, "run_shot_pipeline") or hasattr(run_shot, "main")) def test_f12_03_shot_resolution_logic(self): """F12.3: Verify resolve_start_frame looks up tail frame of preceding shot.""" from production_orchestrator import resolve_start_frame # Shot 02 of a scene should check for shot 01's tail frame resolved = resolve_start_frame("ep01_scene02_shot02", {}) # If no rendered tail frame on disk, it returns fallback or None cleanly without crash self.assertTrue(resolved is None or isinstance(resolved, str)) def test_f12_04_get_shots_for_scene_ordering(self): """F12.4: Verify get_shots_for_scene orders shots numerically.""" from production_orchestrator import get_shots_for_scene shots = get_shots_for_scene("ep01_scene01") if shots: shot_ids = [s[0] for s in shots] nums = [int(re.search(r"shot(\d+)", s).group(1)) for s in shot_ids] self.assertEqual(nums, sorted(nums), "Shots must be sorted numerically by shot number") def test_f12_05_tail_frame_discovery(self): """F12.5: Verify find_tail_frame discovers tail frames in KEYFRAMES_DIR.""" from production_orchestrator import find_tail_frame # Querying an arbitrary shot returns None or Path without raising tf = find_tail_frame("nonexistent_shot") self.assertIsNone(tf) # ------------------------------------------------------------------------- # F13: Master Verification and Assembly Validation # ------------------------------------------------------------------------- def test_f13_01_assemble_feature_script_presence(self): """F13.1: Verify assemble_ep01_feature.py exists and is syntactically valid.""" path = PIPELINE_DIR / "assemble_ep01_feature.py" self.assertTrue(path.exists(), "assemble_ep01_feature.py must exist") # Compile test with open(path, "r", encoding="utf-8") as f: code = f.read() compiled = compile(code, str(path), "exec") self.assertIsNotNone(compiled) def test_f13_02_export_directory_master_files(self): """F13.2: Verify 06_Exports directory exists and holds master outputs.""" self.assertTrue(EXPORTS_DIR.exists(), "06_Exports must exist") def test_f13_03_ffmpeg_stream_mapping_verification(self): """F13.3: Verify assembly scripts map both video [v] and audio [a] streams.""" path = PIPELINE_DIR / "assemble_ep01_feature.py" content = path.read_text(encoding="utf-8") self.assertIn('-map", "[v]"', content) self.assertIn('-map", "[a]"', content) def test_f13_04_no_opencv_concat_violation(self): """F13.4: Verify assembly scripts DO NOT use OpenCV cv2.VideoWriter for master concat.""" for script in PIPELINE_DIR.glob("assemble_*.py"): content = script.read_text(encoding="utf-8") self.assertNotIn( "cv2.VideoWriter", content, f"{script.name} must not use cv2.VideoWriter (violates audio preservation)" ) def test_f13_05_master_audio_aac_sample_rate(self): """F13.5: Verify assembly scripts encode audio with AAC at 48000 Hz.""" path = PIPELINE_DIR / "assemble_ep01_feature.py" content = path.read_text(encoding="utf-8") self.assertIn("aac", content.lower()) self.assertIn("48000", content) if __name__ == "__main__": unittest.main()