"""Phase 2 App Runtime unit tests. No websocket, no Opus.""" from types import SimpleNamespace from core.app_runtime.runtime import ( _plain_user_text, app_sd_layout_paths, is_echo_of_prompt, lesson_mp3_clips, match_voice_switch, repeater_lesson_mp3, ) from core.app_runtime.session import ( TaskSession, parse_skill_event, should_report_asr_now, ) def test_match_voice_switch_shadowing(): assert match_voice_switch("开始跟读") == ("english_learning", "shadowing") def test_match_voice_switch_menu(): assert match_voice_switch("切换模式") == ("__menu__", "") assert match_voice_switch("模式切换") == ("__menu__", "") def test_match_voice_switch_chat_return(): assert match_voice_switch("我们来聊天吧") == ("chat", "conversation") assert match_voice_switch("我们聊天吧") == ("chat", "conversation") assert match_voice_switch("不想跟读了") == ("chat", "conversation") assert match_voice_switch("退出翻译") == ("chat", "conversation") assert match_voice_switch("返回聊天") == ("chat", "conversation") assert match_voice_switch("退出跟读") == ("chat", "conversation") assert match_voice_switch("退出泛听") == ("chat", "conversation") def test_enter_prompts_announce_mode_without_exit_echo(): from core.app_runtime.registry import get_task _, repeater = get_task("english_learning", "repeater") _, shadowing = get_task("english_learning", "shadowing") repeater_prompt = (repeater or {}).get("enter_prompt") or "" shadowing_prompt = (shadowing or {}).get("enter_prompt") or "" assert "现在已经切换至泛听模式" in repeater_prompt assert "退出泛听" not in repeater_prompt assert "现在已经切换至跟读模式" in shadowing_prompt assert "退出跟读" not in shadowing_prompt def test_voice_menu_text_uses_task_menu_names(): from core.app_runtime.runtime import voice_menu_text menu = voice_menu_text() assert "泛听" in menu assert "跟读" in menu assert "翻译" in menu assert "中英对话" in menu assert "我们来聊天吧" in menu def test_exit_phrases_beat_short_enter_phrases(): from core.app_runtime.runtime import is_exit_to_chat assert match_voice_switch("跟读") == ("english_learning", "shadowing") assert match_voice_switch("泛听") == ("english_learning", "repeater") assert match_voice_switch("我想退出跟读") == ("chat", "conversation") assert is_exit_to_chat("退出翻译") is True assert is_exit_to_chat("开始跟读") is False assert is_exit_to_chat("退出泛听") is True def test_new_task_phrases_come_from_registry_only(monkeypatch): from copy import deepcopy from core.app_runtime import registry as reg data = deepcopy(reg.get_registry()) data["apps"]["story_retell"] = { "name": "故事复述", "default_task": "play", "tasks": { "play": { "menu_name": "故事复述", "voice_enter": ["开始故事复述", "故事复述"], "voice_exit": ["退出复述"], "enter_prompt": "现在已经切换至故事复述。", } }, } monkeypatch.setattr(reg, "_REGISTRY", data) monkeypatch.setattr(reg, "_VOICE_INDEX", None) assert match_voice_switch("开始故事复述") == ("story_retell", "play") assert match_voice_switch("退出复述") == ("chat", "conversation") menu = reg.voice_menu_text() assert "故事复述" in menu def test_match_voice_switch_select_after_menu(): assert match_voice_switch("跟读") == ("english_learning", "shadowing") assert match_voice_switch("翻译") == ("english_translation", "live_translate") def test_match_voice_switch_translate(): assert match_voice_switch("打开翻译") == ("english_translation", "live_translate") def test_reserved_voice_enter_stays_in_registry(): assert match_voice_switch("成语接龙") == ("__reserved__", "") assert match_voice_switch("数字背诵") == ("__reserved__", "") def test_is_echo_of_prompt_same_sentence(): prompt = "Hello, how are you today?" assert is_echo_of_prompt(prompt, prompt) is True assert is_echo_of_prompt("unrelated park visit", prompt) is False def test_plain_user_text_extracts_json_content(): assert _plain_user_text('{"content":"hello world","language":"en"}') == "hello world" assert _plain_user_text("just text") == "just text" def test_parse_skill_event_completed(): event = parse_skill_event( "[skill_event]audio.completed|english_learning|repeater|1234" ) assert event is not None assert event["name"] == "audio.completed" assert event["app_id"] == "english_learning" assert event["task_id"] == "repeater" assert event["duration_ms"] == "1234" def test_parse_skill_event_stopped_and_reject_plain(): event = parse_skill_event("[skill_event]audio.stopped|english_learning|repeater") assert event["name"] == "audio.stopped" assert parse_skill_event("我们来聊天吧") is None assert parse_skill_event("[app_task]chat|conversation|voice") is None def test_task_session_save_load(tmp_path, monkeypatch): monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path)) session = TaskSession( task_session_id="sess-demo", ws_session_id="ws-1", device_id="mac", app_id="english_learning", task_id="shadowing", source="voice", interaction_mode="SHADOWING", started_at=1, status="running", ) session.add_turn(0, "Hello, how are you today?", "hello how are you today", False) session.add_turn(1, "I like to read English books.", "i like to read", False) session.finish("finished") session.save() loaded = TaskSession.load("sess-demo") assert loaded is not None assert loaded.status == "completed" assert loaded.end_reason == "finished" assert len(loaded.turns) == 2 assert loaded.turns[0]["prompt_text"] == "Hello, how are you today?" assert loaded.turns[1]["asr_text"] == "i like to read" def test_should_report_asr_now_defers_play_and_shadowing(): assert should_report_asr_now(SimpleNamespace(interaction_mode="CHAT")) is True assert should_report_asr_now(SimpleNamespace(interaction_mode="PLAY_AUDIO")) is False assert should_report_asr_now(SimpleNamespace(interaction_mode="SHADOWING")) is False assert should_report_asr_now(SimpleNamespace(interaction_mode="TRANSLATION")) is True def test_last_shadowing_turn_is_saved_before_chat(tmp_path, monkeypatch): from core.app_runtime.runtime import apply_shadowing_user_text from core.app_runtime.session import begin_session monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path)) conn = SimpleNamespace( session_id="ws-1", device_id="mac", interaction_mode="SHADOWING", client_is_speaking=False, shadowing_index=0, shadowing_prompt="Hello, how are you today?", shadowing_clips=[], shadowing_awaiting_user=True, config={"app_runtime": {"shadowing_sentences": ["Hello, how are you today?"]}}, last_asr_audio_snapshot=[], sentence_id=None, task_session=None, active_app_id="english_learning", current_task_id="shadowing", ) begin_session(conn, "english_learning", "shadowing", "voice", "SHADOWING") outcome = apply_shadowing_user_text(conn, "I went to the park yesterday") assert outcome == "recorded" assert conn.shadowing_index == 0 loaded = TaskSession.load(conn.task_session.task_session_id) assert loaded.status == "running" assert len(loaded.turns) == 1 assert loaded.turns[0]["asr_text"] == "I went to the park yesterday" def test_lesson_mp3_clips_picks_first_lesson_folder(): names = [ "/english_learning/repeater/lessons/l1/001/cn/01.mp3", "/english_learning/shadowing/lessons/l2/002/en/02.mp3", "/english_learning/shadowing/lessons/l1/001/en/02.mp3", "/english_learning/shadowing/lessons/l1/001/en/01.mp3", "/english_learning/shadowing/lessons/l1/001/full.mp3", "/english_learning/shadowing/lessons/l1/001/en/03.mp3", "loose.mp3", ] clips = lesson_mp3_clips(names, "english_learning", "shadowing") assert clips == [ "/english_learning/shadowing/lessons/l1/001/en/01.mp3", "/english_learning/shadowing/lessons/l1/001/en/02.mp3", "/english_learning/shadowing/lessons/l1/001/en/03.mp3", ] only_l2 = lesson_mp3_clips(names, "english_learning", "shadowing", "l2") assert only_l2 == ["/english_learning/shadowing/lessons/l2/002/en/02.mp3"] assert repeater_lesson_mp3(names, "english_learning", "repeater") == ( "/english_learning/repeater/lessons/l1/001/cn/01.mp3" ) assert repeater_lesson_mp3(names, "english_learning", "shadowing") == ( "/english_learning/shadowing/lessons/l1/001/full.mp3" ) def test_task_session_to_dict_has_phase3_fields(): session = TaskSession( task_session_id="abc123", ws_session_id="ws-1", device_id="mac", agent_id="", app_id="english_learning", task_id="shadowing", source="voice", interaction_mode="SHADOWING", started_at=1, status="running", ) session.add_turn(0, "/english_learning/shadowing/lessons/l1/001/en/01.mp3", "hello", False) session.add_event("audio.completed", "english_learning", "shadowing") data = session.to_dict() assert data["task_session_id"] == "abc123" assert data["device_id"] == "mac" assert "agent_id" in data assert data["turns"][0]["prompt_text"].endswith("01.mp3") assert data["events"][0]["name"] == "audio.completed" def test_turn_audio_id_keeps_local_recording_path(): session = TaskSession( task_session_id="abc123", app_id="english_learning", task_id="shadowing", status="running", ) long_path = "14-c1-9f-49-43-94/shadowing/2026-09-18/091728893_asr_b5a60ebacb9544bfba4782a5cc978fb9_0.wav" session.upsert_turn(0, prompt_text="01.mp3", asr_text="hello", audio_id=long_path) assert session.to_dict()["turns"][0]["audio_id"] == long_path assert len(long_path) > 32 def test_save_survives_upsert_scheduler_error(tmp_path, monkeypatch): monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path)) def boom(_payload): raise RuntimeError("manage_api down") monkeypatch.setattr("core.app_runtime.session._schedule_upsert", boom) session = TaskSession( task_session_id="sess-upsert-fail", ws_session_id="ws-1", device_id="mac", app_id="english_learning", task_id="shadowing", source="voice", interaction_mode="SHADOWING", started_at=1, status="running", ) session.add_turn(0, "Hello", "hello", False) session.finish("finished") session.save() loaded = TaskSession.load("sess-upsert-fail") assert loaded is not None assert loaded.status == "completed" assert len(loaded.turns) == 1 def test_repeater_queues_play_until_listen_start(tmp_path, monkeypatch): import asyncio from core.app_runtime.runtime import maybe_flush_pending_play_after_tts, start_app_task plays = [] async def fake_play(conn, params): plays.append(params) return None async def fake_dirs(conn, app_id): return None async def fake_list(conn): return [ "/english_learning/shadowing/lessons/l1/001/full.mp3", "/english_learning/repeater/lessons/l1/001/cn/01.mp3", "/english_learning/shadowing/lessons/l1/001/en/01.mp3", ] monkeypatch.setattr("core.app_runtime.runtime._execute_named_tool_play", fake_play) monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", fake_dirs) monkeypatch.setattr("core.app_runtime.runtime._list_sd_mp3_names", fake_list) monkeypatch.setattr("core.app_runtime.playback.get_project_dir", lambda: str(tmp_path)) async def fake_ensure(*args, **kwargs): return True monkeypatch.setattr("core.app_runtime.runtime._ensure_lesson_mp3s", fake_ensure) spoken = [] monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text)) conn = SimpleNamespace(config={}, prompt="", func_handler=None, device_id="mac-queue") asyncio.run(start_app_task(conn, "english_learning", "repeater", {"lesson_id": "l1.001"}, source="voice")) assert spoken and "泛听" in spoken[0] assert plays == [] assert conn.pending_play_after_tts and conn.pending_play_after_tts["kind"] == "repeater" asyncio.run(maybe_flush_pending_play_after_tts(conn)) assert len(plays) == 1 assert plays[0]["name"] == "/english_learning/repeater/lessons/l1/001/cn/01.mp3" assert plays[0]["repeat"] == 1 assert plays[0]["speed"] == 1.0 assert plays[0].get("finish_task") is False assert len(plays[0]["items"]) == 4 assert plays[0]["items"][1]["name"].endswith("/l1/001/en/01.mp3") assert plays[0]["items"][1]["speed"] == 0.8 assert plays[0]["items"][3]["repeat"] == 5 assert plays[0]["items"][3]["speed"] == 1.0 assert conn.repeater_playlist == [] assert conn.pending_play_after_tts is None def test_sd_ready_skips_sync_and_queued_play(tmp_path, monkeypatch): import asyncio from core.app_runtime.runtime import start_app_task plays = [] spoken = [] dirs = [] ensures = [] async def fake_play(conn, params): plays.append(params) return None async def fake_dirs(conn, app_id): dirs.append(app_id) return None async def fake_ensure(*args, **kwargs): ensures.append(args) return True monkeypatch.setattr("core.app_runtime.runtime._execute_named_tool_play", fake_play) monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", fake_dirs) monkeypatch.setattr("core.app_runtime.runtime._ensure_lesson_mp3s", fake_ensure) monkeypatch.setattr("core.app_runtime.playback.get_project_dir", lambda: str(tmp_path)) monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text)) conn = SimpleNamespace(config={}, prompt="", func_handler=None, device_id="mac-sd-ready") asyncio.run(start_app_task( conn, "english_learning", "repeater", {"lesson_id": "l1.001", "sd_ready": True}, source="mqtt_immediate", )) assert plays == [] assert dirs == [] assert ensures == [] assert getattr(conn, "pending_play_after_tts", None) is None assert conn.interaction_mode == "PLAY_AUDIO" assert spoken == ["开始泛听任务"] assert not any("同步" in str(text) for text in spoken) def test_shadowing_queues_whole_playlist_until_listen_start(tmp_path, monkeypatch): import asyncio from core.app_runtime.runtime import maybe_flush_pending_play_after_tts, start_app_task plays = [] async def fake_play(conn, params): plays.append(params) return None async def fake_dirs(conn, app_id): return None async def fake_ensure(*args, **kwargs): return True monkeypatch.setattr("core.app_runtime.runtime._execute_named_tool_play", fake_play) monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", fake_dirs) monkeypatch.setattr("core.app_runtime.runtime._ensure_lesson_mp3s", fake_ensure) monkeypatch.setattr("core.app_runtime.playback.list_lesson_clips", lambda *_args, **_kwargs: None) spoken = [] monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text)) conn = SimpleNamespace(config={}, prompt="", func_handler=None, device_id="mac-shadow-queue") asyncio.run(start_app_task(conn, "english_learning", "shadowing", {"lesson_id": "l1.001"}, source="voice")) assert spoken and "跟读" in spoken[0] assert plays == [] assert conn.pending_play_after_tts and conn.pending_play_after_tts["kind"] == "shadowing" queued = conn.pending_play_after_tts["params"]["items"] assert len(queued) == 58 assert queued[0]["repeat"] == 1 assert queued[0]["speed"] == 0.8 assert queued[-1]["name"].endswith("/58.mp3") assert queued[-1]["repeat"] == 1 asyncio.run(maybe_flush_pending_play_after_tts(conn)) assert len(plays) == 1 assert len(plays[0]["items"]) == 58 assert plays[0]["items"][0]["repeat"] == 1 assert plays[0]["items"][-1]["repeat"] == 1 assert plays[0].get("finish_task") is False assert plays[0].get("listen_ratio") == 1.2 assert conn.pending_play_after_tts is None def test_shadowing_ensures_mp3s_even_when_sd_empty(tmp_path, monkeypatch): import asyncio from core.app_runtime.runtime import start_app_task ensure_calls = [] async def fake_play(conn, params): return None async def fake_dirs(conn, app_id): return None async def fake_list(conn): return [] async def fake_ensure(conn, names, task_id, params): ensure_calls.append((list(names), task_id, dict(params or {}))) return True monkeypatch.setattr("core.app_runtime.runtime._execute_named_tool_play", fake_play) monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", fake_dirs) monkeypatch.setattr("core.app_runtime.runtime._list_sd_mp3_names", fake_list) monkeypatch.setattr("core.app_runtime.runtime._ensure_lesson_mp3s", fake_ensure) monkeypatch.setattr("core.app_runtime.playback.list_lesson_clips", lambda *_args, **_kwargs: None) monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: None) conn = SimpleNamespace(config={}, prompt="", func_handler=None, device_id="mac-shadow-sync") asyncio.run(start_app_task( conn, "english_learning", "shadowing", {"lesson_id": "l1.001"}, source="mqtt_immediate", )) assert len(ensure_calls) == 1 names, task_id, params = ensure_calls[0] assert task_id == "shadowing" assert params.get("lesson_id") == "l1.001" assert len(names) == 58 assert names[0].endswith("/en/01.mp3") assert names[-1].endswith("/58.mp3") assert conn.pending_play_after_tts and conn.pending_play_after_tts["kind"] == "shadowing" def test_shadowing_expands_default_playlist_into_listen_turns(tmp_path, monkeypatch): import asyncio from core.app_runtime.runtime import _prepare_shadowing_sd monkeypatch.setattr("core.app_runtime.playback.list_lesson_clips", lambda *_args, **_kwargs: None) conn = SimpleNamespace( config={}, active_app_id="english_learning", current_task_id="shadowing", ) ok = asyncio.run(_prepare_shadowing_sd(conn, {"lesson_id": "l1.001"})) assert ok is True clip_01 = "/english_learning/shadowing/lessons/l1/001/en/01.mp3" clip_58 = "/english_learning/shadowing/lessons/l1/001/en/58.mp3" assert conn.shadowing_clips.count(clip_01) == 1 assert conn.shadowing_clips[-1] == clip_58 assert len(conn.shadowing_clips) == 58 assert conn.shadowing_prompt == clip_01 def test_voice_repeater_without_last_lesson_prompts_parent(tmp_path, monkeypatch): import asyncio from core.app_runtime.runtime import start_app_task spoken = [] monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text)) monkeypatch.setattr("core.app_runtime.playback.get_project_dir", lambda: str(tmp_path)) async def fake_dirs(conn, app_id): return None monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", fake_dirs) conn = SimpleNamespace(config={}, prompt="", func_handler=None, device_id="mac-empty") asyncio.run(start_app_task(conn, "english_learning", "repeater", {}, source="voice")) assert spoken and "还没有下发过泛听课文" in spoken[0] assert getattr(conn, "pending_play_after_tts", None) is None def test_voice_repeater_replays_last_parent_task(tmp_path, monkeypatch): import asyncio from core.app_runtime.runtime import start_app_task spoken = [] monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text)) monkeypatch.setattr("core.app_runtime.playback.get_project_dir", lambda: str(tmp_path)) async def fake_dirs(conn, app_id): return None async def fake_ensure(*args, **kwargs): return True monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", fake_dirs) monkeypatch.setattr("core.app_runtime.runtime._ensure_lesson_mp3s", fake_ensure) conn = SimpleNamespace(config={}, prompt="", func_handler=None, device_id="mac-last") asyncio.run( start_app_task( conn, "english_learning", "repeater", {"lesson_id": "l2.010", "en_passes": [{"speed": 0.8, "repeat": 1}]}, source="mqtt_immediate", ) ) conn.active_app_id = "chat" conn.current_task_id = "conversation" conn.current_source = "default_chat" conn.interaction_mode = "CHAT" spoken.clear() asyncio.run(start_app_task(conn, "english_learning", "repeater", {}, source="voice")) pending = conn.pending_play_after_tts assert pending and pending["kind"] == "repeater" assert pending["params"]["name"].endswith("/l2/010/cn/01.mp3") assert spoken and "泛听" in spoken[0] def test_voice_shadowing_replays_last_parent_task(tmp_path, monkeypatch): import asyncio from core.app_runtime.runtime import start_app_task spoken = [] monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text)) monkeypatch.setattr("core.app_runtime.playback.get_project_dir", lambda: str(tmp_path)) async def fake_dirs(conn, app_id): return None async def fake_ensure(*args, **kwargs): return True monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", fake_dirs) monkeypatch.setattr("core.app_runtime.runtime._ensure_lesson_mp3s", fake_ensure) conn = SimpleNamespace(config={}, prompt="", func_handler=None, device_id="mac-last-shd") asyncio.run( start_app_task( conn, "english_learning", "shadowing", {"lesson_id": "l1.001", "language": "en", "speed": 0.8}, source="schedule", ) ) conn.active_app_id = "chat" conn.current_task_id = "conversation" conn.current_source = "default_chat" conn.interaction_mode = "CHAT" conn.shadowing_clips = [] spoken.clear() asyncio.run(start_app_task(conn, "english_learning", "shadowing", {}, source="voice")) pending = conn.pending_play_after_tts assert pending and pending["kind"] == "shadowing" assert pending["params"]["items"][0]["name"].endswith("/l1/001/en/01.mp3") assert spoken and "跟读" in spoken[0] and "还没有下发过" not in spoken[0] def test_app_sd_layout_paths_from_registry(): from core.app_runtime.registry import get_app paths = app_sd_layout_paths("english_learning", get_app("english_learning")) assert set(paths) == { "/english_learning/repeater/lessons", "/english_learning/shadowing/lessons", } assert app_sd_layout_paths("chat", get_app("chat")) == [] assert app_sd_layout_paths("english_translation", get_app("english_translation")) == [] def test_boot_default_chat_speaks_greeting(monkeypatch): import asyncio from core.app_runtime.runtime import start_app_task spoken = [] monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text)) async def _ready(conn, timeout=8): return True monkeypatch.setattr("core.app_runtime.runtime.wait_for_tts_channels", _ready) conn = SimpleNamespace( config={ "enable_proactive_greeting": True, "proactive_chat_greeting": "我们来聊天吧。", }, prompt="", func_handler=None, just_woken_up=True, chat_greeted=False, active_app_id="chat", current_task_id="conversation", current_source="default_chat", interaction_mode="CHAT", ) conn.change_system_prompt = lambda text: setattr(conn, "prompt", text) asyncio.run(start_app_task(conn, "chat", "conversation", {}, source="default_chat")) assert spoken == ["我们来聊天吧。"] assert conn.chat_greeted is True def test_boot_unbound_skips_greeting_and_plays_bind_code(monkeypatch): import asyncio from core.app_runtime.runtime import start_app_task spoken = [] bind_codes = [] monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text)) async def _ready(conn, timeout=8): return True monkeypatch.setattr("core.app_runtime.runtime.wait_for_tts_channels", _ready) async def fake_bind(conn): bind_codes.append(conn.bind_code) monkeypatch.setattr("core.app_runtime.runtime._play_bind_prompt", fake_bind) conn = SimpleNamespace( config={ "enable_proactive_greeting": True, "proactive_chat_greeting": "我们来聊天吧。", }, prompt="", func_handler=None, just_woken_up=False, chat_greeted=False, active_app_id="chat", current_task_id="conversation", current_source="default_chat", interaction_mode="CHAT", need_bind=True, bind_code="123456", bind_completed_event=asyncio.Event(), ) conn.bind_completed_event.set() conn.change_system_prompt = lambda text: setattr(conn, "prompt", text) asyncio.run(start_app_task(conn, "chat", "conversation", {}, source="default_chat")) assert spoken == [] assert bind_codes == ["123456"] assert conn.chat_greeted is False def test_client_hello_is_recognized_before_bind(): from core.websocket_handshake import is_client_hello assert is_client_hello('{"type":"hello","version":1}') is True assert is_client_hello('{"type":"listen","state":"detect"}') is False assert is_client_hello(b"\x00\x01") is False def test_merge_hello_audio_params_keeps_server_output_rate(): from core.websocket_handshake import merge_hello_audio_params merged = merge_hello_audio_params( {"format": "opus", "sample_rate": 24000, "channels": 1, "frame_duration": 60}, {"format": "opus", "sample_rate": 16000, "channels": 1, "frame_duration": 60}, ) assert merged["sample_rate"] == 24000 assert merged["frame_duration"] == 60 assert merged["format"] == "opus" def test_priority_listen_detect_is_not_dropped_during_bind(): from core.websocket_handshake import is_priority_listen_detect assert is_priority_listen_detect( '{"type":"listen","state":"detect","text":"[app_task]chat|conversation|default_chat"}' ) is True assert is_priority_listen_detect( '{"type":"listen","state":"detect","text":"[lesson_sync]"}' ) is True assert is_priority_listen_detect( '{"type":"listen","state":"detect","text":"你好小智"}' ) is False def test_websocket_upgrade_accepts_keep_alive_connection_header(): from core.websocket_handshake import is_websocket_upgrade assert is_websocket_upgrade({"connection": "Upgrade", "upgrade": "websocket"}) is True assert is_websocket_upgrade({"connection": "keep-alive, Upgrade", "upgrade": "websocket"}) is True assert is_websocket_upgrade({"connection": "keep-alive"}) is False def test_app_sd_layout_paths_custom_subscribed_app(): app = { "tasks": { "listen": { "interaction_mode": "PLAY_AUDIO", "skills": [{"name": "self.mp3.play"}], }, "chatty": {"interaction_mode": "CHAT", "skills": []}, } } assert app_sd_layout_paths("story_time", app) == ["/story_time/listen/lessons"] def test_language_pair_uses_bilingual_prompt_and_unknown_falls_back(): from core.app_runtime.runtime import BILINGUAL_PROMPT, _apply_prompt conn = SimpleNamespace(prompt="base", original_system_prompt=None) conn.change_system_prompt = lambda text: setattr(conn, "prompt", text) app = {"system_prompt_mode": "bilingual"} _apply_prompt(conn, app, {"language_pair": "zh-en"}) assert conn.prompt == BILINGUAL_PROMPT assert conn.language_pair_fallback is False conn2 = SimpleNamespace(prompt="base", original_system_prompt=None) conn2.change_system_prompt = lambda text: setattr(conn2, "prompt", text) _apply_prompt(conn2, app, {"language_pair": "zh-ko"}) assert conn2.prompt == BILINGUAL_PROMPT assert conn2.language_pair_fallback is True assert conn2.language_pair == "zh-ko" def test_unknown_repeater_lesson_does_not_queue_play(monkeypatch): import asyncio from core.app_runtime.runtime import start_app_task spoken = [] async def _true(*args, **kwargs): return True async def _false(*args, **kwargs): return False monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text)) monkeypatch.setattr("core.app_runtime.runtime.wait_for_tts_channels", _true) monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", _true) monkeypatch.setattr("core.app_runtime.runtime.prompt_bind_if_needed", _false) conn = SimpleNamespace( config={}, prompt="", func_handler=None, just_woken_up=False, chat_greeted=True, active_app_id="chat", current_task_id="conversation", current_source="default_chat", interaction_mode="CHAT", pending_play_after_tts=None, ) conn.change_system_prompt = lambda text: setattr(conn, "prompt", text) asyncio.run( start_app_task( conn, "english_learning", "repeater", {"lesson_id": "unit99"}, source="mqtt_immediate", ) ) assert conn.pending_play_after_tts is None assert any("SD卡" in text for text in spoken) def test_parent_translation_clears_abort_and_announces(tmp_path, monkeypatch): import asyncio from core.app_runtime.runtime import start_app_task spoken = [] async def _false(*args, **kwargs): return False monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path)) monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text)) monkeypatch.setattr("core.app_runtime.runtime.prompt_bind_if_needed", _false) monkeypatch.setattr("core.app_runtime.runtime.schedule_close_parent_except", lambda *a, **k: None) monkeypatch.setattr("core.app_runtime.session._schedule_close_parent", lambda *a, **k: None) conn = SimpleNamespace( config={}, prompt="base", func_handler=None, device_id="mac-translate", client_abort=True, tts=SimpleNamespace(tts_stop_request=True), ) conn.change_system_prompt = lambda text: setattr(conn, "prompt", text) asyncio.run(start_app_task( conn, "english_translation", "live_translate", {"language_pair": "zh-en"}, source="mqtt_immediate", )) assert conn.client_abort is False assert conn.tts.tts_stop_request is False assert spoken and "翻译模式" in spoken[0] assert conn.interaction_mode == "TRANSLATION" def test_schedule_talk_announces_start_phrase(tmp_path, monkeypatch): import asyncio from core.app_runtime.runtime import start_app_task spoken = [] async def _false(*args, **kwargs): return False monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path)) monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text)) monkeypatch.setattr("core.app_runtime.runtime.prompt_bind_if_needed", _false) monkeypatch.setattr("core.app_runtime.runtime.schedule_close_parent_except", lambda *a, **k: None) monkeypatch.setattr("core.app_runtime.session._schedule_close_parent", lambda *a, **k: None) conn = SimpleNamespace( config={"enable_proactive_greeting": True}, prompt="base", func_handler=None, device_id="mac-schedule-talk", client_abort=False, tts=SimpleNamespace(tts_stop_request=False), active_app_id="chat", current_task_id="conversation", current_source="default_chat", interaction_mode="CHAT", chat_greeted=True, ) conn.change_system_prompt = lambda text: setattr(conn, "prompt", text) asyncio.run(start_app_task(conn, "english_translation", "live_translate", {}, source="schedule")) assert spoken == ["开始翻译任务"] assert conn.interaction_mode == "TRANSLATION" spoken.clear() conn.active_app_id = "chat" conn.current_task_id = "conversation" conn.interaction_mode = "CHAT" asyncio.run(start_app_task(conn, "english_conversation", "bilingual_chat", {}, source="schedule")) assert spoken == ["开始中英对话任务"] assert conn.interaction_mode == "BILINGUAL" spoken.clear() asyncio.run(start_app_task(conn, "chat", "conversation", {}, source="schedule")) assert spoken == ["开始对话任务"] assert conn.chat_greeted is True assert conn.interaction_mode == "CHAT" def test_voice_exit_overrides_parent_bilingual_and_translation(tmp_path, monkeypatch): import asyncio from core.app_runtime.runtime import start_app_task spoken = [] async def _false(*args, **kwargs): return False monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path)) monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text)) monkeypatch.setattr("core.app_runtime.runtime.prompt_bind_if_needed", _false) monkeypatch.setattr("core.app_runtime.runtime.schedule_close_parent_except", lambda *a, **k: None) monkeypatch.setattr("core.app_runtime.session._schedule_close_parent", lambda *a, **k: None) conn = SimpleNamespace( config={}, prompt="base", func_handler=None, device_id="mac-exit", client_abort=False, tts=SimpleNamespace(tts_stop_request=False), ) conn.change_system_prompt = lambda text: setattr(conn, "prompt", text) asyncio.run(start_app_task(conn, "english_conversation", "bilingual_chat", {}, source="schedule")) assert conn.current_source == "schedule" assert conn.interaction_mode == "BILINGUAL" spoken.clear() asyncio.run(start_app_task(conn, "chat", "conversation", {}, source="voice")) assert conn.interaction_mode == "CHAT" assert conn.current_source == "voice" assert spoken and "对话模式" in spoken[0] asyncio.run(start_app_task(conn, "english_translation", "live_translate", {}, source="mqtt_immediate")) assert conn.interaction_mode == "TRANSLATION" spoken.clear() asyncio.run(start_app_task(conn, "english_learning", "repeater", {}, source="voice")) assert conn.interaction_mode == "TRANSLATION" spoken.clear() asyncio.run(start_app_task(conn, "chat", "conversation", {}, source="voice")) assert conn.interaction_mode == "CHAT" assert spoken and "对话模式" in spoken[0]