531b57e08d
#3b Langzeit-Erinnerungen: - Store: memories-Tabelle + add/get/delete_memory (pro Nutzer) - API: GET/POST/DELETE /api/me/memories - chat.py injiziert Nutzer-Erinnerungen als System-Kontext ins LLM (sessionunabhaengig) - ?debug zeigt memories_len #4 Echtzeit (erster Increment): - WS /ws/chat: dauerhafter Kanal, Event-Folge ack -> semantic -> audio (binaer) -> done - Auth (Token-Query), Session-Gedaechtnis und Erinnerungen wie bei POST /api/chat - Fehler als error-Event (422/403/502) - Tests: 38 gruen (Erinnerungs-CRUD/Injektion, WebSocket-Streaming/Auth) - Doku aktualisiert (README, BEDIENUNGSANLEITUNG, Architektur) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
import app.dependencies as deps
|
|
from app.main import app
|
|
from app.config import settings
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def _install_stubs(monkeypatch, captured):
|
|
class MemLLM:
|
|
async def complete(self, text, history=None, session_id=None):
|
|
captured["history"] = list(history or [])
|
|
return f"Echo: {text}"
|
|
|
|
class StubTTS:
|
|
async def synthesize(self, text, voice=None, audio_format="pcm"):
|
|
return b"WSAUDIO"
|
|
|
|
monkeypatch.setitem(deps.LLM_REGISTRY, "mem", lambda s: MemLLM())
|
|
monkeypatch.setitem(deps.TTS_REGISTRY, "stub", lambda s: StubTTS())
|
|
return {"llm_provider": "mem", "tts_provider": "stub", "output_endpoint": "loopback"}
|
|
|
|
|
|
def _drain_turn(ws):
|
|
assert ws.receive_json()["type"] == "ack"
|
|
sem = ws.receive_json()
|
|
assert ws.receive_bytes() == b"WSAUDIO"
|
|
assert ws.receive_json()["type"] == "done"
|
|
return sem
|
|
|
|
|
|
def test_ws_streams_events_and_remembers(monkeypatch):
|
|
captured = {}
|
|
base = _install_stubs(monkeypatch, captured)
|
|
|
|
with client.websocket_connect("/ws/chat?session_id=wsconv") as ws:
|
|
ws.send_json({"text": "Hallo", **base})
|
|
sem1 = _drain_turn(ws)
|
|
assert sem1["type"] == "semantic" and sem1["text"] == "Echo: Hallo"
|
|
|
|
ws.send_json({"text": "Weiter", **base})
|
|
_drain_turn(ws)
|
|
|
|
# Zweiter Turn hat den Verlauf des ersten erhalten.
|
|
assert captured["history"] == [
|
|
{"role": "user", "content": "Hallo"},
|
|
{"role": "assistant", "content": "Echo: Hallo"},
|
|
]
|
|
|
|
|
|
def test_ws_unknown_provider_sends_error_event(monkeypatch):
|
|
captured = {}
|
|
_install_stubs(monkeypatch, captured)
|
|
with client.websocket_connect("/ws/chat") as ws:
|
|
ws.send_json({"text": "x", "llm_provider": "gibtsnicht"})
|
|
err = ws.receive_json()
|
|
assert err["type"] == "error" and err["status"] == 422
|
|
|
|
|
|
def test_ws_requires_token_when_auth_enabled(monkeypatch):
|
|
monkeypatch.setattr(settings, "auth_enabled", True)
|
|
monkeypatch.setattr(settings, "admin_api_key", "k")
|
|
with pytest.raises(WebSocketDisconnect):
|
|
with client.websocket_connect("/ws/chat"):
|
|
pass
|