my_voice_assistant_v3/tests/test_memory.py

from fastapi.testclient import TestClient

import app.dependencies as deps
from app.main import app

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"Antwort auf: {text}"

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

    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 test_history_accumulates_and_flows_to_llm(monkeypatch):
    captured = {}
    base = _install_stubs(monkeypatch, captured)

    # Turn 1: noch kein Verlauf.
    r1 = client.post("/api/chat?debug=true&session_id=conv1", json={"text": "Hallo", **base})
    assert r1.status_code == 200
    assert r1.json()["history_len"] == 0
    assert captured["history"] == []

    # Turn 2: Verlauf enthaelt User+Assistant aus Turn 1.
    r2 = client.post("/api/chat?debug=true&session_id=conv1", json={"text": "Und weiter?", **base})
    assert r2.status_code == 200
    assert r2.json()["history_len"] == 2
    assert captured["history"] == [
        {"role": "user", "content": "Hallo"},
        {"role": "assistant", "content": "Antwort auf: Hallo"},
    ]


def test_no_session_id_is_stateless(monkeypatch):
    captured = {}
    base = _install_stubs(monkeypatch, captured)

    client.post("/api/chat?debug=true", json={"text": "A", **base})
    client.post("/api/chat?debug=true", json={"text": "B", **base})
    assert captured["history"] == []  # ohne session_id kein Gedaechtnis


def test_memories_injected_as_context(monkeypatch):
    captured = {}
    base = _install_stubs(monkeypatch, captured)

    # Erinnerung fuer den (anonymen) Nutzer ablegen.
    deps.get_store().ensure_anonymous_user()
    deps.get_store().add_memory("anonymous", "heisst Anna")

    # Ohne session_id -> kein Verlauf, aber Erinnerung wird trotzdem injiziert.
    r = client.post("/api/chat?debug=true", json={"text": "Hallo", **base})
    assert r.status_code == 200
    assert r.json()["memories_len"] == 1
    assert captured["history"][0]["role"] == "system"
    assert "heisst Anna" in captured["history"][0]["content"]


def test_history_limited_by_setting(monkeypatch):
    from app.config import settings

    captured = {}
    base = _install_stubs(monkeypatch, captured)
    monkeypatch.setattr(settings, "history_max_messages", 2)

    for text in ("eins", "zwei", "drei"):
        client.post("/api/chat?debug=true&session_id=limit", json={"text": text, **base})

    # Vor dem 3. Turn liegen 4 Nachrichten vor; geladen werden nur die letzten 2.
    assert len(captured["history"]) == 2
    assert captured["history"][-1] == {"role": "assistant", "content": "Antwort auf: zwei"}