Coverage for tests/streamwise_app/test_streamshort.py: 100%
250 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 StreamShort.
4"""
6import os
7import sys
8import json
9import pytest
10import aiofiles
12from http import HTTPStatus
14from quart import Quart
16from openai import APIConnectionError
18from unittest.mock import patch
19from unittest.mock import MagicMock
20from unittest.mock import AsyncMock
22# Add current path
23sys.path.append(os.getcwd())
25from file_utils import binary_to_base64
26from media_utils import get_video_with_text
28from tests.test_utils import temp_sys_path
29from tests.torch_mock import TorchMock
30from tests.streamwise_app.app_test_helpers import check_app_root
31from tests.streamwise_app.app_test_helpers import check_health
32from tests.streamwise_app.app_test_helpers import check_files
33from tests.streamwise_app.app_test_helpers import check_unknown_route
34from tests.streamwise_app.app_test_helpers import check_job_submit_page
35from tests.streamwise_app.app_test_helpers import check_job_status_page
36from tests.streamwise_app.app_test_helpers import check_api_job_status
37from tests.streamwise_app.app_test_helpers import check_api_job_requests
39mock_torch = TorchMock()
41mock_cv2 = MagicMock()
43mock_scenedetect = MagicMock()
44mock_scenedetect.VideoManager = MagicMock()
45mock_scenedetect.SceneManager = MagicMock()
46mock_scenedetect.detectors = MagicMock()
47mock_scenedetect.detectors.ContentDetector = MagicMock()
48mock_scenedetect.stats_manager = MagicMock()
49mock_scenedetect.stats_manager.StatsManager = MagicMock()
51mock_modules = {
52 "cv2": mock_cv2,
53 "scenedetect": mock_scenedetect,
54 "scenedetect.detectors": mock_scenedetect.detectors,
55 "scenedetect.stats_manager": mock_scenedetect.stats_manager,
56}
57mock_modules.update(mock_torch.get_sub_modules())
59with patch.dict(sys.modules, mock_modules):
60 with temp_sys_path("apps", "apps/streamshort"):
61 from apps.streamshort.streamshort import StreamShortApp
62 from apps.streamshort.streamshort_job import StreamShortJob
63 from apps.streamshort.streamshort_job import SceneSegment
65streamshort_app = StreamShortApp()
68@pytest.fixture(name="test_app")
69def _test_app() -> Quart:
70 return streamshort_app.app
73@pytest.mark.asyncio
74async def test_app(test_app: Quart) -> None:
75 """Check that GET / returns 200."""
76 await check_app_root(test_app, "StreamShort")
79@pytest.mark.asyncio
80async def test_health(test_app: Quart) -> None:
81 """Check /health."""
82 await check_health(test_app)
85@pytest.mark.asyncio
86async def test_files(test_app: Quart) -> None:
87 """Check /files endpoint."""
88 await check_files(test_app, "streamshort")
91@pytest.mark.asyncio
92async def test_unknown_route(test_app: Quart) -> None:
93 """Check that an unknown route returns 404."""
94 await check_unknown_route(test_app)
97@pytest.mark.asyncio
98async def test_job_submit_page(test_app: Quart) -> None:
99 """Check the web page for job submission."""
100 await check_job_submit_page(test_app)
103@pytest.mark.asyncio
104async def test_job_status_page(test_app: Quart) -> None:
105 """Check the web page for job status."""
106 await check_job_status_page(test_app)
109@pytest.mark.asyncio
110async def test_api_job_status(test_app: Quart) -> None:
111 """Check the API for job status (returns UNKNOWN for nonexistent jobs)."""
112 await check_api_job_status(test_app)
115@pytest.mark.asyncio
116async def test_api_job_requests(test_app: Quart) -> None:
117 """Check the API for job requests listing (returns empty for nonexistent jobs)."""
118 await check_api_job_requests(test_app)
121@pytest.mark.asyncio
122async def test_submit_job(test_app: Quart) -> None:
123 """Check the API for job requests."""
124 client = test_app.test_client()
126 # Mock the service manager
127 streamshort_app.service_manager = MagicMock()
128 streamshort_app.service_manager.get_service_url = MagicMock(
129 return_value="http://mock_service_url:1234"
130 )
132 response = await client.post("/api/job", json={"pdf_base64": "AAAA"})
133 assert response.status_code == HTTPStatus.BAD_REQUEST
134 response_json = await response.get_json()
135 assert "error" in response_json
136 assert response_json["error"] == "Missing 'video_base64' in request"
138 # Bad video data
139 video_base64 = binary_to_base64(b"binary")
140 response = await client.post("/api/job", json={"video_base64": video_base64})
141 assert response.status_code == HTTPStatus.OK
142 response_json = await response.get_json()
143 assert response_json is not None
144 # assert "job_id" in response_json
145 # assert "error" in response_json
146 # assert "Ensure file is valid video" in response_json["error"]
148 # Success case with a valid video file
149 video_binary = await get_video_with_text(
150 width=100, height=100,
151 text="Test Video",
152 duration_seconds=1,
153 )
154 video_base64 = binary_to_base64(video_binary)
155 response = await client.post("/api/job", json={"video_base64": video_base64})
156 assert response.status_code == HTTPStatus.OK
157 response_json = await response.get_json()
158 assert "job_id" in response_json
159 assert response_json["status"] == "success"
162@pytest.mark.asyncio
163async def test_pick_key_frames() -> None:
164 service_manager = MagicMock()
165 service_manager.get_service_url = MagicMock(
166 return_value="http://mock_service_url:1234"
167 )
168 job_id = "test_pick_key_frames"
169 job = StreamShortJob(job_id, service_manager)
171 job.scenes = [
172 SceneSegment(scene_id=0, start_frame=0, end_frame=1, start_sec=0.0, end_sec=1.0),
173 SceneSegment(scene_id=1, start_frame=2, end_frame=3, start_sec=1.0, end_sec=2.0),
174 SceneSegment(scene_id=2, start_frame=4, end_frame=5, start_sec=2.0, end_sec=3.0),
175 ]
177 key_frames = job.pick_key_frames()
178 assert key_frames == [0, 1, 2, 3, 4]
180 await job.close()
183@pytest.mark.asyncio
184async def test_find_scene_for_frame() -> None:
185 service_manager = AsyncMock()
186 service_manager.get_service_url = MagicMock(
187 return_value="http://mock_service_url:1234"
188 )
189 job_id = "tert_find_scene_for_frame"
190 job = StreamShortJob(job_id, service_manager)
192 scene = job.find_scene_for_frame(1)
193 assert scene is None
195 job.scenes = [
196 SceneSegment(scene_id=0, start_frame=0, end_frame=10, start_sec=0.0, end_sec=1.0),
197 SceneSegment(scene_id=1, start_frame=10, end_frame=20, start_sec=1.0, end_sec=2.0),
198 SceneSegment(scene_id=2, start_frame=20, end_frame=30, start_sec=2.0, end_sec=3.0),
199 ]
200 scene = job.find_scene_for_frame(15)
201 assert scene is not None
202 assert scene.scene_id == 1
204 scene = job.find_scene_for_frame(0)
205 assert scene is not None
206 assert scene.scene_id == 0
208 scene = job.find_scene_for_frame(25)
209 assert scene is not None
210 assert scene.scene_id == 2
212 scene = job.find_scene_for_frame(-10)
213 assert scene is None
215 scene = job.find_scene_for_frame(100)
216 assert scene is None
218 await job.close()
221@pytest.mark.asyncio
222async def test_describe_frames() -> None:
223 service_manager = AsyncMock()
224 service_manager.get_service_url = MagicMock(
225 return_value="http://mock_service_url:1234"
226 )
227 job_id = "test_describe_frames"
228 job = StreamShortJob(job_id, service_manager)
230 with pytest.raises(FileNotFoundError, match="frame_0001.jpg"):
231 await job.describe_frames([1, 2, 3])
233 await job.close()
236@pytest.mark.asyncio
237async def test_choose_scenes_for_highlight() -> None:
238 service_manager = AsyncMock()
239 service_manager.get_service_url = MagicMock(
240 return_value="http://mock_service_url:1234"
241 )
242 job_id = "test_choose_scenes_for_highlight"
243 job = StreamShortJob(job_id, service_manager)
245 with pytest.raises(APIConnectionError):
246 await job.choose_scenes_for_highlight()
248 job.scenes = [
249 SceneSegment(scene_id=0, start_frame=0, end_frame=10, start_sec=0.0, end_sec=1.0),
250 SceneSegment(scene_id=1, start_frame=10, end_frame=20, start_sec=1.0, end_sec=2.0),
251 ]
252 with pytest.raises(APIConnectionError):
253 await job.choose_scenes_for_highlight()
255 await job.close()
258@pytest.mark.asyncio
259async def test_save_highlight_short() -> None:
260 service_manager = AsyncMock()
261 service_manager.get_service_url = MagicMock(
262 return_value="http://mock_service_url:1234"
263 )
264 job_id = "test_save_highlight_short"
265 job = StreamShortJob(job_id, service_manager)
267 with pytest.raises(FileNotFoundError, match="video.mp4"):
268 await job.save_highlight_short([])
270 await job.close()
273@pytest.mark.asyncio
274async def test_save_selected_scenes() -> None:
275 service_manager = AsyncMock()
276 service_manager.get_service_url = MagicMock(
277 return_value="http://mock_service_url:1234"
278 )
279 job_id = "test_save_selected_scenes"
280 job = StreamShortJob(job_id, service_manager)
281 try:
282 chosen = [0, 2, 4]
283 await job.save_selected_scenes(chosen)
285 selected_path = f"{job.job_path}/selected_scenes.json"
286 async with aiofiles.open(selected_path) as f:
287 data = json.loads(await f.read())
288 assert data == chosen
290 # Overwrite with a new selection
291 chosen2 = [1, 3]
292 await job.save_selected_scenes(chosen2)
293 async with aiofiles.open(selected_path) as f:
294 data2 = json.loads(await f.read())
295 assert data2 == chosen2
297 # Empty selection is also valid
298 await job.save_selected_scenes([])
299 async with aiofiles.open(selected_path) as f:
300 data3 = json.loads(await f.read())
301 assert data3 == []
302 finally:
303 await job.close()
306@pytest.mark.asyncio
307async def test_chunk_audio_into_scenes() -> None:
308 service_manager = AsyncMock()
309 service_manager.get_service_url = MagicMock(
310 return_value="http://mock_service_url:1234"
311 )
312 job_id = "test_chunk_audio_into_scenes"
313 job = StreamShortJob(job_id, service_manager)
315 await job.chunk_audio_into_scenes()
317 await job.close()
320@pytest.mark.asyncio
321async def test_describe_frames_batch() -> None:
322 service_manager = AsyncMock()
323 service_manager.get_service_url = MagicMock(
324 return_value="http://mock_service_url:1234"
325 )
326 job_id = "test_describe_frames_batch"
327 job = StreamShortJob(job_id, service_manager)
328 try:
329 with pytest.raises(FileNotFoundError, match="frame_0007.jpg"):
330 await job.describe_frames_batch([7, 8, 9])
332 # Mock frame files
333 for frame_id in [1, 2, 3, 4, 5]:
334 frame_path = f"{job.job_path}/frame_{frame_id:04d}.jpg"
335 with open(frame_path, "wb") as f:
336 f.write(b"binary")
337 with pytest.raises(APIConnectionError):
338 await job.describe_frames_batch([1, 2, 3, 4, 5])
339 finally:
340 await job.close()
343@pytest.mark.asyncio
344async def test_transcribe_audio() -> None:
345 service_manager = AsyncMock()
346 service_manager.get_service_url = MagicMock(
347 return_value="http://mock_service_url:1234"
348 )
349 job_id = "test_transcribe_audio"
350 job = StreamShortJob(job_id, service_manager)
351 try:
352 transcription = await job.transcribe_audio()
353 assert transcription == ""
355 # No audio
356 job.scenes = [
357 SceneSegment(scene_id=0, start_frame=0, end_frame=10, start_sec=0.0, end_sec=1.0),
358 SceneSegment(scene_id=1, start_frame=10, end_frame=20, start_sec=1.0, end_sec=2.0),
359 ]
360 transcription = await job.transcribe_audio()
361 assert transcription == ""
363 # Non-existent audio files
364 for scene in job.scenes:
365 scene.audio_path = "non_existent_audio.wav"
366 transcription = await job.transcribe_audio()
367 assert transcription == ""
369 # Mock audio files
370 for scene in job.scenes:
371 scene_id = scene.scene_id
372 scene.audio_path = f"{job.job_path}/scene_{scene_id:04d}_audio.wav"
373 async with aiofiles.open(scene.audio_path, "wb") as f:
374 await f.write(b"binary")
375 transcription = await job.transcribe_audio()
376 assert transcription == ""
378 # Mock transcript generation
379 job.gen.gen_audio_transcript = AsyncMock(return_value=("This is a test transcription.", "en"))
380 transcription = await job.transcribe_audio()
381 assert transcription == "This is a test transcription.\nThis is a test transcription.\n"
382 for scene in job.scenes:
383 if scene.audio_path:
384 assert scene.language == "en"
385 finally:
386 await job.close()
389def test_scene_segement() -> None:
390 scene = SceneSegment(
391 scene_id=1,
392 start_frame=10,
393 end_frame=20,
394 start_sec=1.0,
395 end_sec=2.0
396 )
397 assert scene.scene_id == 1
398 assert scene.start_frame == 10
399 assert scene.end_frame == 20
400 assert scene.start_sec == 1.0
401 assert scene.end_sec == 2.0
402 assert scene.duration_sec == 1
403 assert scene.transcript is None
405 assert scene.get_start() == "00:00:01"
406 assert scene.get_end() == "00:00:02"
407 assert str(scene) == "[ 10- 20, 1.0- 2.0]"
409 scene.add_image_path("/path/to/image.jpg")
410 assert scene.frame_image_paths == ["/path/to/image.jpg"]
412 assert scene.descriptions == []
413 scene.add_description("")
414 assert scene.descriptions == []
416 scene.add_description("A sample description.")
417 assert scene.descriptions == ["A sample description."]
419 assert str(scene) == "[ 10- 20, 1.0- 2.0] | A sample description.... | 1 images"