"""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]
|
