echo-core/tests/test_tts.py

"""Tests for tools/tts.py — rutare engine, fallback pockettts -> supertonic, retry lang=na."""

from unittest.mock import MagicMock, patch

import httpx
import pytest

from tools.tts import _PocketTTSUnavailable, _synthesize_pockettts, _synthesize_supertonic, synthesize


def _fake_response(status_code=200, content=b"RIFF....WAVE", text=""):
    resp = MagicMock()
    resp.status_code = status_code
    resp.content = content
    resp.text = text
    if status_code >= 400:
        resp.raise_for_status.side_effect = httpx.HTTPStatusError(
            "error", request=MagicMock(), response=resp
        )
    else:
        resp.raise_for_status = MagicMock()
    return resp


class TestEngineRouting:
    @patch("tools.tts._load_voice_catalog")
    @patch("tools.tts._synthesize_pockettts")
    def test_catalogued_voice_uses_pockettts(self, mock_pockettts, mock_catalog):
        mock_catalog.return_value = {
            "Marius 1": {"engine": "pockettts", "state_path": "models/voices/marius-1.safetensors"}
        }
        mock_pockettts.return_value = {"ok": True, "path": "/tmp/x.wav", "size_bytes": 10}

        result = synthesize("salut", voice="Marius 1")

        mock_pockettts.assert_called_once()
        assert result["engine_used"] == "pockettts"
        assert result["ok"] is True

    @patch("tools.tts._load_voice_catalog")
    @patch("tools.tts._synthesize_supertonic")
    def test_catalogued_voice_uses_supertonic(self, mock_supertonic, mock_catalog):
        mock_catalog.return_value = {"M1": {"engine": "supertonic"}}
        mock_supertonic.return_value = {"ok": True, "path": "/tmp/x.wav", "size_bytes": 10}

        result = synthesize("salut", voice="M1")

        mock_supertonic.assert_called_once()
        assert result["engine_used"] == "supertonic"

    @patch("tools.tts._config_get")
    @patch("tools.tts._load_voice_catalog")
    @patch("tools.tts._synthesize_pockettts")
    def test_uncatalogued_voice_falls_back_to_default_engine_config(
        self, mock_pockettts, mock_catalog, mock_config_get
    ):
        mock_catalog.return_value = {}
        mock_config_get.return_value = "pockettts"
        mock_pockettts.return_value = {"ok": True, "path": "/tmp/x.wav", "size_bytes": 10}

        result = synthesize("salut", voice="voce-inexistenta")

        mock_config_get.assert_called_with("tts.default_engine", "supertonic")
        assert result["engine_used"] == "pockettts"


class TestPockettsFallback:
    @patch("tools.tts._load_voice_catalog")
    @patch("tools.tts._synthesize_supertonic")
    @patch("tools.tts._synthesize_pockettts")
    def test_connect_error_falls_back_to_supertonic(
        self, mock_pockettts, mock_supertonic, mock_catalog
    ):
        mock_catalog.return_value = {"Marius 1": {"engine": "pockettts"}}
        mock_pockettts.side_effect = _PocketTTSUnavailable("connection refused")
        mock_supertonic.return_value = {"ok": True, "path": "/tmp/x.wav", "size_bytes": 10}

        result = synthesize("salut", voice="Marius 1")

        mock_supertonic.assert_called_once()
        assert result["engine_used"] == "supertonic"
        assert result["ok"] is True

    @patch("tools.tts._load_voice_catalog")
    @patch("tools.tts._synthesize_supertonic")
    @patch("tools.tts._synthesize_pockettts")
    def test_5xx_falls_back_to_supertonic(self, mock_pockettts, mock_supertonic, mock_catalog):
        mock_catalog.return_value = {"Marius 1": {"engine": "pockettts"}}
        mock_pockettts.side_effect = _PocketTTSUnavailable("HTTP 503: unavailable")
        mock_supertonic.return_value = {"ok": True, "path": "/tmp/x.wav", "size_bytes": 10}

        result = synthesize("salut", voice="Marius 1")

        mock_supertonic.assert_called_once()
        assert result["engine_used"] == "supertonic"

    @patch("tools.tts._load_voice_catalog")
    @patch("tools.tts._synthesize_supertonic")
    def test_content_error_does_not_fall_back_to_supertonic(self, mock_supertonic, mock_catalog):
        """Missing state_path file -> ok:false dict, not a raised _PocketTTSUnavailable — no fallback."""
        mock_catalog.return_value = {
            "Marius 1": {"engine": "pockettts", "state_path": "models/voices/does-not-exist.safetensors"}
        }

        result = synthesize("salut", voice="Marius 1")

        mock_supertonic.assert_not_called()
        assert result["ok"] is False
        assert result["engine_used"] == "pockettts"
        assert "lipsă" in result["error"]

    @patch("httpx.post")
    def test_synthesize_pockettts_raises_on_connect_error(self, mock_post):
        mock_post.side_effect = httpx.ConnectError("connection refused")
        with pytest.raises(_PocketTTSUnavailable):
            _synthesize_pockettts("salut", {})

    @patch("httpx.post")
    def test_synthesize_pockettts_raises_on_5xx(self, mock_post):
        mock_post.return_value = _fake_response(status_code=503, text="unavailable")
        with pytest.raises(_PocketTTSUnavailable):
            _synthesize_pockettts("salut", {})

    @patch("httpx.post")
    def test_synthesize_pockettts_returns_error_dict_on_4xx(self, mock_post):
        mock_post.return_value = _fake_response(status_code=400, text="bad request")
        result = _synthesize_pockettts("salut", {})
        assert result["ok"] is False
        assert "400" in result["error"]

    @patch("httpx.post")
    def test_romanian_diacritics_rejected_without_calling_server(self, mock_post):
        result = _synthesize_pockettts("Salut, cum îți este ziua?", {})
        assert result["ok"] is False
        assert "engleză" in result["error"]
        mock_post.assert_not_called()

    @patch("tools.tts._load_voice_catalog")
    @patch("httpx.post")
    def test_romanian_text_does_not_fall_back_to_supertonic(self, mock_post, mock_catalog):
        mock_catalog.return_value = {"Marius 1": {"engine": "pockettts"}}

        result = synthesize("Bună dimineața!", voice="Marius 1")

        mock_post.assert_not_called()
        assert result["ok"] is False
        assert result["engine_used"] == "pockettts"


class TestSupertonicLangNaRetry:
    @patch("tools.tts._load_voice_catalog")
    @patch("httpx.post")
    def test_ro_failure_retries_with_lang_na(self, mock_post, mock_catalog):
        mock_catalog.return_value = {}

        def side_effect(url, json, timeout):
            if json.get("lang") == "ro":
                return _fake_response(status_code=500, text="ro synthesis failed")
            return _fake_response(status_code=200, content=b"RIFF....WAVE")

        mock_post.side_effect = side_effect

        result = _synthesize_supertonic("salut", voice="M1", lang="ro")

        assert result["ok"] is True
        assert mock_post.call_count == 2
        langs_requested = [call.kwargs["json"]["lang"] for call in mock_post.call_args_list]
        assert langs_requested == ["ro", "na"]

    @patch("tools.tts._load_voice_catalog")
    @patch("httpx.post")
    def test_na_failure_does_not_retry_again(self, mock_post, mock_catalog):
        """Regression: retry only happens once (ro -> na), na failure returns the error directly."""
        mock_catalog.return_value = {}
        mock_post.return_value = _fake_response(status_code=500, text="still failing")

        result = _synthesize_supertonic("salut", voice="M1", lang="na")

        assert result["ok"] is False
        assert mock_post.call_count == 1