test_speak_resident.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Resident-Piper checks for speak() (#12).
speak() must load the Piper voice once and reuse it across calls in one process
(the latency win), keep the missing-voice error path with its download hint, and
fall back to the CLI when the in-process synth fails instead of crashing the loop.
These run hardware-free: a fake `piper` module is injected so no ONNX voice loads
and no audio backend is touched.
Run: .venv/bin/python test_speak_resident.py
Exit code is 0 only if every check passes.
"""
from __future__ import annotations
import sys
import tempfile
import types
import warnings
from pathlib import Path
import pipeline
PASS, FAIL = "PASS", "FAIL"
results: list[bool] = []
def check(name: str, ok: bool, detail: str) -> bool:
results.append(bool(ok))
print(f" [{PASS if ok else FAIL}] {name}: {detail}")
return bool(ok)
class _FakeVoice:
"""Writes a tiny valid WAV so speak()'s wave context manager succeeds."""
def synthesize_wav(self, text, wav_file, **kwargs):
wav_file.setnchannels(1)
wav_file.setsampwidth(2)
wav_file.setframerate(22050)
wav_file.writeframes(b"\x00\x00" * 16)
def _install_fake_piper() -> tuple[types.ModuleType, list[str]]:
"""Build a fake `piper` module whose PiperVoice.load records each call."""
calls: list[str] = []
mod = types.ModuleType("piper")
class _FakePiperVoice:
@classmethod
def load(cls, model_path, config_path=None, **kwargs):
calls.append(str(model_path))
return _FakeVoice()
mod.PiperVoice = _FakePiperVoice # type: ignore[attr-defined]
return mod, calls
def _voices_dir(tmp: Path, stem: str = "testvoice") -> Path:
tmp.mkdir(parents=True, exist_ok=True)
(tmp / f"{stem}.onnx").write_bytes(b"") # only existence is checked
return tmp
def test_voice_loads_once_across_calls() -> None:
tmp = Path(tempfile.mkdtemp())
voices = _voices_dir(tmp)
fake_piper, calls = _install_fake_piper()
saved_mod = sys.modules.get("piper")
saved_dir = pipeline.VOICES_DIR
pipeline.VOICES_DIR = voices
pipeline._piper_cache.clear()
sys.modules["piper"] = fake_piper
try:
outs = [str(tmp / f"out{i}.wav") for i in range(3)]
for out in outs:
pipeline.speak("hello there", out, voice_model="testvoice")
finally:
pipeline.VOICES_DIR = saved_dir
pipeline._piper_cache.clear()
if saved_mod is not None:
sys.modules["piper"] = saved_mod
else:
sys.modules.pop("piper", None)
check(
"piper voice loaded exactly once across 3 speak() calls",
len(calls) == 1,
f"load calls={len(calls)}",
)
check(
"every speak() call wrote a non-empty WAV",
all(Path(o).exists() and Path(o).stat().st_size > 0 for o in outs),
f"sizes={[Path(o).stat().st_size for o in outs if Path(o).exists()]}",
)
def test_missing_voice_raises_with_hint() -> None:
tmp = Path(tempfile.mkdtemp()) # empty: no onnx present
saved_dir = pipeline.VOICES_DIR
pipeline.VOICES_DIR = tmp
raised: Exception | None = None
try:
pipeline.speak("hi", str(tmp / "o.wav"), voice_model="nope")
except FileNotFoundError as e:
raised = e
finally:
pipeline.VOICES_DIR = saved_dir
check(
"missing voice raises FileNotFoundError",
isinstance(raised, FileNotFoundError),
f"raised={type(raised).__name__ if raised else None}",
)
check(
"missing-voice error keeps the download hint",
raised is not None and "download" in str(raised),
f"msg={str(raised)[:70]!r}",
)
def test_falls_back_to_cli_when_inprocess_fails() -> None:
tmp = Path(tempfile.mkdtemp())
voices = _voices_dir(tmp)
saved_dir = pipeline.VOICES_DIR
saved_get = pipeline._get_piper
saved_sub = pipeline._speak_subprocess
fell_back: list[tuple] = []
def boom(model_path):
raise RuntimeError("simulated bad piper build")
def fake_sub(text, onnx, out_wav_path):
fell_back.append((text, str(onnx), out_wav_path))
Path(out_wav_path).write_bytes(b"RIFF") # pretend the CLI wrote a wav
pipeline.VOICES_DIR = voices
pipeline._get_piper = boom # type: ignore[assignment]
pipeline._speak_subprocess = fake_sub # type: ignore[assignment]
pipeline._piper_cache.clear()
out = str(tmp / "fb.wav")
err: Exception | None = None
try:
with warnings.catch_warnings():
warnings.simplefilter("ignore")
pipeline.speak("hello", out, voice_model="testvoice")
except Exception as e: # noqa: BLE001 - the point is that it should NOT raise
err = e
finally:
pipeline.VOICES_DIR = saved_dir
pipeline._get_piper = saved_get # type: ignore[assignment]
pipeline._speak_subprocess = saved_sub # type: ignore[assignment]
check("in-process failure does not crash speak()", err is None, f"err={err!r}")
check(
"speak() fell back to the CLI path exactly once",
len(fell_back) == 1,
f"fallback calls={len(fell_back)}",
)
check(
"fallback produced the output wav",
Path(out).exists(),
f"exists={Path(out).exists()}",
)
def test_evicts_and_falls_back_when_synth_fails() -> None:
"""Load succeeds and caches the voice, then synth raises: speak() evicts the
entry and falls back to the CLI. Unlike the boom-on-load test above, this runs
the real _get_piper cache and the _piper_cache.pop eviction line, which is the
exact seam residency introduces."""
tmp = Path(tempfile.mkdtemp())
voices = _voices_dir(tmp)
onnx_key = str(voices / "testvoice.onnx")
calls: list[str] = []
mod = types.ModuleType("piper")
class _BoomVoice:
def synthesize_wav(self, text, wav_file, **kwargs):
raise RuntimeError("simulated synth failure after a good load")
class _FakePiperVoice:
@classmethod
def load(cls, model_path, config_path=None, **kwargs):
calls.append(str(model_path))
return _BoomVoice()
mod.PiperVoice = _FakePiperVoice # type: ignore[attr-defined]
saved_mod = sys.modules.get("piper")
saved_dir = pipeline.VOICES_DIR
saved_sub = pipeline._speak_subprocess
fell_back: list[tuple] = []
def fake_sub(text, onnx, out_wav_path):
fell_back.append((text, str(onnx), out_wav_path))
Path(out_wav_path).write_bytes(b"RIFF") # pretend the CLI wrote a wav
pipeline.VOICES_DIR = voices
pipeline._speak_subprocess = fake_sub # type: ignore[assignment]
pipeline._piper_cache.clear()
sys.modules["piper"] = mod
out = str(tmp / "boom.wav")
err: Exception | None = None
cached_after = True
try:
with warnings.catch_warnings():
warnings.simplefilter("ignore")
pipeline.speak("hello", out, voice_model="testvoice")
cached_after = onnx_key in pipeline._piper_cache
except Exception as e: # noqa: BLE001 - it must NOT raise
err = e
finally:
pipeline.VOICES_DIR = saved_dir
pipeline._speak_subprocess = saved_sub # type: ignore[assignment]
pipeline._piper_cache.clear()
if saved_mod is not None:
sys.modules["piper"] = saved_mod
else:
sys.modules.pop("piper", None)
check(
"the voice loaded once via the real cache before synth failed",
len(calls) == 1,
f"load calls={len(calls)}",
)
check(
"synth failure after load does not crash speak()", err is None, f"err={err!r}"
)
check(
"the failed voice entry was evicted from the cache",
cached_after is False,
f"cached_after={cached_after}",
)
check(
"speak() fell back to the CLI after the synth failure",
len(fell_back) == 1,
f"fallback calls={len(fell_back)}",
)
check(
"fallback produced the output wav",
Path(out).exists(),
f"exists={Path(out).exists()}",
)
def main() -> int:
print("=== resident-piper speak() checks ===")
for test in (
test_voice_loads_once_across_calls,
test_missing_voice_raises_with_hint,
test_falls_back_to_cli_when_inprocess_fails,
test_evicts_and_falls_back_when_synth_fails,
):
try:
test()
except Exception as e: # noqa: BLE001 - report, don't abort the suite
check(
f"{test.__name__} ran without error", False, f"{type(e).__name__}: {e}"
)
n_pass = sum(1 for r in results if r)
print(f"\n=== SUMMARY: {n_pass}/{len(results)} checks passed ===")
return 0 if n_pass == len(results) else 1
if __name__ == "__main__":
sys.exit(main())