68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
"""Unit tests for SearchService."""
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from memory_manager.core.services import SearchService
|
|
from memory_manager.db.models import MemoryCategory
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_repository():
|
|
return AsyncMock()
|
|
|
|
|
|
@pytest.fixture
|
|
def search_service(mock_repository):
|
|
return SearchService(mock_repository)
|
|
|
|
|
|
class TestSearchService:
|
|
@pytest.mark.asyncio
|
|
async def test_search_basic(self, search_service, mock_repository):
|
|
mock_entries = [
|
|
MagicMock(to_dict=lambda: {"id": 1, "title": "Test Entry", "content": "Test content"}),
|
|
]
|
|
mock_repository.search_entries = AsyncMock(return_value=mock_entries)
|
|
|
|
result = await search_service.search(query="test")
|
|
|
|
assert len(result) == 1
|
|
assert result[0]["title"] == "Test Entry"
|
|
mock_repository.search_entries.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_with_category(self, search_service, mock_repository):
|
|
mock_repository.search_entries = AsyncMock(return_value=[])
|
|
|
|
await search_service.search(
|
|
query="test",
|
|
category=MemoryCategory.DECISION,
|
|
)
|
|
|
|
call_args = mock_repository.search_entries.call_args
|
|
assert call_args.kwargs["category"] == MemoryCategory.DECISION
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_with_filters(self, search_service, mock_repository):
|
|
mock_repository.search_entries = AsyncMock(return_value=[])
|
|
|
|
await search_service.search(
|
|
query="test",
|
|
agent_id="test-agent",
|
|
project_path="/test",
|
|
limit=50,
|
|
)
|
|
|
|
call_args = mock_repository.search_entries.call_args
|
|
assert call_args.kwargs["agent_id"] == "test-agent"
|
|
assert call_args.kwargs["project_path"] == "/test"
|
|
assert call_args.kwargs["limit"] == 50
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_empty_results(self, search_service, mock_repository):
|
|
mock_repository.search_entries = AsyncMock(return_value=[])
|
|
|
|
result = await search_service.search(query="nonexistent")
|
|
|
|
assert len(result) == 0
|