"""Phase 4: listen-window clock + deferred quantitative assessment."""
|
|
import asyncio
|
from types import SimpleNamespace
|
|
from core.app_runtime.assessment import (
|
build_metrics,
|
clamp_listen_ratio,
|
listen_window_ms,
|
resolve_expected_text,
|
text_similarity_v0,
|
)
|
from core.app_runtime.session import TaskSession, begin_session
|
|
|
CLIPS = [
|
"/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/l1/001/en/01.mp3"],
|
}
|
expected = resolve_expected_text(task, CLIPS, 0)
|
assert expected == "Hello, how are you today?"
|
assert not expected.endswith(".mp3")
|
assert resolve_expected_text(task, CLIPS, 0) != CLIPS[0]
|
|
|
def test_resolve_expected_text_rejects_path_as_original():
|
task = {
|
"transcripts": {"01.mp3": "/english_learning/shadowing/lessons/l1/001/en/01.mp3"},
|
"sentences": [CLIPS[0]],
|
}
|
assert resolve_expected_text(task, CLIPS, 0) is None
|
|
|
def test_resolve_expected_text_from_cache_txt(tmp_path):
|
(tmp_path / "01.txt").write_text("I like to read English books.", encoding="utf-8")
|
expected = resolve_expected_text({}, CLIPS, 0, cache_dir=str(tmp_path))
|
assert expected == "I like to read English books."
|
|
|
def test_resolve_expected_text_sentences_fallback():
|
task = {"sentences": ["Hello, how are you today?", "I like to read English books."]}
|
assert resolve_expected_text(task, CLIPS, 1) == "I like to read English books."
|
|
|
def test_text_similarity_metrics_have_no_passed_or_grade():
|
metrics = build_metrics("Hello, how are you today?", "Hello, how are you today?")
|
assert len(metrics) == 1
|
assert metrics[0]["method"] == "text_similarity_v0"
|
assert 0.0 <= metrics[0]["value"] <= 1.0
|
assert "passed" not in metrics[0]
|
assert "grade" not in metrics[0]
|
ratio = text_similarity_v0("hello how are you today", "Hello, how are you today?")
|
assert ratio > 0.9
|
|
|
def test_build_metrics_empty_when_missing_asr_or_expected():
|
assert build_metrics("", "Hello") == []
|
assert build_metrics("Hello", "") == []
|
assert build_metrics("Hello", CLIPS[0]) == []
|
|
|
def test_listen_ratio_is_clamped():
|
assert clamp_listen_ratio(0.5) == 1.2
|
assert clamp_listen_ratio(3.0) == 1.5
|
assert clamp_listen_ratio(1.3) == 1.3
|
assert listen_window_ms(1000, 0.5) == max(800, int(1000 * 1.2))
|
assert listen_window_ms(1000, 3.0) == int(1000 * 1.5)
|
assert listen_window_ms(100, 1.2) == 800
|
# 听窗随上一句实际播放时长变化，不是固定 2.5s；倍速已含在 wall-clock duration 里。
|
assert listen_window_ms(2050, 1.2) == int(2050 * 1.2)
|
assert listen_window_ms(4000, 1.2) == int(4000 * 1.2)
|
assert listen_window_ms(2050, 1.2, speed=0.9, wall_clock=True) == int(2050 * 1.2)
|
assert listen_window_ms(1800, 1.2, speed=0.9, wall_clock=False) == int(round(1800 / 0.9) * 1.2)
|
|
|
def _shadowing_conn(tmp_path, monkeypatch):
|
monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path))
|
monkeypatch.setattr("core.app_runtime.session._schedule_upsert", lambda _payload: None)
|
monkeypatch.setattr("core.audio_archive.get_project_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=CLIPS[0],
|
shadowing_clips=list(CLIPS),
|
shadowing_awaiting_user=True,
|
shadowing_clip_gen=1,
|
shadowing_assess_index=0,
|
shadowing_listen_pcm=[],
|
shadowing_audio_saved_index=-1,
|
asr_audio=[],
|
config={"audio_archive": {"enabled": True, "dir": "data/recordings"}},
|
last_asr_audio_snapshot=[],
|
sentence_id=None,
|
task_session=None,
|
active_app_id="english_learning",
|
current_task_id="shadowing",
|
loop=None,
|
prompt="",
|
)
|
begin_session(conn, "english_learning", "shadowing", "voice", "SHADOWING")
|
return conn
|
|
|
def test_asr_does_not_play_next_only_timer_advances(tmp_path, monkeypatch):
|
from core.app_runtime.runtime import (
|
_on_shadowing_listen_window_closed,
|
apply_shadowing_user_text,
|
)
|
|
played = []
|
|
async def fake_play(conn, name):
|
played.append(name)
|
return True
|
|
monkeypatch.setattr("core.app_runtime.runtime._play_shadowing_clip", fake_play)
|
conn = _shadowing_conn(tmp_path, monkeypatch)
|
outcome = apply_shadowing_user_text(conn, "today weather is bad")
|
assert outcome == "recorded"
|
assert conn.shadowing_index == 0
|
assert played == []
|
asyncio.run(_on_shadowing_listen_window_closed(conn))
|
assert conn.shadowing_index == 1
|
assert played == []
|
assert conn.task_session.find_turn(0)["asr_text"] == "today weather is bad"
|
|
|
def test_skip_listen_stop_asr_while_shadowing_clip_plays():
|
from core.app_runtime.runtime import should_skip_listen_stop_asr
|
|
playing = SimpleNamespace(interaction_mode="SHADOWING", shadowing_awaiting_user=False)
|
listening = SimpleNamespace(interaction_mode="SHADOWING", shadowing_awaiting_user=True)
|
chat = SimpleNamespace(interaction_mode="CHAT", shadowing_awaiting_user=False)
|
repeater = SimpleNamespace(interaction_mode="PLAY_AUDIO", shadowing_awaiting_user=False)
|
assert should_skip_listen_stop_asr(playing) is True
|
assert should_skip_listen_stop_asr(listening) is False
|
assert should_skip_listen_stop_asr(chat) is False
|
assert should_skip_listen_stop_asr(repeater) is True
|
|
|
def test_timer_advances_after_assess_and_upsert_errors(tmp_path, monkeypatch):
|
from core.app_runtime.runtime import _on_shadowing_listen_window_closed
|
|
played = []
|
|
async def fake_play(conn, name):
|
played.append(name)
|
return True
|
|
def boom_assess(*_args, **_kwargs):
|
raise RuntimeError("assess exploded")
|
|
monkeypatch.setattr("core.app_runtime.runtime._play_shadowing_clip", fake_play)
|
monkeypatch.setattr("core.app_runtime.runtime.apply_deferred_assessment", boom_assess)
|
conn = _shadowing_conn(tmp_path, monkeypatch)
|
|
def boom_save():
|
raise RuntimeError("upsert exploded")
|
|
conn.task_session.save = boom_save
|
asyncio.run(_on_shadowing_listen_window_closed(conn))
|
assert conn.shadowing_index == 1
|
assert played == []
|
|
|
def test_timer_advances_without_asr(tmp_path, monkeypatch):
|
from core.app_runtime.runtime import _on_shadowing_listen_window_closed
|
|
played = []
|
|
async def fake_play(conn, name):
|
played.append(name)
|
return True
|
|
monkeypatch.setattr("core.app_runtime.runtime._play_shadowing_clip", fake_play)
|
conn = _shadowing_conn(tmp_path, monkeypatch)
|
asyncio.run(_on_shadowing_listen_window_closed(conn))
|
assert conn.shadowing_index == 1
|
turn = conn.task_session.find_turn(0)
|
assert turn is not None
|
assert turn["asr_text"] == ""
|
names = [event["name"] for event in conn.task_session.events]
|
assert "listen_window_closed" in names
|
assert played == []
|
|
|
def test_upsert_turn_keeps_last_asr_same_index(tmp_path, monkeypatch):
|
monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path))
|
session = TaskSession(task_session_id="dup", app_id="english_learning", task_id="shadowing")
|
session.upsert_turn(0, CLIPS[0], "first try")
|
session.upsert_turn(0, CLIPS[0], "second try")
|
assert len(session.turns) == 1
|
assert session.turns[0]["asr_text"] == "second try"
|
|
|
def test_deferred_assessment_writes_metrics_not_passed(tmp_path, monkeypatch):
|
from core.app_runtime.runtime import apply_deferred_assessment, apply_shadowing_user_text
|
|
conn = _shadowing_conn(tmp_path, monkeypatch)
|
apply_shadowing_user_text(conn, "Hello, how are you today?")
|
apply_deferred_assessment(conn, 0)
|
turn = conn.task_session.find_turn(0)
|
assert turn["expected_text"] == "Hello, how are you today?"
|
assert not str(turn["expected_text"]).endswith(".mp3")
|
assert turn["metrics"][0]["method"] == "text_similarity_v0"
|
assert 0.0 <= turn["metrics"][0]["value"] <= 1.0
|
assert "passed" not in turn
|
assert "grade" not in turn
|
assert "passed" not in turn["metrics"][0]
|
|
|
def test_upsert_empty_metrics_do_not_wipe_existing(tmp_path, monkeypatch):
|
monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path))
|
session = TaskSession(task_session_id="keep-m", app_id="english_learning", task_id="shadowing")
|
session.upsert_turn(
|
0,
|
CLIPS[0],
|
"hello",
|
expected_text="Hello, how are you today?",
|
metrics=[{"method": "text_similarity_v0", "value": 0.9}],
|
assessed_at=1,
|
)
|
session.upsert_turn(0, CLIPS[0], "hello")
|
assert session.turns[0]["metrics"][0]["value"] == 0.9
|
assert "passed" not in session.turns[0]
|
assert "passed" not in session.turns[0]["metrics"][0]
|
|
|
def test_skill_event_starts_listen_window_before_save(tmp_path, monkeypatch):
|
from core.app_runtime.runtime import handle_skill_event, listen_window_ms
|
|
scheduled = []
|
|
def fake_schedule(conn, duration_ms):
|
scheduled.append(duration_ms)
|
|
monkeypatch.setattr(
|
"core.app_runtime.runtime._schedule_shadowing_listen_timeout", fake_schedule
|
)
|
conn = _shadowing_conn(tmp_path, monkeypatch)
|
conn.shadowing_awaiting_user = False
|
asyncio.run(
|
handle_skill_event(
|
conn, "[skill_event]audio.completed|english_learning|shadowing|2050"
|
)
|
)
|
assert scheduled == [2050]
|
assert listen_window_ms(2050, 1.2) == int(2050 * 1.2)
|
assert conn.task_session.events[-1]["name"] == "audio.completed"
|
|
|
def test_repeater_completed_returns_to_chat_even_if_event_says_shadowing(monkeypatch):
|
from core.app_runtime.runtime import handle_skill_event
|
|
switched = []
|
|
async def fake_start(conn, app_id, task_id, params, source="voice", **kwargs):
|
switched.append(
|
(app_id, task_id, kwargs.get("speak_switch"), kwargs.get("stop_skills"))
|
)
|
return None
|
|
async def fake_continue(conn):
|
return False
|
|
monkeypatch.setattr("core.app_runtime.runtime.start_app_task", fake_start)
|
monkeypatch.setattr("core.app_runtime.runtime._continue_repeater_if_needed", fake_continue)
|
spoken = []
|
monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
|
conn = SimpleNamespace(
|
interaction_mode="PLAY_AUDIO",
|
active_app_id="english_learning",
|
current_task_id="repeater",
|
task_session=None,
|
repeater_speed_queue=[],
|
repeater_play_name="",
|
prompt="",
|
config={},
|
device_id="mac",
|
)
|
asyncio.run(
|
handle_skill_event(
|
conn, "[skill_event]audio.completed|english_learning|shadowing|6130"
|
)
|
)
|
assert switched == [("chat", "conversation", False, False)]
|
assert spoken == ["泛听结束。我们来聊天吧。"]
|
|
|
def test_late_asr_after_listen_window_still_scores_previous_sentence(tmp_path, monkeypatch):
|
from core.app_runtime.runtime import apply_shadowing_user_text
|
|
conn = _shadowing_conn(tmp_path, monkeypatch)
|
conn.shadowing_awaiting_user = False
|
conn.shadowing_index = 1
|
conn.shadowing_assess_index = 0
|
conn.shadowing_prompt = CLIPS[1]
|
conn.client_is_speaking = False
|
outcome = apply_shadowing_user_text(conn, "hello how are you today")
|
assert outcome == "recorded"
|
assert conn.shadowing_index == 1
|
turn = conn.task_session.find_turn(0)
|
assert turn["asr_text"] == "hello how are you today"
|
assert conn.task_session.find_turn(1) is None
|
|
|
def test_evaluate_turn_saves_wav_and_metrics(tmp_path, monkeypatch):
|
from core.app_runtime.assessment import pcm_snapshot_to_wav
|
from core.app_runtime.runtime import apply_deferred_assessment, apply_shadowing_user_text
|
|
captured = {}
|
|
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"]
|
|
monkeypatch.setattr("core.app_runtime.runtime.save_turn_wav", fake_save)
|
conn = _shadowing_conn(tmp_path, monkeypatch)
|
# 16-bit silence ~ 20ms
|
conn.last_asr_audio_snapshot = [b"\x00\x00" * 160]
|
apply_shadowing_user_text(conn, "Hello, how are you today?")
|
apply_deferred_assessment(conn, 0, audio_snapshot=conn.last_asr_audio_snapshot)
|
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):
|
from core.app_runtime.runtime import handle_skill_event
|
|
switched = []
|
spoken = []
|
played = []
|
|
async def fake_play(conn, name):
|
played.append(name)
|
return True
|
|
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"
|
conn.active_app_id = app_id
|
conn.current_task_id = task_id
|
return None
|
|
monkeypatch.setattr("core.app_runtime.runtime._play_shadowing_clip", fake_play)
|
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 = _shadowing_conn(tmp_path, monkeypatch)
|
conn.shadowing_index = 1
|
conn.shadowing_prompt = CLIPS[1]
|
conn.shadowing_awaiting_user = False
|
asyncio.run(
|
handle_skill_event(
|
conn, "[skill_event]audio.playlist_completed|english_learning|shadowing"
|
)
|
)
|
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):
|
"""WS 掉线后设备可能在新的 CHAT 连接上补发 playlist_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.playlist_completed|english_learning|shadowing"
|
)
|
)
|
assert switched == [("chat", "conversation", False)]
|
assert spoken == ["跟读结束。我们来聊天吧。"]
|
|
|
def test_handle_shadowing_turn_does_not_enter_chat(tmp_path, monkeypatch):
|
from core.app_runtime.runtime import handle_shadowing_turn
|
|
conn = _shadowing_conn(tmp_path, monkeypatch)
|
conn.last_asr_audio_snapshot = [b"\x00\x00" * 80]
|
asyncio.run(handle_shadowing_turn(conn, "Hello, how are you today?"))
|
assert conn.interaction_mode == "SHADOWING"
|
assert conn.shadowing_index == 0
|
turn = conn.task_session.find_turn(0)
|
assert turn["asr_text"] == "Hello, how are you today?"
|
|
|
def test_listen_window_saves_wav_even_if_user_silent(tmp_path, monkeypatch):
|
from pathlib import Path
|
from core.app_runtime.runtime import _on_shadowing_listen_window_closed
|
|
conn = _shadowing_conn(tmp_path, monkeypatch)
|
conn.shadowing_listen_pcm = []
|
conn.asr_audio = []
|
asyncio.run(_on_shadowing_listen_window_closed(conn))
|
turn = conn.task_session.find_turn(0)
|
assert turn is not None
|
assert turn.get("audio_id")
|
assert "mac/shadowing/" in turn["audio_id"]
|
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):
|
from pathlib import Path
|
from core.app_runtime.runtime import save_shadowing_listen_audio
|
|
conn = _shadowing_conn(tmp_path, monkeypatch)
|
conn.shadowing_clips = CLIPS[:1] * 15
|
saved = []
|
for i in range(15):
|
conn.shadowing_assess_index = i
|
conn.shadowing_index = i
|
conn.shadowing_audio_saved_index = -1
|
conn.shadowing_listen_pcm = [b"\x00\x00" * (80 + i)]
|
save_shadowing_listen_audio(conn, i)
|
turn = conn.task_session.find_turn(i)
|
assert turn.get("audio_id")
|
saved.append(turn["audio_id"])
|
assert (tmp_path / "data" / "recordings" / Path(turn["audio_id"])).is_file()
|
assert len(saved) == 15
|
assert len(set(saved)) == 15
|
|
|
def test_shadowing_save_does_not_merge_previous_asr_audio(tmp_path, monkeypatch):
|
from pathlib import Path
|
import wave
|
from core.app_runtime.runtime import save_shadowing_listen_audio
|
|
conn = _shadowing_conn(tmp_path, monkeypatch)
|
leftover = b"\x11\x00" * 8000
|
current = b"\x22\x00" * 160
|
conn.asr_audio = [leftover]
|
conn.shadowing_listen_pcm = [current]
|
conn.shadowing_assess_index = 0
|
conn.shadowing_audio_saved_index = -1
|
save_shadowing_listen_audio(conn, 0)
|
turn = conn.task_session.find_turn(0)
|
wav_path = tmp_path / "data" / "recordings" / Path(turn["audio_id"])
|
with wave.open(str(wav_path), "rb") as handle:
|
pcm = handle.readframes(handle.getnframes())
|
assert leftover not in pcm
|
assert current in pcm
|
assert len(pcm) == len(current)
|
|
|
def test_one_step_writes_only_one_wav(tmp_path, monkeypatch):
|
from core.app_runtime.runtime import save_shadowing_listen_audio
|
|
conn = _shadowing_conn(tmp_path, monkeypatch)
|
conn.shadowing_listen_pcm = [b"\x00\x00" * 80]
|
save_shadowing_listen_audio(conn, 0)
|
first = conn.task_session.find_turn(0)["audio_id"]
|
conn.shadowing_listen_pcm = [b"\x00\x00" * 800]
|
conn.asr_audio = [b"\x00\x00" * 8000]
|
save_shadowing_listen_audio(conn, 0)
|
assert conn.task_session.find_turn(0)["audio_id"] == first
|
recordings = list((tmp_path / "data" / "recordings").rglob("*.wav"))
|
assert len(recordings) == 1
|
|
|
def test_listen_stop_drains_queued_pcm_instead_of_silent_placeholder(tmp_path, monkeypatch):
|
import queue as queue_mod
|
import wave
|
from pathlib import Path
|
from core.app_runtime.runtime import close_shadowing_listen_window
|
|
conn = _shadowing_conn(tmp_path, monkeypatch)
|
conn.shadowing_awaiting_user = True
|
conn.shadowing_index = 0
|
conn.shadowing_listen_pcm = []
|
conn.asr_audio_queue = queue_mod.Queue()
|
real = b"\x22\x00" * 1600
|
conn.asr_audio_queue.put(real)
|
conn.asr_audio_queue.put(b"\x33\x00" * 1600)
|
asyncio.run(close_shadowing_listen_window(conn))
|
turn = conn.task_session.find_turn(0)
|
wav_path = tmp_path / "data" / "recordings" / Path(turn["audio_id"])
|
with wave.open(str(wav_path), "rb") as handle:
|
pcm = handle.readframes(handle.getnframes())
|
assert real in pcm
|
assert len(pcm) == 1600 * 2 * 2
|
assert conn.asr_audio_queue.empty()
|
|
|
def test_listen_stop_advances_index_for_next_wav(tmp_path, monkeypatch):
|
from pathlib import Path
|
from core.app_runtime.runtime import close_shadowing_listen_window, save_shadowing_listen_audio
|
|
conn = _shadowing_conn(tmp_path, monkeypatch)
|
conn.shadowing_clips = CLIPS[:1] * 15
|
conn.shadowing_awaiting_user = True
|
conn.shadowing_index = 0
|
conn.shadowing_listen_pcm = [b"\x00\x00" * 80]
|
asyncio.run(close_shadowing_listen_window(conn))
|
assert conn.shadowing_index == 1
|
assert conn.shadowing_awaiting_user is False
|
first = conn.task_session.find_turn(0)["audio_id"]
|
conn.shadowing_listen_pcm = [b"\x11\x00" * 80]
|
conn.shadowing_audio_saved_index = -1
|
save_shadowing_listen_audio(conn, 1)
|
second = conn.task_session.find_turn(1)["audio_id"]
|
assert first != second
|
recordings = list((tmp_path / "data" / "recordings").rglob("*.wav"))
|
assert len(recordings) == 2
|
