Newer
Older
navi-1 / tests / unit / tools / test_memory.py
"""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 current_session_id, current_user_id


@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.fixture(autouse=True)
def _fake_ctx():
    token = current_session_id.set("sess1")
    token2 = current_user_id.set("user1")
    yield
    current_session_id.reset(token)
    current_user_id.reset(token2)


@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,
    })
    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",
    })
    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",
    })
    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",
    })
    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",
    })
    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",
    })
    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",
    })
    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"})
    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"})
    assert result.success is True
    assert "empty" in result.output.lower()