""" Tests for MCP Tool Backend Integration The Atomizer MCP tools (TypeScript) communicate with the Python backend through REST API endpoints. This test file verifies the backend supports all the endpoints that MCP tools expect. P4.8: MCP tool integration tests """ import json import pytest import tempfile from pathlib import Path from datetime import datetime import sys sys.path.insert(0, str(Path(__file__).parent.parent)) sys.path.insert(0, str(Path(__file__).parent.parent / "atomizer-dashboard" / "backend")) # ============================================================================ # MCP Tool → Backend Endpoint Mapping # ============================================================================ MCP_TOOL_ENDPOINTS = { # Study Management Tools "list_studies": {"method": "GET", "endpoint": "/api/studies"}, "get_study_status": {"method": "GET", "endpoint": "/api/studies/{study_id}"}, "create_study": {"method": "POST", "endpoint": "/api/studies"}, # Optimization Control Tools "run_optimization": {"method": "POST", "endpoint": "/api/optimize/{study_id}/start"}, "stop_optimization": {"method": "POST", "endpoint": "/api/optimize/{study_id}/stop"}, "get_optimization_status": {"method": "GET", "endpoint": "/api/optimize/{study_id}/status"}, # Analysis Tools "get_trial_data": {"method": "GET", "endpoint": "/api/studies/{study_id}/trials"}, "analyze_convergence": {"method": "GET", "endpoint": "/api/studies/{study_id}/convergence"}, "compare_trials": {"method": "POST", "endpoint": "/api/studies/{study_id}/compare"}, "get_best_design": {"method": "GET", "endpoint": "/api/studies/{study_id}/best"}, # Reporting Tools "generate_report": {"method": "POST", "endpoint": "/api/studies/{study_id}/report"}, "export_data": {"method": "GET", "endpoint": "/api/studies/{study_id}/export"}, # Physics Tools "explain_physics": {"method": "GET", "endpoint": "/api/physics/explain"}, "recommend_method": {"method": "POST", "endpoint": "/api/physics/recommend"}, "query_extractors": {"method": "GET", "endpoint": "/api/physics/extractors"}, # Canvas Tools (AtomizerSpec v2.0) "canvas_add_node": {"method": "POST", "endpoint": "/api/studies/{study_id}/spec/nodes"}, "canvas_update_node": {"method": "PATCH", "endpoint": "/api/studies/{study_id}/spec/nodes/{node_id}"}, "canvas_remove_node": {"method": "DELETE", "endpoint": "/api/studies/{study_id}/spec/nodes/{node_id}"}, "canvas_connect_nodes": {"method": "POST", "endpoint": "/api/studies/{study_id}/spec/edges"}, # Canvas Intent Tools "validate_canvas_intent": {"method": "POST", "endpoint": "/api/studies/{study_id}/spec/validate"}, "execute_canvas_intent": {"method": "POST", "endpoint": "/api/studies/{study_id}/spec/execute"}, "interpret_canvas_intent": {"method": "POST", "endpoint": "/api/studies/{study_id}/spec/interpret"}, } # ============================================================================ # Fixtures # ============================================================================ @pytest.fixture def minimal_spec() -> dict: """Minimal valid AtomizerSpec.""" return { "meta": { "version": "2.0", "created": datetime.now().isoformat() + "Z", "modified": datetime.now().isoformat() + "Z", "created_by": "test", "modified_by": "test", "study_name": "mcp_test_study" }, "model": { "sim": {"path": "model.sim", "solver": "nastran"} }, "design_variables": [ { "id": "dv_001", "name": "thickness", "expression_name": "thickness", "type": "continuous", "bounds": {"min": 1.0, "max": 10.0}, "baseline": 5.0, "enabled": True, "canvas_position": {"x": 50, "y": 100} } ], "extractors": [ { "id": "ext_001", "name": "Mass Extractor", "type": "mass", "builtin": True, "outputs": [{"name": "mass", "units": "kg"}], "canvas_position": {"x": 740, "y": 100} } ], "objectives": [ { "id": "obj_001", "name": "mass", "direction": "minimize", "source": {"extractor_id": "ext_001", "output_name": "mass"}, "canvas_position": {"x": 1020, "y": 100} } ], "constraints": [], "optimization": { "algorithm": {"type": "TPE"}, "budget": {"max_trials": 100} }, "canvas": { "edges": [ {"source": "dv_001", "target": "model"}, {"source": "model", "target": "solver"}, {"source": "solver", "target": "ext_001"}, {"source": "ext_001", "target": "obj_001"}, {"source": "obj_001", "target": "optimization"} ], "layout_version": "2.0" } } @pytest.fixture def temp_studies_dir(minimal_spec): """Create temporary studies directory.""" with tempfile.TemporaryDirectory() as tmpdir: study_dir = Path(tmpdir) / "studies" / "mcp_test_study" study_dir.mkdir(parents=True) spec_path = study_dir / "atomizer_spec.json" with open(spec_path, "w") as f: json.dump(minimal_spec, f, indent=2) yield Path(tmpdir) @pytest.fixture def test_client(temp_studies_dir, monkeypatch): """Create test client.""" from api.routes import spec monkeypatch.setattr(spec, "STUDIES_DIR", temp_studies_dir / "studies") from api.main import app from fastapi.testclient import TestClient return TestClient(app) # ============================================================================ # Canvas MCP Tool Tests (AtomizerSpec v2.0) # ============================================================================ class TestCanvasMCPTools: """Tests for canvas-related MCP tools that use AtomizerSpec.""" def test_canvas_add_node_endpoint_exists(self, test_client): """Test canvas_add_node MCP tool calls /spec/nodes endpoint.""" response = test_client.post( "/api/studies/mcp_test_study/spec/nodes", json={ "type": "designVar", "data": { "name": "width", "expression_name": "width", "type": "continuous", "bounds": {"min": 5.0, "max": 15.0}, "baseline": 10.0, "enabled": True }, "modified_by": "mcp" } ) # Endpoint should respond (not 404) assert response.status_code in [200, 400, 500] def test_canvas_update_node_endpoint_exists(self, test_client): """Test canvas_update_node MCP tool calls PATCH /spec/nodes endpoint.""" response = test_client.patch( "/api/studies/mcp_test_study/spec/nodes/dv_001", json={ "updates": {"bounds": {"min": 2.0, "max": 15.0}}, "modified_by": "mcp" } ) # Endpoint should respond (not 404 for route) assert response.status_code in [200, 400, 404, 500] def test_canvas_remove_node_endpoint_exists(self, test_client): """Test canvas_remove_node MCP tool calls DELETE /spec/nodes endpoint.""" response = test_client.delete( "/api/studies/mcp_test_study/spec/nodes/dv_001", params={"modified_by": "mcp"} ) # Endpoint should respond assert response.status_code in [200, 400, 404, 500] def test_canvas_connect_nodes_endpoint_exists(self, test_client): """Test canvas_connect_nodes MCP tool calls POST /spec/edges endpoint.""" response = test_client.post( "/api/studies/mcp_test_study/spec/edges", params={ "source": "ext_001", "target": "obj_001", "modified_by": "mcp" } ) # Endpoint should respond assert response.status_code in [200, 400, 500] class TestIntentMCPTools: """Tests for canvas intent MCP tools.""" def test_validate_canvas_intent_endpoint_exists(self, test_client): """Test validate_canvas_intent MCP tool.""" response = test_client.post("/api/studies/mcp_test_study/spec/validate") # Endpoint should respond assert response.status_code in [200, 400, 404, 500] def test_get_spec_endpoint_exists(self, test_client): """Test that MCP tools can fetch spec.""" response = test_client.get("/api/studies/mcp_test_study/spec") assert response.status_code in [200, 404] # ============================================================================ # Physics MCP Tool Tests # ============================================================================ class TestPhysicsMCPTools: """Tests for physics explanation MCP tools.""" def test_explain_physics_concepts(self): """Test that physics extractors are available.""" # Import extractors module from optimization_engine import extractors # Check that key extractor functions exist (using actual exports) assert hasattr(extractors, 'extract_solid_stress') assert hasattr(extractors, 'extract_part_mass') assert hasattr(extractors, 'ZernikeOPDExtractor') def test_query_extractors_available(self): """Test that extractor functions are importable.""" from optimization_engine.extractors import ( extract_solid_stress, extract_part_mass, extract_zernike_opd, ) # Functions should be callable assert callable(extract_solid_stress) assert callable(extract_part_mass) assert callable(extract_zernike_opd) # ============================================================================ # Method Recommendation Tests # ============================================================================ class TestMethodRecommendation: """Tests for optimization method recommendation logic.""" def test_method_selector_exists(self): """Test that method selector module exists.""" from optimization_engine.core import method_selector # Check key classes exist assert hasattr(method_selector, 'AdaptiveMethodSelector') assert hasattr(method_selector, 'MethodRecommendation') def test_algorithm_types_defined(self): """Test that algorithm types are defined for recommendations.""" from optimization_engine.config.spec_models import AlgorithmType # Check all expected algorithm types exist (using actual enum names) assert AlgorithmType.TPE is not None assert AlgorithmType.CMA_ES is not None assert AlgorithmType.NSGA_II is not None assert AlgorithmType.RANDOM_SEARCH is not None # ============================================================================ # Canvas Intent Validation Tests # ============================================================================ class TestCanvasIntentValidation: """Tests for canvas intent validation logic.""" def test_valid_intent_structure(self): """Test that valid intent passes validation.""" intent = { "version": "1.0", "source": "canvas", "timestamp": datetime.now().isoformat(), "model": {"path": "model.sim", "type": "sim"}, "solver": {"type": "SOL101"}, "design_variables": [ {"name": "thickness", "min": 1.0, "max": 10.0, "unit": "mm"} ], "extractors": [ {"id": "E5", "name": "Mass", "config": {}} ], "objectives": [ {"name": "mass", "direction": "minimize", "weight": 1.0, "extractor": "E5"} ], "constraints": [], "optimization": {"method": "TPE", "max_trials": 100} } # Validate required fields assert intent["model"]["path"] is not None assert intent["solver"]["type"] is not None assert len(intent["design_variables"]) > 0 assert len(intent["objectives"]) > 0 def test_invalid_intent_missing_model(self): """Test that missing model is detected.""" intent = { "version": "1.0", "source": "canvas", "model": {}, # Missing path "solver": {"type": "SOL101"}, "design_variables": [{"name": "x", "min": 0, "max": 1}], "objectives": [{"name": "y", "direction": "minimize", "extractor": "E5"}], "extractors": [{"id": "E5", "name": "Mass"}], } # Check validation would catch this assert intent["model"].get("path") is None def test_invalid_bounds(self): """Test that invalid bounds are detected.""" dv = {"name": "x", "min": 10.0, "max": 5.0} # min > max # Validation should catch this assert dv["min"] >= dv["max"] # ============================================================================ # MCP Tool Schema Documentation Tests # ============================================================================ class TestMCPToolDocumentation: """Tests to ensure MCP tools are properly documented.""" def test_all_canvas_tools_have_endpoints(self): """Verify canvas MCP tools map to backend endpoints.""" canvas_tools = [ "canvas_add_node", "canvas_update_node", "canvas_remove_node", "canvas_connect_nodes" ] for tool in canvas_tools: assert tool in MCP_TOOL_ENDPOINTS, f"Tool {tool} should be documented" assert "endpoint" in MCP_TOOL_ENDPOINTS[tool] assert "method" in MCP_TOOL_ENDPOINTS[tool] def test_all_intent_tools_have_endpoints(self): """Verify intent MCP tools map to backend endpoints.""" intent_tools = [ "validate_canvas_intent", "execute_canvas_intent", "interpret_canvas_intent" ] for tool in intent_tools: assert tool in MCP_TOOL_ENDPOINTS, f"Tool {tool} should be documented" # ============================================================================ # Run Tests # ============================================================================ if __name__ == "__main__": pytest.main([__file__, "-v"])