my_voice_assistant_v3/tests/test_quota_safety.py

from fastapi.testclient import TestClient

import app.dependencies as deps
from app.main import app
from app.config import settings
from app.safety.emergency import detect

client = TestClient(app)


def _stub_chat(monkeypatch, answer="ok"):
    class StubLLM:
        async def complete(self, text, history=None, session_id=None, language=None):
            return answer

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

    monkeypatch.setitem(deps.LLM_REGISTRY, "l", lambda s: StubLLM())
    monkeypatch.setitem(deps.TTS_REGISTRY, "t", lambda s: StubTTS())
    return {"llm_provider": "l", "tts_provider": "t"}


# --- Notfall-Erkennung (Einheit) ------------------------------------------

def test_detect_categories():
    assert detect("Ich habe Brustschmerzen")[0] == "medical"
    assert detect("Bitte ruf einen Arzt")[0] == "help"
    assert detect("I want to die")[0] == "self_harm"
    assert detect("Wie wird das Wetter morgen?") is None
    assert detect("") is None


# --- Quota -----------------------------------------------------------------

def test_quota_blocks_after_limit(monkeypatch):
    base = _stub_chat(monkeypatch)
    monkeypatch.setattr(settings, "daily_request_limit", 1)
    body = {"text": "Hallo", **base}
    assert client.post("/api/chat?debug=true", json=body).status_code == 200
    assert client.post("/api/chat?debug=true", json=body).status_code == 429


def test_quota_unlimited_by_default(monkeypatch):
    base = _stub_chat(monkeypatch)
    body = {"text": "Hallo", **base}
    for _ in range(3):
        assert client.post("/api/chat?debug=true", json=body).status_code == 200


# --- Notfall ueber Endpunkt ------------------------------------------------

def test_emergency_surfaced_in_response(monkeypatch):
    base = _stub_chat(monkeypatch, answer="Bleiben Sie ruhig.")
    resp = client.post(
        "/api/chat?debug=true",
        json={"text": "Ich habe starke Schmerzen in der Brust", **base},
    )
    assert resp.status_code == 200
    assert resp.json()["emergency"]["category"] == "medical"


def test_emergency_bypasses_quota(monkeypatch):
    base = _stub_chat(monkeypatch)
    monkeypatch.setattr(settings, "daily_request_limit", 1)
    body = {**base}
    assert client.post("/api/chat?debug=true", json={"text": "hallo", **body}).status_code == 200
    assert client.post("/api/chat?debug=true", json={"text": "hallo", **body}).status_code == 429
    rescue = client.post("/api/chat?debug=true", json={"text": "ich kann nicht atmen", **body})
    assert rescue.status_code == 200
    assert rescue.json()["emergency"]["category"] == "medical"


def test_ws_emergency_event(monkeypatch):
    base = _stub_chat(monkeypatch, answer="Ruhig bleiben, Hilfe kommt.")
    with client.websocket_connect("/ws/chat") as ws:
        ws.send_json({"text": "Ich bin gestürzt und komme nicht hoch", **base})
        event = ws.receive_json()
        assert event["type"] == "emergency" and event["category"] == "medical"
        assert ws.receive_json()["type"] == "ack"