Anto01
6c30224341
Config Layer: - spec_models.py: Pydantic models for AtomizerSpec v2.0 - spec_validator.py: Semantic validation with detailed error reporting Extractors: - custom_extractor_loader.py: Runtime custom extractor loading - spec_extractor_builder.py: Build extractors from spec definitions Tools: - migrate_to_spec_v2.py: CLI tool for batch migration Tests: - test_migrator.py: Migration tests - test_spec_manager.py: SpecManager service tests - test_spec_api.py: REST API tests - test_mcp_tools.py: MCP tool tests - test_e2e_unified_config.py: End-to-end config tests
388 lines
14 KiB
Python
388 lines
14 KiB
Python
"""
|
|
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"])
|