Coverage for tests/streamwise_app/test_streamedit.py: 100%
199 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 StreamEdit.
4"""
6import base64
7import os
8import sys
9import tempfile
10import pytest
12from http import HTTPStatus
14from PIL import Image
16from quart import Quart
18from unittest.mock import patch
19from unittest.mock import AsyncMock
20from unittest.mock import MagicMock
22# Add current path
23sys.path.append(os.getcwd())
25from media_utils import video_frames_to_base64
27from tests.test_utils import temp_sys_path
28from tests.torch_mock import TorchMock
29from tests.streamwise_app.app_test_helpers import check_app_root
30from tests.streamwise_app.app_test_helpers import check_health
31from tests.streamwise_app.app_test_helpers import check_files
32from tests.streamwise_app.app_test_helpers import check_unknown_route
33from tests.streamwise_app.app_test_helpers import check_job_submit_page
34from tests.streamwise_app.app_test_helpers import check_job_status_page
35from tests.streamwise_app.app_test_helpers import check_api_job_status
36from tests.streamwise_app.app_test_helpers import check_api_job_requests
38mock_torch = TorchMock()
40mock_modules = {}
41mock_modules.update(mock_torch.get_sub_modules())
43with patch.dict(sys.modules, mock_modules):
44 with temp_sys_path("apps", "apps/streamedit"):
45 from apps.streamedit.streamedit import StreamEditApp
47scene_mocks_base = {
48 'scenedetect': MagicMock(),
49 'scenedetect.detectors': MagicMock(),
50 'scenedetect.stats_manager': MagicMock(),
51}
53with patch.dict(sys.modules, {**mock_modules, **scene_mocks_base}):
54 with temp_sys_path("apps", "apps/streamedit"):
55 from apps.streamedit.streamedit_job import StreamEditJob
56 import apps.streamedit.streamedit_job as _sei_module # keep reference for patch.object
58with temp_sys_path("apps"):
59 from scene import SceneSegment
62streamedit_app = StreamEditApp()
65@pytest.fixture(name="test_app")
66def _test_app() -> Quart:
67 return streamedit_app.app
70@pytest.mark.asyncio
71async def test_app(test_app: Quart) -> None:
72 """Check that GET / returns 200."""
73 await check_app_root(test_app, "StreamEdit")
76@pytest.mark.asyncio
77async def test_health(test_app: Quart) -> None:
78 """Check /health."""
79 await check_health(test_app)
82@pytest.mark.asyncio
83async def test_files(test_app: Quart) -> None:
84 """Check /files endpoint."""
85 await check_files(test_app, "streamedit")
88@pytest.mark.asyncio
89async def test_unknown_route(test_app: Quart) -> None:
90 """Check that an unknown route returns 404."""
91 await check_unknown_route(test_app)
94@pytest.mark.asyncio
95async def test_job_submit_page(test_app: Quart) -> None:
96 """Check the web page for job submission."""
97 await check_job_submit_page(test_app)
100@pytest.mark.asyncio
101async def test_job_status_page(test_app: Quart) -> None:
102 """Check the web page for job status."""
103 await check_job_status_page(test_app)
106@pytest.mark.asyncio
107async def test_api_job_status(test_app: Quart) -> None:
108 """Check the API for job status (returns UNKNOWN for nonexistent jobs)."""
109 await check_api_job_status(test_app)
112@pytest.mark.asyncio
113async def test_api_job_requests(test_app: Quart) -> None:
114 """Check the API for job requests listing (returns empty for nonexistent jobs)."""
115 await check_api_job_requests(test_app)
118@pytest.mark.asyncio
119async def test_submit_job(test_app: Quart) -> None:
120 """Check the API for job requests."""
121 client = test_app.test_client()
123 response = await client.post("/api/job", json={"video_base64": "AAAA"})
124 assert response.status_code == HTTPStatus.BAD_REQUEST
125 response_json = await response.get_json()
126 assert "error" in response_json
127 assert response_json["error"] == "Service manager not initialized"
129 # Mock the service manager
130 streamedit_app.service_manager = MagicMock()
131 streamedit_app.service_manager.get_service_url = MagicMock(
132 return_value="http://mock_service_url:1234"
133 )
135 # Bad video data – scenedetect raises "Ensure file is valid video" (or similar)
136 response = await client.post("/api/job", json={"video_base64": "AAAA"})
137 assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
138 response_json = await response.get_json()
139 assert response_json["status"] == "error"
140 assert "error" in response_json
142 # Generate fake video for testing
143 video_frames = [
144 Image.new("RGB", (640, 480), color="blue")
145 for _ in range(8)
146 ]
147 video_base64 = video_frames_to_base64(video_frames)
149 response = await client.post("/api/job", json={"video_base64": video_base64})
150 assert response.status_code == HTTPStatus.OK
151 response_json = await response.get_json()
152 assert response_json["status"] == "success"
153 assert "job_id" in response_json
156def test_edit_prompt() -> None:
157 """Test that EDIT_PROMPT is imported and non-empty."""
158 from apps.streamedit.edit_prompts import EDIT_PROMPT
159 assert EDIT_PROMPT
160 assert "video editor" in EDIT_PROMPT.lower() or "edit" in EDIT_PROMPT.lower()
163@pytest.mark.asyncio
164async def test_streamedit_job_no_video() -> None:
165 """StreamEditJob.gen_edit with missing video raises ValueError."""
166 job = StreamEditJob(
167 job_id="test_no_video",
168 service_manager=MagicMock(),
169 )
170 with pytest.raises(ValueError, match="Missing 'video_base64'"):
171 await job.gen_edit(video_base64=None) # type: ignore[arg-type]
174@pytest.mark.asyncio
175async def test_streamedit_job_detect_scenes_missing_file() -> None:
176 """StreamEditJob.detect_scenes raises FileNotFoundError when video file is absent."""
177 job = StreamEditJob(
178 job_id="test_detect_no_file",
179 service_manager=MagicMock(),
180 )
181 with pytest.raises(FileNotFoundError):
182 await job.detect_scenes("/nonexistent/path/video.mp4")
185@pytest.mark.asyncio
186async def test_streamedit_job_chunk_audio_into_scenes() -> None:
187 """chunk_audio_into_scenes saves scene_{id:03d}.wav and sets scene.audio_path."""
188 with tempfile.TemporaryDirectory() as tmp_dir:
189 job = StreamEditJob(
190 job_id="test_chunk_audio",
191 service_manager=MagicMock(),
192 )
193 job.job_path = tmp_dir
195 # Create a fake input video file
196 video_path = f"{tmp_dir}/video.mp4"
197 with open(video_path, "wb") as f:
198 f.write(b"fake_video_data")
200 fake_audio_base64 = base64.b64encode(b"dummy_audio").decode()
202 job.scenes = [
203 SceneSegment(scene_id=0, start_frame=0, end_frame=30, start_sec=0.0, end_sec=1.0),
204 SceneSegment(scene_id=1, start_frame=30, end_frame=60, start_sec=1.0, end_sec=2.0),
205 ]
207 with patch.object(_sei_module, "extract_audio_from_video", new_callable=AsyncMock,
208 return_value=f"{tmp_dir}/audio.wav"), \
209 patch.object(_sei_module, "read_file_base64", new_callable=AsyncMock,
210 return_value=fake_audio_base64), \
211 patch.object(_sei_module, "chunk_audio_base64", return_value=fake_audio_base64):
213 chunks = await job.chunk_audio_into_scenes()
215 # Both scenes should have audio_path set with consistent naming
216 assert job.scenes[0].audio_path == "scene_000.wav"
217 assert job.scenes[1].audio_path == "scene_001.wav"
219 # Per-scene WAV files must exist on disk
220 assert os.path.exists(f"{tmp_dir}/scene_000.wav")
221 assert os.path.exists(f"{tmp_dir}/scene_001.wav")
223 # Two audio chunks returned
224 assert len(chunks) == 2
227@pytest.mark.asyncio
228async def test_streamedit_job_gen_edit_scene_chunked() -> None:
229 """_gen_edit_scene_chunked splits audio and generates sub-chunk videos."""
230 with tempfile.TemporaryDirectory() as tmp_dir:
231 job = StreamEditJob(
232 job_id="test_scene_chunked",
233 service_manager=MagicMock(),
234 )
235 job.job_path = tmp_dir
237 # Fake scene audio file
238 fake_audio_base64 = base64.b64encode(b"dummy_audio_long").decode()
239 with open(f"{tmp_dir}/scene_005.wav", "wb") as f:
240 f.write(b"dummy_audio_long")
242 scene = SceneSegment(
243 scene_id=5,
244 start_frame=0,
245 end_frame=300,
246 start_sec=0.0,
247 end_sec=12.0, # > MAX_FT_DURATION_SECS (~5.1 s)
248 audio_path="scene_005.wav",
249 )
251 fake_frame = Image.new("RGB", (64, 64), color="red")
252 fake_sub_edit = b"sub_edited_video"
253 fake_concat = b"concatenated_video"
255 with patch.object(_sei_module, "get_audio_chunks_by_silences",
256 return_value=[(0.0, 5.0), (5.0, 10.0), (10.0, 12.0)]), \
257 patch.object(_sei_module, "chunk_video_binary", return_value=b"sub_scene_vid"), \
258 patch.object(_sei_module, "get_video_frames", new_callable=AsyncMock,
259 return_value=[fake_frame] * 10), \
260 patch.object(_sei_module, "chunk_audio_base64", return_value=fake_audio_base64), \
261 patch.object(_sei_module, "concatenate_videos", new_callable=AsyncMock,
262 return_value=fake_concat), \
263 patch.object(job.gen, "gen_video_audio_from_video", new_callable=AsyncMock,
264 return_value=fake_sub_edit):
266 result = await job._gen_edit_scene_chunked(
267 scene=scene,
268 scene_binary=b"scene_vid",
269 scene_audio_path=f"{tmp_dir}/scene_005.wav",
270 scene_audio_base64=fake_audio_base64,
271 edit_prompt="edit",
272 )
274 assert result == fake_concat
277@pytest.mark.asyncio
278async def test_streamedit_job_gen_edit_scene_uses_chunked_for_long_scene() -> None:
279 """gen_edit_scene uses the chunked path when scene duration > MAX_FT_DURATION_SECS."""
280 with tempfile.TemporaryDirectory() as tmp_dir:
281 job = StreamEditJob(
282 job_id="test_long_scene",
283 service_manager=MagicMock(),
284 )
285 job.job_path = tmp_dir
287 # Fake input video file
288 video_path = f"{tmp_dir}/video.mp4"
289 with open(video_path, "wb") as f:
290 f.write(b"fake_video_data")
292 # Fake scene audio file
293 fake_audio_base64 = base64.b64encode(b"dummy_audio_long").decode()
294 with open(f"{tmp_dir}/scene_003.wav", "wb") as f:
295 f.write(b"dummy_audio_long")
297 scene = SceneSegment(
298 scene_id=3,
299 start_frame=0,
300 end_frame=360,
301 start_sec=0.0,
302 end_sec=12.0, # > MAX_FT_DURATION_SECS (~5.1 s)
303 audio_path="scene_003.wav",
304 )
306 fake_concat = b"chunked_edit_result"
308 chunked_mock = AsyncMock(return_value=fake_concat)
309 with patch.object(_sei_module, "chunk_video_binary", return_value=b"scene_vid"), \
310 patch.object(_sei_module, "read_file_base64", new_callable=AsyncMock,
311 return_value=fake_audio_base64), \
312 patch.object(job, "_gen_edit_scene_chunked", chunked_mock):
314 result_path = await job.gen_edit_scene(scene)
316 # Chunked path must have been invoked
317 chunked_mock.assert_called_once()
319 # The edited file must be saved
320 assert os.path.exists(f"{tmp_dir}/scene_003_edit.mp4")
321 assert result_path == f"{tmp_dir}/scene_003_edit.mp4"
324@pytest.mark.asyncio
325async def test_streamedit_job_gen_edit_scene_saves_video() -> None:
326 """gen_edit_scene reads audio from scene.audio_path and saves the edited video."""
327 with tempfile.TemporaryDirectory() as tmp_dir:
328 job = StreamEditJob(
329 job_id="test_scene_edit",
330 service_manager=MagicMock(),
331 )
332 job.job_path = tmp_dir
334 # Create a fake input video file so read_file_bytes can read it
335 video_path = f"{tmp_dir}/video.mp4"
336 with open(video_path, "wb") as f:
337 f.write(b"fake_video_data")
339 # Pre-create the per-scene audio file (as chunk_audio_into_scenes would)
340 fake_audio_base64 = base64.b64encode(b"dummy_audio").decode()
341 with open(f"{tmp_dir}/scene_002.wav", "wb") as f:
342 f.write(b"dummy_audio")
344 scene = SceneSegment(
345 scene_id=2,
346 start_frame=60,
347 end_frame=120,
348 start_sec=2.0,
349 end_sec=4.0,
350 audio_path="scene_002.wav",
351 )
353 with patch.object(_sei_module, "chunk_video_binary", return_value=b"scene_vid"), \
354 patch.object(_sei_module, "get_video_frames", new_callable=AsyncMock, return_value=[]), \
355 patch.object(_sei_module, "read_file_base64", new_callable=AsyncMock,
356 return_value=fake_audio_base64), \
357 patch.object(job.gen, "gen_video_audio_from_video", new_callable=AsyncMock,
358 return_value=b"edited_video"):
360 result_path = await job.gen_edit_scene(scene)
362 # The edited scene file must be saved with consistent naming
363 assert os.path.exists(f"{tmp_dir}/scene_002_edit.mp4")
365 # The returned path must point to the edited scene file
366 assert result_path == f"{tmp_dir}/scene_002_edit.mp4"
369@pytest.mark.asyncio
370async def test_extract_scene_frames_saves_pngs() -> None:
371 """extract_scene_frames saves scene_{id:03d}_frame.png and populates frame_image_paths."""
372 with tempfile.TemporaryDirectory() as tmp_dir:
373 job = StreamEditJob(
374 job_id="test_extract_frames",
375 service_manager=MagicMock(),
376 )
377 job.job_path = tmp_dir
379 job.scenes = [
380 SceneSegment(scene_id=0, start_frame=0, end_frame=30, start_sec=0.0, end_sec=1.0),
381 SceneSegment(scene_id=1, start_frame=30, end_frame=60, start_sec=1.0, end_sec=2.0),
382 ]
384 # Patch cv2 so we don't need a real video file
385 mock_cv2 = MagicMock()
386 mock_cap = MagicMock()
387 mock_cap.read.return_value = (True, MagicMock()) # ok=True, dummy frame
388 mock_cv2.VideoCapture.return_value = mock_cap
390 def _fake_imwrite(path: str, _frame: object) -> bool:
391 with open(path, "wb") as fh:
392 fh.write(b"pngdata")
393 return True
395 mock_cv2.imwrite.side_effect = _fake_imwrite
397 with patch.dict(sys.modules, {"cv2": mock_cv2}):
398 await job.extract_scene_frames()
400 # Each scene should now have exactly one image path
401 assert job.scenes[0].frame_image_paths == ["scene_000_frame.png"]
402 assert job.scenes[1].frame_image_paths == ["scene_001_frame.png"]
405@pytest.mark.asyncio
406async def test_extract_scene_frames_skips_failed_frame() -> None:
407 """extract_scene_frames skips a scene when cv2 cannot read the frame."""
408 with tempfile.TemporaryDirectory() as tmp_dir:
409 job = StreamEditJob(
410 job_id="test_extract_frames_fail",
411 service_manager=MagicMock(),
412 )
413 job.job_path = tmp_dir
415 job.scenes = [
416 SceneSegment(scene_id=0, start_frame=0, end_frame=30, start_sec=0.0, end_sec=1.0),
417 ]
419 mock_cv2 = MagicMock()
420 mock_cap = MagicMock()
421 mock_cap.read.return_value = (False, None) # read fails
422 mock_cv2.VideoCapture.return_value = mock_cap
424 with patch.dict(sys.modules, {"cv2": mock_cv2}):
425 await job.extract_scene_frames()
427 # frame_image_paths must remain empty since reading failed
428 assert job.scenes[0].frame_image_paths == []