""" Tier 2: Boundary & Corner Cases Test Suite (F1 - F13) Contains >= 5 distinct, rigorous boundary/edge-case tests per feature (Total: 65 test cases). """ import os import sys import re import json import tempfile import unittest from pathlib import Path from typing import Dict, List, Optional, Tuple, Any # 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, find_male_crying_violations, create_synthetic_wav, create_synthetic_video_mp4 ) class TestTier2Boundaries(unittest.TestCase): """ Tier 2 tests push each subsystem against adverse, extreme, and edge conditions. """ # ------------------------------------------------------------------------- # F1 Boundaries: Screenplay & Pacing # ------------------------------------------------------------------------- def test_f1_b01_empty_scene_description_detection(self): """F1.B1: Screenplay parser detects and rejects scenes with empty content.""" malformed_script = "#### CẢNH 01: NGOẠI. BẦU TRỜI - NGÀY\n\n#### CẢNH 02: NỘI. PHÒNG KHÁCH - NGÀY\nNội dung cảnh 2..." scenes = re.split(r"(####\s*CẢNH\s*\d+[^:\n]*:[^\n]*)", malformed_script) empty_scenes = [] for i in range(1, len(scenes), 2): header = scenes[i].strip() content = scenes[i + 1].strip() if i + 1 < len(scenes) else "" if len(content) == 0: empty_scenes.append(header) self.assertEqual(len(empty_scenes), 1, "Must detect empty scene content") def test_f1_b02_oversized_single_shot_duration_rejection(self): """F1.B2: Validator rejects individual shots declared with duration > 15s.""" def validate_shot_duration(dur: float) -> bool: return 0 < dur <= 15.0 self.assertTrue(validate_shot_duration(10.0)) self.assertFalse(validate_shot_duration(30.0)) self.assertFalse(validate_shot_duration(0.0)) self.assertFalse(validate_shot_duration(-5.0)) def test_f1_b03_malformed_act_heading_parser_resilience(self): """F1.B3: Act parser tolerates case variance, missing spaces, and punctuation.""" variations = ["HỒI 1: KHỞI ĐẦU", "Hồi 1 - Khởi đầu", "hồi 1", "HỒI 01 :", "## HỒI 1"] pattern = re.compile(r"hồi\s*0?1\b", re.IGNORECASE) for var in variations: self.assertTrue(bool(pattern.search(var)), f"Pattern failed on variation: {var}") def test_f1_b04_negative_or_zero_shot_indexing_rejection(self): """F1.B4: Shot parser rejects non-positive shot index (shot00, shot-1).""" def is_valid_shot_index(shot_id: str) -> bool: m = re.search(r"shot(\d+)", shot_id, re.IGNORECASE) return m is not None and int(m.group(1)) > 0 self.assertTrue(is_valid_shot_index("ep01_scene01_shot01")) self.assertFalse(is_valid_shot_index("ep01_scene01_shot00")) self.assertFalse(is_valid_shot_index("ep01_scene01_shot")) def test_f1_b05_maximum_runtime_ceiling_guard(self): """F1.B5: Calculated EP01 total runtime does not exceed maximum ceiling of 35 minutes (2100s).""" from tests import load_muse_prompts_data muse_data = load_muse_prompts_data() ep01_shots = [k for k in muse_data.get("motion_prompts", {}) if "ep01" in k.lower()] total_sec = len(ep01_shots) * 10 self.assertLessEqual(total_sec, 2100, f"EP01 runtime {total_sec}s must not exceed 2100s (35m)") # ------------------------------------------------------------------------- # F2 Boundaries: Đạm Tiên Grave Sequence # ------------------------------------------------------------------------- def test_f2_b01_premature_weeping_detection(self): """F2.B1: Validator detects and flags premature weeping in sub-beat A.""" def check_beat_a_tears(beat_a_text: str) -> bool: return bool(re.search(r"(khóc|rơi lệ|nước mắt|châu sa)", beat_a_text, re.IGNORECASE)) # Beat A should be serene stroll; weeping here is premature self.assertTrue(check_beat_a_tears("Kiều đi dạo và rơi nước mắt")) self.assertFalse(check_beat_a_tears("Kiều thả bộ dọc ngọn tiểu khê ngắm cảnh thanh thanh")) def test_f2_b02_missing_sub_beat_flagging(self): """F2.B2: Validator detects missing sub-beats in Scene 05.""" def audit_scene05_sub_beats(scene_text: str) -> Dict[str, bool]: return { "beat_a": bool(re.search(r"(tiểu khê|suối|bước dần)", scene_text, re.IGNORECASE)), "beat_b": bool(re.search(r"(sè sè|nấm mồ|vắng tanh)", scene_text, re.IGNORECASE)), "beat_c": bool(re.search(r"(bạc mệnh|dấu hài|dấu giày|in rêu)", scene_text, re.IGNORECASE)), } incomplete_scene = "Kiều đi dạo bên suối rồi hỏi nấm mồ là của ai." audit = audit_scene05_sub_beats(incomplete_scene) self.assertTrue(audit["beat_a"]) self.assertTrue(audit["beat_b"]) self.assertFalse(audit["beat_c"]) def test_f2_b03_sub_beat_shot_partition_minimum(self): """F2.B3: Each sub-beat in Scene 05 must contain >= 4 distinct shots.""" sub_beats_shots = { "C05-A": ["shot01", "shot02", "shot03", "shot04"], "C05-B": ["shot05", "shot06", "shot07", "shot08"], "C05-C": ["shot09", "shot10", "shot11", "shot12"] } for beat, shots in sub_beats_shots.items(): self.assertGreaterEqual(len(shots), 4, f"{beat} must have at least 4 shots") def test_f2_b04_inversion_of_sub_beats_rejection(self): """F2.B4: Sequence validator flags inverted dramatic order (tears before mound encounter).""" def validate_sub_beat_order(text: str) -> bool: m_a = re.search(r"tiểu khê", text, re.IGNORECASE) m_b = re.search(r"sè sè", text, re.IGNORECASE) m_c = re.search(r"dấu hài", text, re.IGNORECASE) if not (m_a and m_b and m_c): return False return m_a.start() < m_b.start() < m_c.start() valid_sample = "Đi bên tiểu khê thấy nấm mồ sè sè rồi thấy dấu hài." inverted_sample = "Thấy dấu hài rồi đi bên tiểu khê thấy nấm mồ sè sè." self.assertTrue(validate_sub_beat_order(valid_sample)) self.assertFalse(validate_sub_beat_order(inverted_sample)) def test_f2_b05_empty_dialogue_in_sub_beat_b(self): """F2.B5: Beat B must contain dialogue exchanges between Thúy Kiều and Vương Quan.""" def extract_dialogue_speakers(text: str) -> List[str]: speakers = [] if "THÚY KIỀU" in text: speakers.append("THÚY KIỀU") if "VƯƠNG QUAN" in text: speakers.append("VƯƠNG QUAN") return speakers sample = "**THÚY KIỀU**: Sao vắng tanh?\n**VƯƠNG QUAN**: Đây là mộ Đạm Tiên." speakers = extract_dialogue_speakers(sample) self.assertIn("THÚY KIỀU", speakers) self.assertIn("VƯƠNG QUAN", speakers) # ------------------------------------------------------------------------- # F3 Boundaries: Anti-Male-Tears # ------------------------------------------------------------------------- def test_f3_b01_female_tears_preserved_without_false_positive(self): """F3.B1: Anti-Male-Tears detector does NOT flag female characters weeping.""" female_sample = """ **THÚY KIỀU** *(Quỳ sụp xuống bên nấm đất, giọt nước mắt lăn dài trên má)* Đau đớn thay phận đàn bà! """ violations = find_male_crying_violations(female_sample) self.assertEqual(len(violations), 0, "Female weeping must NOT trigger male tears violation") def test_f3_b02_figurative_tear_metaphors_in_narration_delineation(self): """F3.B2: Poetic/narrator verses on tears do not flag male character violations.""" narration_sample = """ **NGƯỜI DẪN CHUYỆN (V.O.)** Trải qua một cuộc bể dâu, những điều trông thấy mà đau đớn lòng rơi lệ... """ violations = find_male_crying_violations(narration_sample) self.assertEqual(len(violations), 0, "Narrator V.O. verses must not be attributed to male characters") def test_f3_b03_case_insensitive_tear_keywords_coverage(self): """F3.B3: Anti-Male-Tears validator detects uppercase, lowercase, and mixed-case crying keywords.""" sample = "**KIM TRỌNG**: *(KHÓC NỨC NỞ)*\n**VƯƠNG QUAN**: *(Rơi Lệ)*" violations = find_male_crying_violations(sample) self.assertEqual(len(violations), 2, "Must detect crying regardless of case") def test_f3_b04_punctuation_adjacent_crying_cues_detection(self): """F3.B4: Detector catches crying keywords immediately adjacent to punctuation.""" sample = "KIM TRỌNG: (nước_mắt? Không, rơi.lệ! khóc...)" # Test individual keywords inside regex engine kw_found = any(kw in sample.lower() for kw in ["khóc", "rơi.lệ", "nước"]) self.assertTrue(kw_found, "Must detect keywords with surrounding punctuation") def test_f3_b05_composite_male_actions_validation(self): """F3.B5: Action beat validator flags mixed cues (stoic + crying).""" sample = "**TỪ HẢI**: *(Cơ mặt đanh lại nhưng ngấn lệ trong khóe mắt)*" violations = find_male_crying_violations(sample) self.assertEqual(len(violations), 1, "Must flag even when mixed with stoic phrases") # ------------------------------------------------------------------------- # F4 Boundaries: Universal Acting Directives in Bible # ------------------------------------------------------------------------- def test_f4_b01_bible_character_coverage_completeness(self): """F4.B1: Project Bible registers 32 characters in CHARACTER_BIBLE.md.""" bible_path = BIBLE_DIR / "CHARACTER_BIBLE.md" self.assertTrue(bible_path.exists()) content = bible_path.read_text(encoding="utf-8") # Check character numbers up to at least 25 found_chars = re.findall(r"###\s*\d+[\.\:]\s*([^\n]+)", content) self.assertGreaterEqual(len(found_chars), 15, "Character bible must enumerate full roster") def test_f4_b02_missing_acting_directive_validator_flag(self): """F4.B2: Validator raises error if bible text has zero acting guidelines.""" def check_acting_guidelines_present(text: str) -> bool: return bool(re.search(r"(acting|diễn xuất|quy chuẩn biểu cảm|micro-expression)", text, re.IGNORECASE)) self.assertFalse(check_acting_guidelines_present("Văn bản chỉ mô tả bối cảnh và vũ khí.")) self.assertTrue(check_acting_guidelines_present("Quy chuẩn diễn xuất và micro-expression.")) def test_f4_b03_empty_directive_content_detection(self): """F4.B3: Acting directives section must contain >= 100 substantive words.""" bible_path = BIBLE_DIR / "CHARACTER_BIBLE.md" content = bible_path.read_text(encoding="utf-8") if bible_path.exists() else "" words = content.split() self.assertGreaterEqual(len(words), 500, "Bible must contain extensive directive content") def test_f4_b04_prohibited_acting_keywords_list(self): """F4.B4: Bible enumerates prohibited melodramatic theatrical habits.""" master_bible = BIBLE_DIR / "MASTER_BIBLE.md" char_bible = BIBLE_DIR / "CHARACTER_BIBLE.md" combined = (master_bible.read_text(encoding="utf-8") if master_bible.exists() else "") + \ (char_bible.read_text(encoding="utf-8") if char_bible.exists() else "") has_prohibitions = any(w in combined.lower() for w in ["tránh", "không", "cấm", "tuyệt đối"]) self.assertTrue(has_prohibitions, "Bible must delineate prohibited acting patterns") def test_f4_b05_age_appropriate_acting_rules_contrast(self): """F4.B5: Bible contrasts young male (Kim Trọng 18yo) vs elder male (Vương Ông 55yo) acting demeanor.""" bible_path = BIBLE_DIR / "CHARACTER_BIBLE.md" content = bible_path.read_text(encoding="utf-8") if bible_path.exists() else "" self.assertIn("18", content, "Bible must address 18yo Kim Trọng demeanor") self.assertIn("55", content, "Bible must address 55yo Vương Ông demeanor") # ------------------------------------------------------------------------- # F5 Boundaries: Sweet-Spot Start Frame Protocol # ------------------------------------------------------------------------- def test_f5_b01_invalid_resolution_rejection(self): """F5.B1: Validator rejects resolutions that do not meet 720p sweet spot (1280x720).""" def validate_start_frame_res(res_str: str) -> bool: return res_str in ["1280x720", "720x1280"] self.assertTrue(validate_start_frame_res("1280x720")) self.assertTrue(validate_start_frame_res("720x1280")) self.assertFalse(validate_start_frame_res("1920x1080")) self.assertFalse(validate_start_frame_res("640x480")) self.assertFalse(validate_start_frame_res("invalid")) def test_f5_b02_missing_environment_anchor_detection(self): """F5.B2: Validator detects prompts specifying character but missing environmental anchor.""" def has_both_anchors(prompt_dict: Dict) -> bool: char = prompt_dict.get("character_anchor", "") env = prompt_dict.get("environment_anchor", "") return len(char.strip()) > 0 and len(env.strip()) > 0 self.assertTrue(has_both_anchors({"character_anchor": "Thúy Kiều", "environment_anchor": "Suối nhỏ"})) self.assertFalse(has_both_anchors({"character_anchor": "Thúy Kiều", "environment_anchor": ""})) self.assertFalse(has_both_anchors({"character_anchor": "", "environment_anchor": "Suối nhỏ"})) def test_f5_b03_aspect_ratio_consistency_16_9(self): """F5.B3: Verify 1280x720 matches exactly 16:9 aspect ratio.""" w, h = 1280, 720 ratio = w / h expected = 16.0 / 9.0 self.assertAlmostEqual(ratio, expected, places=3) def test_f5_b04_nonexistent_start_frame_path_handling(self): """F5.B4: Production runner handles nonexistent explicit start frame path cleanly.""" from production_orchestrator import resolve_start_frame res = resolve_start_frame("ep01_scene01_shot01", {}, explicit_frame="nonexistent_frame_12345.png") self.assertTrue(res is None or isinstance(res, str)) def test_f5_b05_forbidden_motion_phrases_scanner(self): """F5.B5: Scanner flags prompts with ambiguous motion risking reverse optical flow.""" def scan_dangerous_motion_phrases(prompt: str) -> List[str]: forbidden = ["bay lùi", "đi lùi", "moonwalk", "bay ngược"] return [w for w in forbidden if w in prompt.lower()] self.assertEqual(len(scan_dangerous_motion_phrases("Đàn én bay ngược chiều")), 1) self.assertEqual(len(scan_dangerous_motion_phrases("Đàn én bay lướt về phía trước")), 0) # ------------------------------------------------------------------------- # F6 Boundaries: Lip-Sync Guard # ------------------------------------------------------------------------- def test_f6_b01_pure_instrumental_shots_exemption(self): """F6.B1: Pure instrumental music shots without dialogue are exempt from lip-sync guard.""" def requires_lip_guard(audio_prompt: str) -> bool: return any(w in audio_prompt.lower() for w in ["từ xa", "ngoài khung", "v.o."]) self.assertFalse(requires_lip_guard("Tiếng đàn tỳ bà độc tấu réo rắt")) self.assertTrue(requires_lip_guard("Tiếng Kiều gọi từ xa: Chàng Kim ơi!")) def test_f6_b02_mixed_on_off_dialogue_handling(self): """F6.B2: Handler distinguishes between on-screen speaker and off-screen listener.""" sample_shot = { "on_screen_character": "Kim Trọng", "speaker": "Thúy Kiều (off-screen)", "audio": "Tiếng Kiều gọi ngoài khung" } # In this scenario, Kim Trọng's lips must be closed should_lock_lips = (sample_shot["on_screen_character"] != sample_shot["speaker"]) self.assertTrue(should_lock_lips) def test_f6_b03_whitespace_padded_lip_guard_matching(self): """F6.B3: Lip-guard regex matches despite arbitrary whitespace or newlines.""" clause = " [KHÓA KHẨU HÌNH BẮT BUỘC]: \n Đôi môi khép chặt tự nhiên " pattern = re.compile(r"khép\s*chặt|lips\s*closed", re.IGNORECASE) self.assertTrue(bool(pattern.search(clause))) def test_f6_b04_on_screen_speaking_not_over_guarded(self): """F6.B4: On-screen speaker shots should NOT have closed lips clause.""" on_screen_shot = { "motion_prompt": "Thúy Kiều mấp máy môi đọc câu thơ tuyệt mệnh.", "type": "on_screen" } has_closed_lips = "khép chặt" in on_screen_shot["motion_prompt"] self.assertFalse(has_closed_lips) def test_f6_b05_truncated_guard_clause_detection(self): """F6.B5: Validator flags partially truncated guard clauses.""" def is_full_guard(text: str) -> bool: return "khép" in text and ("chặt" in text or "môi" in text) self.assertTrue(is_full_guard("[KHÓA KHẨU HÌNH]: Môi khép chặt")) self.assertFalse(is_full_guard("[KHÓA KHẨU HÌNH]:")) # ------------------------------------------------------------------------- # F7 Boundaries: Prompt Registry # ------------------------------------------------------------------------- def test_f7_b01_whitespace_in_shot_ids_rejection(self): """F7.B1: Validator flags shot IDs with leading or trailing whitespace.""" def is_clean_shot_id(shot_id: str) -> bool: return shot_id == shot_id.strip() and " " not in shot_id self.assertTrue(is_clean_shot_id("ep01_scene01_shot01")) self.assertFalse(is_clean_shot_id(" ep01_scene01_shot01")) self.assertFalse(is_clean_shot_id("ep01_scene01_shot01 ")) self.assertFalse(is_clean_shot_id("ep01_scene01 shot01")) def test_f7_b02_duplicate_shot_id_detection(self): """F7.B2: Validator detects duplicate shot IDs in prompt list.""" shot_list = ["ep01_scene01_shot01", "ep01_scene01_shot02", "ep01_scene01_shot01"] duplicates = set([x for x in shot_list if shot_list.count(x) > 1]) self.assertEqual(len(duplicates), 1) self.assertIn("ep01_scene01_shot01", duplicates) def test_f7_b03_corrupted_json_syntax_resilience(self): """F7.B3: JSON loader handles malformed JSON string without unhandled crash.""" corrupted = '{"motion_prompts": {"shot01": "test",}}' # trailing comma try: json.loads(corrupted) self.fail("Should raise JSONDecodeError") except json.JSONDecodeError: pass # Expected def test_f7_b04_missing_audio_guard_flagging(self): """F7.B4: Validator catches single prompt missing audio guard.""" prompts = { "shot01": {"motion": "ok", "audio_guard": "Tuyệt đối không sinh nhạc nền"}, "shot02": {"motion": "missing guard"} } def check_guards(p_dict: Dict) -> List[str]: return [k for k, v in p_dict.items() if "tuyệt đối không sinh nhạc nền" not in json.dumps(v, ensure_ascii=False).lower()] missing = check_guards(prompts) self.assertEqual(missing, ["shot02"]) def test_f7_b05_extremely_long_prompt_truncation_warning(self): """F7.B5: Prompt validator flags prompts exceeding 2000 character API limit.""" oversized_prompt = "A" * 2500 self.assertGreater(len(oversized_prompt), 2000) # ------------------------------------------------------------------------- # F8 Boundaries: Audio Continuity Engine # ------------------------------------------------------------------------- def test_f8_b01_zero_length_input_list_handling(self): """F8.B1: stitch_with_audio_crossfade returns False on empty video list.""" from audio_continuity_engine import AudioContinuityEngine engine = AudioContinuityEngine() result = engine.stitch_with_audio_crossfade([], "output.mp4") self.assertFalse(result) def test_f8_b02_single_file_crossfade_passthrough(self): """F8.B2: Filter string generator handles single file without acrossfade filter crash.""" video_paths = ["file1.mp4"] filter_parts = [] if len(video_paths) == 1: filter_parts.append("[0:a]anull[a_faded]") self.assertIn("[0:a]anull[a_faded]", filter_parts) def test_f8_b03_nonexistent_file_handling(self): """F8.B3: inspect_shot_audio returns has_audio=False for nonexistent path.""" from audio_continuity_engine import AudioContinuityEngine engine = AudioContinuityEngine() info = engine.inspect_shot_audio("nonexistent_path_xyz.mp4") self.assertFalse(info["has_audio"]) self.assertEqual(info.get("max_volume_db", -99.0), -99.0) def test_f8_b04_zero_audio_stream_handling(self): """F8.B4: inspect_shot_audio assigns default volume -99.0 dB when file has 0 audio streams.""" from audio_continuity_engine import AudioContinuityEngine engine = AudioContinuityEngine() # Verify default return on missing audio rep = {"file": "silent.mp4", "has_audio": False, "max_volume_db": -99.0, "mean_volume_db": -99.0} self.assertEqual(rep["mean_volume_db"], -99.0) def test_f8_b05_negative_crossfade_duration_validation(self): """F8.B5: Validator rejects crossfade duration <= 0.""" def validate_crossfade(dur: float) -> bool: return dur > 0.0 self.assertTrue(validate_crossfade(1.0)) self.assertTrue(validate_crossfade(0.5)) self.assertFalse(validate_crossfade(0.0)) self.assertFalse(validate_crossfade(-1.0)) # ------------------------------------------------------------------------- # F9 Boundaries: Continuous Festival Ambience # ------------------------------------------------------------------------- def test_f9_b01_missing_ambience_fallback_graceful(self): """F9.B1: layer_bgm_with_ducking returns False if input files do not exist.""" from audio_continuity_engine import AudioContinuityEngine engine = AudioContinuityEngine() res = engine.layer_bgm_with_ducking("none1.mp4", "none2.mp3", "out.mp4") self.assertFalse(res) def test_f9_b02_corrupted_audio_file_detection(self): """F9.B2: Audio inspector flags 0-byte audio file as unusable.""" with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tf: tf_path = tf.name try: # 0-byte file from audio_continuity_engine import AudioContinuityEngine engine = AudioContinuityEngine() info = engine.inspect_shot_audio(tf_path) self.assertFalse(info.get("has_audio", False)) finally: if os.path.exists(tf_path): os.unlink(tf_path) def test_f9_b03_ambience_volume_ducking_bounds(self): """F9.B3: Ducking level is bounded between -24.0 dB and -6.0 dB.""" def is_valid_duck_level(duck_db: float) -> bool: return -24.0 <= duck_db <= -6.0 self.assertTrue(is_valid_duck_level(-14.0)) self.assertFalse(is_valid_duck_level(-50.0)) self.assertFalse(is_valid_duck_level(0.0)) def test_f9_b04_looping_behavior_on_short_ambience(self): """F9.B4: FFmpeg command uses -stream_loop -1 for continuous looping of background track.""" import inspect from audio_continuity_engine import AudioContinuityEngine src = inspect.getsource(AudioContinuityEngine.layer_bgm_with_ducking) self.assertIn("-stream_loop", src) self.assertIn("-1", src) def test_f9_b05_stereo_channel_consistency_validation(self): """F9.B5: Audio inspector checks stereo channel count (channels == 2).""" mock_stream = {"channels": 2, "sample_rate": 48000} self.assertEqual(mock_stream["channels"], 2) # ------------------------------------------------------------------------- # F10 Boundaries: EBU R128 Normalization # ------------------------------------------------------------------------- def test_f10_b01_extreme_quiet_input_gain_limit(self): """F10.B1: Loudnorm filter string includes standard target -14 even for quiet inputs.""" from audio_continuity_engine import AudioContinuityEngine engine = AudioContinuityEngine() self.assertIsNotNone(engine) def test_f10_b02_extreme_loud_input_attenuation(self): """F10.B2: True peak limit ensures no clipping above -1.0 dBTP.""" tp_limit = -1.0 self.assertLess(tp_limit, 0.0, "True peak must be strictly below 0 dBFS/dBTP") def test_f10_b03_pure_silence_input_handling(self): """F10.B3: Inspector returns -99.0 dB for silent material.""" rep = {"max_volume_db": -99.0, "mean_volume_db": -99.0} self.assertEqual(rep["mean_volume_db"], -99.0) def test_f10_b04_short_duration_material_resilience(self): """F10.B4: normalize_loudness handles non-existent or truncated inputs safely.""" from audio_continuity_engine import AudioContinuityEngine engine = AudioContinuityEngine() res = engine.normalize_loudness("nonexistent_short.mp4", "out.mp4") self.assertFalse(res) def test_f10_b05_non_standard_sample_rate_resampling(self): """F10.B5: Output encoding forces -ar 48000 for standard resampling.""" import inspect from audio_continuity_engine import AudioContinuityEngine src = inspect.getsource(AudioContinuityEngine.normalize_loudness) self.assertIn("48000", src) # ------------------------------------------------------------------------- # F11 Boundaries: Web Review Studio # ------------------------------------------------------------------------- def test_f11_b01_concurrent_generate_conflict_409(self): """F11.B1: /api/generate returns 409 Conflict when status is already generating.""" from server import app, gen_state from starlette.testclient import TestClient client = TestClient(app) prev_status = gen_state["status"] try: gen_state["status"] = "generating" res = client.post("/api/generate", json={"prompt": "test"}) self.assertEqual(res.status_code, 409) finally: gen_state["status"] = prev_status def test_f11_b02_invalid_episode_path_traversal_404(self): """F11.B2: /api/episode/{filename} rejects path traversal attempts with 404.""" from server import app from starlette.testclient import TestClient client = TestClient(app) res = client.get("/api/episode/../../windows/system32/cmd.exe") self.assertEqual(res.status_code, 404) def test_f11_b03_empty_prompt_generation_validation(self): """F11.B3: Request schema validates prompt field presence.""" from server import GenerateRequest from pydantic import ValidationError try: GenerateRequest(prompt="valid prompt") except ValidationError: self.fail("Valid prompt should not raise ValidationError") def test_f11_b04_library_cache_invalidation_window(self): """F11.B4: Verify build_library_data cache interval is 3.0 seconds.""" import server self.assertEqual(server._cache_time >= 0, True) def test_f11_b05_nonexistent_video_category_query(self): """F11.B5: Querying nonexistent category returns empty list rather than 500 error.""" from server import app from starlette.testclient import TestClient client = TestClient(app) res = client.get("/api/videos?category=nonexistent_xyz") self.assertEqual(res.status_code, 200) self.assertEqual(res.json()["videos"], []) # ------------------------------------------------------------------------- # F12 Boundaries: Production Orchestrator CLI # ------------------------------------------------------------------------- def test_f12_b01_invalid_shot_id_handling(self): """F12.B1: find_rendered_video returns None for nonexistent shot ID.""" from production_orchestrator import find_rendered_video v = find_rendered_video("shot_nonexistent_999999") self.assertIsNone(v) def test_f12_b02_missing_input_frame_graceful_fallback(self): """F12.B2: resolve_start_frame does not crash when explicit frame is missing.""" from production_orchestrator import resolve_start_frame res = resolve_start_frame("ep01_scene01_shot01", {}, explicit_frame="nonexistent.png") self.assertTrue(res is None or isinstance(res, str)) def test_f12_b03_first_shot_of_scene_tail_frame_lookup(self): """F12.B3: Shot 01 of scene does not attempt to lookup Shot 00.""" from production_orchestrator import resolve_start_frame # Shot 01 has shot_num == 1, so it shouldn't look for shot 0 res = resolve_start_frame("ep01_scene02_shot01", {}) self.assertTrue(res is None or isinstance(res, str)) def test_f12_b04_empty_scene_id_query(self): """F12.B4: get_shots_for_scene with empty string returns empty list.""" from production_orchestrator import get_shots_for_scene shots = get_shots_for_scene("") self.assertEqual(len(shots), 0) def test_f12_b05_unsupported_cli_command_error_code(self): """F12.B5: CLI exits with code 2 on unknown argument.""" import subprocess proc = subprocess.run( [sys.executable, str(PIPELINE_DIR / "production_orchestrator.py"), "--invalid-flag-xyz"], capture_output=True, text=True, encoding="utf-8", errors="replace" ) self.assertEqual(proc.returncode, 2) # ------------------------------------------------------------------------- # F13 Boundaries: Master Assembly # ------------------------------------------------------------------------- def test_f13_b01_zero_byte_video_detection(self): """F13.B1: find_rendered_video filters out 0-byte or corrupted files.""" from production_orchestrator import find_rendered_video with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tf: tf_path = Path(tf.name) try: # 0-byte file should not be accepted as rendered video self.assertEqual(tf_path.stat().st_size, 0) finally: if tf_path.exists(): tf_path.unlink() def test_f13_b02_mismatched_resolution_inputs_handling(self): """F13.B2: Concat filter handles scale normalization for differing resolutions.""" cmd_filter = "[0:v]scale=1280:720[v0];[1:v]scale=1280:720[v1];[v0][v1]concat=n=2:v=1:a=0[v]" self.assertIn("scale=1280:720", cmd_filter) def test_f13_b03_audio_channel_count_mismatch_handling(self): """F13.B3: Audio filter enforces 2 channels via aformat or aresample.""" cmd_filter = "aformat=channel_layouts=stereo:sample_rates=48000" self.assertIn("stereo", cmd_filter) def test_f13_b04_truncated_video_stream_detection(self): """F13.B4: Validator detects if video file is missing either video or audio stream.""" mock_probe = {"streams": [{"codec_type": "video"}]} has_audio = any(s.get("codec_type") == "audio" for s in mock_probe["streams"]) self.assertFalse(has_audio) def test_f13_b05_export_path_collision_protection(self): """F13.B5: FFmpeg invocation includes -y flag to safely overwrite or handle existing target.""" import inspect from production_orchestrator import concat_scene_shots src = inspect.getsource(concat_scene_shots) self.assertIn('"-y"', src) if __name__ == "__main__": unittest.main()