Coverage for tests/streamwise_app/test_streamlecture.py: 94%
189 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 StreamLecture.
4"""
6import os
7import sys
8import pytest
10from http import HTTPStatus
12from PIL import Image # noqa: F401 - import before patch.dict to keep PIL in sys.modules
14from quart import Quart
16from unittest.mock import patch
17from unittest.mock import MagicMock
19# Add current path
20sys.path.append(os.getcwd())
22from file_utils import read_file_base64
24from tests.test_utils import temp_sys_path
25from tests.torch_mock import TorchMock
26from tests.streamwise_app.app_test_helpers import check_app_root
27from tests.streamwise_app.app_test_helpers import check_health
28from tests.streamwise_app.app_test_helpers import check_files
29from tests.streamwise_app.app_test_helpers import check_unknown_route
30from tests.streamwise_app.app_test_helpers import check_job_submit_page
31from tests.streamwise_app.app_test_helpers import check_job_status_page
32from tests.streamwise_app.app_test_helpers import check_api_job_status
33from tests.streamwise_app.app_test_helpers import check_api_job_requests
35mock_torch = TorchMock()
37mock_modules = {}
38mock_modules.update(mock_torch.get_sub_modules())
40with patch.dict(sys.modules, mock_modules):
41 with temp_sys_path("apps", "apps/streamlecture"):
42 from apps.streamlecture.streamlecture import StreamLectureApp
43 from apps.streamlecture.streamlecture_job import StreamLectureJob
44 from apps.streamlecture.streamlecture_job import JobStatus
45 from apps.streamlecture.lecture_prompts import IMG_PROMPT
46 from apps.streamlecture.lecture_prompts import VIDEO_PROMPT
47 from tests.streamwise_app.lmm_generator_mock import LMMGeneratorMock
50streamlecture_app = StreamLectureApp()
53@pytest.fixture(name="test_app")
54def _test_app() -> Quart:
55 return streamlecture_app.app
58@pytest.mark.asyncio
59async def test_app(test_app: Quart) -> None:
60 """Check that GET / returns 200."""
61 await check_app_root(test_app, "StreamLecture")
64@pytest.mark.asyncio
65async def test_health(test_app: Quart) -> None:
66 """Check /health."""
67 await check_health(test_app)
70@pytest.mark.asyncio
71async def test_files(test_app: Quart) -> None:
72 """Check /files endpoint."""
73 await check_files(test_app, "streamlecture")
76@pytest.mark.asyncio
77async def test_unknown_route(test_app: Quart) -> None:
78 """Check that an unknown route returns 404."""
79 await check_unknown_route(test_app)
82@pytest.mark.asyncio
83async def test_job_submit_page(test_app: Quart) -> None:
84 """Check the web page for job submission."""
85 await check_job_submit_page(test_app)
88@pytest.mark.asyncio
89async def test_job_status_page(test_app: Quart) -> None:
90 """Check the web page for job status."""
91 await check_job_status_page(test_app)
94@pytest.mark.asyncio
95async def test_api_job_status(test_app: Quart) -> None:
96 """Check the API for job status (returns UNKNOWN for nonexistent jobs)."""
97 await check_api_job_status(test_app)
100@pytest.mark.asyncio
101async def test_api_job_requests(test_app: Quart) -> None:
102 """Check the API for job requests listing (returns empty for nonexistent jobs)."""
103 await check_api_job_requests(test_app)
106@pytest.mark.asyncio
107async def test_submit_job(test_app: Quart) -> None:
108 """Check the API for job requests."""
109 client = test_app.test_client()
111 response = await client.post("/api/job", json={"video_base64": "AAAA"})
112 assert response.status_code == HTTPStatus.BAD_REQUEST
113 response_json = await response.get_json()
114 assert "error" in response_json
115 assert response_json["error"] == "Service manager not initialized"
117 # Mock the service manager
118 streamlecture_app.service_manager = MagicMock()
119 streamlecture_app.service_manager.get_service_url = MagicMock(
120 return_value="http://mock_service_url:1234"
121 )
123 pdf_path = "tests/data/blank.pdf"
124 pdf_base64 = await read_file_base64(pdf_path)
125 response = await client.post("/api/job", json={"pdf_base64": pdf_base64})
126 # The job is accepted; it may fail later if external services are unavailable.
127 response_json = await response.get_json()
128 assert response_json is not None
129 assert "status" in response_json
130 # If the job started before any service call fails, we get job_id back
131 # If a service call fails synchronously (e.g. DNS error), we get an error
132 if response.status_code == HTTPStatus.OK:
133 assert "job_id" in response_json
134 assert response_json["status"] == "success"
135 else:
136 assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
137 assert "error" in response_json
140def test_lecture_prompts() -> None:
141 """Test that lecture prompts are defined and non-empty."""
142 assert IMG_PROMPT
143 assert "classroom" in IMG_PROMPT.lower() or "professor" in IMG_PROMPT.lower()
144 assert VIDEO_PROMPT
145 assert len(VIDEO_PROMPT) > 0
148@pytest.mark.asyncio
149async def test_gen_lecture_no_pdf() -> None:
150 """StreamLectureJob.gen_lecture with missing pdf raises ValueError."""
151 service_manager = MagicMock()
152 job = StreamLectureJob(
153 job_id="test_no_pdf",
154 service_manager=service_manager,
155 )
156 with pytest.raises(ValueError, match="Missing 'pdf_base64' in request"):
157 await job.gen_lecture(pdf_base64=None)
158 job_status = await job.get_status()
159 assert job_status == JobStatus.FAILED
161 del job
162 del service_manager
165@pytest.mark.asyncio
166async def test_gen_lecture_invalid_pdf() -> None:
167 """StreamLectureJob.gen_lecture with invalid base64 raises ValueError."""
168 service_manager = MagicMock()
169 job = StreamLectureJob(
170 job_id="test_invalid_pdf",
171 service_manager=service_manager,
172 )
173 with pytest.raises(ValueError, match="Invalid base64-encoded string"):
174 await job.gen_lecture(pdf_base64="AAAAA")
175 job_status = await job.get_status()
176 assert job_status == JobStatus.FAILED
178 del job
179 del service_manager
182@pytest.mark.asyncio
183async def test_gen_lecture_mock() -> None:
184 """StreamLectureJob.gen_lecture with mocked services completes or fails gracefully."""
185 service_manager = MagicMock()
186 job = StreamLectureJob(
187 job_id="test_gen_lecture_mock",
188 service_manager=service_manager,
189 )
190 job.gen = LMMGeneratorMock()
192 pdf_path = "tests/data/blank.pdf"
193 pdf_base64 = await read_file_base64(pdf_path)
194 try:
195 await job.gen_lecture(pdf_base64=pdf_base64)
196 job_status = await job.get_status()
197 assert job_status == JobStatus.COMPLETED
198 except (FileNotFoundError, ValueError):
199 # ffmpeg not available in this environment; job fails gracefully
200 job_status = await job.get_status()
201 assert job_status == JobStatus.FAILED
203 del job
204 del service_manager
207@pytest.mark.asyncio
208async def test_gen_scene_no_image() -> None:
209 """StreamLectureJob.gen_scene with no image raises ValueError."""
210 service_manager = MagicMock()
211 job = StreamLectureJob(
212 job_id="test_gen_scene_no_image",
213 service_manager=service_manager,
214 )
215 job.gen = LMMGeneratorMock()
216 # With no image set, gen_scene raises ValueError after generating audio
217 with pytest.raises(ValueError, match="Classroom image not available for video generation"):
218 await job.gen_scene(scene_id=0, text="Test text.")
220 del job
221 del service_manager
224@pytest.mark.asyncio
225async def test_gen_scene_audio_only() -> None:
226 """StreamLectureJob.gen_scene in audio_only mode returns audio path."""
227 service_manager = MagicMock()
228 job = StreamLectureJob(
229 job_id="test_gen_scene_audio_only",
230 service_manager=service_manager,
231 )
232 job.gen = LMMGeneratorMock()
233 job.config["output_mode"] = "audio_only"
234 # Set a mock classroom image (required even in audio_only mode)
235 job.image = Image.new("RGB", (640, 400), color="white")
237 result = await job.gen_scene(scene_id=0, text="Test lecture text.")
238 assert result.endswith(".wav")
239 assert os.path.exists(result)
241 del job
242 del service_manager
245@pytest.mark.asyncio
246async def test_gen_scene_video_audio_synced() -> None:
247 """StreamLectureJob.gen_scene in video_audio_synced mode with short audio uses gen_video_audio_from_img."""
248 service_manager = MagicMock()
249 job = StreamLectureJob(
250 job_id="test_gen_scene_synced",
251 service_manager=service_manager,
252 )
253 job.gen = LMMGeneratorMock()
254 job.config["output_mode"] = "video_audio_synced"
255 job.image = Image.new("RGB", (160, 100), color="white")
257 try:
258 result = await job.gen_scene(scene_id=0, text="Test synced scene.")
259 # If ffmpeg is available the result is an mp4 path
260 assert result.endswith(".mp4")
261 assert os.path.exists(result)
262 except FileNotFoundError:
263 pass # ffmpeg not available in test environment
265 del job
266 del service_manager
269@pytest.mark.asyncio
270async def test_gen_scene_video_audio_unsynced() -> None:
271 """StreamLectureJob.gen_scene in video_audio_unsynced mode uses gen_video."""
272 service_manager = MagicMock()
273 job = StreamLectureJob(
274 job_id="test_gen_scene_unsynced",
275 service_manager=service_manager,
276 )
277 job.gen = LMMGeneratorMock()
278 job.config["output_mode"] = "video_audio_unsynced"
279 job.image = Image.new("RGB", (160, 100), color="white")
281 try:
282 result = await job.gen_scene(scene_id=0, text="Test unsynced scene.")
283 assert result.endswith(".mp4")
284 assert os.path.exists(result)
285 except FileNotFoundError:
286 pass # ffmpeg not available in test environment
288 del job
289 del service_manager
292@pytest.mark.asyncio
293async def test_gen_lecture_audio_only() -> None:
294 """StreamLectureJob.gen_lecture in audio_only mode skips video generation."""
295 service_manager = MagicMock()
296 job = StreamLectureJob(
297 job_id="test_gen_lecture_audio_only",
298 service_manager=service_manager,
299 )
300 job.gen = LMMGeneratorMock()
301 job.config["output_mode"] = "audio_only"
303 pdf_path = "tests/data/blank.pdf"
304 pdf_base64 = await read_file_base64(pdf_path)
305 try:
306 await job.gen_lecture(pdf_base64=pdf_base64)
307 job_status = await job.get_status()
308 assert job_status == JobStatus.COMPLETED
309 except (FileNotFoundError, ValueError):
310 job_status = await job.get_status()
311 assert job_status == JobStatus.FAILED
313 del job
314 del service_manager