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

tts_stream_stitched is the sole synthesis primitive: it synthesizes an ordered list of
(voice, text) spans serially into one continuous stream — the first span verbatim, later
spans header-stripped (DEC-11 two-voice split). A single-span list is a verbatim passthrough
(no buffering, no header rewrite), so the single-voice / dialogue-only case is unchanged. No
affect dials (dots has no emotion knob). The mid-stream degrade policy spans the sequence.
"""

import httpx
import pytest
import respx

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

_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
)
# A second span's WAV with distinct PCM — its header is stripped when stitched after span 0.
_WAV2 = (
    b"RIFF\xff\xff\xff\xffWAVEfmt \x10\x00\x00\x00" + b"\x00" * 20
    + b"data\xff\xff\xff\xff" + b"\x33\x44" * 32
)
_PCM2 = b"\x33\x44" * 32  # the part of _WAV2 after `data`+size (what stitching keeps)


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


def _json_voice(route, i: int) -> str:
    import json as _json

    return _json.loads(route.calls[i].request.content)["voice"]


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 TestTtsStreamStitched:
    @respx.mock
    async def test_single_span_verbatim_and_posts_openai_body(self) -> None:
        # A single-span list is a verbatim passthrough (INV-TTS-6) with the OpenAI body.
        route = respx.post(_URL).mock(return_value=httpx.Response(200, content=_WAV))
        async with httpx.AsyncClient() as client:
            out = await _drain(
                tts_stream_stitched([("donut", "hello there")], 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_two_spans_stitched_one_header(self) -> None:
        # Span 0 verbatim (its WAV header + PCM), span 1 header-STRIPPED → one continuous
        # stream with exactly one leading header (INV-TTS-7). Distinct voices per span.
        route = respx.post(_URL).mock(
            side_effect=[httpx.Response(200, content=_WAV), httpx.Response(200, content=_WAV2)]
        )
        async with httpx.AsyncClient() as client:
            out = await _drain(
                tts_stream_stitched(
                    [("miranda", "spoken bit"), ("emmie", "narrated bit")],
                    client=client, url=_URL,
                )
            )
        assert out == _WAV + _PCM2                      # span1's header dropped, PCM kept
        assert out.count(b"RIFF") == 1                  # exactly one WAV header
        assert _json_voice(route, 0) == "miranda" and _json_voice(route, 1) == "emmie"

    @respx.mock
    async def test_empty_span_skipped(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_stitched(
                    [("miranda", "   "), ("donut", "real")], client=client, url=_URL
                )
            )
        assert out == _WAV and len(route.calls) == 1     # blank span never synthesized

    @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_stitched([("donut", "hi")], 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_stitched([("glados", "hi")], 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_stitched([("glados", "hi")], 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. 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_stitched([("donut", "hi")], client=client, url=_URL))
        assert out == _WAV  # head kept, no raise

    @respx.mock
    async def test_later_span_open_fail_after_commit_degrades(self) -> None:
        # Span 0 commits a 200 + bytes; span 1's OPEN then 500s. Because the stream is already
        # committed, this DEGRADES (keep span 0), never raises into the 200 (INV-TTS-4).
        route = respx.post(_URL).mock(
            side_effect=[httpx.Response(200, content=_WAV), httpx.Response(500, content=b"boom")]
        )
        async with httpx.AsyncClient() as client:
            out = await _drain(
                tts_stream_stitched([("miranda", "a"), ("emmie", "b")], client=client, url=_URL)
            )
        assert out == _WAV and len(route.calls) == 2  # span0 kept, span1 attempted then dropped