5种场景模式家长端ok

wangzhibo

10 天以前 220e4fab27b7f419b3e89e2f2d28ae789aac09d7

 xiaozhi-esp32-server/main/xiaozhi-server/tests/test_assessment.py

@@ -14,15 +14,15 @@


CLIPS = [
    "/english_learning/shadowing/lessons/unit01/01.mp3",
    "/english_learning/shadowing/lessons/unit01/02.mp3",
    "/english_learning/shadowing/lessons/l1/001/en/01.mp3",
    "/english_learning/shadowing/lessons/l1/001/en/02.mp3",
]


def test_resolve_expected_text_from_transcripts_not_path():
    task = {
        "transcripts": {"01.mp3": "Hello, how are you today?", "01": "Hello, how are you today?"},
        "sentences": ["/english_learning/shadowing/lessons/unit01/01.mp3"],
        "sentences": ["/english_learning/shadowing/lessons/l1/001/en/01.mp3"],
    }
    expected = resolve_expected_text(task, CLIPS, 0)
    assert expected == "Hello, how are you today?"
@@ -32,7 +32,7 @@

def test_resolve_expected_text_rejects_path_as_original():
    task = {
        "transcripts": {"01.mp3": "/english_learning/shadowing/lessons/unit01/01.mp3"},
        "transcripts": {"01.mp3": "/english_learning/shadowing/lessons/l1/001/en/01.mp3"},
        "sentences": [CLIPS[0]],
    }
    assert resolve_expected_text(task, CLIPS, 0) is None
@@ -294,7 +294,7 @@
        )
    )
    assert switched == [("chat", "conversation", False, False)]
    assert spoken == ["复读结束。我们来聊天吧。"]
    assert spoken == ["泛听结束。我们来聊天吧。"]


def test_late_asr_after_listen_window_still_scores_previous_sentence(tmp_path, monkeypatch):
@@ -320,8 +320,9 @@

    captured = {}

    def fake_save(session_id, step_index, wav_bytes, conn=None):
    def fake_save(session_id, step_index, wav_bytes, conn=None, text=None):
        captured["wav"] = wav_bytes
        captured["text"] = text
        captured["id"] = f"{session_id}_{step_index}.wav"
        return captured["id"]

@@ -334,10 +335,33 @@
    turn = conn.task_session.find_turn(0)
    assert turn["metrics"][0]["method"] == "text_similarity_v0"
    assert turn["audio_id"] == captured["id"]
    assert captured["text"] == "Hello, how are you today?"
    wav = pcm_snapshot_to_wav(conn.last_asr_audio_snapshot)
    assert captured["wav"][:4] == b"RIFF"
    assert wav[8:12] == b"WAVE"
    assert captured["wav"] == wav


def test_later_asr_updates_sidecar_without_new_wav(tmp_path, monkeypatch):
    from pathlib import Path
    from core.app_runtime.runtime import (
        apply_deferred_assessment,
        apply_shadowing_user_text,
        save_shadowing_listen_audio,
    )

    conn = _shadowing_conn(tmp_path, monkeypatch)
    conn.shadowing_listen_pcm = [b"\x00\x00" * 80]
    save_shadowing_listen_audio(conn, 0)
    turn = conn.task_session.find_turn(0)
    wav_path = tmp_path / "data" / "recordings" / Path(turn["audio_id"])
    txt_path = wav_path.with_suffix(".txt")
    assert txt_path.read_text(encoding="utf-8").strip() == ""
    apply_shadowing_user_text(conn, "Hello, how are you today?")
    apply_deferred_assessment(conn, 0, persist_audio=False)
    recordings = list((tmp_path / "data" / "recordings").rglob("*.wav"))
    assert len(recordings) == 1
    assert txt_path.read_text(encoding="utf-8").strip() == "Hello, how are you today?"


def test_shadowing_playlist_end_announces_then_chat(tmp_path, monkeypatch):
@@ -373,6 +397,42 @@
    assert played == []
    assert switched == [("chat", "conversation", False)]
    assert spoken == ["跟读结束。我们来聊天吧。"]


def test_repeater_completed_returns_chat_after_reconnect(monkeypatch):
    """WS 掉线后设备可能在新的 CHAT 连接上补发 audio.completed。"""
    from core.app_runtime.runtime import handle_skill_event

    switched = []
    spoken = []

    async def fake_start(conn, app_id, task_id, params, source="voice", **kwargs):
        switched.append((app_id, task_id, kwargs.get("speak_switch")))
        conn.interaction_mode = "CHAT"
        return None

    monkeypatch.setattr("core.app_runtime.runtime.start_app_task", fake_start)
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
    conn = SimpleNamespace(
        interaction_mode="CHAT",
        active_app_id="chat",
        current_task_id="conversation",
        task_session=None,
        repeater_speed_queue=[],
        repeater_playlist=[],
        shadowing_awaiting_user=False,
        prompt="",
        config={},
        device_id="mac",
        tts=None,
    )
    asyncio.run(
        handle_skill_event(
            conn, "[skill_event]audio.completed|english_learning|repeater|101290"
        )
    )
    assert switched == [("chat", "conversation", False)]
    assert spoken == ["泛听结束。我们来聊天吧。"]


def test_playlist_completed_returns_chat_after_reconnect(monkeypatch):
@@ -438,6 +498,9 @@
    wav_path = tmp_path / "data" / "recordings" / Path(turn["audio_id"])
    assert wav_path.is_file()
    assert wav_path.stat().st_size > 44
    txt_path = wav_path.with_suffix(".txt")
    assert txt_path.is_file()
    assert txt_path.read_text(encoding="utf-8").strip() == ""


def test_fifteen_listen_windows_write_fifteen_wavs(tmp_path, monkeypatch):