1eb79c1f09
- Quota (app/quota.py): Anfragen pro Nutzer/Tag (usage-Tabelle), DAILY_REQUEST_LIMIT, pro Nutzer via prefs uebersteuerbar; 429 (REST) bzw. error-Event (WS); Metrik - Notfall (app/safety/emergency.py): heuristische Erkennung (de/en); Log im Store (emergency_events) + optionaler Webhook (best-effort) + X-Emergency/emergency-Event; Metrik emergency_total; Notfaelle umgehen das Kontingent - verdrahtet in chat/speak/transcribe + WS-Turns - Tests: 64 gruen (+6); Doku aktualisiert (README, BEDIENUNGSANLEITUNG, Architektur, .env.example) Hinweis: Notfall-Erkennung ist eine Heuristik (kein Lebensretter); erkannte Texte sind sensibel -> DSGVO beachten. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
81 lines
3 KiB
Python
81 lines
3 KiB
Python
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):
|
|
return answer
|
|
|
|
class StubTTS:
|
|
async def synthesize(self, text, voice=None, audio_format="pcm"):
|
|
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"
|