Coverage for tests/streamwise_app/test_streammovie.py: 99%
333 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 04:47 +0000
1#!/usr/bin/env python3
2"""
3Unit tests for StreamMovie.
4"""
6import os
7import sys
8import json
9import pytest
10import aiofiles
11import aiofiles.os
13from PIL import Image # noqa: F401 - import before patch.dict to keep PIL in sys.modules
15from http import HTTPStatus
17from typing import AsyncGenerator
18from typing import Dict
19from typing import Any
21from quart import Quart
23from unittest.mock import patch
24from unittest.mock import MagicMock
26# Add current path
27sys.path.append(os.getcwd())
29from tests.test_utils import temp_sys_path
30from tests.torch_mock import TorchMock
31from tests.k8s_mock import K8sMock
32from tests.fantasytalking_mock import FantasyTalkingMock
33from tests.streamwise_app.app_test_helpers import check_app_root
34from tests.streamwise_app.app_test_helpers import check_health
35from tests.streamwise_app.app_test_helpers import check_files
36from tests.streamwise_app.app_test_helpers import check_unknown_route
37from tests.streamwise_app.app_test_helpers import check_job_submit_page
38from tests.streamwise_app.app_test_helpers import check_job_status_page
39from tests.streamwise_app.app_test_helpers import check_api_job_status
40from tests.streamwise_app.app_test_helpers import check_api_job_requests
42mock_torch = TorchMock()
43mock_k8s = K8sMock()
44mock_ft = FantasyTalkingMock()
46mock_modules = {
47 "imageio": MagicMock(),
48 "tabulate": MagicMock(),
49 "soundfile": MagicMock(),
50}
51mock_modules.update(mock_torch.get_sub_modules())
52mock_modules.update(mock_k8s.get_sub_modules())
53mock_modules.update(mock_ft.get_sub_modules())
55with patch.dict(sys.modules, mock_modules):
56 with temp_sys_path("apps", "apps/streammovie"):
57 from apps.streammovie.streammovie import StreamMovieApp
58 from apps.streammovie.streammovie_job import StreamMovieJob
59 from apps.streammovie.streammovie_job import JobStatus
61 from tests.streamwise_app.lmm_generator_mock import LMMGeneratorMock
64streammovie_app = StreamMovieApp()
67@pytest.fixture(name="test_app")
68def _test_app() -> Quart:
69 return streammovie_app.app
72# ---------------------------------------------------------------------------
73# Helper: a mock LMMGeneratorMock that also supports gen_text_stream
74# ---------------------------------------------------------------------------
76# Sample JSONL script returned by the LLM mock
77MOCK_SHOT_0 = {
78 "type": "shot_description",
79 "shot_id": "S001",
80 "visual_prompt": "A wide-angle cinematic shot of a cyberpunk city at night.",
81 "negative_prompt": "blur, watermark",
82 "dialogue": None,
83 "technical_specs": {"duration_seconds": 4},
84}
86MOCK_SHOT_1 = {
87 "type": "shot_description",
88 "shot_id": "S002",
89 "visual_prompt": "Close-up of the hero's determined face.",
90 "negative_prompt": "blur",
91 "dialogue": "I will stop them.",
92 "technical_specs": {"duration_seconds": 3},
93}
96def _make_mock_script_chunks() -> Any:
97 """Yield JSONL lines as streaming chunks."""
98 lines = [
99 json.dumps({"type": "movie_metadata", "title": "Test Movie"}),
100 json.dumps(MOCK_SHOT_0),
101 json.dumps(MOCK_SHOT_1),
102 ]
103 for line in lines:
104 yield line + "\n"
107def _make_mock_script_chunks_with_noise() -> Any:
108 """Yield a mix of valid JSONL and prose/noise lines."""
109 items = [
110 "Okay, here's the Movie Bible for a test film.", # prose noise
111 json.dumps({"type": "movie_metadata", "title": "Noisy Movie"}),
112 "```jsonl", # markdown fence noise
113 json.dumps(MOCK_SHOT_0),
114 "This is an explanatory paragraph that should be filtered out.", # prose noise
115 json.dumps(MOCK_SHOT_1),
116 "```", # markdown fence noise
117 "And now, some notes about the next act...", # prose noise
118 ]
119 for item in items:
120 yield item + "\n"
123class LMMGeneratorMovieMock(LMMGeneratorMock):
124 """LMMGeneratorMock that handles gen_text_stream for movie script generation."""
126 def __init__(self) -> None:
127 super().__init__()
128 self.last_messages: list = []
130 async def gen_text_stream(
131 self,
132 *args: Any,
133 **kwargs: Any,
134 ) -> AsyncGenerator[str, None]:
135 self.last_messages = kwargs.get("messages", [])
136 for chunk in _make_mock_script_chunks():
137 yield chunk
139 async def gen_text(
140 self,
141 *args: Any,
142 **kwargs: Any,
143 ) -> str:
144 return ""
146 async def stop(self) -> None:
147 pass
150class LMMGeneratorMovieNoisyMock(LMMGeneratorMock):
151 """LMMGeneratorMock that emits prose noise mixed with valid JSONL."""
153 async def gen_text_stream(
154 self,
155 *args: Any,
156 **kwargs: Any,
157 ) -> AsyncGenerator[str, None]:
158 for chunk in _make_mock_script_chunks_with_noise():
159 yield chunk
161 async def gen_text(
162 self,
163 *args: Any,
164 **kwargs: Any,
165 ) -> str:
166 return ""
168 async def stop(self) -> None:
169 pass
172def _make_job(job_id: str, config: Dict[str, Any] | None = None) -> StreamMovieJob:
173 """Create a StreamMovieJob with a mocked service manager and generator."""
174 service_manager = MagicMock()
175 service_manager.get_service_url = MagicMock(return_value="http://mock:1234")
176 job = StreamMovieJob(job_id=job_id, config=config or {}, service_manager=service_manager)
177 job.gen = LMMGeneratorMovieMock()
178 return job
181def _make_noisy_job(job_id: str, config: Dict[str, Any] | None = None) -> StreamMovieJob:
182 """Create a StreamMovieJob backed by the noisy LLM mock."""
183 service_manager = MagicMock()
184 service_manager.get_service_url = MagicMock(return_value="http://mock:1234")
185 job = StreamMovieJob(job_id=job_id, config=config or {}, service_manager=service_manager)
186 job.gen = LMMGeneratorMovieNoisyMock()
187 return job
190# ---------------------------------------------------------------------------
191# HTTP route tests
192# ---------------------------------------------------------------------------
194@pytest.mark.asyncio
195async def test_app(test_app: Quart) -> None:
196 """Check that GET / returns 200 with the correct content."""
197 await check_app_root(test_app, "StreamMovie")
200@pytest.mark.asyncio
201async def test_health(test_app: Quart) -> None:
202 """Check /health."""
203 await check_health(test_app)
206@pytest.mark.asyncio
207async def test_files(test_app: Quart) -> None:
208 """Check /files endpoint."""
209 await check_files(test_app, "streammovie")
212@pytest.mark.asyncio
213async def test_unknown_route(test_app: Quart) -> None:
214 """Check that an unknown route returns 404."""
215 await check_unknown_route(test_app)
218@pytest.mark.asyncio
219async def test_job_status_page(test_app: Quart) -> None:
220 """Check the web page for job status."""
221 await check_job_status_page(test_app)
224@pytest.mark.asyncio
225async def test_api_job_status(test_app: Quart) -> None:
226 """Check the API for job status (returns UNKNOWN for nonexistent jobs)."""
227 await check_api_job_status(test_app)
230@pytest.mark.asyncio
231async def test_api_job_requests(test_app: Quart) -> None:
232 """Check the API for job requests listing (returns empty for nonexistent jobs)."""
233 await check_api_job_requests(test_app)
236@pytest.mark.asyncio
237async def test_job_submit_page(test_app: Quart) -> None:
238 """Check the web page for job submission."""
239 await check_job_submit_page(test_app)
242@pytest.mark.asyncio
243async def test_submit_job_page(test_app: Quart) -> None:
244 """Check that GET /job returns the submit job form with movie_description field."""
245 client = test_app.test_client()
246 response = await client.get("/job")
247 assert response.status_code == HTTPStatus.OK
248 response_html = await response.get_data(as_text=True)
249 assert "movie_description" in response_html
252@pytest.mark.asyncio
253async def test_submit_job_no_service_manager(test_app: Quart) -> None:
254 """POST /api/job without service manager should return 400."""
255 # Reset service manager to None
256 original = streammovie_app.service_manager
257 streammovie_app.service_manager = None
258 try:
259 client = test_app.test_client()
260 response = await client.post("/api/job", json={"movie_description": "Test movie"})
261 assert response.status_code == HTTPStatus.BAD_REQUEST
262 response_json = await response.get_json()
263 assert "error" in response_json
264 assert response_json["error"] == "Service manager not initialized"
265 finally:
266 streammovie_app.service_manager = original
269@pytest.mark.asyncio
270async def test_submit_job_no_description(test_app: Quart) -> None:
271 """POST /api/job without movie_description triggers job failure, returning 400."""
272 streammovie_app.service_manager = MagicMock()
273 streammovie_app.service_manager.get_service_url = MagicMock(
274 return_value="http://mock_service_url:1234"
275 )
276 client = test_app.test_client()
277 response = await client.post("/api/job", json={"video_base64": "AAAA"})
278 # missing movie_description causes the job to fail with ValueError → 400 BAD REQUEST
279 assert response.status_code == HTTPStatus.BAD_REQUEST
280 response_json = await response.get_json()
281 assert response_json is not None
282 assert response_json["status"] == "error"
283 assert "error" in response_json
284 assert "movie_description" in response_json["error"]
287# ---------------------------------------------------------------------------
288# StreamMovieJob unit tests
289# ---------------------------------------------------------------------------
292@pytest.mark.asyncio
293async def test_get_shot_deadline() -> None:
294 """Deadline should equal submission_time + offset + buffer."""
295 job = _make_job("test_deadline")
296 t0 = job.get_submission_time()
298 deadline_0 = job._get_shot_deadline(0, 4.0)
299 deadline_1 = job._get_shot_deadline(1, 4.0)
300 deadline_5 = job._get_shot_deadline(5, 4.0)
302 assert deadline_0 == pytest.approx(t0 + 0 * 4.0 + 120.0)
303 assert deadline_1 == pytest.approx(t0 + 1 * 4.0 + 120.0)
304 assert deadline_5 == pytest.approx(t0 + 5 * 4.0 + 120.0)
306 # Later shots have a later deadline (gives earlier shots higher priority)
307 assert deadline_1 > deadline_0
308 assert deadline_5 > deadline_1
310 await job.close()
313@pytest.mark.asyncio
314async def test_try_parse_json() -> None:
315 """Valid JSON parses correctly; invalid JSON returns None."""
316 job = _make_job("test_parse_json")
318 result = job._try_parse_json('{"type": "shot_description", "shot_id": 1}', 1)
319 assert result is not None
320 assert result["type"] == "shot_description"
322 result = job._try_parse_json("not json at all", 2)
323 assert result is None
325 result = job._try_parse_json("", 3)
326 assert result is None
328 await job.close()
331@pytest.mark.asyncio
332async def test_stream_movie_script() -> None:
333 """_stream_movie_script should return shot_description objects from LLM stream."""
334 job = _make_job("test_stream_script")
336 shots = await job._stream_movie_script("A cyberpunk heist movie.")
338 # Two shot_description objects are yielded by the mock
339 assert len(shots) == 2
340 assert shots[0]["shot_id"] == "S001"
341 assert shots[1]["shot_id"] == "S002"
343 # Script file should be saved
344 script_path = f"{job.job_path}/movie_script.jsonl"
345 assert await aiofiles.os.path.exists(script_path)
346 async with aiofiles.open(script_path) as f:
347 content = await f.read()
348 # All three JSONL lines should be present (metadata + 2 shots)
349 lines = [line for line in content.strip().splitlines() if line]
350 assert len(lines) == 3
352 await job.close()
355@pytest.mark.asyncio
356async def test_stream_movie_script_returns_all_shots() -> None:
357 """_stream_movie_script returns ALL parsed shots; max_shots cap is applied in gen_movie."""
358 job = _make_job("test_all_shots", config={"max_shots": 1})
360 shots = await job._stream_movie_script("A sci-fi adventure.")
361 # The LLM mock emits 2 shots; _stream_movie_script collects all of them.
362 # The max_shots cap (1) is enforced by gen_movie, not _stream_movie_script.
363 assert len(shots) == 2
365 await job.close()
368@pytest.mark.asyncio
369async def test_gen_shot_no_dialogue() -> None:
370 """_gen_shot without dialogue should produce a plain video."""
371 job = _make_job("test_gen_shot_no_dialogue", config={"output_mode": "video_audio_synced"})
373 shot = {
374 "visual_prompt": "A wide panoramic shot of mountains.",
375 "negative_prompt": "",
376 "dialogue": None,
377 "technical_specs": {"duration_seconds": 4.0},
378 }
380 shot_path = await job._gen_shot(0, shot)
381 assert shot_path is not None
382 assert shot_path.endswith(".mp4")
383 assert await aiofiles.os.path.exists(shot_path)
385 # Image for the shot should also have been saved
386 image_path = f"{job.job_path}/shot_000.png"
387 assert await aiofiles.os.path.exists(image_path)
389 await job.close()
392@pytest.mark.asyncio
393async def test_gen_shot_with_dialogue_synced() -> None:
394 """_gen_shot with dialogue in VIDEO_AUDIO_SYNCED mode uses gen_video_audio_from_img."""
395 job = _make_job("test_gen_shot_synced", config={"output_mode": "video_audio_synced"})
397 shot = {
398 "visual_prompt": "Close-up of the hero.",
399 "negative_prompt": "blur",
400 "dialogue": "I will stop them.",
401 "technical_specs": {"duration_seconds": 3.0},
402 }
404 shot_path = await job._gen_shot(0, shot)
405 assert shot_path is not None
406 assert shot_path.endswith(".mp4")
407 assert await aiofiles.os.path.exists(shot_path)
409 await job.close()
412@pytest.mark.asyncio
413async def test_gen_shot_with_dialogue_unsynced() -> None:
414 """_gen_shot with dialogue in VIDEO_AUDIO_UNSYNCED mode merges audio separately."""
415 job = _make_job("test_gen_shot_unsynced", config={"output_mode": "video_audio_unsynced"})
417 shot = {
418 "visual_prompt": "Hero walking through rain.",
419 "negative_prompt": "",
420 "dialogue": "We have to go now.",
421 "technical_specs": {"duration_seconds": 3.0},
422 }
424 shot_path = await job._gen_shot(0, shot)
425 assert shot_path is not None
426 assert shot_path.endswith(".mp4")
427 assert await aiofiles.os.path.exists(shot_path)
429 await job.close()
432@pytest.mark.asyncio
433async def test_gen_shot_with_dialogue_audio_only() -> None:
434 """In AUDIO_ONLY mode, dialogue presence doesn't trigger video+audio from img path."""
435 job = _make_job("test_gen_shot_audio_only", config={"output_mode": "audio_only"})
437 shot = {
438 "visual_prompt": "Hero close-up.",
439 "negative_prompt": "",
440 "dialogue": "Audio only dialogue.",
441 "technical_specs": {"duration_seconds": 4.0},
442 }
444 shot_path = await job._gen_shot(0, shot)
445 assert shot_path is not None
446 assert shot_path.endswith(".mp4")
447 assert await aiofiles.os.path.exists(shot_path)
449 await job.close()
452@pytest.mark.asyncio
453async def test_gen_movie_missing_description() -> None:
454 """gen_movie with no description should fail immediately with ValueError."""
455 job = _make_job("test_gen_movie_no_desc")
457 with pytest.raises(ValueError, match="Missing 'movie_description'"):
458 await job.gen_movie(None)
460 job_status = await job.get_status()
461 assert job_status == JobStatus.FAILED
463 await job.close()
466@pytest.mark.asyncio
467async def test_gen_movie_success() -> None:
468 """gen_movie end-to-end: produces a final .mp4 file."""
469 job = _make_job("test_gen_movie_success", config={"output_mode": "video_audio_synced"})
471 await job.gen_movie("A short heist movie in Neo-Tokyo.")
473 job_status = await job.get_status()
474 assert job_status == JobStatus.COMPLETED
476 final_path = f"{job.job_path}/{job.job_id}.mp4"
477 assert await aiofiles.os.path.exists(final_path)
479 await job.close()
482@pytest.mark.asyncio
483async def test_gen_movie_max_shots_limits_output() -> None:
484 """When max_shots=1, only one shot should be generated."""
485 job = _make_job("test_gen_movie_max1", config={"max_shots": 1})
487 await job.gen_movie("A one-shot thriller.")
489 # Only shot_000.mp4 should exist; shot_001 should not
490 assert await aiofiles.os.path.exists(f"{job.job_path}/shot_000.mp4")
491 assert not await aiofiles.os.path.exists(f"{job.job_path}/shot_001.mp4")
493 job_status = await job.get_status()
494 assert job_status == JobStatus.COMPLETED
496 await job.close()
499@pytest.mark.asyncio
500async def test_gen_movie_script_saved() -> None:
501 """After gen_movie, the raw JSONL script should be on disk."""
502 job = _make_job("test_gen_movie_script_saved")
504 await job.gen_movie("A space adventure.")
506 script_path = f"{job.job_path}/movie_script.jsonl"
507 assert await aiofiles.os.path.exists(script_path)
508 async with aiofiles.open(script_path) as f:
509 content = await f.read()
510 assert "shot_description" in content
512 await job.close()
515def test_build_movie_messages() -> None:
516 """Test that build_movie_messages includes SYSTEM_PROMPT and user description."""
517 messages = StreamMovieJob.build_movie_messages("a sci-fi thriller")
518 assert len(messages) == 2
519 assert messages[0]["role"] == "system"
520 assert "filmmaker" in messages[0]["content"]
521 assert messages[1]["role"] == "user"
522 assert "sci-fi thriller" in messages[1]["content"]
523 # No shot count instruction when max_shots is not specified
524 assert "EXACTLY" not in messages[1]["content"]
527def test_build_movie_messages_with_max_shots() -> None:
528 """build_movie_messages with max_shots includes the exact-count instruction."""
529 messages = StreamMovieJob.build_movie_messages("a heist drama", max_shots=3)
530 assert len(messages) == 2
531 user_content = messages[1]["content"]
532 assert "heist drama" in user_content
533 assert "EXACTLY 3 shots" in user_content
536@pytest.mark.asyncio
537async def test_api_get_movie_script_jsonl(test_app: Quart) -> None:
538 """GET /api/job/{job_id}/movie_script.jsonl returns the raw JSONL content as text."""
539 job = _make_job("test_api_jsonl")
540 await job._stream_movie_script("A noir detective story.")
542 client = test_app.test_client()
543 response = await client.get(f"/api/job/{job.job_id}/movie_script.jsonl")
544 assert response.status_code == HTTPStatus.OK
545 content = await response.get_data(as_text=True)
546 assert "shot_description" in content
547 assert "movie_metadata" in content
549 await job.close()
552@pytest.mark.asyncio
553async def test_api_get_movie_script_jsonl_not_found(test_app: Quart) -> None:
554 """GET /api/job/{job_id}/movie_script.jsonl returns error JSON when file is missing."""
555 client = test_app.test_client()
556 response = await client.get("/api/job/nonexistent_job/movie_script.jsonl")
557 assert response.status_code == HTTPStatus.OK
558 data = await response.get_json()
559 assert data is not None
560 assert data["status"] == "error"
561 assert "not found" in data["error"]
564@pytest.mark.asyncio
565async def test_stream_movie_script_filters_noise() -> None:
566 """Non-JSON prose lines from the LLM are excluded from the saved script file."""
567 job = _make_noisy_job("test_noise_filter")
569 shots = await job._stream_movie_script("A noisy sci-fi drama.")
571 # 2 shot_description objects should still be extracted despite the noise
572 assert len(shots) == 2
573 assert shots[0]["shot_id"] == "S001"
574 assert shots[1]["shot_id"] == "S002"
576 # The saved script file must contain ONLY valid JSON lines
577 script_path = f"{job.job_path}/movie_script.jsonl"
578 assert await aiofiles.os.path.exists(script_path)
579 async with aiofiles.open(script_path) as f:
580 content = await f.read()
582 lines = [line for line in content.strip().splitlines() if line]
583 # Only 3 valid JSON lines: movie_metadata + 2 shot_descriptions
584 assert len(lines) == 3
585 for line in lines:
586 parsed = json.loads(line) # must not raise
587 assert "type" in parsed
589 # Noise strings must NOT appear in the file at all
590 assert "Okay, here" not in content
591 assert "explanatory paragraph" not in content
592 assert "notes about the next act" not in content
593 assert "```" not in content
595 await job.close()
598@pytest.mark.asyncio
599async def test_stream_movie_script_max_shots_instruction() -> None:
600 """When max_shots is set, the LLM receives an exact shot-count instruction."""
601 job = _make_job("test_max_shots_instruction", config={"max_shots": 3})
602 mock_gen = job.gen
604 await job._stream_movie_script("A thriller.", max_shots=3)
606 # The last messages sent to the LLM must include the exact-count instruction
607 assert mock_gen.last_messages, "No messages were captured by the mock"
608 user_message = mock_gen.last_messages[-1]
609 assert user_message["role"] == "user"
610 assert "EXACTLY 3 shots" in user_message["content"]
612 await job.close()
615@pytest.mark.asyncio
616async def test_stream_movie_script_no_shot_instruction_when_unset() -> None:
617 """When max_shots is not set, the LLM message has no exact-count instruction."""
618 job = _make_job("test_no_shot_instruction")
619 mock_gen = job.gen
621 await job._stream_movie_script("A comedy.", max_shots=-1)
623 assert mock_gen.last_messages, "No messages were captured by the mock"
624 user_message = mock_gen.last_messages[-1]
625 assert "EXACTLY" not in user_message["content"]
627 await job.close()