"""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 actually registered with the FastMCP app."""
@pytest.mark.asyncio
async def test_resources_registered_with_fastmcp(self, resource_handler, fastmcp_app):
"""Verify resources are registered by listing them from the FastMCP app."""
templates = await fastmcp_app.list_resource_templates()
template_uris = [t.uriTemplate for t in templates]
# The resource handler registers URI patterns during __init__
assert any("record" in uri for uri in template_uris), (
f"Expected a 'record' resource template, got: {template_uris}"
)
assert any("search" in uri for uri in template_uris), (
f"Expected a 'search' resource template, got: {template_uris}"
)
assert any("fields" in uri for uri in template_uris), (
f"Expected a 'fields' resource template, got: {template_uris}"
)
