jarvis-assist/Claude outputs/tests/test_memory.py

import os
import sys
from unittest.mock import AsyncMock, patch, MagicMock

# Mock psycopg2 before importing main to allow tests to run in environments
# where psycopg2-binary cannot be installed (e.g., local Python 3.14).
# In the documented Docker test environment (JARVIS_HANDOFF.md), psycopg2-binary
# is available, so this mocking is not needed there. Since pg_pool is only
# used at runtime in on_startup(), not during module import, this mocking
# does not affect test correctness.
sys.modules["psycopg2"] = MagicMock()
sys.modules["psycopg2.pool"] = MagicMock()
sys.modules["psycopg2.extras"] = MagicMock()

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

import pytest

import main


@pytest.mark.asyncio
async def test_insert_memory_fact_returns_new_id():
    with patch.object(main, "db_query", new=AsyncMock(return_value={"id": 42})) as mock_query:
        fact_id = await main.insert_memory_fact(1, "Mag lieber Tee als Kaffee")

    assert fact_id == 42
    args, kwargs = mock_query.call_args
    assert "INSERT INTO memory_facts" in args[0]
    assert args[1] == (1, "Mag lieber Tee als Kaffee")


@pytest.mark.asyncio
async def test_search_memory_facts_uses_ilike():
    rows = [{"id": 1, "content": "Hund heisst Bruno"}]
    with patch.object(main, "db_query", new=AsyncMock(return_value=rows)) as mock_query:
        result = await main.search_memory_facts("Bruno")

    assert result == rows
    args, kwargs = mock_query.call_args
    assert "ILIKE" in args[0]
    assert args[1] == ("%Bruno%",)


@pytest.mark.asyncio
async def test_delete_memory_fact_deletes_by_id():
    with patch.object(main, "db_query", new=AsyncMock(return_value=None)) as mock_query:
        await main.delete_memory_fact(7)

    args, kwargs = mock_query.call_args
    assert "DELETE FROM memory_facts" in args[0]
    assert args[1] == (7,)


@pytest.mark.asyncio
async def test_get_all_memory_facts_returns_rows():
    rows = [{"id": 1, "content": "A"}, {"id": 2, "content": "B"}]
    with patch.object(main, "db_query", new=AsyncMock(return_value=rows)):
        result = await main.get_all_memory_facts()

    assert result == rows


@pytest.mark.asyncio
async def test_upsert_conversation_summary_inserts_with_vector_literal():
    with patch.object(main, "db_query", new=AsyncMock(return_value=None)) as mock_query:
        await main.upsert_conversation_summary(5, "Kurze Zusammenfassung", [0.1, 0.2])

    args, kwargs = mock_query.call_args
    assert "INSERT INTO conversation_summaries" in args[0]
    assert "ON CONFLICT (conversation_id) DO UPDATE" in args[0]
    assert args[1] == (5, "Kurze Zusammenfassung", "[0.1,0.2]")


@pytest.mark.asyncio
async def test_search_similar_conversation_summaries_excludes_current_conversation():
    rows = [{"conversation_id": 3, "summary": "Ueber Kalender", "distance": 0.1}]
    with patch.object(main, "db_query", new=AsyncMock(return_value=rows)) as mock_query:
        result = await main.search_similar_conversation_summaries([0.1, 0.2], exclude_conversation_id=9, limit=3)

    assert result == rows
    args, kwargs = mock_query.call_args
    assert "conversation_id != %s" in args[0]
    assert args[1] == ("[0.1,0.2]", 9, 3)


@pytest.mark.asyncio
async def test_remember_fact_inserts_and_returns_content():
    main.DEFAULT_USER_ID = 1
    with patch.object(main, "insert_memory_fact", new=AsyncMock(return_value=99)):
        result = await main.remember_fact("Hund heisst Bruno")

    assert result == {"id": 99, "content": "Hund heisst Bruno"}


@pytest.mark.asyncio
async def test_forget_fact_no_match_returns_message():
    with patch.object(main, "search_memory_facts", new=AsyncMock(return_value=[])):
        result = await main.forget_fact("Katze")

    assert result["deleted"] is False
    assert result["matches"] == []


@pytest.mark.asyncio
async def test_forget_fact_single_match_deletes():
    matches = [{"id": 5, "content": "Hund heisst Bruno"}]
    with patch.object(main, "search_memory_facts", new=AsyncMock(return_value=matches)), patch.object(
        main, "delete_memory_fact", new=AsyncMock()
    ) as mock_delete:
        result = await main.forget_fact("Bruno")

    mock_delete.assert_called_once_with(5)
    assert result == {"deleted": True, "content": "Hund heisst Bruno"}


@pytest.mark.asyncio
async def test_forget_fact_multiple_matches_does_not_delete():
    matches = [
        {"id": 5, "content": "Hund heisst Bruno"},
        {"id": 6, "content": "Bruno ist der Nachbar"},
    ]
    with patch.object(main, "search_memory_facts", new=AsyncMock(return_value=matches)), patch.object(
        main, "delete_memory_fact", new=AsyncMock()
    ) as mock_delete:
        result = await main.forget_fact("Bruno")

    mock_delete.assert_not_called()
    assert result["deleted"] is False
    assert len(result["matches"]) == 2


@pytest.mark.asyncio
async def test_execute_tool_dispatches_remember_and_forget_fact():
    import json

    with patch.object(main, "remember_fact", new=AsyncMock(return_value={"id": 1, "content": "X"})):
        result = await main.execute_tool("remember_fact", {"fact": "X"})
    assert json.loads(result) == {"id": 1, "content": "X"}

    with patch.object(main, "forget_fact", new=AsyncMock(return_value={"deleted": True, "content": "X"})):
        result = await main.execute_tool("forget_fact", {"query": "X"})
    assert json.loads(result) == {"deleted": True, "content": "X"}


@pytest.mark.asyncio
async def test_build_memory_context_empty_when_nothing_stored():
    with patch.object(main, "get_all_memory_facts", new=AsyncMock(return_value=[])), patch.object(
        main, "get_embedding", new=AsyncMock(return_value=[0.1])
    ), patch.object(main, "search_similar_conversation_summaries", new=AsyncMock(return_value=[])):
        context = await main.build_memory_context("Hallo", conversation_id=1)

    assert context == ""


@pytest.mark.asyncio
async def test_build_memory_context_includes_facts_section():
    facts = [{"id": 1, "content": "Hund heisst Bruno"}]
    with patch.object(main, "get_all_memory_facts", new=AsyncMock(return_value=facts)), patch.object(
        main, "get_embedding", new=AsyncMock(return_value=[0.1])
    ), patch.object(main, "search_similar_conversation_summaries", new=AsyncMock(return_value=[])):
        context = await main.build_memory_context("Wie geht es meinem Hund?", conversation_id=1)

    assert "Bekannte Fakten ueber den Nutzer" in context
    assert "Hund heisst Bruno" in context
    assert "Relevante fruehere Gespraeche" not in context


@pytest.mark.asyncio
async def test_build_memory_context_includes_summaries_section():
    summaries = [{"conversation_id": 2, "summary": "Ging um den Kalender", "distance": 0.05}]
    with patch.object(main, "get_all_memory_facts", new=AsyncMock(return_value=[])), patch.object(
        main, "get_embedding", new=AsyncMock(return_value=[0.1])
    ), patch.object(main, "search_similar_conversation_summaries", new=AsyncMock(return_value=summaries)):
        context = await main.build_memory_context("Was war nochmal mit dem Termin?", conversation_id=1)

    assert "Relevante fruehere Gespraeche" in context
    assert "Ging um den Kalender" in context


@pytest.mark.asyncio
async def test_build_memory_context_skips_summaries_when_embedding_fails():
    with patch.object(main, "get_all_memory_facts", new=AsyncMock(return_value=[])), patch.object(
        main, "get_embedding", new=AsyncMock(side_effect=RuntimeError("Ollama down"))
    ):
        context = await main.build_memory_context("Hallo", conversation_id=1)

    assert context == ""