my_voice_assistant_v2/tests/test_chatterbox_tts.py

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, "de") == "/default.wav"
    assert p._ref_voice("Zephyr", "de") == "/default.wav"      # Cloud-Name -> Default
    assert p._ref_voice("/custom.wav", "de") == "/custom.wav"  # Pfad -> uebernommen


def test_ref_voice_per_language(tmp_path):
    # Native Sprach-Referenz gewinnt über die Default-Stimme, fehlt sie -> Default.
    (tmp_path / "fr.wav").write_bytes(b"RIFF")
    p = ChatterboxTTSProvider("http://x:9999", voice="/default.wav", voices_dir=str(tmp_path))
    assert p._ref_voice(None, "fr") == str(tmp_path / "fr.wav")  # native fr-Stimme
    assert p._ref_voice(None, "de") == "/default.wav"            # keine de.wav -> Default
    assert p._ref_voice("/custom.wav", "fr") == "/custom.wav"    # expliziter Pfad gewinnt