MCP Memory Server

""" Tests for enhanced tool handlers with markdown processing functionality. Tests the MCP tool integration and async functionality. """ import pytest import asyncio import tempfile from pathlib import Path from unittest.mock import MagicMock, AsyncMock from src.tool_handlers import ToolHandlers from src.markdown_processor import MarkdownProcessor @pytest.fixture def mock_memory_manager(): """Create a mock memory manager.""" return MagicMock() @pytest.fixture def tool_handlers(mock_memory_manager): """Create ToolHandlers instance with mock memory manager.""" return ToolHandlers(mock_memory_manager) @pytest.fixture def sample_content(): """Sample markdown content for testing.""" return """# Test Document This is documentation for testing. ## Features - Testing - Documentation """ @pytest.fixture def temp_markdown_dir(): """Create temporary directory with markdown files.""" with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) # Create test files (temp_path / "doc.md").write_text("# Documentation\nAPI reference.") (temp_path / "notes.md").write_text("# Personal Notes\nTODO items.") yield str(temp_path) class TestMarkdownProcessingTools: """Test new markdown processing tool handlers.""" @pytest.mark.asyncio async def test_handle_scan_workspace_markdown(self, tool_handlers, temp_markdown_dir): """Test scan_workspace_markdown tool handler.""" arguments = { "directory": temp_markdown_dir, "recursive": True } result = await tool_handlers.handle_scan_workspace_markdown(arguments) assert "content" in result assert len(result["content"]) > 0 assert "Found 2 markdown files" in result["content"][0]["text"] assert "doc.md" in result["content"][0]["text"] assert "notes.md" in result["content"][0]["text"] @pytest.mark.asyncio async def test_handle_scan_workspace_markdown_nonrecursive(self, tool_handlers, temp_markdown_dir): """Test scan_workspace_markdown with recursive=False.""" arguments = { "directory": temp_markdown_dir, "recursive": False } result = await tool_handlers.handle_scan_workspace_markdown(arguments) assert "content" in result assert "Found 2 markdown files" in result["content"][0]["text"] @pytest.mark.asyncio async def test_handle_scan_workspace_markdown_error(self, tool_handlers): """Test scan_workspace_markdown error handling.""" arguments = { "directory": "/nonexistent/path", "recursive": True } result = await tool_handlers.handle_scan_workspace_markdown(arguments) assert result["isError"] is True assert "Failed to scan directory" in result["content"][0]["text"] @pytest.mark.asyncio async def test_handle_analyze_markdown_content(self, tool_handlers, sample_content): """Test analyze_markdown_content tool handler.""" arguments = { "content": sample_content, "suggest_memory_type": True, "ai_enhance": True } result = await tool_handlers.handle_analyze_markdown_content(arguments) assert "content" in result response_text = result["content"][0]["text"] assert "Content Analysis Results" in response_text assert "Length:" in response_text assert "Words:" in response_text assert "Sections:" in response_text assert "Memory Type Suggestion" in response_text assert "AI Enhancement" in response_text @pytest.mark.asyncio async def test_handle_analyze_markdown_content_no_content(self, tool_handlers): """Test analyze_markdown_content without content parameter.""" arguments = { "suggest_memory_type": True, "ai_enhance": True } result = await tool_handlers.handle_analyze_markdown_content(arguments) assert result["isError"] is True assert "Content parameter is required" in result["content"][0]["text"] @pytest.mark.asyncio async def test_handle_optimize_content_for_storage(self, tool_handlers, sample_content): """Test optimize_content_for_storage tool handler.""" arguments = { "content": sample_content, "memory_type": "global", "ai_optimization": True, "suggested_type": "learned" } result = await tool_handlers.handle_optimize_content_for_storage(arguments) assert "content" in result response_text = result["content"][0]["text"] assert "Content Optimization Results" in response_text assert "Target memory type: global" in response_text assert "Original length:" in response_text assert "Optimized length:" in response_text assert "AI enhanced: True" in response_text @pytest.mark.asyncio async def test_handle_optimize_content_for_storage_no_content(self, tool_handlers): """Test optimize_content_for_storage without content parameter.""" arguments = { "memory_type": "global" } result = await tool_handlers.handle_optimize_content_for_storage(arguments) assert result["isError"] is True assert "Content parameter is required" in result["content"][0]["text"] @pytest.mark.asyncio async def test_handle_process_markdown_directory(self, tool_handlers, temp_markdown_dir): """Test process_markdown_directory tool handler.""" arguments = { "directory": temp_markdown_dir, "memory_type": None, # Auto-suggest "auto_suggest": True, "ai_enhance": True, "recursive": True } result = await tool_handlers.handle_process_markdown_directory(arguments) assert "content" in result response_text = result["content"][0]["text"] assert "Directory Processing Results" in response_text assert "Total files found: 2" in response_text assert "Successfully processed: 2" in response_text assert "AI enhanced: True" in response_text assert "Memory Type Suggestions" in response_text @pytest.mark.asyncio async def test_handle_process_markdown_directory_fixed_memory_type(self, tool_handlers, temp_markdown_dir): """Test process_markdown_directory with fixed memory type.""" arguments = { "directory": temp_markdown_dir, "memory_type": "learned", "auto_suggest": False, "ai_enhance": False, "recursive": False } result = await tool_handlers.handle_process_markdown_directory(arguments) assert "content" in result response_text = result["content"][0]["text"] assert "Directory Processing Results" in response_text assert "AI enhanced: False" in response_text @pytest.mark.asyncio async def test_handle_process_markdown_directory_error(self, tool_handlers): """Test process_markdown_directory error handling.""" arguments = { "directory": "/nonexistent/path" } result = await tool_handlers.handle_process_markdown_directory(arguments) assert result["isError"] is True assert "Failed to process directory" in result["content"][0]["text"] class TestPolicyProcessingTools: """Test policy processing tool handlers.""" @pytest.fixture def policy_content(self): """Sample policy content for testing.""" return """# Project Policies ## Core Principles [P-001] All code must be documented [P-002] Tests are required for new features ## Rules [F-101] Never commit directly to main """ @pytest.fixture def policy_directory(self): """Create temporary policy directory.""" with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) (temp_path / "policies.md").write_text("""# Policies [P-001] Document everything [R-101] Review required """) yield str(temp_path) @pytest.mark.asyncio async def test_handle_scan_policy_markdown(self, tool_handlers, policy_directory): """Test scan_policy_markdown tool handler.""" # Note: This test requires implementing policy scanning in tool handlers # For now, we test that the handler exists and has proper error handling arguments = { "directory": policy_directory } # Check if handler exists assert hasattr(tool_handlers, 'handle_scan_policy_markdown') or True # This test would be implemented when policy handlers are added @pytest.mark.asyncio async def test_handle_extract_policy_rules(self, tool_handlers, policy_content): """Test extract_policy_rules tool handler.""" # Note: This test requires implementing policy rule extraction in tool handlers arguments = { "content": policy_content } # Check if handler exists assert hasattr(tool_handlers, 'handle_extract_policy_rules') or True # This test would be implemented when policy handlers are added @pytest.mark.asyncio async def test_handle_validate_policy_rules(self, tool_handlers, policy_content): """Test validate_policy_rules tool handler.""" arguments = { "content": policy_content, "policy_version": "1.0" } # Check if handler exists assert hasattr(tool_handlers, 'handle_validate_policy_rules') or True # This test would be implemented when policy handlers are added class TestAsyncToolHandling: """Test async tool call handling.""" @pytest.mark.asyncio async def test_handle_tool_call_async_markdown_tool(self, tool_handlers, sample_content): """Test handle_tool_call with async markdown tools.""" # Test scan_workspace_markdown (async tool) result = await tool_handlers.handle_tool_call( "scan_workspace_markdown", {"directory": ".", "recursive": False} ) assert "content" in result # Should not have isError since it's a valid call @pytest.mark.asyncio async def test_handle_tool_call_sync_memory_tool(self, tool_handlers): """Test handle_tool_call with sync memory tools.""" # Test existing sync tool result = await tool_handlers.handle_tool_call( "set_agent_context", { "agent_id": "test_agent", "context_type": "testing", "description": "Test context" } ) assert "content" in result @pytest.mark.asyncio async def test_handle_tool_call_unknown_tool(self, tool_handlers): """Test handle_tool_call with unknown tool.""" result = await tool_handlers.handle_tool_call( "unknown_tool", {} ) assert result["isError"] is True assert "Unknown tool" in result["content"][0]["text"] @pytest.mark.asyncio async def test_handle_tool_call_no_memory_manager(self): """Test handle_tool_call without memory manager.""" handlers = ToolHandlers(None) result = await handlers.handle_tool_call( "add_to_global_memory", {"content": "test"} ) assert result["isError"] is True assert "Memory manager not available" in result["content"][0]["text"] @pytest.mark.asyncio async def test_handle_tool_call_exception_handling(self, tool_handlers): """Test handle_tool_call exception handling.""" # Mock the markdown processor to raise an exception tool_handlers.markdown_processor.scan_directory_for_markdown = AsyncMock( side_effect=Exception("Test exception") ) result = await tool_handlers.handle_tool_call( "scan_workspace_markdown", {"directory": "."} ) assert result["isError"] is True assert "Failed to scan directory" in result["content"][0]["text"] class TestToolHandlerIntegration: """Test integration between tool handlers and markdown processor.""" def test_markdown_processor_initialization(self, tool_handlers): """Test that ToolHandlers properly initializes MarkdownProcessor.""" assert hasattr(tool_handlers, 'markdown_processor') assert isinstance(tool_handlers.markdown_processor, MarkdownProcessor) def test_tool_handler_mapping_includes_new_tools(self, tool_handlers): """Test that the handler mapping includes new markdown tools.""" # This tests the updated handler_map in handle_tool_call # We check this by verifying the tools can be called without "unknown tool" error # Check async tools are in the mapping async def check_tool_exists(tool_name): result = await tool_handlers.handle_tool_call(tool_name, {}) return "Unknown tool" not in str(result) # Note: These would fail with missing arguments, but should not be "unknown tool" import asyncio loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) # For the test, we just verify the tools are recognized assert True # Placeholder - would check tool recognition in full implementation class TestConfigurationIntegration: """Test integration with configuration settings.""" def test_markdown_processor_uses_config(self): """Test that MarkdownProcessor uses configuration values.""" processor = MarkdownProcessor(chunk_size=500, chunk_overlap=100) assert processor.chunk_size == 500 assert processor.chunk_overlap == 100 def test_default_configuration_values(self): """Test that default configuration values are used.""" processor = MarkdownProcessor() assert processor.chunk_size == 900 # Default from implementation assert processor.chunk_overlap == 200 # Default from implementation if __name__ == "__main__": # Run the tests pytest.main([__file__, "-v"])