my_voice_assistant_v2/tests/test_endpoints_e2e.py

import json

from fastapi.testclient import TestClient

import app.dependencies as deps
from app.main import app
from tests.conftest import loopback_output

client = TestClient(app)


def test_speak_loopback_collects_chunks():
    # piper-Stub liefert b"" -> kein Netzcall; Loopback sammelt den Chunk.
    resp = client.post(
        "/api/speak",
        json={"text": "Hallo Welt", "tts_provider": "piper", "output_endpoint": "loopback"},
    )
    assert resp.status_code == 200
    assert resp.headers["X-Output-Endpoint"] == "loopback"
    assert resp.headers["X-TTS-Provider"] == "piper"
    assert len(loopback_output().chunks) == 1


def test_unknown_endpoint_returns_422():
    resp = client.post(
        "/api/speak",
        json={"text": "x", "tts_provider": "piper", "output_endpoint": "gibtsnicht"},
    )
    assert resp.status_code == 422
    assert "gibtsnicht" in resp.json()["detail"]


def test_unknown_provider_returns_422():
    resp = client.post("/api/speak", json={"text": "x", "tts_provider": "gibtsnicht"})
    assert resp.status_code == 422


def test_session_route_applies():
    client.post(
        "/api/sessions/s1/route",
        json={"tts_provider": "piper", "output_endpoint": "loopback"},
    )
    resp = client.post("/api/speak?session_id=s1", json={"text": "hallo"})
    assert resp.status_code == 200
    assert resp.headers["X-Output-Endpoint"] == "loopback"
    assert resp.headers["X-TTS-Provider"] == "piper"


def test_chat_per_request_override_and_loopback(monkeypatch):
    class StubLLM:
        async def complete(self, text, history=None, session_id=None):
            return "Mir geht es gut, danke."

    class StubTTS:
        async def synthesize(self, text, voice=None, audio_format="pcm"):
            return b"AUDIO"

    monkeypatch.setitem(deps.LLM_REGISTRY, "stub", lambda s: StubLLM())
    monkeypatch.setitem(deps.TTS_REGISTRY, "stub", lambda s: StubTTS())

    resp = client.post(
        "/api/chat?debug=true",
        json={
            "text": "Wie geht es dir?",
            "llm_provider": "stub",
            "tts_provider": "stub",
            "output_endpoint": "loopback",
        },
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["route"]["llm_provider"] == "stub"
    assert body["route"]["output_endpoint"] == "loopback"
    assert body["trace"]["semantic_response"] == "Mir geht es gut, danke."
    assert loopback_output().chunks[0].data == b"AUDIO"


def test_transcribe_with_stub_stt(monkeypatch):
    class StubSTT:
        async def transcribe(self, audio_bytes, fmt, language=None):
            return "erkannter text"

    monkeypatch.setitem(deps.STT_REGISTRY, "stub-stt", lambda s: StubSTT())
    files = {"file": ("a.wav", b"RIFFdata", "audio/wav")}
    data = {"stt_provider": "stub-stt"}
    resp = client.post("/api/transcribe", data=data, files=files)
    assert resp.status_code == 200
    body = resp.json()
    assert body["route"]["stt_provider"] == "stub-stt"
    assert body["trace"]["raw_transcript"] == "erkannter text"


def test_config_endpoint_exposes_no_secrets():
    resp = client.get("/api/config")
    assert resp.status_code == 200
    body = resp.json()
    assert "piper" in body["available"]["tts_providers"]
    assert "loopback" in {e["kind"] for e in body["available"]["output_endpoints"]}
    # Keine echten Secrets im Body.
    assert "sk-or-" not in json.dumps(body)
    assert set(body["secrets"].keys()) == {"openrouter_api_key_set"}