""" Tier 4: Real-World Application Workloads Test Suite Contains >= 7 realistic, end-to-end workload test cases (Total: 10 test cases). """ import os import sys import re import json import tempfile import subprocess 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, create_synthetic_wav, create_synthetic_video_mp4 ) class TestTier4Workloads(unittest.TestCase): """ Tier 4 tests execute realistic end-to-end application workloads across the pipeline. """ def test_workload_01_full_ep01_screenplay_parse_and_metrics(self): """Workload 1: Parse entire EP01 screenplay, extracting metadata, scene breakdown, and metrics.""" screenplay = read_ep01_screenplay() self.assertGreater(len(screenplay), 2000, "EP01 screenplay must contain substantive content") # Parse character roster char_roster = re.findall(r"\d+\.\s*\*\*([^*]+)\*\*", screenplay) self.assertGreaterEqual(len(char_roster), 5, "EP01 must list main character roster") # Parse scenes scenes = get_screenplay_scenes(screenplay) self.assertGreaterEqual(len(scenes), 12, "EP01 must contain at least 12 scenes") # Parse act breakdown acts = re.findall(r"HỒI\s*(\d+)", screenplay, re.IGNORECASE) self.assertGreaterEqual(len(acts), 3, "Screenplay must delineate 3 acts") def test_workload_02_prompt_registries_schema_and_cross_reference_integrity(self): """Workload 2: Validate JSON registries for Gemini Banana and Muse.ai prompts.""" banana_data = load_banana_prompts_data() muse_data = load_muse_prompts_data() self.assertIsInstance(banana_data, dict, "gemini_banana_prompts.json must be a JSON dictionary") self.assertIsInstance(muse_data, dict, "muse_ai_video_prompts.json must be a JSON dictionary") motion_prompts = muse_data.get("motion_prompts", {}) self.assertGreater(len(motion_prompts), 0, "Motion prompts registry must not be empty") # Inspect prompt content integrity for shot_id, shot_info in list(motion_prompts.items())[:20]: self.assertTrue(isinstance(shot_id, str) and len(shot_id) > 0) if isinstance(shot_info, dict): self.assertIn("motion_prompt", shot_info) def test_workload_03_synthetic_audio_mastering_two_pass_lufs(self): """Workload 3: Generate synthetic audio and execute EBU R128 loudness normalization via FFmpeg.""" from audio_continuity_engine import AudioContinuityEngine, get_ffmpeg ffmpeg = get_ffmpeg() with tempfile.TemporaryDirectory() as tmpdir: tmp_path = Path(tmpdir) input_video = str(tmp_path / "test_input.mp4") output_video = str(tmp_path / "test_normalized.mp4") # Create test video with audio created = create_synthetic_video_mp4(input_video, duration_sec=3.0) self.assertTrue(created, "Must generate synthetic video") # Execute normalization engine = AudioContinuityEngine() success = engine.normalize_loudness(input_video, output_video, target_lufs=-14.0) self.assertTrue(success, "normalize_loudness must succeed on valid video") self.assertTrue(os.path.exists(output_video), "Output video must exist") # Measure loudness via FFmpeg ebur128 measure_cmd = [ ffmpeg, "-i", output_video, "-filter_complex", "ebur128=peak=true", "-f", "null", "-" ] res = subprocess.run(measure_cmd, capture_output=True, text=True) self.assertEqual(res.returncode, 0) stderr = res.stderr # Verify Integrated loudness line exists in summary self.assertIn("Integrated loudness:", stderr) def test_workload_04_synthetic_multi_shot_video_crossfade_assembly(self): """Workload 4: Assemble multiple synthetic video clips using equal-power audio crossfade.""" from audio_continuity_engine import AudioContinuityEngine with tempfile.TemporaryDirectory() as tmpdir: tmp_path = Path(tmpdir) v1 = str(tmp_path / "shot01.mp4") v2 = str(tmp_path / "shot02.mp4") out_master = str(tmp_path / "scene_crossfaded.mp4") # Create 2 synthetic video clips (2.5s each) create_synthetic_video_mp4(v1, duration_sec=2.5) create_synthetic_video_mp4(v2, duration_sec=2.5) engine = AudioContinuityEngine() success = engine.stitch_with_audio_crossfade([v1, v2], out_master, crossfade_dur=0.5) self.assertTrue(success, "stitch_with_audio_crossfade must succeed") self.assertTrue(os.path.exists(out_master)) self.assertGreater(os.path.getsize(out_master), 1000) def test_workload_05_web_review_studio_api_full_catalog_walk(self): """Workload 5: Comprehensive HTTP walk across all Web Review Studio endpoints.""" from server import app from starlette.testclient import TestClient client = TestClient(app) # 1. GET / r_index = client.get("/") self.assertEqual(r_index.status_code, 200) # 2. GET /api/library r_lib = client.get("/api/library") self.assertEqual(r_lib.status_code, 200) lib_data = r_lib.json() self.assertIn("summary", lib_data) self.assertIn("raw_scenes", lib_data) # 3. GET /api/videos r_vids = client.get("/api/videos") self.assertEqual(r_vids.status_code, 200) # 4. GET /api/characters r_chars = client.get("/api/characters") self.assertEqual(r_chars.status_code, 200) # 5. GET /api/prompts r_prompts = client.get("/api/prompts") self.assertEqual(r_prompts.status_code, 200) # 6. GET /api/episodes r_eps = client.get("/api/episodes") self.assertEqual(r_eps.status_code, 200) # 7. GET /api/status r_status = client.get("/api/status") self.assertEqual(r_status.status_code, 200) def test_workload_06_complete_anti_male_tears_audit_across_all_acts(self): """Workload 6: Exhaustive audit of all male dialogue lines and parentheticals across Acts 1-3.""" screenplay = read_ep01_screenplay() violations = find_male_crying_violations(screenplay) if len(violations) > 0: msg = "\n".join([f"Line {v['line_num']}: {v['character']} -> {v['text']}" for v in violations]) self.fail(f"Anti-Male-Tears audit failed ({len(violations)} violations):\n{msg}") def test_workload_07_full_scene05_sub_beat_dramatic_arc_verification(self): """Workload 7: Verify dramatic progression of Scene 05 from stroll to grave discovery to grief.""" screenplay = read_ep01_screenplay() scenes = get_screenplay_scenes(screenplay) sc5_list = [c for h, c in scenes if "CẢNH 05" in h or "CẢNH 5" in h] self.assertTrue(len(sc5_list) > 0, "Scene 05 must exist") sc5 = sc5_list[0] # Check required poetic / dramatic elements self.assertTrue(bool(re.search(r"tiểu khê|suối", sc5, re.IGNORECASE)), "Must feature pastoral stroll") self.assertTrue(bool(re.search(r"sè sè|nấm mồ", sc5, re.IGNORECASE)), "Must feature discovery of mound") self.assertTrue(bool(re.search(r"đau đớn thay|bạc mệnh", sc5, re.IGNORECASE)), "Must feature mourning for fate") self.assertTrue(bool(re.search(r"dấu hài|dấu giày|in rêu", sc5, re.IGNORECASE)), "Must feature footprints in moss") def test_workload_08_production_orchestrator_scene_batch_simulation(self): """Workload 8: Simulate orchestrator batch workflow (list shots, resolve frames, continuity).""" from production_orchestrator import get_shots_for_scene, resolve_start_frame shots = get_shots_for_scene("ep01_scene01") self.assertGreater(len(shots), 0, "Scene 01 must contain registered shots") for shot_id, shot_data in shots: start_frame = resolve_start_frame(shot_id, shot_data) self.assertTrue(start_frame is None or isinstance(start_frame, str)) def test_workload_09_master_feature_grand_assembly_verification(self): """Workload 9: Verify feature assembly pipeline script and FFmpeg command construction.""" script_path = PIPELINE_DIR / "assemble_ep01_feature.py" self.assertTrue(script_path.exists()) code = script_path.read_text(encoding="utf-8") self.assertIn("filter_complex", code) self.assertIn("concat=n=", code) self.assertIn("libx264", code) self.assertIn("aac", code) def test_workload_10_end_to_end_project_health_audit(self): """Workload 10: Health audit of core project directories, files, and licenses.""" core_dirs = [ BASE_DIR / "00_Project_Bible", BASE_DIR / "01_Scripts_And_Episodes", BASE_DIR / "02_AI_Prompts", BASE_DIR / "04_Assets", BASE_DIR / "05_Production_Pipeline", BASE_DIR / "06_Exports", BASE_DIR / "FilmMaker", BASE_DIR / "web_review", BASE_DIR / "license" ] for d in core_dirs: self.assertTrue(d.exists() and d.is_dir(), f"Core directory missing: {d.name}") if __name__ == "__main__": unittest.main()