Add memory_manager source files and tests
Some checks failed
CI / test (push) Failing after 15s
CI / build (push) Has been skipped

This commit is contained in:
2026-03-22 16:19:05 +00:00
parent adb8cb48bb
commit 3f36745745

View File

@@ -0,0 +1,97 @@
import pytest
from unittest.mock import AsyncMock, MagicMock
from memory_manager.core.services import MemoryService, SearchService, CommitService
from memory_manager.db.models import MemoryCategory
@pytest.fixture
def mock_repository():
return AsyncMock()
@pytest.fixture
def memory_service(mock_repository):
return MemoryService(mock_repository)
class TestMemoryService:
@pytest.mark.asyncio
async def test_create_entry(self, memory_service, mock_repository):
mock_entry = MagicMock()
mock_entry.to_dict.return_value = {
"id": 1,
"title": "Test Entry",
"content": "Test content",
"category": "decision",
"tags": ["test"],
"agent_id": "test-agent",
"project_path": "/test",
"created_at": "2024-01-01T00:00:00",
"updated_at": "2024-01-01T00:00:00",
}
mock_repository.create_entry = AsyncMock(return_value=mock_entry)
result = await memory_service.create_entry(
title="Test Entry",
content="Test content",
category=MemoryCategory.DECISION,
tags=["test"],
agent_id="test-agent",
project_path="/test",
)
assert result["title"] == "Test Entry"
assert result["id"] == 1
mock_repository.create_entry.assert_called_once()
@pytest.mark.asyncio
async def test_get_entry(self, memory_service, mock_repository):
mock_entry = MagicMock()
mock_entry.to_dict.return_value = {"id": 1, "title": "Test"}
mock_repository.get_entry = AsyncMock(return_value=mock_entry)
result = await memory_service.get_entry(1)
assert result["id"] == 1
mock_repository.get_entry.assert_called_once_with(1)
@pytest.mark.asyncio
async def test_get_entry_not_found(self, memory_service, mock_repository):
mock_repository.get_entry = AsyncMock(return_value=None)
result = await memory_service.get_entry(999)
assert result is None
@pytest.mark.asyncio
async def test_update_entry(self, memory_service, mock_repository):
mock_entry = MagicMock()
mock_entry.to_dict.return_value = {"id": 1, "title": "Updated"}
mock_repository.update_entry = AsyncMock(return_value=mock_entry)
result = await memory_service.update_entry(entry_id=1, title="Updated")
assert result["title"] == "Updated"
mock_repository.update_entry.assert_called_once()
@pytest.mark.asyncio
async def test_delete_entry(self, memory_service, mock_repository):
mock_repository.delete_entry = AsyncMock(return_value=True)
result = await memory_service.delete_entry(1)
assert result is True
mock_repository.delete_entry.assert_called_once_with(1)
@pytest.mark.asyncio
async def test_list_entries(self, memory_service, mock_repository):
mock_entries = [
MagicMock(to_dict=lambda: {"id": 1, "title": "Entry 1"}),
MagicMock(to_dict=lambda: {"id": 2, "title": "Entry 2"}),
]
mock_repository.list_entries = AsyncMock(return_value=mock_entries)
result = await memory_service.list_entries()
assert len(result) == 2
assert result[0]["id"] == 1