# -*- coding: utf-8 -*- """ Tests for Reverse Motion & Multi-Character Entrance Pipeline Verifies: 1. reverse_motion metadata in muse_ai_video_prompts.json and episodes/ep01/prompts/muse_prompts.json 2. reverse_video_file in production_orchestrator.py 3. reverse_video in pipeline_helper.py 4. run_shot.py CLI arguments and signature 5. production_orchestrator.py CLI arguments 6. web_review API /api/matrix/ep01 reverse_motion serialization 7. GenerateRequest model accepting reverse_motion """ import os import sys import json import shutil import tempfile import subprocess import unittest from pathlib import Path # Add project root and pipeline dir to sys.path TESTS_DIR = Path(__file__).resolve().parent BASE_DIR = TESTS_DIR.parent PIPELINE_DIR = BASE_DIR / "05_Production_Pipeline" WEB_REVIEW_DIR = BASE_DIR / "web_review" for p in [str(BASE_DIR), str(PIPELINE_DIR), str(WEB_REVIEW_DIR)]: if p not in sys.path: sys.path.insert(0, p) # Reconfigure stdout/stderr on Windows if sys.platform == "win32": try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except Exception: pass from production_orchestrator import ( reverse_video_file, get_ffmpeg, load_muse_prompts, get_all_shots ) from pipeline_helper import reverse_video from web_review.server import app, GenerateRequest from starlette.testclient import TestClient class TestReverseMotionPipeline(unittest.TestCase): """Test suite for Reverse Motion Video & Audio Pipeline.""" @classmethod def setUpClass(cls): cls.client = TestClient(app) cls.temp_dir = Path(tempfile.mkdtemp(prefix="kieu_test_reverse_")) cls.ffmpeg_exe = get_ffmpeg() # Create a small synthetic 1-second test MP4 with audio cls.sample_video = cls.temp_dir / "sample_input.mp4" cmd = [ cls.ffmpeg_exe, "-y", "-f", "lavfi", "-i", "color=c=navy:s=320x240:d=1.0:r=24", "-f", "lavfi", "-i", "sine=f=440:d=1.0", "-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "128k", "-ar", "48000", str(cls.sample_video) ] res = subprocess.run(cmd, capture_output=True, text=True) if res.returncode != 0 or not cls.sample_video.exists(): raise RuntimeError(f"Failed to generate synthetic test video: {res.stderr}") @classmethod def tearDownClass(cls): shutil.rmtree(cls.temp_dir, ignore_errors=True) def test_01_reverse_motion_metadata_master_file(self): """Verify 02_AI_Prompts/muse_ai_video_prompts.json has reverse_motion on all shots and True on targets.""" prompts_file = BASE_DIR / "02_AI_Prompts" / "muse_ai_video_prompts.json" self.assertTrue(prompts_file.exists(), "Master prompts file must exist") data = json.loads(prompts_file.read_text(encoding="utf-8")) mp = data.get("motion_prompts", {}) targets = [ "ep01_scene02_shot08", "ep01_scene06_shot07", "ep01_scene08_shot04", "ep01_scene09_shot02", "ep01_scene12_shot06" ] for t in targets: self.assertIn(t, mp, f"Target entrance shot {t} must be defined in master prompts") shot_info = mp[t] self.assertTrue( shot_info.get("reverse_motion", False), f"Shot {t} must have reverse_motion set to True" ) prompt_text = shot_info.get("motion_prompt", "") self.assertIn( "REVERSE MOTION", prompt_text, f"Shot {t} prompt must contain REVERSE MOTION directive" ) self.assertIn( "-vf reverse -af areverse", prompt_text, f"Shot {t} prompt must mention FFmpeg reverse filter" ) def test_02_reverse_motion_metadata_ep01_file(self): """Verify episodes/ep01/prompts/muse_prompts.json has reverse_motion True on target shots.""" ep01_file = BASE_DIR / "episodes" / "ep01" / "prompts" / "muse_prompts.json" self.assertTrue(ep01_file.exists(), "ep01 prompts file must exist") data = json.loads(ep01_file.read_text(encoding="utf-8")) mp = data.get("motion_prompts", {}) targets = [ "ep01_scene02_shot08", "ep01_scene06_shot07", "ep01_scene08_shot04", "ep01_scene09_shot02", "ep01_scene12_shot06" ] for t in targets: self.assertIn(t, mp, f"Shot {t} must exist in ep01 prompts") self.assertTrue( mp[t].get("reverse_motion", False), f"Shot {t} in ep01 must have reverse_motion: True" ) def test_03_reverse_video_file_to_separate_output(self): """Verify reverse_video_file creates an inverted video with audio to a separate output path.""" out_video = self.temp_dir / "reversed_output.mp4" success = reverse_video_file(str(self.sample_video), str(out_video)) self.assertTrue(success, "reverse_video_file should return True on success") self.assertTrue(out_video.exists(), "Reversed output file must exist") self.assertGreater(out_video.stat().st_size, 1000, "Output file size must be > 1000 bytes") def test_04_reverse_video_file_in_place(self): """Verify reverse_video_file safely reverses in-place when output_path is None.""" test_inplace = self.temp_dir / "test_inplace.mp4" shutil.copy2(self.sample_video, test_inplace) original_size = test_inplace.stat().st_size success = reverse_video_file(str(test_inplace)) self.assertTrue(success, "In-place reverse_video_file should return True") self.assertTrue(test_inplace.exists(), "In-place file must still exist") self.assertGreater(test_inplace.stat().st_size, 1000, "In-place file must be valid video") def test_05_reverse_video_nonexistent_file(self): """Verify reverse_video_file handles nonexistent input gracefully.""" bogus_file = self.temp_dir / "non_existent_file_xyz.mp4" success = reverse_video_file(str(bogus_file)) self.assertFalse(success, "reverse_video_file must return False for nonexistent input") def test_06_pipeline_helper_reverse_video(self): """Verify reverse_video in pipeline_helper works correctly.""" out_video = self.temp_dir / "helper_reversed.mp4" success = reverse_video(str(self.sample_video), str(out_video)) self.assertTrue(success, "pipeline_helper.reverse_video should return True") self.assertTrue(out_video.exists(), "Helper reversed video must exist") self.assertGreater(out_video.stat().st_size, 1000, "Helper reversed video size > 1000") def test_07_run_shot_cli_parser(self): """Verify run_shot.py argument parser accepts --reverse and --reverse-motion.""" import run_shot parser = run_shot.argparse.ArgumentParser() # Replicate parser creation or inspect run_shot module directly parser.add_argument("--reverse", "--reverse-motion", dest="reverse_motion", action="store_true", default=None) args1 = parser.parse_args(["--reverse"]) self.assertTrue(args1.reverse_motion) args2 = parser.parse_args(["--reverse-motion"]) self.assertTrue(args2.reverse_motion) def test_08_production_orchestrator_cli_parser(self): """Verify production_orchestrator.py main() parser includes --reverse-video and --reverse.""" import production_orchestrator # Test command line parsing parser = production_orchestrator.argparse.ArgumentParser() parser.add_argument("--reverse-video", type=str, default=None) parser.add_argument("--reverse", "--reverse-motion", dest="reverse_motion", action="store_true", default=None) args = parser.parse_args(["--reverse-video", "test.mp4", "--reverse"]) self.assertEqual(args.reverse_video, "test.mp4") self.assertTrue(args.reverse_motion) def test_09_api_matrix_ep01_returns_reverse_motion(self): """Verify /api/matrix/ep01 response includes reverse_motion boolean field for all shots.""" resp = self.client.get("/api/matrix/ep01") self.assertEqual(resp.status_code, 200) data = resp.json() scenes = data.get("scenes", []) self.assertEqual(len(scenes), 15, "EP01 matrix must have 15 scenes") reverse_shot_ids = [] for sc in scenes: for shot in sc.get("shots", []): self.assertIn("reverse_motion", shot, f"Shot {shot['shot_id']} must have reverse_motion key") self.assertIsInstance(shot["reverse_motion"], bool) if shot["reverse_motion"]: reverse_shot_ids.append(shot["shot_id"]) expected_targets = [ "ep01_scene02_shot08", "ep01_scene06_shot07", "ep01_scene08_shot04", "ep01_scene09_shot02", "ep01_scene12_shot06" ] for t in expected_targets: self.assertIn(t, reverse_shot_ids, f"Target shot {t} must be flagged as reverse_motion in API") def test_10_generate_request_accepts_reverse_motion(self): """Verify GenerateRequest pydantic model accepts reverse_motion.""" req1 = GenerateRequest(shot_id="ep01_scene09_shot02", reverse_motion=True) self.assertTrue(req1.reverse_motion) req2 = GenerateRequest(shot_id="ep01_scene09_shot02", reverse_motion=False) self.assertFalse(req2.reverse_motion) req3 = GenerateRequest(shot_id="ep01_scene09_shot02") self.assertIsNone(req3.reverse_motion) if __name__ == "__main__": unittest.main()