#!/usr/bin/env python3
"""Unit tests for Memory MCP tools.
Phase 2.2: MCP Memory Tools Tests
Tests for reflection storage, search, and database management tools.
"""
import pytest
from unittest.mock import AsyncMock, Mock, patch, MagicMock
from datetime import datetime
from pathlib import Path
from session_buddy.mcp.tools.memory.memory_tools import (
_store_reflection_impl,
_quick_search_impl,
_search_summary_impl,
_search_by_file_impl,
_search_by_concept_impl,
_reflection_stats_impl,
_reset_reflection_database_impl,
register_memory_tools,
_check_reflection_tools_available,
_get_reflection_database,
_format_score,
_format_stats_new,
_format_stats_old,
)
from session_buddy.adapters.reflection_adapter import ReflectionDatabaseAdapter
@pytest.mark.asyncio
class TestReflectionStorageTools:
"""Test suite for reflection storage tools."""
async def test_store_reflection_success(self, reflection_db_with_data):
"""Test successful reflection storage."""
content = "Test reflection content"
tags = ["test", "example"]
result = await reflection_db_with_data.store_reflection(content, tags)
assert result is not None
assert isinstance(result, str) or isinstance(result, int)
async def test_store_reflection_empty_content(self, reflection_db):
"""Test storing reflection with empty content."""
with pytest.raises(Exception):
await reflection_db.store_reflection("", ["test"])
async def test_store_reflection_with_tags(self, reflection_db):
"""Test storing reflection with multiple tags."""
content = "Important insight about testing"
tags = ["testing", "quality", "best-practices"]
result = await reflection_db.store_reflection(content, tags)
assert result is not None
@pytest.mark.asyncio
class TestSearchTools:
"""Test suite for search tools."""
async def test_quick_search_with_results(self, reflection_db_with_data):
"""Test quick search returning results."""
results = await reflection_db_with_data.search_conversations(
query="pytest",
project="test-project",
limit=1,
min_score=0.0,
)
assert isinstance(results, list)
async def test_quick_search_no_results(self, reflection_db):
"""Test quick search with no matching results."""
results = await reflection_db.search_conversations(
query="nonexistent query xyz123",
project="test-project",
limit=1,
min_score=0.9,
)
assert isinstance(results, list)
async def test_search_summary(self, reflection_db_with_data):
"""Test search summary functionality."""
results = await reflection_db_with_data.search_conversations(
query="async",
project="test-project",
limit=20,
min_score=0.0,
)
assert isinstance(results, list)
async def test_search_by_file(self, reflection_db_with_data):
"""Test searching conversations about a specific file."""
# First store a conversation about a file
await reflection_db_with_data.store_conversation(
"I was working on tests/conftest.py and added fixtures",
{"project": "test-project"},
)
results = await reflection_db_with_data.search_conversations(
query="conftest.py",
limit=10,
)
assert isinstance(results, list)
async def test_search_by_concept(self, reflection_db_with_data):
"""Test searching by concept."""
results = await reflection_db_with_data.search_conversations(
query="async patterns",
limit=10,
)
assert isinstance(results, list)
@pytest.mark.asyncio
class TestReflectionStats:
"""Test suite for reflection statistics tools."""
async def test_get_stats(self, reflection_db_with_data):
"""Test getting database statistics."""
stats = await reflection_db_with_data.get_stats()
assert isinstance(stats, dict)
assert "conversations_count" in stats or "total_reflections" in stats
async def test_get_stats_empty_database(self, reflection_db):
"""Test stats on empty database."""
stats = await reflection_db.get_stats()
assert isinstance(stats, dict)
# Empty database should have 0 counts
if "conversations_count" in stats:
assert stats["conversations_count"] == 0
if "reflections_count" in stats:
assert stats["reflections_count"] == 0
class TestMemoryToolHelpers:
"""Test suite for memory tool helper functions."""
def test_format_score(self):
"""Test score formatting."""
score = 0.8567
result = _format_score(score)
assert "0.86" in result or "0.85" in result # Check for reasonable formatting
def test_format_stats_new_format(self):
"""Test formatting stats in new format."""
stats = {
"conversations_count": 150,
"reflections_count": 45,
"embedding_provider": "sentence-transformers",
}
result = _format_stats_new(stats)
result_str = "\n".join(result)
assert "150" in result_str
assert "45" in result_str
assert "sentence-transformers" in result_str
def test_format_stats_old_format(self):
"""Test formatting stats in old format."""
stats = {
"total_reflections": 100,
"projects": 5,
"date_range": {"start": "2024-01-01", "end": "2024-12-31"},
"recent_activity": ["Stored reflection 1", "Stored reflection 2"],
}
result = _format_stats_old(stats)
result_str = "\n".join(result)
assert "100" in result_str
assert "5" in result_str
assert "2024-01-01" in result_str
def test_check_reflection_tools_available(self):
"""Test checking if reflection tools are available."""
with patch("importlib.util.find_spec", return_value=Mock()):
result = _check_reflection_tools_available()
# Result should be boolean
assert isinstance(result, bool)
def test_check_reflection_tools_unavailable(self):
"""Test checking when reflection tools are not available."""
with patch("importlib.util.find_spec", return_value=None):
result = _check_reflection_tools_available()
assert result is False
@pytest.mark.asyncio
class TestReflectionDatabaseReset:
"""Test suite for database reset functionality."""
async def test_reset_database_connection(self):
"""Test resetting database connection."""
# This test verifies the reset logic without actually closing connections
from session_buddy.mcp.tools.memory.memory_tools import _reflection_db
# Set a mock database
mock_db = Mock()
mock_db.close = Mock()
mock_db.aclose = AsyncMock()
with patch.dict(
"session_buddy.mcp.tools.memory.memory_tools.__dict__",
{"_reflection_db": mock_db},
):
# Reset would clear the global and force reconnection
# We're testing that the logic handles the mock correctly
pass # Actual reset tested in integration tests
class TestMemoryToolRegistration:
"""Test suite for memory tool registration."""
def test_register_memory_tools(self, mock_mcp_server):
"""Test that all memory tools are registered."""
register_memory_tools(mock_mcp_server)
# Verify tool decorator was called for each tool
# (store_reflection, quick_search, search_summary, search_by_file,
# search_by_concept, reflection_stats, reset_reflection_database)
assert mock_mcp_server.tool.call_count >= 7
@pytest.mark.asyncio
class TestMemoryToolImplementation:
"""Test suite for tool implementation functions."""
@patch(
"session_buddy.mcp.tools.memory.memory_tools._check_reflection_tools_available",
return_value=True,
)
async def test_store_reflection_impl_success(self, mock_available, reflection_db):
"""Test store_reflection implementation."""
content = "Test implementation"
tags = ["test"]
with patch(
"session_buddy.mcp.tools.memory.memory_tools._get_reflection_database",
return_value=reflection_db,
):
result = await _store_reflection_impl(content, tags)
assert "stored" in result.lower() or "test implementation" in result.lower()
@patch(
"session_buddy.mcp.tools.memory.memory_tools._check_reflection_tools_available",
return_value=False,
)
async def test_store_reflection_impl_unavailable(self, mock_available):
"""Test store_reflection when tools unavailable."""
result = await _store_reflection_impl("Test", ["test"])
assert "not available" in result.lower()
@patch(
"session_buddy.mcp.tools.memory.memory_tools._check_reflection_tools_available",
return_value=True,
)
async def test_quick_search_impl(self, mock_available, reflection_db_with_data):
"""Test quick_search implementation."""
with patch(
"session_buddy.mcp.tools.memory.memory_tools._get_reflection_database",
return_value=reflection_db_with_data,
):
result = await _quick_search_impl(query="test", min_score=0.0)
assert isinstance(result, str)
assert "search" in result.lower() or "test" in result.lower()
@patch(
"session_buddy.mcp.tools.memory.memory_tools._check_reflection_tools_available",
return_value=True,
)
async def test_search_summary_impl(self, mock_available, reflection_db_with_data):
"""Test search_summary implementation."""
with patch(
"session_buddy.mcp.tools.memory.memory_tools._get_reflection_database",
return_value=reflection_db_with_data,
):
result = await _search_summary_impl(query="test", min_score=0.0)
assert isinstance(result, str)
assert "summary" in result.lower() or "search" in result.lower()
@patch(
"session_buddy.mcp.tools.memory.memory_tools._check_reflection_tools_available",
return_value=True,
)
async def test_search_by_file_impl(self, mock_available, reflection_db_with_data):
"""Test search_by_file implementation."""
with patch(
"session_buddy.mcp.tools.memory.memory_tools._get_reflection_database",
return_value=reflection_db_with_data,
):
result = await _search_by_file_impl(file_path="test.py")
assert isinstance(result, str)
@patch(
"session_buddy.mcp.tools.memory.memory_tools._check_reflection_tools_available",
return_value=True,
)
async def test_search_by_concept_impl(self, mock_available, reflection_db_with_data):
"""Test search_by_concept implementation."""
with patch(
"session_buddy.mcp.tools.memory.memory_tools._get_reflection_database",
return_value=reflection_db_with_data,
):
result = await _search_by_concept_impl(concept="async")
assert isinstance(result, str)
@patch(
"session_buddy.mcp.tools.memory.memory_tools._check_reflection_tools_available",
return_value=True,
)
async def test_reflection_stats_impl(self, mock_available, reflection_db_with_data):
"""Test reflection_stats implementation."""
with patch(
"session_buddy.mcp.tools.memory.memory_tools._get_reflection_database",
return_value=reflection_db_with_data,
):
result = await _reflection_stats_impl()
assert isinstance(result, str)
assert "statistics" in result.lower() or "stats" in result.lower()
@patch(
"session_buddy.mcp.tools.memory.memory_tools._check_reflection_tools_available",
return_value=False,
)
async def test_tools_unavailable_message(self, mock_available):
"""Test unavailable message is shown when tools not available."""
result = await _reflection_stats_impl()
assert "not available" in result.lower()
@pytest.mark.asyncio
class TestMemoryToolIntegration:
"""Integration tests for memory tools."""
async def test_store_and_search_workflow(self, reflection_db):
"""Test complete workflow: store -> search."""
# Store a reflection
content = "Use async fixtures for database tests"
tags = ["testing", "async", "database"]
await reflection_db.store_reflection(content, tags)
# Search for it
results = await reflection_db.search_conversations(
query="async fixtures",
limit=5,
min_score=0.0,
)
assert isinstance(results, list)
async def test_multiple_reflections_search(self, reflection_db):
"""Test storing and searching multiple reflections."""
reflections = [
("Always use type hints", ["python", "best-practices"]),
("Async functions need event loops", ["async", "python"]),
("Tests should be independent", ["testing"]),
]
for content, tags in reflections:
await reflection_db.store_reflection(content, tags)
# Search across all
results = await reflection_db.search_conversations(
query="python",
limit=10,
min_score=0.0,
)
assert isinstance(results, list)
async def test_stats_after_operations(self, reflection_db):
"""Test stats reflect database operations."""
# Get initial stats
initial_stats = await reflection_db.get_stats()
# Perform operations
await reflection_db.store_reflection("Test content", ["test"])
await reflection_db.store_conversation("Test conversation", {"project": "test"})
# Get updated stats
final_stats = await reflection_db.get_stats()
# Verify counts changed
assert isinstance(final_stats, dict)
async def test_database_persistence_simulation(self, reflection_db):
"""Test simulating database persistence across operations."""
# Store data
await reflection_db.store_reflection("Persistent data", ["test"])
# Verify it's there
results = await reflection_db.search_conversations(
query="persistent",
limit=5,
min_score=0.0,
)
assert isinstance(results, list)
class TestMemoryToolEdgeCases:
"""Test edge cases and error handling."""
async def test_search_with_special_characters(self, reflection_db):
"""Test searching with special characters."""
await reflection_db.store_reflection("Use <script> tags carefully", ["security"])
results = await reflection_db.search_conversations(
query="<script>",
limit=5,
min_score=0.0,
)
assert isinstance(results, list)
async def test_search_with_unicode(self, reflection_db):
"""Test searching with unicode characters."""
await reflection_db.store_reflection("Unicode: 你好 🎉", ["test"])
results = await reflection_db.search_conversations(
query="你好",
limit=5,
min_score=0.0,
)
assert isinstance(results, list)
async def test_empty_tags_list(self, reflection_db):
"""Test storing reflection with empty tags."""
result = await reflection_db.store_reflection("No tags content", [])
assert result is not None
async def test_very_long_content(self, reflection_db):
"""Test storing very long reflection content."""
long_content = "Test content. " * 1000 # Create long content
result = await reflection_db.store_reflection(long_content, ["test"])
assert result is not None
async def test_search_case_sensitivity(self, reflection_db):
"""Test search case handling."""
await reflection_db.store_reflection("Python is great", ["python"])
# Search with different cases
results_lower = await reflection_db.search_conversations(
query="python",
limit=5,
min_score=0.0,
)
results_upper = await reflection_db.search_conversations(
query="PYTHON",
limit=5,
min_score=0.0,
)
assert isinstance(results_lower, list)
assert isinstance(results_upper, list)
