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
|