62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
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
|