ratatoskr/tests/test_tts.py

"""Tests for ratatoskr.tts — the STREAMING dots-tts gateway client.

tts_stream proxies the gateway's chunked response verbatim (no buffering, no header
rewrite — the placeholder-size streaming WAV is meant to be played progressively). It
is the sole synthesis primitive: dots streams a whole turn from one call, so there is
no client-side chunk-and-concatenate, and no affect dials (dots has no emotion knob).
The mid-stream degrade policy is folded in.
"""

import httpx
import pytest
import respx

from ratatoskr.tts import (
    DOTS_TTS_URL,
    TtsUnavailable,
    gateway_body,
    tts_stream,
)

_URL = "http://tts.example/v1/audio/speech"
# The gateway's streaming WAV bytes (placeholder 0xFFFFFFFF sizes). We pass them through
# untouched, so the content only has to round-trip.
_WAV = (
    b"RIFF\xff\xff\xff\xffWAVEfmt \x10\x00\x00\x00" + b"\x00" * 20
    + b"data\xff\xff\xff\xff" + b"\x11\x22" * 64
)


async def _drain(gen) -> bytes:
    out = b""
    async for chunk in gen:
        out += chunk
    return out


class _RaisingByteStream(httpx.AsyncByteStream):
    """A 200-body stream that yields `head` then drops mid-stream (an httpx.ReadError, a
    RequestError subclass) — models a gateway connection drop AFTER the response committed."""

    def __init__(self, head: bytes) -> None:
        self._head = head

    async def __aiter__(self):
        yield self._head
        raise httpx.ReadError("mid-stream drop")

    async def aclose(self) -> None:
        pass


class TestGatewayBody:
    def test_openai_dots_schema(self) -> None:
        b = gateway_body("hi", "donut")
        assert b["input"] == "hi"              # OpenAI "input", not chatterbox "text"
        assert b["voice"] == "donut"
        assert b["response_format"] == "wav"   # DEC-3 — "response_format", not "format"
        assert b["stream"] is True             # DEC-2 — play-as-it-arrives

    def test_default_sampling_no_client_side_curbs(self) -> None:
        # No client sampling curbs — the gateway's defaults govern (a client-side curb
        # was counterproductive on the prior backend and dots exposes no such need).
        b = gateway_body("a long turn", "donut")
        for knob in ("temperature", "top_p", "top_k"):
            assert knob not in b

    def test_no_chatterbox_or_zonos_era_fields(self) -> None:
        # The chatterbox bespoke names + Zonos-era fields are gone: no `text`/`format`
        # (chatterbox), no `language` pin, no affect dials (DEC-7 retired).
        b = gateway_body("hi", "glados")
        for dead in ("text", "format", "language", "emotion_valence",
                     "emotion_arousal", "emotion_enabled", "emotion_strength"):
            assert dead not in b


class TestTtsStream:
    @respx.mock
    async def test_streams_chunks_and_posts_openai_body(self) -> None:
        route = respx.post(_URL).mock(return_value=httpx.Response(200, content=_WAV))
        async with httpx.AsyncClient() as client:
            out = await _drain(tts_stream("hello there", voice="donut", client=client, url=_URL))
        assert out == _WAV  # passed through verbatim — no header rewrite
        import json as _json

        body = _json.loads(route.calls.last.request.content)
        assert body["input"] == "hello there"
        assert body["voice"] == "donut"
        assert body["response_format"] == "wav"
        assert body["stream"] is True

    @respx.mock
    async def test_default_url_is_dots(self) -> None:
        route = respx.post(DOTS_TTS_URL).mock(
            return_value=httpx.Response(200, content=_WAV)
        )
        async with httpx.AsyncClient() as client:
            await _drain(tts_stream("hi", voice="donut", client=client))
        assert route.called  # the module default points at the dots-tts gateway

    @respx.mock
    async def test_non_200_open_raises_before_any_chunk(self) -> None:
        respx.post(_URL).mock(return_value=httpx.Response(500, content=b"boom"))
        async with httpx.AsyncClient() as client:
            with pytest.raises(TtsUnavailable) as exc:
                await _drain(tts_stream("hi", voice="Cora", client=client, url=_URL))
        assert exc.value.status == 500

    @respx.mock
    async def test_transport_error_on_open_raises(self) -> None:
        respx.post(_URL).mock(side_effect=httpx.ConnectError("refused"))
        async with httpx.AsyncClient() as client:
            with pytest.raises(TtsUnavailable):
                await _drain(tts_stream("hi", voice="Cora", client=client, url=_URL))

    @respx.mock
    async def test_mid_stream_drop_after_first_byte_degrades_not_raises(self) -> None:
        # The 200 is committed once bytes flow; a later transport drop must DEGRADE
        # (return what streamed), never raise — the pivot is yielded_any, folded in from
        # the retired tts_stream_long. Keeps a committed StreamingResponse from an ASGI trace.
        respx.post(_URL).mock(
            return_value=httpx.Response(200, stream=_RaisingByteStream(_WAV))
        )
        async with httpx.AsyncClient() as client:
            out = await _drain(tts_stream("hi", voice="donut", client=client, url=_URL))
        assert out == _WAV  # head kept, no raise