#!/usr/bin/env python3
"""
MCP Tool Testing Framework (mock fixtures)
Run with: python test_runner.py
This script exercises the CKAN MCP server tooling using deterministic mock responses.
Update the helpers below if you want to point at a real CKAN endpoint.
"""
import asyncio
import json
import time
import os
import sys
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass
import logging
# Add src to path for imports
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
from ckan_mcp.ckan_tools import CkanToolsManager, create_tool_response
from ckan_mcp.types import CkanToolsConfig
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class TestCase:
"""Test case definition."""
name: str
tool: str
parameters: Dict[str, Any]
expected_checks: Dict[str, Any]
class MockMCPClient:
"""Mock MCP client for testing."""
def __init__(self, config: CkanToolsConfig):
"""Initialize mock client with configuration."""
self.config = config
self.manager = CkanToolsManager(config)
async def call_tool(self, tool_name: str, parameters: Dict[str, Any]) -> Any:
"""Mock tool call - returns realistic test data."""
logger.info(f"Calling tool: {tool_name} with parameters: {parameters}")
# Simulate network delay
await asyncio.sleep(0.1)
# Return mock responses based on tool name
if tool_name == "find_relevant_datasets":
max_results = parameters.get("maxResults", 20)
query = parameters.get("query", "")
return {
"query": query,
"total_found": 15,
"returned_count": min(5, max_results),
"datasets": [
{
"id": f"parking-dataset-{i}",
"name": f"parking-data-{i}",
"title": f"Parking Data {i}",
"description": f"Comprehensive parking information dataset {i}",
"organization": "City of Toronto",
"tags": ["parking", "transportation", "city"],
"relevance_score": 15 - i,
"update_frequency": "monthly",
"resource_count": 3,
"has_datastore": True,
"url": f"https://open.toronto.ca/dataset/parking-data-{i}"
}
for i in range(min(5, max_results))
],
"facets": {
"organization": [{"count": 10, "display_name": "City of Toronto"}],
"tags": [{"count": 8, "display_name": "parking"}]
}
}
elif tool_name == "analyze_dataset_updates":
return {
"total_datasets": 12,
"frequency_summary": [
{
"frequency": "monthly",
"count": 6,
"datasets": [
{"id": "traffic-flow", "title": "Traffic Flow Data"},
{"id": "parking-violations", "title": "Parking Violations"}
]
},
{
"frequency": "weekly",
"count": 3,
"datasets": [
{"id": "traffic-incidents", "title": "Traffic Incidents"}
]
},
{
"frequency": "daily",
"count": 2,
"datasets": [
{"id": "real-time-traffic", "title": "Real-time Traffic"}
]
},
{
"frequency": "quarterly",
"count": 1,
"datasets": [
{"id": "annual-traffic-report", "title": "Annual Traffic Report"}
]
}
],
"datasets": [
{
"id": "traffic-flow",
"name": "traffic-flow-data",
"title": "Traffic Flow Data",
"update_frequency": "monthly",
"last_modified": "2024-01-15T10:30:00Z",
"refresh_rate": "Monthly updates",
"organization": "Transportation Services",
"resource_count": 5,
"datastore_resources": 3
},
{
"id": "traffic-incidents",
"name": "traffic-incidents",
"title": "Traffic Incidents",
"update_frequency": "weekly",
"last_modified": "2024-01-20T14:22:00Z",
"refresh_rate": "Weekly updates",
"organization": "Transportation Services",
"resource_count": 2,
"datastore_resources": 2
}
]
}
elif tool_name == "analyze_dataset_structure":
package_id = parameters.get("packageId", "building-permits")
include_preview = parameters.get("includeDataPreview", False)
preview_limit = parameters.get("previewLimit", 5)
return {
"package_id": package_id,
"name": "building-permits",
"title": "Building Permits",
"description": "All building permits issued by the City of Toronto",
"tags": ["building", "permits", "construction", "development"],
"organization": "City Planning",
"created": "2020-01-01T00:00:00Z",
"last_modified": "2024-01-18T09:15:00Z",
"update_frequency": "daily",
"resource_summary": {
"total_resources": 4,
"datastore_resources": 2,
"formats": ["CSV", "GeoJSON", "PDF"],
"total_records": 150000
},
"resources": [
{
"id": "building-permits-csv",
"name": "Building Permits CSV",
"format": "CSV",
"size": 25000000,
"mimetype": "text/csv",
"url": "https://example.com/building-permits.csv",
"created": "2020-01-01T00:00:00Z",
"last_modified": "2024-01-18T09:15:00Z",
"datastore_active": True,
"fields": [
{"id": "permit_number", "type": "text", "info": {"label": "Permit Number"}},
{"id": "permit_type", "type": "text", "info": {"label": "Permit Type"}},
{"id": "issue_date", "type": "timestamp", "info": {"label": "Issue Date"}}
],
"record_count": 150000,
"sample_data": [
{
"permit_number": "20-123456",
"permit_type": "Building",
"issue_date": "2020-01-15T00:00:00Z"
}
] if include_preview else None
}
]
}
elif tool_name == "get_data_categories":
return {
"organizations": [
{
"id": "city-toronto",
"name": "city-toronto",
"title": "City of Toronto",
"description": "Municipal government data and services",
"package_count": 342
},
{
"id": "ttc",
"name": "toronto-transit-commission",
"title": "CKAN Transit Commission",
"description": "Public transit data and information",
"package_count": 87
},
{
"id": "tpb",
"name": "toronto-police-services",
"title": "CKAN Police Board",
"description": "Police services and public safety data",
"package_count": 45
}
],
"groups": [
{
"id": "transportation",
"name": "transportation",
"title": "Transportation",
"description": "All transportation-related datasets",
"package_count": 156
},
{
"id": "health",
"name": "health-and-wellness",
"title": "Health and Wellness",
"description": "Public health and wellness data",
"package_count": 78
},
{
"id": "environment",
"name": "environment",
"title": "Environment",
"description": "Environmental data and sustainability",
"package_count": 92
}
]
}
elif tool_name == "get_dataset_insights":
query = parameters.get("query", "")
max_datasets = parameters.get("maxDatasets", 20)
include_update_freq = parameters.get("includeUpdateFrequency", True)
include_structure = parameters.get("includeDataStructure", True)
return {
"query": query,
"total_found": 35,
"analyzed_datasets": min(3, max_datasets),
"insights": [
{
"id": "ttc-ridership",
"name": "ttc-ridership-data",
"title": "TTC Ridership Data",
"description": "Daily TTC ridership statistics by route and station",
"relevance_score": 18,
"organization": "CKAN Transit Commission",
"tags": ["ttc", "transit", "ridership", "transportation"],
"url": "https://open.toronto.ca/dataset/ttc-ridership-data",
**({"update_info": {
"frequency": "daily",
"last_modified": "2024-01-20T06:00:00Z",
"refresh_rate": "Daily updates"
}} if include_update_freq else {}),
**({"data_structure": {
"record_count": 50000,
"fields": [
{"id": "route_id", "type": "text"},
{"id": "station_id", "type": "text"},
{"id": "ridership_count", "type": "integer"},
{"id": "date", "type": "timestamp"}
],
"resource_count": 3,
"datastore_resources": 2
}} if include_structure else {})
},
{
"id": "road-maintenance",
"name": "road-maintenance-schedule",
"title": "Road Maintenance Schedule",
"description": "Scheduled road maintenance and construction activities",
"relevance_score": 15,
"organization": "Transportation Services",
"tags": ["roads", "maintenance", "construction", "transportation"],
"url": "https://open.toronto.ca/dataset/road-maintenance-schedule",
**({"update_info": {
"frequency": "weekly",
"last_modified": "2024-01-19T12:30:00Z",
"refresh_rate": "Weekly updates"
}} if include_update_freq else {}),
**({"data_structure": {
"record_count": 1200,
"fields": [
{"id": "work_id", "type": "text"},
{"id": "street_name", "type": "text"},
{"id": "work_type", "type": "text"},
{"id": "scheduled_date", "type": "timestamp"}
],
"resource_count": 2,
"datastore_resources": 2
}} if include_structure else {})
}
],
"query_suggestions": {
"organizations": [
"CKAN Transit Commission",
"Transportation Services",
"City of Toronto"
],
"common_tags": [
"transportation",
"data",
"city",
"services",
"public"
]
}
}
else:
# Default mock response for other tools
return {
"mock": True,
"tool": tool_name,
"parameters": parameters,
"message": f"Mock response for {tool_name}"
}
# Test cases matching the TypeScript version
TEST_CASES = [
TestCase(
name="Basic Dataset Search",
tool="find_relevant_datasets",
parameters={
"query": "parking",
"maxResults": 5,
"includeRelevanceScore": True
},
expected_checks={
"hasResults": True,
"minResults": 1,
"hasField": ["datasets", "total_found"],
"noErrors": True
}
),
TestCase(
name="Update Frequency Analysis",
tool="analyze_dataset_updates",
parameters={
"query": "traffic",
"groupByFrequency": True
},
expected_checks={
"hasResults": True,
"hasField": ["frequency_summary", "total_datasets"],
"customCheck": lambda result: len(result.get("frequency_summary", [])) > 0
}
),
TestCase(
name="Data Structure Analysis",
tool="analyze_dataset_structure",
parameters={
"packageId": "building-permits",
"includeDataPreview": False
},
expected_checks={
"hasResults": True,
"hasField": ["resources", "resource_summary"],
"customCheck": lambda result: len(result.get("resources", [])) > 0
}
),
TestCase(
name="Category Discovery",
tool="get_data_categories",
parameters={},
expected_checks={
"hasResults": True,
"hasField": ["organizations", "groups"],
"customCheck": lambda result: len(result.get("organizations", [])) > 0
}
),
TestCase(
name="Comprehensive Insights",
tool="get_dataset_insights",
parameters={
"query": "transportation",
"maxDatasets": 3,
"includeUpdateFrequency": True,
"includeDataStructure": True
},
expected_checks={
"hasResults": True,
"minResults": 1,
"hasField": ["insights", "query_suggestions"],
"customCheck": lambda result: len(result.get("insights", [])) > 0
}
),
]
async def run_test(client: MockMCPClient, test_case: TestCase) -> bool:
"""Run a single test case."""
print(f"\n🧪 Running test: {test_case.name}")
try:
result = await client.call_tool(test_case.tool, test_case.parameters)
print("✅ Tool executed successfully")
# Run validation checks
passed = True
checks = test_case.expected_checks
if checks.get("hasResults") and not result:
print("❌ Expected results but got none")
passed = False
if checks.get("noErrors") and isinstance(result, dict) and result.get("error"):
print(f"❌ Unexpected error: {result['error']}")
passed = False
if "hasField" in checks:
for field in checks["hasField"]:
if field not in result:
print(f"❌ Missing expected field: {field}")
passed = False
if "minResults" in checks:
datasets_count = len(result.get("datasets", []))
if datasets_count < checks["minResults"]:
print(f"❌ Expected at least {checks['minResults']} results, got {datasets_count}")
passed = False
if "maxResults" in checks:
datasets_count = len(result.get("datasets", []))
if datasets_count > checks["maxResults"]:
print(f"❌ Expected at most {checks['maxResults']} results, got {datasets_count}")
passed = False
custom_check = checks.get("customCheck")
if custom_check and not custom_check(result):
print("❌ Custom validation check failed")
passed = False
if passed:
print(f"✅ Test passed: {test_case.name}")
else:
print(f"❌ Test failed: {test_case.name}")
return passed
except Exception as e:
print(f"❌ Test failed with error: {e}")
return False
async def run_all_tests():
"""Run all test cases."""
print("🚀 Starting MCP Tool Testing")
print("ℹ️ Note: This is currently using a mock client that returns test data.")
print("ℹ️ To test against your actual MCP server, update the configuration.")
print("")
# Create test configuration
config = CkanToolsConfig(
ckan_base_url="https://ckan0.cf.opendata.inter.prod-toronto.ca/api/3/action"
)
client = MockMCPClient(config)
passed = 0
total = len(TEST_CASES)
for test_case in TEST_CASES:
result = await run_test(client, test_case)
if result:
passed += 1
print(f"\n📊 Test Results: {passed}/{total} tests passed")
print("\n💡 Next Steps:")
print("1. Set up environment variables:")
print(" export CKAN_BASE_URL='https://ckan0.cf.opendata.inter.prod-toronto.ca/api/3/action'")
print(" export CKAN_SITE_URL='https://ckan0.cf.opendata.inter.prod-toronto.ca'")
print("2. Run pytest for more comprehensive testing: python -m pytest tests/")
print("3. Start the MCP server: python -m ckan_mcp.main")
print("4. Connect to an AI assistant and run manual tests from documentation")
if passed == total:
print("🎉 All mock tests passed! Ready for real deployment testing.")
return 0
else:
print("⚠️ Some mock tests failed - this indicates test framework issues, not your MCP server.")
return 1
async def run_performance_tests(client: MockMCPClient):
"""Run performance tests."""
print("\n⚡ Running Performance Tests")
performance_tests = [
{
"name": "Large Query Response Time",
"tool": "find_relevant_datasets",
"parameters": {"query": "toronto", "maxResults": 50}
},
{
"name": "Complex Analysis Response Time",
"tool": "get_dataset_insights",
"parameters": {"query": "budget financial", "maxDatasets": 10}
}
]
for test in performance_tests:
start_time = time.time()
await client.call_tool(test["tool"], test["parameters"])
duration_ms = (time.time() - start_time) * 1000
print(f"{test['name']}: {duration_ms:.2f}ms")
if duration_ms > 5000: # 5 second threshold for mock
print(f"⚠️ Slow response: {test['name']} took {duration_ms:.2f}ms")
async def main():
"""Main entry point."""
# Create configuration and client
config = CkanToolsConfig(
ckan_base_url=os.getenv("CKAN_BASE_URL", "https://ckan0.cf.opendata.inter.prod-toronto.ca/api/3/action"),
ckan_site_url=os.getenv("CKAN_SITE_URL", "https://ckan0.cf.opendata.inter.prod-toronto.ca")
)
client = MockMCPClient(config)
# Run main tests
exit_code = await run_all_tests()
# Run performance tests
await run_performance_tests(client)
return exit_code
if __name__ == "__main__":
exit_code = asyncio.run(main())
sys.exit(exit_code)
