82 lines
2.9 KiB
Python
82 lines
2.9 KiB
Python
|
import io
|
||
|
|
import wave
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
import app.providers.tts.chatterbox as cb
|
||
|
|
from app.providers.tts.chatterbox import ChatterboxTTSProvider
|
||
|
|
|
||
|
|
|
||
|
|
def _wav(rate=24000, payload=b"\x01\x02" * 2400) -> bytes:
|
||
|
|
buf = io.BytesIO()
|
||
|
|
with wave.open(buf, "wb") as w:
|
||
|
|
w.setnchannels(1)
|
||
|
|
w.setsampwidth(2)
|
||
|
|
w.setframerate(rate)
|
||
|
|
w.writeframes(payload)
|
||
|
|
return buf.getvalue()
|
||
|
|
|
||
|
|
|
||
|
|
def _install(monkeypatch, handler):
|
||
|
|
real = httpx.AsyncClient # echte Klasse sichern (vor dem Patch)
|
||
|
|
|
||
|
|
def make_client(*args, **kwargs):
|
||
|
|
kwargs.pop("timeout", None)
|
||
|
|
return real(transport=httpx.MockTransport(handler))
|
||
|
|
monkeypatch.setattr(cb.httpx, "AsyncClient", make_client)
|
||
|
|
|
||
|
|
|
||
|
|
def _ok_handler(status="done", wav_rate=24000):
|
||
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
||
|
|
p = request.url.path
|
||
|
|
if request.method == "POST" and p == "/speak":
|
||
|
|
return httpx.Response(200, json={"job_id": "j1", "status": "pending"})
|
||
|
|
if p == "/status":
|
||
|
|
return httpx.Response(200, json={"recent_jobs": [{"id": "j1", "status": status,
|
||
|
|
"error": "boom" if status == "error" else None}]})
|
||
|
|
if p == "/audio/j1":
|
||
|
|
return httpx.Response(200, content=_wav(wav_rate), headers={"content-type": "audio/wav"})
|
||
|
|
return httpx.Response(404)
|
||
|
|
return handler
|
||
|
|
|
||
|
|
|
||
|
|
def test_synthesize_returns_pcm(monkeypatch):
|
||
|
|
_install(monkeypatch, _ok_handler())
|
||
|
|
p = ChatterboxTTSProvider("http://x:9999", voice="/ref.wav")
|
||
|
|
pcm = __import__("asyncio").run(p.synthesize("Hallo Welt"))
|
||
|
|
assert pcm == b"\x01\x02" * 2400 # 24000 Hz -> kein Resampling, Frames unveraendert
|
||
|
|
|
||
|
|
|
||
|
|
def test_synthesize_wav_format(monkeypatch):
|
||
|
|
_install(monkeypatch, _ok_handler())
|
||
|
|
p = ChatterboxTTSProvider("http://x:9999")
|
||
|
|
wav = __import__("asyncio").run(p.synthesize("Hallo", audio_format="wav"))
|
||
|
|
assert wav.startswith(b"RIFF") and b"WAVE" in wav[:16]
|
||
|
|
|
||
|
|
|
||
|
|
def test_resample_when_rate_differs(monkeypatch):
|
||
|
|
_install(monkeypatch, _ok_handler(wav_rate=22050))
|
||
|
|
p = ChatterboxTTSProvider("http://x:9999", target_rate=24000)
|
||
|
|
pcm = __import__("asyncio").run(p.synthesize("Hallo"))
|
||
|
|
assert isinstance(pcm, bytes) and len(pcm) > 0
|
||
|
|
|
||
|
|
|
||
|
|
def test_job_error_raises(monkeypatch):
|
||
|
|
_install(monkeypatch, _ok_handler(status="error"))
|
||
|
|
p = ChatterboxTTSProvider("http://x:9999")
|
||
|
|
with pytest.raises(RuntimeError):
|
||
|
|
__import__("asyncio").run(p.synthesize("Hallo"))
|
||
|
|
|
||
|
|
|
||
|
|
def test_empty_text_raises(monkeypatch):
|
||
|
|
p = ChatterboxTTSProvider("http://x:9999")
|
||
|
|
with pytest.raises(ValueError):
|
||
|
|
__import__("asyncio").run(p.synthesize(" "))
|
||
|
|
|
||
|
|
|
||
|
|
def test_ref_voice_selection():
|
||
|
|
p = ChatterboxTTSProvider("http://x:9999", voice="/default.wav")
|
||
|
|
assert p._ref_voice(None) == "/default.wav"
|
||
|
|
assert p._ref_voice("Zephyr") == "/default.wav" # Cloud-Name -> Default
|
||
|
|
assert p._ref_voice("/custom.wav") == "/custom.wav" # Pfad -> uebernommen
|