MCP Server for Odoo

"""Tests for resource URI query parameter handling. This test file specifically tests the fix for issue #4 where resource URIs with query parameters were failing with "Unknown operation" errors. """ import json from unittest.mock import Mock from urllib.parse import quote import pytest from mcp.server.fastmcp import FastMCP from mcp_server_odoo.access_control import AccessController from mcp_server_odoo.config import OdooConfig from mcp_server_odoo.odoo_connection import OdooConnection from mcp_server_odoo.resources import OdooResourceHandler @pytest.fixture def mock_config(): """Create a mock configuration.""" config = Mock(spec=OdooConfig) config.default_limit = 10 config.max_limit = 100 return config @pytest.fixture def mock_connection(): """Create a mock Odoo connection.""" conn = Mock(spec=OdooConnection) conn.is_authenticated = True return conn @pytest.fixture def mock_access_controller(): """Create a mock access controller.""" controller = Mock(spec=AccessController) controller.validate_model_access.return_value = None return controller @pytest.fixture def fastmcp_app(): """Create a real FastMCP app instance.""" return FastMCP(name="test-odoo-mcp") @pytest.fixture def resource_handler(fastmcp_app, mock_connection, mock_access_controller, mock_config): """Create a resource handler instance with real FastMCP app.""" return OdooResourceHandler(fastmcp_app, mock_connection, mock_access_controller, mock_config) class TestResourceQueryParameterHandling: """Test that resource URIs with various query parameter combinations work correctly.""" @pytest.mark.asyncio async def test_search_with_limit_only(self, resource_handler, mock_connection): """Test search resource with only limit parameter (issue #4 case).""" # Setup mocks mock_connection.search_count.return_value = 10 mock_connection.search.return_value = [1, 2] mock_connection.read.return_value = [ {"id": 1, "name": "Record 1"}, {"id": 2, "name": "Record 2"}, ] mock_connection.fields_get.return_value = {} # This should work now with the fix result = await resource_handler._handle_search("res.partner", None, None, 2, None, None) # Verify the search was called with correct limit mock_connection.search.assert_called_once_with( "res.partner", [], limit=2, offset=0, order=None ) # Verify result contains the records assert "Record 1" in result assert "Record 2" in result assert "Showing records 1-2 of 10" in result @pytest.mark.asyncio async def test_search_with_domain_only(self, resource_handler, mock_connection): """Test search resource with only domain parameter.""" # Setup domain domain = [["is_company", "=", True]] domain_encoded = quote(json.dumps(domain)) # Setup mocks mock_connection.search_count.return_value = 3 mock_connection.search.return_value = [1, 2, 3] mock_connection.read.return_value = [ {"id": 1, "name": "Company A"}, {"id": 2, "name": "Company B"}, {"id": 3, "name": "Company C"}, ] mock_connection.fields_get.return_value = {} # Test search with domain only result = await resource_handler._handle_search( "res.partner", domain_encoded, None, None, None, None ) # Verify domain was parsed and used mock_connection.search_count.assert_called_once_with("res.partner", domain) mock_connection.search.assert_called_once_with( "res.partner", domain, limit=10, offset=0, order=None ) assert "Company A" in result assert "is_company = True" in result @pytest.mark.asyncio async def test_search_with_fields_only(self, resource_handler, mock_connection): """Test search resource with only fields parameter.""" fields = "name,email" # Setup mocks mock_connection.search_count.return_value = 1 mock_connection.search.return_value = [1] mock_connection.read.return_value = [ {"id": 1, "name": "Test Partner", "email": "test@example.com"} ] mock_connection.fields_get.return_value = {} # Test search with fields only result = await resource_handler._handle_search( "res.partner", None, fields, None, None, None ) # Verify fields were parsed and used mock_connection.read.assert_called_once_with("res.partner", [1], ["name", "email"]) assert "Fields: name, email" in result assert "Test Partner" in result assert "test@example.com" in result @pytest.mark.asyncio async def test_search_with_pagination_only(self, resource_handler, mock_connection): """Test search resource with limit and offset parameters.""" # Setup mocks mock_connection.search_count.return_value = 100 mock_connection.search.return_value = [21, 22, 23, 24, 25] mock_connection.read.return_value = [ {"id": i, "name": f"Record {i}"} for i in range(21, 26) ] mock_connection.fields_get.return_value = {} # Test with pagination result = await resource_handler._handle_search("res.partner", None, None, 5, 20, None) # Verify pagination mock_connection.search.assert_called_once_with( "res.partner", [], limit=5, offset=20, order=None ) assert "Page 5 of 20" in result # offset 20, limit 5 = page 5 assert "Showing records 21-25 of 100" in result assert "Record 21" in result assert "Record 25" in result @pytest.mark.asyncio async def test_search_with_domain_and_limit(self, resource_handler, mock_connection): """Test search resource with domain and limit parameters.""" domain = [["active", "=", True]] domain_encoded = quote(json.dumps(domain)) # Setup mocks mock_connection.search_count.return_value = 50 mock_connection.search.return_value = [1, 2, 3] mock_connection.read.return_value = [ {"id": i, "name": f"Active Record {i}"} for i in range(1, 4) ] mock_connection.fields_get.return_value = {} # Test with domain and limit result = await resource_handler._handle_search( "res.partner", domain_encoded, None, 3, None, None ) # Verify both domain and limit were used mock_connection.search_count.assert_called_once_with("res.partner", domain) mock_connection.search.assert_called_once_with( "res.partner", domain, limit=3, offset=0, order=None ) assert "active = True" in result assert "Active Record 1" in result assert "Showing records 1-3 of 50" in result # Browse test removed - browse resource not supported due to FastMCP query parameter limitations # Use get_record multiple times or search_records tool instead @pytest.mark.asyncio async def test_count_without_domain(self, resource_handler, mock_connection): """Test count resource without domain parameter.""" # Setup mocks mock_connection.search_count.return_value = 150 # Test count without domain result = await resource_handler._handle_count("res.partner", None) # Verify count was called with empty domain mock_connection.search_count.assert_called_once_with("res.partner", []) assert "Total count: 150 record(s)" in result assert "Search criteria: All records" in result @pytest.mark.asyncio async def test_count_with_domain(self, resource_handler, mock_connection): """Test count resource with domain parameter.""" domain = [["customer_rank", ">", 0]] domain_encoded = quote(json.dumps(domain)) # Setup mocks mock_connection.search_count.return_value = 75 # Test count with domain result = await resource_handler._handle_count("res.partner", domain_encoded) # Verify count was called with parsed domain mock_connection.search_count.assert_called_once_with("res.partner", domain) assert "Total count: 75 record(s)" in result assert "customer_rank > 0" in result class TestResourceRegistration: """Test that resources are properly registered with FastMCP.""" def test_search_resources_registered(self, resource_handler, fastmcp_app): """Verify that multiple search resource patterns are registered.""" # Get registered handlers from the app # Note: FastMCP doesn't expose registered resources directly, # but we can verify they work by checking the handler exists # Due to FastMCP limitations, only these patterns are registered: # - odoo://{model}/search (no parameters) # - odoo://{model}/record/{record_id} # - odoo://{model}/count (no parameters) # - odoo://{model}/fields # Since we can't directly inspect FastMCP's internal registry, # we verify that the handler methods exist assert hasattr(resource_handler, "_handle_search") # Browse resource removed due to FastMCP limitations assert hasattr(resource_handler, "_handle_count") assert hasattr(resource_handler, "_handle_fields") assert hasattr(resource_handler, "_handle_record_retrieval") def test_count_resources_registered(self, resource_handler, fastmcp_app): """Verify that count resource patterns are registered.""" # Count only supports basic pattern due to FastMCP limitations: # - odoo://{model}/count (no parameters) assert hasattr(resource_handler, "_handle_count")