from unittest.mock import AsyncMock, patch from fastapi import HTTPException from httpx import ASGITransport, AsyncClient from app.core.config import settings from app.db.database import get_db from app.main import app from app.services.adapters.text.models import StoryOutput from app.services.voice_transcription_service import VoiceTranscriptionResult async def test_voice_session_create_and_fallback_turn_returns_audio( db_session, auth_token, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with ( patch( "app.services.voice_session_service.generate_story_content", new_callable=AsyncMock, ) as mock_generate, patch( "app.services.voice_session_service.text_to_speech", new_callable=AsyncMock, ) as mock_tts, ): mock_generate.return_value = StoryOutput( mode="generated", title="小猫去太空", story_text="小猫跳上纸飞机，朝着月亮轻轻挥手。", cover_prompt_suggestion="温暖儿童绘本封面，小猫与月亮", ) mock_tts.return_value = b"fake-turn-audio" transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) assert response.status_code == 201 session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{session_id}/turns/fallback", json={"transcript_text": "我想听一个小猫去太空的故事"}, ) assert response.status_code == 202 turn_id = response.json()["turn_id"] response = await client.get( f"/api/voice-sessions/{session_id}/turns/{turn_id}" ) assert response.status_code == 200 turn_data = response.json() assert turn_data["status"] == "audio_ready" assert turn_data["detected_intent"] == "start_story" assert turn_data["assistant_audio_ready"] is True assert turn_data["assistant_audio_url"].endswith("/audio") response = await client.get(turn_data["assistant_audio_url"]) assert response.status_code == 200 assert response.content == b"fake-turn-audio" assert response.headers["content-type"] == "audio/mpeg" response = await client.get(f"/api/voice-sessions/{session_id}") assert response.status_code == 200 session_data = response.json() assert session_data["status"] == "waiting_user" assert session_data["working_title"] == "小猫去太空" assert session_data["can_continue"] is True assert session_data["can_finalize"] is True assert len(session_data["recent_turns"]) == 1 finally: app.dependency_overrides.clear() async def test_voice_session_correct_turn_and_finalize_to_story( db_session, auth_token, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with ( patch( "app.services.voice_session_service.generate_story_content", new_callable=AsyncMock, ) as mock_generate, patch( "app.services.voice_session_service.text_to_speech", new_callable=AsyncMock, ) as mock_tts, patch( "app.services.voice_session_service.queue_story_asset_generation", new_callable=AsyncMock, ) as mock_queue_asset_generation, ): mock_generate.side_effect = [ StoryOutput( mode="generated", title="小猫去太空", story_text="第一段故事：小猫坐着纸飞机飞向月亮。", cover_prompt_suggestion="温暖儿童绘本封面，小猫飞向月亮", ), StoryOutput( mode="generated", title="小猫去太空", story_text="第二段故事：它在月亮上遇见了会发光的新朋友。", cover_prompt_suggestion="温暖儿童绘本封面，小猫与月亮朋友", ), ] mock_tts.side_effect = [b"turn-1-audio", b"turn-2-audio"] mock_queue_asset_generation.return_value = { "id": "cover-job-123", "story_id": 1, "output_mode": "asset_generation", "input_type": "image", "status": "running", "current_step": "request_accepted", "progress_percent": 5, "progress_label": "已接收请求", "is_terminal": False, "can_cancel": True, "can_retry": False, "result_snapshot": {}, "error_message": None, "created_at": "2026-04-20T00:00:00Z", "updated_at": "2026-04-20T00:00:00Z", } transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) assert response.status_code == 201 session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{session_id}/turns/fallback", json={"transcript_text": "我想听一个小猫去太空的故事"}, ) assert response.status_code == 202 response = await client.post( f"/api/voice-sessions/{session_id}/turns/fallback", json={"transcript_text": "不要让它哭了，我想让它找到一个朋友"}, ) assert response.status_code == 202 turn_id = response.json()["turn_id"] response = await client.get( f"/api/voice-sessions/{session_id}/turns/{turn_id}" ) assert response.status_code == 200 assert response.json()["detected_intent"] == "correct_story" response = await client.post( f"/api/voice-sessions/{session_id}/finalize", json={"save_story": True, "generate_cover": True}, ) assert response.status_code == 200 finalize_data = response.json() story_id = finalize_data["story_id"] assert finalize_data["status"] == "completed" assert finalize_data["generation_job_id"] == "cover-job-123" response = await client.get(f"/api/stories/{story_id}") assert response.status_code == 200 story_data = response.json() assert story_data["title"] == "小猫去太空" assert "第一段故事" in story_data["story_text"] assert "第二段故事" in story_data["story_text"] assert story_data["generation_status"] == "partial_ready" response = await client.get(f"/api/voice-sessions/{session_id}") assert response.status_code == 200 session_data = response.json() assert session_data["status"] == "completed" assert session_data["final_story_id"] == story_id assert session_data["can_continue"] is False assert session_data["story_state"]["final_summary"] mock_queue_asset_generation.assert_awaited_once() finally: app.dependency_overrides.clear() async def test_voice_session_abandon_blocks_future_turns( db_session, auth_token, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) assert response.status_code == 201 session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{session_id}/abandon", json={"reason": "孩子先去吃饭了"}, ) assert response.status_code == 200 assert response.json()["status"] == "abandoned" response = await client.post( f"/api/voice-sessions/{session_id}/turns/fallback", json={"transcript_text": "我们继续讲吧"}, ) assert response.status_code == 409 finally: app.dependency_overrides.clear() async def test_voice_session_uploaded_audio_turn_uses_demo_transcript_hint( db_session, auth_token, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with ( patch( "app.services.voice_session_service.generate_story_content", new_callable=AsyncMock, ) as mock_generate, patch( "app.services.voice_session_service.text_to_speech", new_callable=AsyncMock, ) as mock_tts, ): mock_generate.return_value = StoryOutput( mode="generated", title="小鲸鱼找朋友", story_text="小鲸鱼在海面上遇见了一只会唱歌的海鸥。", cover_prompt_suggestion="温暖儿童绘本封面，小鲸鱼和海鸥", ) mock_tts.return_value = b"fake-upload-audio" transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) assert response.status_code == 201 session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{session_id}/turns", files={ "audio_file": ("turn.webm", b"fake-webm-audio", "audio/webm"), }, data={ "duration_ms": "3200", "transcript_hint": "我想听一个小鲸鱼找朋友的故事", }, ) assert response.status_code == 202 turn_data = response.json() assert turn_data["status"] == "audio_ready" assert turn_data["transcription_provider"] == "demo" turn_id = turn_data["turn_id"] response = await client.get( f"/api/voice-sessions/{session_id}/turns/{turn_id}" ) assert response.status_code == 200 detail = response.json() assert detail["user_audio_ready"] is True assert detail["user_audio_url"].endswith("/user-audio") assert detail["transcription_provider"] == "demo" assert detail["assistant_audio_ready"] is True response = await client.get(detail["user_audio_url"]) assert response.status_code == 200 assert response.content == b"fake-webm-audio" assert response.headers["content-type"] == "audio/webm" finally: app.dependency_overrides.clear() async def test_voice_session_low_confidence_turn_requests_confirmation( db_session, auth_token, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with ( patch( "app.services.voice_session_service.generate_story_content", new_callable=AsyncMock, ) as mock_generate, patch( "app.services.voice_session_service.text_to_speech", new_callable=AsyncMock, ) as mock_tts, patch( "app.services.voice_session_service.transcribe_voice_audio", new_callable=AsyncMock, ) as mock_transcribe, ): mock_tts.return_value = b"confirmation-audio" mock_transcribe.return_value = VoiceTranscriptionResult( transcript_text="我想听一个会发光的小恐龙故事", confidence=0.41, provider="openai", ) transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) assert response.status_code == 201 session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{session_id}/turns", files={ "audio_file": ("turn.webm", b"fake-webm-audio", "audio/webm"), }, data={"duration_ms": "1200"}, ) assert response.status_code == 202 turn_id = response.json()["turn_id"] response = await client.get( f"/api/voice-sessions/{session_id}/turns/{turn_id}" ) assert response.status_code == 200 turn_data = response.json() assert turn_data["status"] == "audio_ready" assert turn_data["requires_confirmation"] is True assert turn_data["confirmation_state"] == "pending" assert turn_data["understanding_summary"].startswith("本轮系统理解为") assert "请家长帮忙确认" in turn_data["confirmation_message"] assert turn_data["assistant_text"] == turn_data["confirmation_message"] response = await client.post( f"/api/voice-sessions/{session_id}/turns/fallback", json={"transcript_text": "我要直接继续下一轮"}, ) assert response.status_code == 409 response = await client.get(f"/api/voice-sessions/{session_id}") assert response.status_code == 200 session_data = response.json() assert session_data["latest_requires_confirmation"] is True assert session_data["latest_confirmation_state"] == "pending" assert "请家长帮忙确认" in session_data["latest_confirmation_message"] assert session_data["can_finalize"] is False assert session_data["story_state"]["narrative_segments"] == [] assert any( event["event_type"] == "turn_confirmation_requested" for event in session_data["events"] ) mock_generate.assert_not_awaited() finally: app.dependency_overrides.clear() async def test_voice_session_confirmation_accept_continues_original_turn( db_session, auth_token, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with ( patch( "app.services.voice_session_service.generate_story_content", new_callable=AsyncMock, ) as mock_generate, patch( "app.services.voice_session_service.text_to_speech", new_callable=AsyncMock, ) as mock_tts, patch( "app.services.voice_session_service.transcribe_voice_audio", new_callable=AsyncMock, ) as mock_transcribe, ): mock_generate.return_value = StoryOutput( mode="generated", title="小恐龙的星光之旅", story_text="小恐龙踩着亮晶晶的石头，朝着会唱歌的山谷慢慢走去。", cover_prompt_suggestion="A glowing little dinosaur walking into a musical valley", ) mock_tts.side_effect = [b"confirmation-audio", b"story-audio"] mock_transcribe.return_value = VoiceTranscriptionResult( transcript_text="我想听一个会发光的小恐龙故事", confidence=0.44, provider="openai", ) transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{session_id}/turns", files={ "audio_file": ("turn.webm", b"fake-webm-audio", "audio/webm"), }, data={"duration_ms": "1200"}, ) turn_id = response.json()["turn_id"] response = await client.post( f"/api/voice-sessions/{session_id}/turns/{turn_id}/confirm", json={"action": "accept"}, ) assert response.status_code == 200 turn_data = response.json() assert turn_data["status"] == "audio_ready" assert turn_data["requires_confirmation"] is False assert turn_data["confirmation_state"] == "accepted" assert "小恐龙踩着亮晶晶的石头" in turn_data["assistant_text"] response = await client.get(f"/api/voice-sessions/{session_id}") session_data = response.json() assert session_data["latest_requires_confirmation"] is False assert session_data["can_finalize"] is True assert len(session_data["story_state"]["narrative_segments"]) == 1 finally: app.dependency_overrides.clear() async def test_voice_session_confirmation_switch_to_text_allows_follow_up_turn( db_session, auth_token, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with ( patch( "app.services.voice_session_service.generate_story_content", new_callable=AsyncMock, ) as mock_generate, patch( "app.services.voice_session_service.text_to_speech", new_callable=AsyncMock, ) as mock_tts, patch( "app.services.voice_session_service.transcribe_voice_audio", new_callable=AsyncMock, ) as mock_transcribe, ): mock_generate.return_value = StoryOutput( mode="generated", title="文字修正后的故事", story_text="小熊轻轻推开了云朵门，发现里面藏着一座会发光的图书馆。", cover_prompt_suggestion="A little bear opening a glowing cloud library door", ) mock_tts.side_effect = [b"confirmation-audio", b"story-audio"] mock_transcribe.return_value = VoiceTranscriptionResult( transcript_text="我想听一个小熊和云朵门的故事", confidence=0.4, provider="openai", ) transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{session_id}/turns", files={ "audio_file": ("turn.webm", b"fake-webm-audio", "audio/webm"), }, data={"duration_ms": "1200"}, ) turn_id = response.json()["turn_id"] response = await client.post( f"/api/voice-sessions/{session_id}/turns/{turn_id}/confirm", json={"action": "switch_to_text"}, ) assert response.status_code == 200 assert response.json()["confirmation_state"] == "switch_to_text" response = await client.post( f"/api/voice-sessions/{session_id}/turns/fallback", json={"transcript_text": "我想听一个小熊打开云朵门去冒险的故事"}, ) assert response.status_code == 202 response = await client.get(f"/api/voice-sessions/{session_id}") session_data = response.json() assert session_data["latest_requires_confirmation"] is False assert session_data["can_finalize"] is True finally: app.dependency_overrides.clear() async def test_voice_session_unsafe_transcript_is_redirected_safely( db_session, auth_token, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with patch( "app.services.voice_session_service.text_to_speech", new_callable=AsyncMock, ) as mock_tts, patch( "app.services.voice_session_service.generate_story_content", new_callable=AsyncMock, ) as mock_generate: mock_tts.return_value = b"safe-redirect-audio" transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{session_id}/turns/fallback", json={"transcript_text": "我想听一个拿着炸弹互相打的故事"}, ) assert response.status_code == 202 turn_id = response.json()["turn_id"] response = await client.get( f"/api/voice-sessions/{session_id}/turns/{turn_id}" ) turn_data = response.json() assert turn_data["safety_blocked"] is True assert "violence" in turn_data["safety_flags"] assert "温柔、安全" in turn_data["assistant_text"] response = await client.get(f"/api/voice-sessions/{session_id}") session_data = response.json() assert session_data["story_state"]["narrative_segments"] == [] assert "violence" in session_data["latest_safety_flags"] mock_generate.assert_not_awaited() finally: app.dependency_overrides.clear() async def test_voice_session_analytics_summarize_failures_and_confirmations( db_session, auth_token, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with ( patch( "app.services.voice_session_service.generate_story_content", new_callable=AsyncMock, ) as mock_generate, patch( "app.services.voice_session_service.text_to_speech", new_callable=AsyncMock, ) as mock_tts, patch( "app.services.voice_session_service.transcribe_voice_audio", new_callable=AsyncMock, ) as mock_transcribe, ): mock_generate.side_effect = [ StoryOutput( mode="generated", title="安全故事", story_text="第一段安全故事。", cover_prompt_suggestion="safe cover", ), StoryOutput( mode="generated", title="确认后继续", story_text="第二段确认后顺利继续。", cover_prompt_suggestion="safe cover 2", ), ] mock_tts.side_effect = [ RuntimeError("tts down"), b"confirmation-audio", b"confirmed-story-audio", ] mock_transcribe.side_effect = [ VoiceTranscriptionResult( transcript_text="我想听一个会发光的小恐龙故事", confidence=0.41, provider="openai", ), HTTPException(status_code=503, detail="语音转写服务暂时不可用，请稍后重试。"), ] transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) session_id = response.json()["id"] await client.post( f"/api/voice-sessions/{session_id}/turns/fallback", json={"transcript_text": "先给我一段故事"}, ) response = await client.post( f"/api/voice-sessions/{session_id}/turns", files={ "audio_file": ("turn.webm", b"fake-webm-audio", "audio/webm"), }, data={"duration_ms": "1200"}, ) turn_id = response.json()["turn_id"] await client.post( f"/api/voice-sessions/{session_id}/turns/{turn_id}/confirm", json={"action": "accept"}, ) response = await client.post( f"/api/voice-sessions/{session_id}/turns", files={ "audio_file": ("turn-2.webm", b"fake-webm-audio-2", "audio/webm"), }, ) assert response.status_code == 503 await client.post( f"/api/voice-sessions/{session_id}/finalize", json={"save_story": True, "generate_cover": False}, ) response = await client.get("/api/voice-sessions/analytics?days=30") assert response.status_code == 200 analytics = response.json() assert analytics["total_sessions"] >= 1 assert analytics["successful_turns"] >= 1 assert analytics["tts_failures"] >= 1 assert analytics["low_confidence_turns"] >= 1 assert analytics["asr_failures"] >= 1 assert analytics["finalized_sessions"] >= 1 assert analytics["finalize_conversion_rate"] > 0 assert analytics["text_fallback_turns"] >= 1 assert analytics["uploaded_audio_turns"] >= 1 assert analytics["user_audio_turn_rate"] > 0 assert analytics["assistant_audio_ready_turns"] >= 1 assert analytics["assistant_audio_ready_rate"] > 0 assert analytics["asr_success_rate"] > 0 assert analytics["tts_success_rate"] > 0 assert analytics["avg_transcript_confidence"] > 0 assert analytics["avg_intent_confidence"] > 0 assert analytics["failure_event_counts"]["turn_transcription_failed"] >= 1 assert analytics["failure_event_counts"]["assistant_audio_failed"] >= 1 assert analytics["total_user_audio_duration_ms"] >= 1200 assert analytics["avg_user_audio_duration_ms"] >= 1200 assert analytics["transcription_provider_counts"]["openai"] >= 1 assert analytics["transcription_provider_counts"]["fallback"] >= 1 assert analytics["confirmation_request_rate"] > 0 response = await client.get( "/api/voice-sessions/analytics?days=30&provider=openai" ) assert response.status_code == 200 provider_analytics = response.json() assert provider_analytics["provider"] == "openai" assert provider_analytics["uploaded_audio_turns"] >= 1 assert provider_analytics["text_fallback_turns"] == 0 assert set(provider_analytics["transcription_provider_counts"]) == {"openai"} response = await client.get( "/api/voice-sessions/analytics?days=30&session_status=completed" ) assert response.status_code == 200 status_analytics = response.json() assert status_analytics["session_status"] == "completed" assert status_analytics["total_sessions"] >= 1 assert status_analytics["finalized_sessions"] >= 1 response = await client.get( "/api/voice-sessions/analytics?days=30&session_status=unknown" ) assert response.status_code == 422 finally: app.dependency_overrides.clear() async def test_voice_session_attention_filter_and_analytics_count( db_session, auth_token, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with ( patch( "app.services.voice_session_service.generate_story_content", new_callable=AsyncMock, ) as mock_generate, patch( "app.services.voice_session_service.text_to_speech", new_callable=AsyncMock, ) as mock_tts, patch( "app.services.voice_session_service.transcribe_voice_audio", new_callable=AsyncMock, ) as mock_transcribe, ): mock_generate.side_effect = [ StoryOutput( mode="generated", title="正常故事", story_text="第一段温暖故事。", cover_prompt_suggestion="normal cover", ), RuntimeError("provider down"), ] mock_tts.side_effect = [ b"normal-audio", b"confirmation-audio", b"safety-audio", ] mock_transcribe.return_value = VoiceTranscriptionResult( transcript_text="我想听一个会发光的小恐龙故事", confidence=0.41, provider="openai", ) transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) normal_session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{normal_session_id}/turns/fallback", json={"transcript_text": "先讲一个温暖的普通故事"}, ) assert response.status_code == 202 response = await client.post("/api/voice-sessions", json={}) failed_session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{failed_session_id}/turns/fallback", json={"transcript_text": "这轮会触发 provider 异常"}, ) assert response.status_code == 202 response = await client.post("/api/voice-sessions", json={}) confirmation_session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{confirmation_session_id}/turns", files={ "audio_file": ("turn.webm", b"fake-webm-audio", "audio/webm"), }, ) assert response.status_code == 202 response = await client.post("/api/voice-sessions", json={}) safety_session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{safety_session_id}/turns/fallback", json={"transcript_text": "我想听一个拿着炸弹互相打的故事"}, ) assert response.status_code == 202 response = await client.get( "/api/voice-sessions?needs_attention=true&limit=8" ) assert response.status_code == 200 attention_sessions = response.json() attention_session_ids = {item["id"] for item in attention_sessions} assert attention_session_ids == { failed_session_id, confirmation_session_id, safety_session_id, } assert normal_session_id not in attention_session_ids attention_reason_sets = { item["id"]: set(item["attention_reasons"]) for item in attention_sessions } assert attention_reason_sets[confirmation_session_id] == { "pending_confirmation" } assert attention_reason_sets[safety_session_id] == { "safety_intervention" } assert attention_reason_sets[failed_session_id] == {"failed_turn"} response = await client.get( "/api/voice-sessions?needs_attention=true&attention_reason=pending_confirmation" ) assert response.status_code == 200 confirmation_sessions = response.json() assert [item["id"] for item in confirmation_sessions] == [ confirmation_session_id ] response = await client.get( "/api/voice-sessions?needs_attention=true&attention_reason=safety_intervention" ) assert response.status_code == 200 safety_sessions = response.json() assert [item["id"] for item in safety_sessions] == [safety_session_id] response = await client.get( "/api/voice-sessions?needs_attention=true&attention_reason=failed_turn" ) assert response.status_code == 200 failed_sessions = response.json() assert [item["id"] for item in failed_sessions] == [failed_session_id] response = await client.get("/api/voice-sessions/analytics?days=30") assert response.status_code == 200 analytics = response.json() assert analytics["total_sessions"] == 4 assert analytics["attention_sessions"] == 3 assert analytics["confirmation_attention_sessions"] == 1 assert analytics["safety_attention_sessions"] == 1 assert analytics["failed_attention_sessions"] == 1 assert analytics["failed_turns"] >= 1 assert analytics["low_confidence_turns"] >= 1 assert analytics["safety_interventions"] >= 1 finally: app.dependency_overrides.clear() async def test_voice_session_list_orders_recent_sessions_first( db_session, auth_token, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with ( patch( "app.services.voice_session_service.generate_story_content", new_callable=AsyncMock, ) as mock_generate, patch( "app.services.voice_session_service.text_to_speech", new_callable=AsyncMock, ) as mock_tts, ): mock_generate.side_effect = [ StoryOutput( mode="generated", title="第一场冒险", story_text="第一段故事。", cover_prompt_suggestion="封面一", ), StoryOutput( mode="generated", title="第二场冒险", story_text="第二段故事。", cover_prompt_suggestion="封面二", ), ] mock_tts.side_effect = [b"audio-1", b"audio-2"] transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) first_session_id = response.json()["id"] await client.post( f"/api/voice-sessions/{first_session_id}/turns/fallback", json={"transcript_text": "第一个故事"}, ) response = await client.post("/api/voice-sessions", json={}) second_session_id = response.json()["id"] await client.post( f"/api/voice-sessions/{second_session_id}/turns/fallback", json={"transcript_text": "第二个故事"}, ) response = await client.get("/api/voice-sessions?limit=8") assert response.status_code == 200 sessions = response.json() assert len(sessions) >= 2 assert sessions[0]["id"] == second_session_id assert sessions[1]["id"] == first_session_id assert sessions[0]["total_turns"] == 1 assert sessions[0]["last_turn_status"] == "audio_ready" response = await client.get("/api/voice-sessions?active_only=true") assert response.status_code == 200 active_sessions = response.json() assert {item["id"] for item in active_sessions} >= { first_session_id, second_session_id, } finally: app.dependency_overrides.clear() async def test_voice_session_active_endpoint_returns_latest_active_session( db_session, auth_token, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with ( patch( "app.services.voice_session_service.generate_story_content", new_callable=AsyncMock, ) as mock_generate, patch( "app.services.voice_session_service.text_to_speech", new_callable=AsyncMock, ) as mock_tts, ): mock_generate.return_value = StoryOutput( mode="generated", title="活动会话", story_text="一段活动中的故事。", cover_prompt_suggestion="活动会话封面", ) mock_tts.return_value = b"active-audio" transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) old_session_id = response.json()["id"] await client.post( f"/api/voice-sessions/{old_session_id}/abandon", json={"reason": "旧会话结束"}, ) response = await client.post("/api/voice-sessions", json={}) active_session_id = response.json()["id"] await client.post( f"/api/voice-sessions/{active_session_id}/turns/fallback", json={"transcript_text": "请继续一个新故事"}, ) response = await client.get("/api/voice-sessions/active") assert response.status_code == 200 data = response.json() assert data["id"] == active_session_id assert data["can_continue"] is True assert data["status"] == "waiting_user" finally: app.dependency_overrides.clear() async def test_voice_session_can_retry_failed_turn_from_saved_transcript( db_session, auth_token, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with ( patch( "app.services.voice_session_service.generate_story_content", new_callable=AsyncMock, ) as mock_generate, patch( "app.services.voice_session_service.text_to_speech", new_callable=AsyncMock, ) as mock_tts, ): mock_generate.side_effect = [ RuntimeError("provider down"), StoryOutput( mode="generated", title="重试成功", story_text="重试后的故事终于顺利继续了。", cover_prompt_suggestion="重试封面", ), ] mock_tts.return_value = b"retry-turn-audio" transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{session_id}/turns/fallback", json={"transcript_text": "先讲一个会失败的回合"}, ) assert response.status_code == 202 failed_turn_id = response.json()["turn_id"] response = await client.get( f"/api/voice-sessions/{session_id}/turns/{failed_turn_id}" ) assert response.status_code == 200 assert response.json()["status"] == "failed" response = await client.post( f"/api/voice-sessions/{session_id}/turns/{failed_turn_id}/retry" ) assert response.status_code == 202 retried_turn_id = response.json()["turn_id"] assert retried_turn_id != failed_turn_id response = await client.get( f"/api/voice-sessions/{session_id}/turns/{retried_turn_id}" ) assert response.status_code == 200 retried_turn = response.json() assert retried_turn["status"] == "audio_ready" assert retried_turn["assistant_text"] == "重试后的故事终于顺利继续了。" finally: app.dependency_overrides.clear() async def test_voice_session_can_retry_missing_assistant_audio( db_session, auth_token, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with ( patch( "app.services.voice_session_service.generate_story_content", new_callable=AsyncMock, ) as mock_generate, patch( "app.services.voice_session_service.text_to_speech", new_callable=AsyncMock, ) as mock_tts, ): mock_generate.return_value = StoryOutput( mode="generated", title="补发语音", story_text="这一轮先有文本，稍后再补发语音。", cover_prompt_suggestion="补发语音封面", ) mock_tts.side_effect = [RuntimeError("tts down"), b"recovered-audio"] transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{session_id}/turns/fallback", json={"transcript_text": "请先给我一段只有文本的结果"}, ) assert response.status_code == 202 turn_id = response.json()["turn_id"] response = await client.get( f"/api/voice-sessions/{session_id}/turns/{turn_id}" ) assert response.status_code == 200 turn = response.json() assert turn["status"] == "narrative_ready" assert turn["assistant_audio_ready"] is False response = await client.post( f"/api/voice-sessions/{session_id}/turns/{turn_id}/retry-audio" ) assert response.status_code == 200 retried = response.json() assert retried["status"] == "audio_ready" assert retried["assistant_audio_ready"] is True finally: app.dependency_overrides.clear() async def test_voice_session_uploaded_audio_respects_size_limit( db_session, auth_token, monkeypatch, ): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db monkeypatch.setattr(settings, "voice_turn_max_upload_bytes", 4) transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://test") as client: client.cookies.set("access_token", auth_token) response = await client.post("/api/voice-sessions", json={}) assert response.status_code == 201 session_id = response.json()["id"] response = await client.post( f"/api/voice-sessions/{session_id}/turns", files={ "audio_file": ("turn.webm", b"12345", "audio/webm"), }, data={"transcript_hint": "太长了"}, ) assert response.status_code == 413 finally: app.dependency_overrides.clear()