cca423dac3
PiperTTSProvider ruft das piper-Binary (--output-raw) async auf, liest die native Sample-Rate aus der .onnx.json und resampelt per ffmpeg auf 24000 Hz (Gateway-Norm). Nicht passende Stimmen (z. B. Cloud-Stimme 'Zephyr' aus der Route) fallen auf die konfigurierte Default-Stimme zurueck. Damit ist eine voll-lokale Konstellation (faster-whisper + Ollama + piper) moeglich -> keine API-Kosten, max. Datenschutz. - config: PIPER_BIN/PIPER_VOICES_DIR/PIPER_VOICE/TTS_SAMPLE_RATE (+ .env.example) - dependencies: piper-Factory mit Settings verdrahtet - tests: tests/test_piper_tts.py (offline, Fake-Binary; Resample-Test skippt ohne ffmpeg); e2e/auth-Tests nutzen jetzt einen Stub-TTS statt 'piper' als Pseudo-Stub - docs: README, BEDIENUNGSANLEITUNG (voll-lokal-Beispiel), Architektur-Roadmap Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
112 lines
3.8 KiB
Python
112 lines
3.8 KiB
Python
import json
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app.dependencies as deps
|
|
from app.main import app
|
|
from tests.conftest import loopback_output
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def _stub_tts(monkeypatch, name="stub-tts", audio=b""):
|
|
"""Registriert einen deterministischen TTS-Stub (kein Netz, kein lokales Binary)."""
|
|
class StubTTS:
|
|
async def synthesize(self, text, voice=None, audio_format="pcm"):
|
|
return audio
|
|
monkeypatch.setitem(deps.TTS_REGISTRY, name, lambda s: StubTTS())
|
|
return name
|
|
|
|
|
|
def test_speak_loopback_collects_chunks(monkeypatch):
|
|
# Stub-TTS liefert b"" -> kein Netzcall; Loopback sammelt den Chunk.
|
|
tts = _stub_tts(monkeypatch)
|
|
resp = client.post(
|
|
"/api/speak",
|
|
json={"text": "Hallo Welt", "tts_provider": tts, "output_endpoint": "loopback"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.headers["X-Output-Endpoint"] == "loopback"
|
|
assert resp.headers["X-TTS-Provider"] == tts
|
|
assert len(loopback_output().chunks) == 1
|
|
|
|
|
|
def test_unknown_endpoint_returns_422():
|
|
resp = client.post(
|
|
"/api/speak",
|
|
json={"text": "x", "tts_provider": "piper", "output_endpoint": "gibtsnicht"},
|
|
)
|
|
assert resp.status_code == 422
|
|
assert "gibtsnicht" in resp.json()["detail"]
|
|
|
|
|
|
def test_unknown_provider_returns_422():
|
|
resp = client.post("/api/speak", json={"text": "x", "tts_provider": "gibtsnicht"})
|
|
assert resp.status_code == 422
|
|
|
|
|
|
def test_session_route_applies(monkeypatch):
|
|
tts = _stub_tts(monkeypatch)
|
|
client.post(
|
|
"/api/sessions/s1/route",
|
|
json={"tts_provider": tts, "output_endpoint": "loopback"},
|
|
)
|
|
resp = client.post("/api/speak?session_id=s1", json={"text": "hallo"})
|
|
assert resp.status_code == 200
|
|
assert resp.headers["X-Output-Endpoint"] == "loopback"
|
|
assert resp.headers["X-TTS-Provider"] == tts
|
|
|
|
|
|
def test_chat_per_request_override_and_loopback(monkeypatch):
|
|
class StubLLM:
|
|
async def complete(self, text, history=None, session_id=None):
|
|
return "Mir geht es gut, danke."
|
|
|
|
class StubTTS:
|
|
async def synthesize(self, text, voice=None, audio_format="pcm"):
|
|
return b"AUDIO"
|
|
|
|
monkeypatch.setitem(deps.LLM_REGISTRY, "stub", lambda s: StubLLM())
|
|
monkeypatch.setitem(deps.TTS_REGISTRY, "stub", lambda s: StubTTS())
|
|
|
|
resp = client.post(
|
|
"/api/chat?debug=true",
|
|
json={
|
|
"text": "Wie geht es dir?",
|
|
"llm_provider": "stub",
|
|
"tts_provider": "stub",
|
|
"output_endpoint": "loopback",
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["route"]["llm_provider"] == "stub"
|
|
assert body["route"]["output_endpoint"] == "loopback"
|
|
assert body["trace"]["semantic_response"] == "Mir geht es gut, danke."
|
|
assert loopback_output().chunks[0].data == b"AUDIO"
|
|
|
|
|
|
def test_transcribe_with_stub_stt(monkeypatch):
|
|
class StubSTT:
|
|
async def transcribe(self, audio_bytes, fmt, language=None):
|
|
return "erkannter text"
|
|
|
|
monkeypatch.setitem(deps.STT_REGISTRY, "stub-stt", lambda s: StubSTT())
|
|
files = {"file": ("a.wav", b"RIFFdata", "audio/wav")}
|
|
data = {"stt_provider": "stub-stt"}
|
|
resp = client.post("/api/transcribe", data=data, files=files)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["route"]["stt_provider"] == "stub-stt"
|
|
assert body["trace"]["raw_transcript"] == "erkannter text"
|
|
|
|
|
|
def test_config_endpoint_exposes_no_secrets():
|
|
resp = client.get("/api/config")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert "piper" in body["available"]["tts_providers"]
|
|
assert "loopback" in {e["kind"] for e in body["available"]["output_endpoints"]}
|
|
# Keine echten Secrets im Body.
|
|
assert "sk-or-" not in json.dumps(body)
|
|
assert set(body["secrets"].keys()) == {"openrouter_api_key_set"}
|