rorappandClaude Opus 5.5
f90b7df31c
ci / test (push) Successful in 24s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
117 lines
4.5 KiB
Python
117 lines
4.5 KiB
Python
import io
|
|
import json
|
|
import struct
|
|
import sys
|
|
import wave
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from knaster.audio import rtf, wav_duration
|
|
from knaster.bench import bench_tts
|
|
from knaster.canary import CanaryProber
|
|
from knaster.llm import OpenAIChatClient
|
|
from knaster.tts_server import CommandBackend, create_tts_app
|
|
|
|
|
|
def make_wav(seconds: float, rate: int = 22050) -> bytes:
|
|
buf = io.BytesIO()
|
|
with wave.open(buf, "wb") as w:
|
|
w.setnchannels(1)
|
|
w.setsampwidth(2)
|
|
w.setframerate(rate)
|
|
w.writeframes(struct.pack("<h", 0) * int(seconds * rate))
|
|
return buf.getvalue()
|
|
|
|
|
|
FAKE_PIPER = """
|
|
import sys, wave, struct
|
|
args = sys.argv[1:]
|
|
out = args[args.index("-f") + 1]
|
|
model = args[args.index("-m") + 1]
|
|
text = sys.stdin.read()
|
|
if "FEHLER" in text:
|
|
print("kaputt", file=sys.stderr); sys.exit(3)
|
|
with wave.open(out, "wb") as w:
|
|
w.setnchannels(1); w.setsampwidth(2); w.setframerate(16000)
|
|
w.writeframes(struct.pack("<h", 0) * (160 * len(text))) # 10 ms pro Zeichen
|
|
"""
|
|
|
|
|
|
@pytest.fixture
|
|
def backend(tmp_path):
|
|
script = tmp_path / "fake_piper.py"
|
|
script.write_text(FAKE_PIPER)
|
|
return CommandBackend({"thorsten": "/voices/t.onnx", "kerstin": "/voices/k.onnx"},
|
|
cmd=f"{sys.executable} {script} -m {{model}} -f {{out}}")
|
|
|
|
|
|
def test_wav_helpers():
|
|
assert wav_duration(make_wav(1.5)) == pytest.approx(1.5)
|
|
assert wav_duration(b"kein wav") == 0.0
|
|
assert rtf(3.0, 1.5) == 2.0 and rtf(1.0, 0) is None
|
|
|
|
|
|
def test_tts_server(backend):
|
|
client = TestClient(create_tts_app(backend, token="t"))
|
|
assert client.get("/health").json()["default"] == "thorsten"
|
|
H = {"Authorization": "Bearer t"}
|
|
r = client.post("/v1/audio/speech", json={"input": "Hallo Welt"}, headers=H)
|
|
assert r.status_code == 200 and r.headers["content-type"] == "audio/wav"
|
|
assert wav_duration(r.content) == pytest.approx(0.1)
|
|
assert client.post("/v1/audio/speech", json={"input": "x", "voice": "kerstin"}, headers=H).status_code == 200
|
|
assert client.post("/v1/audio/speech", json={"input": "x", "voice": "nope"}, headers=H).status_code == 400
|
|
assert client.post("/v1/audio/speech", json={"input": "FEHLER"}, headers=H).status_code == 502
|
|
assert client.post("/v1/audio/speech", json={"input": "x", "response_format": "mp3"},
|
|
headers=H).status_code == 400
|
|
assert client.post("/v1/audio/speech", json={"input": "x"}).status_code == 401
|
|
|
|
|
|
def test_backend_requires_voice():
|
|
with pytest.raises(ValueError):
|
|
CommandBackend({})
|
|
|
|
|
|
def tts_transport(seconds: float = 2.0, fail: bool = False):
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
assert request.url.path == "/v1/audio/speech"
|
|
body = json.loads(request.content)
|
|
if fail or body["model"] == "kaputt":
|
|
return httpx.Response(503, text="down")
|
|
return httpx.Response(200, content=make_wav(seconds), headers={"content-type": "audio/wav"})
|
|
|
|
return httpx.Client(transport=httpx.MockTransport(handler))
|
|
|
|
|
|
def test_bench_tts():
|
|
client = OpenAIChatClient("http://x", client=tts_transport(2.0))
|
|
result = bench_tts("http://x", "tts-de", runs=2, client=client)
|
|
assert len(result["runs"]) == 2 and result["errors"] == []
|
|
assert result["runs"][0]["audio_seconds"] == 2.0 and result["rtf_median"] is not None
|
|
bad = bench_tts("http://x", "kaputt", runs=1, client=client)
|
|
assert bad["runs"] == [] and bad["errors"]
|
|
|
|
|
|
def test_canary_probes_tts_slots(registry):
|
|
from prometheus_client import CollectorRegistry
|
|
|
|
from knaster.registry import Registry
|
|
from knaster.store import Store
|
|
|
|
from .conftest import REGISTRY
|
|
|
|
data = {"slots": {**REGISTRY["slots"], "tts-de": {
|
|
"tier": 1, "device": "cpu", "engine": "piper", "litellm_model": "tts-de", "base_model": "piper",
|
|
"max_context": 5000, "kind": "tts"}}}
|
|
reg = Registry.from_dict(data, Store("sqlite://"))
|
|
chat = OpenAIChatClient("http://x", client=tts_transport(0.8))
|
|
prober = CanaryProber(reg, chat, metrics_registry=CollectorRegistry())
|
|
tts = next(prober.probe(s) for s in reg.slots() if s.name == "tts-de")
|
|
assert tts.up and tts.ok
|
|
silent = OpenAIChatClient("http://x", client=tts_transport(0.05))
|
|
assert not CanaryProber(reg, silent, metrics_registry=CollectorRegistry()).probe(reg.slot("tts-de")).ok
|
|
down = OpenAIChatClient("http://x", client=tts_transport(fail=True))
|
|
assert not CanaryProber(reg, down, metrics_registry=CollectorRegistry()).probe(reg.slot("tts-de")).up
|
|
|