"""Unit tests for navi.tools.memory."""
from unittest.mock import AsyncMock
import pytest
from navi.tools.memory import MemoryTool
from navi.tools._internal.base import ToolContext
@pytest.fixture
def fake_store():
store = AsyncMock()
store.upsert_fact = AsyncMock()
store.search_facts = AsyncMock(return_value=[])
store.delete_fact = AsyncMock(return_value=1)
store.get_categories = AsyncMock(return_value=[])
return store
@pytest.mark.asyncio
async def test_save(fake_store):
tool = MemoryTool(fake_store)
result = await tool.execute({
"action": "save",
"category": "technical",
"key": "primary_os",
"value": "Arch Linux",
"source": "tool_call",
"confidence": 95,
}, ctx=ToolContext(session_id="sess1", user_id="user1"))
assert result.success is True
assert "Saved [technical] primary_os" in result.output
fake_store.upsert_fact.assert_awaited_once()
@pytest.mark.asyncio
async def test_save_missing_category(fake_store):
tool = MemoryTool(fake_store)
result = await tool.execute({
"action": "save",
"key": "primary_os",
"value": "Arch Linux",
}, ctx=ToolContext(session_id="sess1", user_id="user1"))
assert result.success is False
assert "category" in result.output.lower()
@pytest.mark.asyncio
async def test_save_invalid_category(fake_store):
tool = MemoryTool(fake_store)
result = await tool.execute({
"action": "save",
"category": "invalid",
"key": "primary_os",
"value": "Arch Linux",
}, ctx=ToolContext(session_id="sess1", user_id="user1"))
assert result.success is False
assert "Invalid category" in result.output
@pytest.mark.asyncio
async def test_search_no_results(fake_store):
tool = MemoryTool(fake_store)
result = await tool.execute({
"action": "search",
"query": "nonexistent",
}, ctx=ToolContext(session_id="sess1", user_id="user1"))
assert result.success is True
assert "No matching facts" in result.output
@pytest.mark.asyncio
async def test_search_with_results(fake_store):
fake_store.search_facts = AsyncMock(return_value=[{
"category": "technical",
"key": "primary_os",
"value": "Arch Linux",
"source": "tool_call",
"confidence": 95,
"source_context": "found via uname",
}])
tool = MemoryTool(fake_store)
result = await tool.execute({
"action": "search",
"query": "os",
}, ctx=ToolContext(session_id="sess1", user_id="user1"))
assert result.success is True
assert "Arch Linux" in result.output
assert "technical" in result.output
@pytest.mark.asyncio
async def test_forget(fake_store):
tool = MemoryTool(fake_store)
result = await tool.execute({
"action": "forget",
"key": "primary_os",
}, ctx=ToolContext(session_id="sess1", user_id="user1"))
assert result.success is True
assert "Deleted 1 fact" in result.output
@pytest.mark.asyncio
async def test_forget_not_found(fake_store):
fake_store.delete_fact = AsyncMock(return_value=0)
tool = MemoryTool(fake_store)
result = await tool.execute({
"action": "forget",
"key": "missing",
}, ctx=ToolContext(session_id="sess1", user_id="user1"))
assert result.success is False
assert "No fact found" in result.output
@pytest.mark.asyncio
async def test_list_categories(fake_store):
fake_store.get_categories = AsyncMock(return_value=["technical", "preferences"])
tool = MemoryTool(fake_store)
result = await tool.execute({"action": "list"}, ctx=ToolContext(session_id="sess1", user_id="user1"))
assert result.success is True
assert "technical" in result.output
assert "preferences" in result.output
@pytest.mark.asyncio
async def test_list_empty(fake_store):
tool = MemoryTool(fake_store)
result = await tool.execute({"action": "list"}, ctx=ToolContext(session_id="sess1", user_id="user1"))
assert result.success is True
assert "empty" in result.output.lower()