ratatoskr/tests/test_tts.py

"""Tests for ratatoskr.tts per docs/contracts/donut_voiced_interview.contract.md
(slice 2 — auto-TTS). Two units:

  - pad_to_dials / PadState / EmotionDials — pure, total, degrade-not-crash.
  - tts_synthesize — the Zonos-gateway client (respx-mocked; no live network).

DEC-3: response_format is ALWAYS "wav". DEC-7: live PAD → emotion dials.
INV-TTS-4: a gateway failure / non-wav body → TtsUnavailable (caller degrades).
"""

import math

import httpx
import pytest
import respx

from ratatoskr.tts import (
    EmotionDials,
    PadState,
    TtsUnavailable,
    pad_to_dials,
    tts_synthesize,
)

# A minimally-valid wav body: the DEC-3 guard only checks the RIFF magic.
_WAV = b"RIFF" + b"\x00" * 40


class TestPadState:
    def test_from_obj_valid_mapping(self) -> None:
        pad = PadState.from_obj({"pleasure": 0.5, "arousal": -0.2, "dominance": 0.1})
        assert pad == PadState(pleasure=0.5, arousal=-0.2, dominance=0.1)

    def test_from_obj_dominance_optional(self) -> None:
        pad = PadState.from_obj({"pleasure": 0.5, "arousal": -0.2})
        assert pad is not None and pad.dominance == 0.0

    def test_from_obj_none_is_none(self) -> None:
        assert PadState.from_obj(None) is None

    def test_from_obj_non_mapping_is_none(self) -> None:
        assert PadState.from_obj("not a mapping") is None
        assert PadState.from_obj([0.1, 0.2]) is None

    def test_from_obj_missing_key_is_none(self) -> None:
        assert PadState.from_obj({"pleasure": 0.5}) is None  # no arousal

    def test_from_obj_non_numeric_is_none(self) -> None:
        assert PadState.from_obj({"pleasure": "hot", "arousal": 0.1}) is None


class TestPadToDials:
    def test_none_pad_is_neutral_disabled(self) -> None:
        d = pad_to_dials(None)
        assert d.emotion_enabled is False
        assert d.to_body() == {}  # a neutral read sends no emotion params

    def test_maps_pleasure_and_arousal(self) -> None:
        d = pad_to_dials(PadState(pleasure=0.4, arousal=0.6))
        assert d.emotion_enabled is True
        assert d.emotion_valence == pytest.approx(0.4)
        assert d.emotion_arousal == pytest.approx(0.6)

    def test_clamps_out_of_range(self) -> None:
        d = pad_to_dials(PadState(pleasure=5.0, arousal=-9.0))
        assert d.emotion_valence == 1.0
        assert d.emotion_arousal == -1.0

    def test_nan_degrades_to_zero_never_raises(self) -> None:
        # INV FN pad_to_dials: total over ANY input (incl. NaN) → valid dials.
        d = pad_to_dials(PadState(pleasure=math.nan, arousal=math.inf))
        assert d.emotion_valence == 0.0
        assert d.emotion_arousal == 1.0  # +inf clamps to the ceiling
        assert not math.isnan(d.emotion_valence)


class TestEmotionDialsToBody:
    def test_disabled_emits_no_params(self) -> None:
        assert EmotionDials(emotion_enabled=False).to_body() == {}

    def test_enabled_emits_valence_arousal_strength(self) -> None:
        body = EmotionDials(
            emotion_enabled=True, emotion_valence=0.3, emotion_arousal=-0.1
        ).to_body()
        assert body["emotion_enabled"] is True
        assert body["emotion_valence"] == 0.3
        assert body["emotion_arousal"] == -0.1
        assert "emotion_strength" in body


class TestTtsSynthesize:
    @respx.mock
    async def test_happy_posts_wav_request_and_returns_bytes(self) -> None:
        route = respx.post("http://tts.example/v1/audio/speech").mock(
            return_value=httpx.Response(200, content=_WAV)
        )
        async with httpx.AsyncClient() as client:
            out = await tts_synthesize(
                "hello there",
                voice="donut",
                dials=pad_to_dials(PadState(pleasure=0.5, arousal=0.2)),
                client=client,
                url="http://tts.example/v1/audio/speech",
            )
        assert out == _WAV
        sent = route.calls.last.request
        import json as _json

        body = _json.loads(sent.content)
        assert body["input"] == "hello there"
        assert body["voice"] == "donut"
        assert body["response_format"] == "wav"  # DEC-3 — ALWAYS wav
        assert body["emotion_valence"] == pytest.approx(0.5)

    @respx.mock
    async def test_neutral_dials_send_no_emotion_params(self) -> None:
        route = respx.post("http://tts.example/v1/audio/speech").mock(
            return_value=httpx.Response(200, content=_WAV)
        )
        async with httpx.AsyncClient() as client:
            await tts_synthesize(
                "hi", voice="Cora", dials=pad_to_dials(None), client=client,
                url="http://tts.example/v1/audio/speech",
            )
        import json as _json

        body = _json.loads(route.calls.last.request.content)
        assert "emotion_valence" not in body
        assert body["response_format"] == "wav"

    @respx.mock
    async def test_non_200_raises_tts_unavailable(self) -> None:
        respx.post("http://tts.example/v1/audio/speech").mock(
            return_value=httpx.Response(500, content=b"boom")
        )
        async with httpx.AsyncClient() as client:
            with pytest.raises(TtsUnavailable) as exc:
                await tts_synthesize(
                    "hi", voice="Cora", dials=pad_to_dials(None), client=client,
                    url="http://tts.example/v1/audio/speech",
                )
        assert exc.value.status == 500

    @respx.mock
    async def test_transport_error_raises_tts_unavailable(self) -> None:
        respx.post("http://tts.example/v1/audio/speech").mock(
            side_effect=httpx.ConnectError("refused")
        )
        async with httpx.AsyncClient() as client:
            with pytest.raises(TtsUnavailable):
                await tts_synthesize(
                    "hi", voice="Cora", dials=pad_to_dials(None), client=client,
                    url="http://tts.example/v1/audio/speech",
                )

    @respx.mock
    async def test_non_wav_body_raises_tts_unavailable(self) -> None:
        # DEC-3 guard: a 200 with a non-RIFF body (an error page, or the
        # mislabeled-PCM mp3/opus trap) is unavailable, not played.
        respx.post("http://tts.example/v1/audio/speech").mock(
            return_value=httpx.Response(200, content=b"<html>error</html>")
        )
        async with httpx.AsyncClient() as client:
            with pytest.raises(TtsUnavailable):
                await tts_synthesize(
                    "hi", voice="Cora", dials=pad_to_dials(None), client=client,
                    url="http://tts.example/v1/audio/speech",
                )