"""
Comprehensive MCP integration tests for the Smithsonian MCP server.
This module tests end-to-end MCP workflows, protocol adherence, and cross-tool dependencies.
"""
# TODO: DEBUG THIS TEST. STILL PASSING, BUT WE NEED TO FIGURE OUT THE 21 ISSUES.
import pytest
import asyncio
from datetime import datetime
from unittest.mock import AsyncMock, patch, MagicMock
from typing import List, Dict, Any, Optional
from smithsonian_mcp.models import (
SmithsonianObject,
SearchResult,
CollectionSearchFilter,
SmithsonianUnit,
CollectionStats,
APIError,
)
from smithsonian_mcp.api_client import SmithsonianAPIClient
pytest.importorskip("pytest_asyncio")
class TestMCPProtocolAdherence:
"""Test MCP protocol adherence and tool interface compliance."""
@pytest.mark.asyncio
async def test_tool_signature_compliance(self):
"""Test that MCP tools have proper signatures by checking the server module."""
from smithsonian_mcp import tools as tools_module
from smithsonian_mcp import resources as resources_module
import inspect
# Get the actual function implementations (the decorated tools are directly callable)
tools_to_test = [
("search_collections", tools_module.search_collections),
("simple_search", tools_module.simple_search),
("find_and_describe", tools_module.find_and_describe),
("search_and_get_first_details", tools_module.search_and_get_first_details),
("search_and_get_details", tools_module.search_and_get_details),
("simple_explore", tools_module.simple_explore),
("continue_explore", tools_module.continue_explore),
("summarize_search_results", tools_module.summarize_search_results),
("get_object_ids", tools_module.get_object_ids),
("get_first_object_id", tools_module.get_first_object_id),
("validate_object_id", tools_module.validate_object_id),
("resolve_museum_name", tools_module.resolve_museum_name),
("get_object_details", tools_module.get_object_details),
("get_object_url", tools_module.get_object_url),
("get_smithsonian_units", tools_module.get_smithsonian_units),
("get_collection_statistics", tools_module.get_collection_statistics),
("search_by_unit", tools_module.search_by_unit),
("get_objects_on_view", tools_module.get_objects_on_view),
("get_museum_highlights_on_view", tools_module.get_museum_highlights_on_view),
("get_museum_collection_types", tools_module.get_museum_collection_types),
("check_museum_has_object_type", tools_module.check_museum_has_object_type),
("get_search_context", resources_module.get_search_context),
("get_object_context", resources_module.get_object_context),
("get_on_view_context", resources_module.get_on_view_context),
("get_units_context", resources_module.get_units_context),
("get_stats_context", resources_module.get_stats_context),
]
for tool_name, tool_func in tools_to_test:
# Get the actual function from the FunctionTool wrapper
actual_func = tool_func.fn
# Verify tool has proper signature
sig = inspect.signature(actual_func)
assert "ctx" in sig.parameters, f"Tool {tool_name} missing ctx parameter"
assert (
sig.parameters["ctx"].default is None
), f"Tool {tool_name} ctx parameter should default to None"
@pytest.mark.asyncio
async def test_tool_return_types(self):
"""Test that tools return expected types."""
mock_client_instance = AsyncMock(spec=SmithsonianAPIClient)
with patch("smithsonian_mcp.context._global_api_client", None):
with patch("smithsonian_mcp.context.create_client") as mock_create_client:
mock_create_client.return_value = mock_client_instance
from smithsonian_mcp import tools as tools_module
# Test search_collections returns SearchResult
mock_client_instance.search_collections.return_value = SearchResult(
objects=[],
total_count=0,
returned_count=0,
offset=0,
has_more=False,
next_offset=None,
)
result = await tools_module.search_collections.fn(query="test")
assert isinstance(
result, SearchResult
), "search_collections should return SearchResult"
# Test get_smithsonian_units returns list
mock_client_instance.get_units.return_value = [
SmithsonianUnit(code="NMAH", name="American History Museum"),
SmithsonianUnit(code="NMNH", name="Natural History Museum"),
]
result = await tools_module.get_smithsonian_units.fn()
assert isinstance(
result, list
), "get_smithsonian_units should return list"
# Test get_collection_statistics returns CollectionStats
mock_client_instance.get_collection_stats.return_value = (
CollectionStats(
total_objects=1000,
total_digitized=500,
total_cc0=200,
total_with_images=400,
last_updated=datetime(2024, 1, 1),
units=[],
)
)
result = await tools_module.get_collection_statistics.fn()
assert isinstance(
result, CollectionStats
), "get_collection_statistics should return CollectionStats"
class TestEndToEndMCPWorkflows:
"""Test complete user interaction patterns and workflows."""
@pytest.mark.asyncio
async def test_full_discovery_workflow(self):
"""Test complete discovery workflow: explore -> search -> details -> context."""
mock_client_instance = AsyncMock(spec=SmithsonianAPIClient)
with patch("smithsonian_mcp.context._global_api_client", None):
with patch("smithsonian_mcp.context.create_client") as mock_create_client:
mock_create_client.return_value = mock_client_instance
from smithsonian_mcp import (
tools as tools_module,
resources as resources_module,
)
# Step 1: Simple exploration
mock_client_instance.search_collections.return_value = SearchResult(
objects=[
SmithsonianObject(
id="test-123",
title="Test Object",
unit_code="NMNH",
unit_name="Natural History Museum",
is_on_view=True,
object_type="fossil",
)
],
total_count=1,
returned_count=1,
offset=0,
has_more=False,
next_offset=None,
)
explore_result = await tools_module.simple_explore.fn(topic="dinosaurs")
assert explore_result.objects
object_id = explore_result.objects[0].id
# Step 2: Get detailed object information
mock_client_instance.get_object_by_id.return_value = SmithsonianObject(
id=object_id,
title="Detailed Test Object",
unit_code="NMNH",
description="A detailed description",
images=[{"url": "http://example.com/image.jpg"}],
)
detail_result = await tools_module.get_object_details.fn(
object_id=object_id
)
assert detail_result.id == object_id
# Step 3: Get object context
context_result = await resources_module.get_object_context.fn(
object_id=object_id
)
assert "Detailed Test Object" in context_result
assert "Images: 1 available" in context_result
@pytest.mark.asyncio
async def test_exhibition_planning_workflow(self):
"""Test exhibition planning workflow: units -> on-view -> search -> context."""
mock_client_instance = AsyncMock(spec=SmithsonianAPIClient)
with patch("smithsonian_mcp.context._global_api_client", None):
with patch("smithsonian_mcp.context.create_client") as mock_create_client:
mock_create_client.return_value = mock_client_instance
from smithsonian_mcp import (
tools as tools_module,
resources as resources_module,
)
# Step 1: Get available museums
mock_client_instance.get_units.return_value = [
SmithsonianUnit(code="NMAH", name="American History Museum"),
SmithsonianUnit(code="NMNH", name="Natural History Museum"),
]
units = await tools_module.get_smithsonian_units.fn()
assert len(units) >= 2
# Step 2: Find on-view items at specific museum
mock_client_instance.search_collections.return_value = SearchResult(
objects=[
SmithsonianObject(
id="exhibit-1",
title="Exhibition Item 1",
unit_code="NMAH",
is_on_view=True,
exhibition_title="American Innovation",
)
],
total_count=1,
returned_count=1,
offset=0,
has_more=False,
next_offset=None,
)
on_view_result = await tools_module.get_objects_on_view.fn(
unit_code="NMAH"
)
assert on_view_result.objects
assert all(obj.is_on_view for obj in on_view_result.objects)
# Step 3: Get on-view context
context_result = await resources_module.get_on_view_context.fn(
museum="NMAH"
)
assert "Currently on exhibit" in context_result
@pytest.mark.asyncio
async def test_research_workflow_with_pagination(self):
"""Test research workflow with pagination: search -> continue -> details."""
mock_client_instance = AsyncMock(spec=SmithsonianAPIClient)
with patch("smithsonian_mcp.context._global_api_client", None):
with patch("smithsonian_mcp.context.create_client") as mock_create_client:
mock_create_client.return_value = mock_client_instance
from smithsonian_mcp import tools as tools_module
# Initial search
seen_ids = []
# Mock first batch of results
mock_client_instance.search_collections.return_value = SearchResult(
objects=[
SmithsonianObject(
id=f"obj-{i}", title=f"Object {i}", unit_code="NMNH"
)
for i in range(5)
],
total_count=20,
returned_count=5,
offset=0,
has_more=True,
next_offset=5,
)
first_result = await tools_module.simple_explore.fn(
topic="fossils", max_samples=5
)
assert len(first_result.objects) == 5
seen_ids.extend([obj.id for obj in first_result.objects])
# Continue exploration with deduplication
mock_client_instance.search_collections.return_value = SearchResult(
objects=[
SmithsonianObject(
id=f"obj-{i+10}", title=f"Object {i+10}", unit_code="NMNH"
)
for i in range(5)
],
total_count=20,
returned_count=5,
offset=5,
has_more=True,
next_offset=10,
)
continue_result = await tools_module.continue_explore.fn(
topic="fossils", previously_seen_ids=seen_ids, max_samples=5
)
# Verify no duplicates
new_ids = [obj.id for obj in continue_result.objects]
assert not any(
obj_id in seen_ids for obj_id in new_ids
), "Continue explore should not return duplicate objects"
class TestCrossToolDependencies:
"""Test interactions and dependencies between different MCP tools."""
@pytest.mark.asyncio
async def test_search_to_details_dependency(self):
"""Test that search results can be used with get_object_details."""
mock_client_instance = AsyncMock(spec=SmithsonianAPIClient)
with patch("smithsonian_mcp.context._global_api_client", None):
with patch("smithsonian_mcp.context.create_client") as mock_create_client:
mock_create_client.return_value = mock_client_instance
from smithsonian_mcp import tools as tools_module
# Search returns objects
mock_client_instance.search_collections.return_value = SearchResult(
objects=[
SmithsonianObject(
id="interconnected-123",
title="Search Result Object",
unit_code="NMAH",
)
],
total_count=1,
returned_count=1,
offset=0,
has_more=False,
next_offset=None,
)
search_result = await tools_module.search_collections.fn(
query="technology"
)
assert search_result.objects
# Get details for search result object
mock_client_instance.get_object_by_id.return_value = SmithsonianObject(
id="interconnected-123",
title="Detailed Search Result Object",
unit_code="NMAH",
description="A detailed description",
images=[{"url": "http://example.com/image.jpg"}],
)
detail_result = await tools_module.get_object_details.fn(
object_id="interconnected-123"
)
assert detail_result.id == "interconnected-123"
assert "detailed description" in detail_result.description
@pytest.mark.asyncio
async def test_unit_search_integration(self):
"""Test integration between get_smithsonian_units and search_by_unit."""
mock_client_instance = AsyncMock(spec=SmithsonianAPIClient)
with patch("smithsonian_mcp.context._global_api_client", None):
with patch("smithsonian_mcp.context.create_client") as mock_create_client:
mock_create_client.return_value = mock_client_instance
from smithsonian_mcp import tools as tools_module
# Get units first
mock_client_instance.get_units.return_value = [
SmithsonianUnit(
code="SAAM", name="Smithsonian American Art Museum"
),
SmithsonianUnit(code="NPG", name="National Portrait Gallery"),
]
units = await tools_module.get_smithsonian_units.fn()
saam_unit = next(unit for unit in units if unit.code == "SAAM")
# Use unit code for targeted search
mock_client_instance.search_collections.return_value = SearchResult(
objects=[
SmithsonianObject(
id="art-123",
title="American Art Piece",
unit_code="SAAM",
unit_name=saam_unit.name,
)
],
total_count=1,
returned_count=1,
offset=0,
has_more=False,
next_offset=None,
)
unit_search_result = await tools_module.search_by_unit.fn(
unit_code="SAAM", query="painting"
)
assert unit_search_result.objects
assert all(
obj.unit_code == "SAAM" for obj in unit_search_result.objects
)
@pytest.mark.asyncio
async def test_on_view_search_reliability(self):
"""Test that find_on_view_items provides more reliable results than basic on_view filter."""
mock_client_instance = AsyncMock(spec=SmithsonianAPIClient)
with patch("smithsonian_mcp.context._global_api_client", None):
with patch("smithsonian_mcp.context.create_client") as mock_create_client:
mock_create_client.return_value = mock_client_instance
from smithsonian_mcp import tools as tools_module
# Mock reliable local filtering results
mock_reliable_objects = [
SmithsonianObject(
id="reliable-1", title="Reliable On-View Item", is_on_view=True
),
SmithsonianObject(
id="reliable-2", title="Another On-View Item", is_on_view=True
),
]
mock_client_instance.search_collections.return_value = SearchResult(
objects=mock_reliable_objects,
total_count=2,
returned_count=2,
offset=0,
has_more=False,
next_offset=None,
)
# Test find_on_view_items (reliable approach)
reliable_result = await tools_module.find_on_view_items.fn(
query="exhibition"
)
assert all(obj.is_on_view for obj in reliable_result.objects)
assert len(reliable_result.objects) == 2
class TestErrorHandlingAndEdgeCases:
"""Test error handling and edge cases in MCP workflows."""
@pytest.mark.asyncio
async def test_empty_search_results_handling(self):
"""Test handling of empty search results across tools."""
mock_client_instance = AsyncMock(spec=SmithsonianAPIClient)
with patch("smithsonian_mcp.context._global_api_client", None):
with patch("smithsonian_mcp.context.create_client") as mock_create_client:
mock_create_client.return_value = mock_client_instance
from smithsonian_mcp import tools as tools_module
# Mock empty results
empty_result = SearchResult(
objects=[],
total_count=0,
returned_count=0,
offset=0,
has_more=False,
next_offset=None,
)
mock_client_instance.search_collections.return_value = empty_result
# Test various tools with empty results
search_result = await tools_module.search_collections.fn(
query="nonexistenttopic12345"
)
assert search_result.objects == []
assert search_result.total_count == 0
explore_result = await tools_module.simple_explore.fn(
topic="nonexistenttopic12345"
)
assert explore_result.objects == []
on_view_result = await tools_module.get_objects_on_view.fn()
assert on_view_result.objects == []
# Test get_museum_highlights_on_view with empty results
highlights_result = await tools_module.get_museum_highlights_on_view.fn()
assert highlights_result.objects == []
@pytest.mark.asyncio
async def test_invalid_object_id_handling(self):
"""Test handling of invalid object IDs."""
mock_client_instance = AsyncMock(spec=SmithsonianAPIClient)
with patch("smithsonian_mcp.context._global_api_client", None):
with patch("smithsonian_mcp.context.create_client") as mock_create_client:
mock_create_client.return_value = mock_client_instance
from smithsonian_mcp import (
tools as tools_module,
resources as resources_module,
)
# Mock not found result
mock_client_instance.get_object_by_id.return_value = None
result = await tools_module.get_object_details.fn(
object_id="invalid-id-12345"
)
assert result is None
context_result = await resources_module.get_object_context.fn(
object_id="invalid-id-12345"
)
assert "not found" in context_result.lower()
@pytest.mark.asyncio
async def test_rate_limiting_simulation(self):
"""Test behavior under simulated rate limiting conditions."""
mock_client_instance = AsyncMock(spec=SmithsonianAPIClient)
with patch("smithsonian_mcp.context._global_api_client", None):
with patch("smithsonian_mcp.context.create_client") as mock_create_client:
mock_create_client.return_value = mock_client_instance
from smithsonian_mcp import tools as tools_module
# Simulate rate limit error
mock_client_instance.search_collections.side_effect = Exception(
"Rate limit exceeded"
)
with pytest.raises(Exception) as exc_info:
await tools_module.search_collections.fn(query="test")
assert "Rate limit exceeded" in str(exc_info.value)
class TestMCPContextTools:
"""Test context-specific tools that provide formatted data."""
@pytest.mark.asyncio
async def test_context_tools_formatting(self):
"""Test that context tools return properly formatted strings."""
mock_client_instance = AsyncMock(spec=SmithsonianAPIClient)
with patch("smithsonian_mcp.context._global_api_client", None):
with patch("smithsonian_mcp.context.create_client") as mock_create_client:
mock_create_client.return_value = mock_client_instance
from smithsonian_mcp import resources as resources_module
# Mock search context data
mock_search_result = SearchResult(
objects=[
SmithsonianObject(
id="ctx-123",
title="Context Test Object",
unit_name="Test Museum",
unit_code="TEST",
)
],
total_count=1,
returned_count=1,
offset=0,
has_more=False,
next_offset=None,
)
mock_client_instance.search_collections.return_value = (
mock_search_result
)
# Test search context formatting
search_context = await resources_module.get_search_context.fn(
query="test"
)
assert "Search Results for 'test'" in search_context
assert "Context Test Object" in search_context
assert "Test Museum" in search_context
# Test units context formatting
mock_units = [
SmithsonianUnit(
code="TEST", name="Test Museum", description="A test museum"
),
]
mock_client_instance.get_units.return_value = mock_units
units_context = await resources_module.get_units_context.fn()
assert "Smithsonian Institution Museums" in units_context
@pytest.mark.asyncio
async def test_on_view_context_filtering(self):
"""Test that on_view_context properly filters and formats on-view objects."""
mock_client_instance = AsyncMock(spec=SmithsonianAPIClient)
with patch("smithsonian_mcp.context._global_api_client", None):
with patch("smithsonian_mcp.context.create_client") as mock_create_client:
mock_create_client.return_value = mock_client_instance
from smithsonian_mcp import resources as resources_module
# Mix of on-view and not-on-view objects
mock_mixed_objects = [
SmithsonianObject(
id="on-view-1",
title="On View Object",
unit_name="Test Museum",
is_on_view=True,
object_type="Painting",
),
SmithsonianObject(
id="storage-1",
title="In Storage Object",
unit_name="Test Museum",
is_on_view=False,
),
]
mock_result = SearchResult(
objects=mock_mixed_objects,
total_count=2,
returned_count=2,
offset=0,
has_more=False,
next_offset=None,
)
mock_client_instance.search_collections.return_value = mock_result
on_view_context = await resources_module.get_on_view_context.fn()
# Should only include on-view objects
assert "On View Object" in on_view_context
assert "In Storage Object" not in on_view_context
assert "Currently on exhibit" in on_view_context
assert "Type: Painting" in on_view_context
if __name__ == "__main__":
pytest.main([__file__, "-v"])
