Coverage for tests/streamwise_app/test_streamchat.py: 100%
252 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 StreamChat.
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 tests.test_utils import temp_sys_path
23from tests.torch_mock import TorchMock
24from tests.streamwise_app.app_test_helpers import check_app_root
25from tests.streamwise_app.app_test_helpers import check_health
26from tests.streamwise_app.app_test_helpers import check_files
27from tests.streamwise_app.app_test_helpers import check_unknown_route
28from tests.streamwise_app.app_test_helpers import check_job_submit_page
29from tests.streamwise_app.app_test_helpers import check_job_status_page
30from tests.streamwise_app.app_test_helpers import check_api_job_status
31from tests.streamwise_app.app_test_helpers import check_api_job_requests
33mock_torch = TorchMock()
35mock_modules = {}
36mock_modules.update(mock_torch.get_sub_modules())
38with patch.dict(sys.modules, mock_modules):
39 with temp_sys_path("apps", "apps/streamchat"):
40 from apps.streamchat.streamchat import StreamChatApp
41 from apps.streamchat.streamchat import get_chat_history_from_file
42 from apps.streamchat.streamchat import parse_chat_history
43 from apps.streamchat.streamchat_job import StreamChatJob
44 from apps.streamchat.streamchat_job import remove_emojis
45 from apps.streamchat.streamchat_job import JobStatus
46 from character import Character
47 from tests.streamwise_app.lmm_generator_mock import LMMGeneratorMock
50streamchat_app = StreamChatApp()
53@pytest.fixture(name="test_app")
54def _test_app() -> Quart:
55 return streamchat_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, "StreamChat")
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, "streamchat")
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 streamchat_app.service_manager = MagicMock()
119 streamchat_app.service_manager.get_service_url = MagicMock(
120 return_value="http://mock_service_url:1234"
121 )
123 response = await client.post("/api/job", json={"video_base64": "AAAA"})
124 # assert response.status_code == HTTPStatus.OK
125 assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
126 response_json = await response.get_json()
127 # assert "job_id" in response_json
128 # assert response_json["status"] == "success"
129 assert response_json["status"] == "error"
130 assert "error" in response_json
131 assert "Error generating image" in response_json["error"]
134def test_remove_emojis() -> None:
135 """Test the remove_emojis function."""
136 text_with_emojis = "Hello, world! 😊🚀🌟"
137 text_without_emojis = "Hello, world! "
139 assert remove_emojis(text_with_emojis) == text_without_emojis
140 assert remove_emojis(text_without_emojis) == text_without_emojis
143@pytest.mark.asyncio
144async def test_gen_chat_base_mock() -> None:
145 """StreamChatJob.gen_chat_base with mocked services generates response."""
146 service_manager = MagicMock()
147 job = StreamChatJob(
148 job_id="test_gen_chat_base_mock",
149 service_manager=service_manager,
150 )
151 job.gen = LMMGeneratorMock()
153 await job.gen_chat_base()
154 job_status = await job.get_status()
155 # The gen_chat_base starts the character/image and sends a chat message
156 # With mock services that don't fail it should at minimum start
157 assert job_status in (JobStatus.COMPLETED, JobStatus.FAILED)
159 del job
160 del service_manager
163@pytest.mark.asyncio
164async def test_gen_chat_history() -> None:
165 """StreamChatJob.get_chat_history returns messages list."""
166 service_manager = MagicMock()
167 job = StreamChatJob(
168 job_id="test_gen_chat_history",
169 service_manager=service_manager,
170 )
171 history = await job.get_chat_history()
172 assert isinstance(history, list)
173 # System prompt is set in __init__
174 assert len(history) >= 1
175 assert history[0]["role"] == "system"
177 del job
178 del service_manager
181@pytest.mark.asyncio
182async def test_get_msg_id() -> None:
183 """StreamChatJob._get_msg_id returns 0 before any chat turns."""
184 service_manager = MagicMock()
185 job = StreamChatJob(job_id="test_msg_id", service_manager=service_manager)
186 assert job._get_msg_id() == 0
187 del job
188 del service_manager
191@pytest.mark.asyncio
192async def test_gen_chat_text() -> None:
193 """StreamChatJob.gen_chat_text calls gen.gen_text and appends messages."""
194 service_manager = MagicMock()
195 job = StreamChatJob(job_id="test_gen_chat_text", service_manager=service_manager)
196 job.gen = LMMGeneratorMock()
198 response_text = await job.gen_chat_text(user_message="Hello!", msg_id=0)
199 assert isinstance(response_text, str)
200 assert len(response_text) > 0
201 # Messages should now contain system + user + assistant
202 assert len(job.messages) == 3
203 assert job.messages[1]["role"] == "user"
204 assert job.messages[2]["role"] == "assistant"
206 del job
207 del service_manager
210@pytest.mark.asyncio
211async def test_gen_chat_audio() -> None:
212 """StreamChatJob.gen_chat_audio produces a WAV file."""
214 service_manager = MagicMock()
215 job = StreamChatJob(job_id="test_gen_chat_audio", service_manager=service_manager)
216 job.gen = LMMGeneratorMock()
217 job.character = Character(name="Alice", gender="Female", speech_speed=1.0)
219 audio_base64 = await job.gen_chat_audio(response_text="Hello world.", msg_id=0)
220 assert isinstance(audio_base64, str)
221 assert len(audio_base64) > 0
223 del job
224 del service_manager
227@pytest.mark.asyncio
228async def test_gen_chat() -> None:
229 """StreamChatJob.gen_chat returns a reply dict."""
231 service_manager = MagicMock()
232 job = StreamChatJob(job_id="test_gen_chat", service_manager=service_manager)
233 job.gen = LMMGeneratorMock()
234 job.image = Image.new("RGB", (160, 100), color="white")
235 job.character = Character(name="Alice", gender="Female", speech_speed=1.0)
237 result = await job.gen_chat(user_message="Say something.")
238 assert "reply" in result
239 assert isinstance(result["reply"], str)
240 assert "id" in result
242 del job
243 del service_manager
246@pytest.mark.asyncio
247async def test_chat_route_not_found(test_app: Quart) -> None:
248 """POST /chat/<job_id> for unknown job returns 404."""
249 client = test_app.test_client()
250 response = await client.post("/chat/unknown_job_id", data={"message": "Hi"})
251 assert response.status_code == HTTPStatus.NOT_FOUND
252 response_json = await response.get_json()
253 assert "error" in response_json
256@pytest.mark.asyncio
257async def test_chat_route_no_message(test_app: Quart) -> None:
258 """POST /chat/<job_id> with no text and no audio returns 400."""
260 service_manager = MagicMock()
261 job = StreamChatJob(job_id="test_no_msg_job", service_manager=service_manager)
262 job.gen = LMMGeneratorMock()
263 job.image = Image.new("RGB", (160, 100), color="white")
264 job.character = Character(name="Alice", gender="Female", speech_speed=1.0)
265 streamchat_app.jobs["test_no_msg_job"] = job
267 client = test_app.test_client()
268 response = await client.post("/chat/test_no_msg_job", data={})
269 assert response.status_code == HTTPStatus.BAD_REQUEST
270 response_json = await response.get_json()
271 assert "error" in response_json
273 del streamchat_app.jobs["test_no_msg_job"]
274 del job
275 del service_manager
278@pytest.mark.asyncio
279async def test_chat_route_with_message(test_app: Quart) -> None:
280 """POST /chat/<job_id> with a text message returns a reply."""
282 service_manager = MagicMock()
283 job = StreamChatJob(job_id="test_chat_msg_job", service_manager=service_manager)
284 job.gen = LMMGeneratorMock()
285 job.image = Image.new("RGB", (160, 100), color="white")
286 job.character = Character(name="Alice", gender="Female", speech_speed=1.0)
287 streamchat_app.jobs["test_chat_msg_job"] = job
289 client = test_app.test_client()
290 response = await client.post(
291 "/chat/test_chat_msg_job",
292 form={"message": "Hello!"},
293 )
294 assert response.status_code == HTTPStatus.OK
295 response_json = await response.get_json()
296 assert response_json["status"] == "ok"
297 assert "reply" in response_json
299 del streamchat_app.jobs["test_chat_msg_job"]
300 del job
301 del service_manager
304@pytest.mark.asyncio
305async def test_chat_history_route(test_app: Quart) -> None:
306 """GET /chat/<job_id>/history returns the chat history."""
308 service_manager = MagicMock()
309 job = StreamChatJob(job_id="test_history_job", service_manager=service_manager)
310 job.gen = LMMGeneratorMock()
311 job.character = Character(name="Alice", gender="Female", speech_speed=1.0)
312 streamchat_app.jobs["test_history_job"] = job
314 client = test_app.test_client()
315 response = await client.get("/chat/test_history_job/history")
316 assert response.status_code == HTTPStatus.OK
317 response_json = await response.get_json()
318 assert response_json["status"] == "ok"
319 assert "history" in response_json
320 assert isinstance(response_json["history"], list)
322 del streamchat_app.jobs["test_history_job"]
323 del job
324 del service_manager
327@pytest.mark.asyncio
328async def test_parse_chat_history() -> None:
329 """parse_chat_history returns empty list for nonexistent file."""
330 history = await parse_chat_history("/nonexistent/file.jsonl")
331 assert history == []
334@pytest.mark.asyncio
335async def test_get_chat_history_from_file_not_found() -> None:
336 """get_chat_history_from_file raises FileNotFoundError for missing job path."""
337 with pytest.raises(FileNotFoundError):
338 await get_chat_history_from_file("/nonexistent/path", "fake_job_id")
341@pytest.mark.asyncio
342async def test_transcribe_audio() -> None:
343 """StreamChatJob.transcribe_audio returns a transcript string."""
344 service_manager = MagicMock()
345 job = StreamChatJob(job_id="test_transcribe", service_manager=service_manager)
346 job.gen = LMMGeneratorMock()
348 audio_path = "tests/data/audio_4675.wav"
349 transcript = await job.transcribe_audio(audio_path)
350 assert isinstance(transcript, str)
351 assert len(transcript) > 0
353 del job
354 del service_manager
357@pytest.mark.asyncio
358async def test_gen_chat_base_with_config() -> None:
359 """StreamChatJob.gen_chat_base with style/scene/custom prompt config branches."""
360 service_manager = MagicMock()
361 job = StreamChatJob(job_id="test_gen_chat_base_config", service_manager=service_manager)
362 job.gen = LMMGeneratorMock()
363 job.config["style_prompt"] = "cinematic"
364 job.config["scene_prompt"] = "office"
365 job.config["custom_prompt"] = "daytime lighting"
367 await job.gen_chat_base()
368 job_status = await job.get_status()
369 assert job_status in (JobStatus.COMPLETED, JobStatus.FAILED)
371 del job
372 del service_manager
375@pytest.mark.asyncio
376async def test_chat_history_route_nonexistent_job(test_app: Quart) -> None:
377 """GET /chat/<job_id>/history for nonexistent job falls back to file lookup."""
378 client = test_app.test_client()
379 # job_id not in jobs dict, and no history file exists → FileNotFoundError handled by app
380 response = await client.get("/chat/definitely_nonexistent_job/history")
381 # The app should either return 200 with empty history or 500 if FileNotFoundError propagates
382 assert response.status_code in (HTTPStatus.OK, HTTPStatus.INTERNAL_SERVER_ERROR)
385@pytest.mark.asyncio
386async def test_parse_chat_history_with_data() -> None:
387 """parse_chat_history correctly parses a JSONL file."""
388 import json
389 import aiofiles
391 tmp_path = "/tmp/test_chat_history.jsonl"
392 messages = [
393 {"role": "user", "content": "Hello"},
394 {"role": "assistant", "content": "Hi there!"},
395 ]
396 async with aiofiles.open(tmp_path, "w") as f:
397 for msg in messages:
398 await f.write(json.dumps(msg) + "\n")
400 history = await parse_chat_history(tmp_path)
401 assert len(history) == 2
402 assert history[0]["role"] == "user"
403 assert history[1]["role"] == "assistant"