| 1 | """Regression tests for silent wrong-output bugs in the script2video render path.""" |
| 2 | |
| 3 | import asyncio |
| 4 | import os |
| 5 | import tempfile |
| 6 | import unittest |
| 7 | from unittest.mock import AsyncMock, MagicMock |
| 8 | |
| 9 | from agents.reference_image_selector import select_pairs_by_indices |
| 10 | from agents.storyboard_artist import validate_char_idxs |
| 11 | from interfaces.camera import Camera |
| 12 | from interfaces.shot_description import ShotDescription |
| 13 | from pipelines.script2video_pipeline import ( |
| 14 | Script2VideoPipeline, |
| 15 | _collect_priority_shot_idxs, |
| 16 | _group_shots_into_cameras, |
| 17 | ) |
| 18 | from utils.text import safe_path_component |
| 19 | |
| 20 | |
| 21 | def _shot(idx, cam_idx, variation_type="small", ff_chars=None, lf_chars=None): |
| 22 | return ShotDescription( |
| 23 | idx=idx, |
| 24 | is_last=False, |
| 25 | cam_idx=cam_idx, |
| 26 | visual_desc=f"shot {idx}", |
| 27 | variation_type=variation_type, |
| 28 | variation_reason="r", |
| 29 | ff_desc=f"first frame {idx}", |
| 30 | ff_vis_char_idxs=ff_chars or [], |
| 31 | lf_desc=f"last frame {idx}", |
| 32 | lf_vis_char_idxs=lf_chars or [], |
| 33 | motion_desc="m", |
| 34 | audio_desc="a", |
| 35 | ) |
| 36 | |
| 37 | |
| 38 | class TestCameraGrouping(unittest.TestCase): |
| 39 | def test_out_of_order_camera_indices_group_correctly(self): |
| 40 | # Shot 0 uses camera 1, shot 1 uses camera 0, shot 2 uses camera 1 again. |
| 41 | shots = [_shot(0, cam_idx=1), _shot(1, cam_idx=0), _shot(2, cam_idx=1)] |
| 42 | cameras = _group_shots_into_cameras(shots) |
| 43 | by_idx = {camera.idx: camera for camera in cameras} |
| 44 | self.assertEqual(by_idx[1].active_shot_idxs, [0, 2]) |
| 45 | self.assertEqual(by_idx[0].active_shot_idxs, [1]) |
| 46 | |
| 47 | |
| 48 | class TestPriorityShotIdxs(unittest.TestCase): |
| 49 | def test_priorities_are_shot_indices_not_camera_indices(self): |
| 50 | # Camera 2 depends on shot 7 of camera 0: shot 7 must be prioritized. |
| 51 | camera_tree = [ |
| 52 | Camera(idx=0, active_shot_idxs=[7, 8]), |
| 53 | Camera(idx=2, active_shot_idxs=[9], parent_cam_idx=0, parent_shot_idx=7), |
| 54 | ] |
| 55 | self.assertEqual(_collect_priority_shot_idxs(camera_tree), [7]) |
| 56 | |
| 57 | |
| 58 | class TestEventDictsAreInstanceState(unittest.TestCase): |
| 59 | def _pipeline(self, working_dir): |
| 60 | return Script2VideoPipeline( |
| 61 | chat_model=MagicMock(), |
| 62 | image_generator=MagicMock(), |
| 63 | video_generator=MagicMock(), |
| 64 | working_dir=working_dir, |
| 65 | ) |
| 66 | |
| 67 | def test_two_pipelines_do_not_share_events(self): |
| 68 | with tempfile.TemporaryDirectory() as tmp: |
| 69 | p1 = self._pipeline(os.path.join(tmp, "a")) |
| 70 | p2 = self._pipeline(os.path.join(tmp, "b")) |
| 71 | p1.frame_events[0] = {"first_frame": asyncio.Event()} |
| 72 | p1.shot_desc_events[0] = asyncio.Event() |
| 73 | p1.character_portrait_events[0] = asyncio.Event() |
| 74 | self.assertEqual(p2.frame_events, {}) |
| 75 | self.assertEqual(p2.shot_desc_events, {}) |
| 76 | self.assertEqual(p2.character_portrait_events, {}) |
| 77 | |
| 78 | def test_no_class_level_mutable_event_dicts(self): |
| 79 | for name in ("frame_events", "shot_desc_events", "character_portrait_events"): |
| 80 | self.assertNotIsInstance( |
| 81 | Script2VideoPipeline.__dict__.get(name), dict, |
| 82 | f"{name} must not be shared class state", |
| 83 | ) |
| 84 | |
| 85 | |
| 86 | class TestResumeIncludesNewCameraReference(unittest.IsolatedAsyncioTestCase): |
| 87 | async def test_existing_new_camera_image_is_still_offered_to_selector(self): |
| 88 | with tempfile.TemporaryDirectory() as tmp: |
| 89 | pipeline = Script2VideoPipeline( |
| 90 | chat_model=MagicMock(), |
| 91 | image_generator=MagicMock(), |
| 92 | video_generator=MagicMock(), |
| 93 | working_dir=tmp, |
| 94 | ) |
| 95 | shots = [_shot(0, cam_idx=0), _shot(1, cam_idx=1)] |
| 96 | camera = Camera( |
| 97 | idx=1, active_shot_idxs=[1], |
| 98 | parent_cam_idx=0, parent_shot_idx=0, |
| 99 | missing_info="wrong background", |
| 100 | ) |
| 101 | parent_done = asyncio.Event() |
| 102 | parent_done.set() |
| 103 | pipeline.frame_events = { |
| 104 | 0: {"first_frame": parent_done}, |
| 105 | 1: {"first_frame": asyncio.Event()}, |
| 106 | } |
| 107 | |
| 108 | # Resume state: transition video and new-camera image already on disk. |
| 109 | shot_dir = os.path.join(tmp, "shots", "1") |
| 110 | os.makedirs(shot_dir, exist_ok=True) |
| 111 | new_camera_path = os.path.join(shot_dir, "new_camera_1.png") |
| 112 | open(os.path.join(shot_dir, "transition_video_from_shot_0.mp4"), "wb").close() |
| 113 | open(new_camera_path, "wb").close() |
| 114 | |
| 115 | selector = AsyncMock(return_value={"reference_image_path_and_text_pairs": [], "text_prompt": "p"}) |
| 116 | pipeline.reference_image_selector = MagicMock(select_reference_images_and_generate_prompt=selector) |
| 117 | fake_image = MagicMock() |
| 118 | pipeline.image_generator.generate_single_image = AsyncMock(return_value=fake_image) |
| 119 | |
| 120 | await pipeline.generate_frames_for_single_camera( |
| 121 | camera=camera, |
| 122 | shot_descriptions=shots, |
| 123 | characters=[], |
| 124 | character_portraits_registry={}, |
| 125 | priority_shot_idxs=[], |
| 126 | ) |
| 127 | |
| 128 | selector.assert_awaited_once() |
| 129 | offered = selector.await_args.kwargs["available_image_path_and_text_pairs"] |
| 130 | offered_paths = [pair[0] for pair in offered] |
| 131 | self.assertIn(new_camera_path, offered_paths, |
| 132 | "resumed runs must offer the new-camera reference image to the selector") |
| 133 | |
| 134 | |
| 135 | class TestCharIdxValidation(unittest.TestCase): |
| 136 | def test_valid_indices_pass(self): |
| 137 | validate_char_idxs([0, 1], 2, "ff_vis_char_idxs") |
| 138 | |
| 139 | def test_out_of_range_rejected(self): |
| 140 | with self.assertRaises(ValueError): |
| 141 | validate_char_idxs([0, 2], 2, "ff_vis_char_idxs") |
| 142 | |
| 143 | def test_negative_rejected(self): |
| 144 | with self.assertRaises(ValueError): |
| 145 | validate_char_idxs([-1], 2, "lf_vis_char_idxs") |
| 146 | |
| 147 | |
| 148 | class TestReferenceSelectorIndices(unittest.TestCase): |
| 149 | def test_valid_selection(self): |
| 150 | pairs = [("a.png", "a"), ("b.png", "b")] |
| 151 | self.assertEqual(select_pairs_by_indices(pairs, [1]), [("b.png", "b")]) |
| 152 | |
| 153 | def test_negative_index_rejected(self): |
| 154 | with self.assertRaises(ValueError): |
| 155 | select_pairs_by_indices([("a.png", "a")], [-1]) |
| 156 | |
| 157 | def test_out_of_range_rejected(self): |
| 158 | with self.assertRaises(ValueError): |
| 159 | select_pairs_by_indices([("a.png", "a")], [3]) |
| 160 | |
| 161 | |
| 162 | class TestSafePathComponent(unittest.TestCase): |
| 163 | def test_clean_names_unchanged(self): |
| 164 | self.assertEqual(safe_path_component("Alice"), "Alice") |
| 165 | self.assertEqual(safe_path_component("Bob_2"), "Bob_2") |
| 166 | |
| 167 | def test_unicode_names_preserved(self): |
| 168 | self.assertEqual(safe_path_component("李雷"), "李雷") |
| 169 | |
| 170 | def test_path_separators_removed(self): |
| 171 | self.assertNotIn("/", safe_path_component("a/b")) |
| 172 | self.assertNotIn("\\", safe_path_component("a\\b")) |
| 173 | |
| 174 | def test_traversal_neutralized(self): |
| 175 | cleaned = safe_path_component("../../etc/passwd") |
| 176 | self.assertNotIn("/", cleaned) |
| 177 | self.assertFalse(cleaned.startswith(".")) |
| 178 | |
| 179 | def test_empty_becomes_placeholder(self): |
| 180 | self.assertEqual(safe_path_component(""), "unnamed") |
| 181 | |
| 182 | |
| 183 | if __name__ == "__main__": |
| 184 | unittest.main() |
| 185 |