6e20f2cd79
The STT/TTS worker _ensure() treated a spawned-but-not-yet-handshaked subprocess as ready, so a voice turn arriving during warmup read the same stdout StreamReader concurrently with the warmup handshake and crashed with "readuntil() called while another coroutine is already waiting for incoming data". Add a _start_lock + _ready flag so (re)start and the ready handshake run atomically and callers wait for real readiness before reading stdout. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
134 lines
4.4 KiB
Python
134 lines
4.4 KiB
Python
"""Unit tests for the WhisperSTT source plumbing.
|
|
|
|
These do NOT load a model or spawn the worker (that needs the whisper312 venv and
|
|
is exercised by the manual TTS->STT round trip). They pin the SpeechToText
|
|
contract: how `utterances()` turns an audio source into Utterances, and how it
|
|
behaves with no source wired yet.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
from typing import AsyncIterator
|
|
|
|
from wsai.backends.whisper import WhisperSTT
|
|
from wsai.interfaces import SpeechToText, Utterance
|
|
|
|
|
|
def _collect(stt: WhisperSTT) -> list[Utterance]:
|
|
async def run():
|
|
return [u async for u in stt.utterances()]
|
|
|
|
return asyncio.run(asyncio.wait_for(run(), timeout=5))
|
|
|
|
|
|
async def _paths(items) -> AsyncIterator[str]:
|
|
for it in items:
|
|
yield it
|
|
|
|
|
|
def test_is_speech_to_text():
|
|
assert isinstance(WhisperSTT(), SpeechToText)
|
|
|
|
|
|
def test_no_audio_source_yields_nothing():
|
|
# Discord voice receiver not wired yet -> the loop idles and returns.
|
|
assert _collect(WhisperSTT(audio_source=None)) == []
|
|
|
|
|
|
def test_utterances_transcribes_each_chunk(monkeypatch):
|
|
stt = WhisperSTT(audio_source=_paths(["a.wav", "b.wav"]))
|
|
|
|
async def fake_transcribe(wav_path, *, language=None):
|
|
return f"text::{wav_path}"
|
|
|
|
monkeypatch.setattr(stt, "transcribe", fake_transcribe)
|
|
|
|
utts = _collect(stt)
|
|
assert [u.text for u in utts] == ["text::a.wav", "text::b.wav"]
|
|
assert all(u.source == "voice" for u in utts)
|
|
|
|
|
|
def test_empty_transcript_is_skipped(monkeypatch):
|
|
# Silence / VAD-filtered audio transcribes to "" and must not become a turn.
|
|
stt = WhisperSTT(audio_source=_paths(["silent.wav", "real.wav"]))
|
|
|
|
async def fake_transcribe(wav_path, *, language=None):
|
|
return "" if wav_path == "silent.wav" else "안녕"
|
|
|
|
monkeypatch.setattr(stt, "transcribe", fake_transcribe)
|
|
|
|
utts = _collect(stt)
|
|
assert [u.text for u in utts] == ["안녕"]
|
|
|
|
|
|
def test_request_during_warmup_does_not_overlap_stdout(monkeypatch):
|
|
"""Regression: a transcribe() arriving while warmup() is still awaiting the
|
|
worker's ready line must NOT read the same stdout StreamReader concurrently.
|
|
|
|
Before the fix, _ensure()'s fast path returned as soon as the subprocess was
|
|
spawned (proc set, returncode None) even though the ready handshake was still
|
|
in flight, so the request's stdout.readline() overlapped warmup's and asyncio
|
|
raised "readuntil() called while another coroutine is already waiting for
|
|
incoming data" — the exact crash seen in the Discord voice server."""
|
|
|
|
async def run():
|
|
stt = WhisperSTT()
|
|
stdout = asyncio.StreamReader()
|
|
stderr = asyncio.StreamReader()
|
|
stderr.feed_eof() # nothing on stderr; let the drain task finish cleanly
|
|
|
|
class FakeStdin:
|
|
def write(self, _b):
|
|
pass
|
|
|
|
async def drain(self):
|
|
pass
|
|
|
|
class FakeProc:
|
|
returncode = None
|
|
|
|
def __init__(self):
|
|
self.stdin = FakeStdin()
|
|
self.stdout = stdout
|
|
self.stderr = stderr
|
|
|
|
def terminate(self):
|
|
self.returncode = 0
|
|
|
|
async def wait(self):
|
|
return 0
|
|
|
|
spawns = []
|
|
|
|
async def fake_create(*_a, **_k):
|
|
spawns.append(1)
|
|
return FakeProc()
|
|
|
|
monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create)
|
|
|
|
# warmup enters _ensure and blocks awaiting the ready line on stdout.
|
|
warm = asyncio.create_task(stt.warmup())
|
|
await asyncio.sleep(0.05)
|
|
|
|
# A concurrent request lands mid-warmup. It must wait for readiness, not
|
|
# crash and not read stdout yet.
|
|
tr = asyncio.create_task(stt.transcribe("x.wav"))
|
|
await asyncio.sleep(0.05)
|
|
assert not tr.done() # blocked on the start lock, no overlapping read
|
|
|
|
# Complete the handshake -> warmup finishes and releases the request.
|
|
stdout.feed_data(
|
|
(json.dumps({"ready": True, "ms": 1, "device": "cpu"}) + "\n").encode()
|
|
)
|
|
await asyncio.wait_for(warm, timeout=1)
|
|
await asyncio.sleep(0.02)
|
|
stdout.feed_data(
|
|
(json.dumps({"ok": True, "text": "안녕", "ms": 2}) + "\n").encode()
|
|
)
|
|
assert await asyncio.wait_for(tr, timeout=1) == "안녕"
|
|
assert sum(spawns) == 1 # one worker, not one-per-concurrent-caller
|
|
|
|
await stt.aclose()
|
|
|
|
asyncio.run(asyncio.wait_for(run(), timeout=5))
|