QML-MCP

#!/usr/bin/env python3 """ End-to-end integration test for the QML MCP server. Tests all tools through the actual server interface. """ import asyncio import json import sys import os # Add the project directory to the path sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from server import call_tool from qiskit import QuantumCircuit from qiskit.qasm3 import dumps as qasm3_dumps async def test_full_workflow(): """Test complete quantum ML workflow through the server.""" print("\n" + "=" * 70) print("QML MCP Server - End-to-End Integration Test") print("=" * 70) # Test 1: Run quantum circuit print("\n[Test 1] Running quantum circuit...") circuit = QuantumCircuit(2) circuit.h(0) circuit.cx(0, 1) circuit.measure_all() qasm = qasm3_dumps(circuit) result = await call_tool("run_quantum_circuit", {"qasm": qasm, "shots": 500}) response = json.loads(result[0].text) assert response["success"] is True, "Circuit execution failed" assert response["shots"] == 500, "Incorrect shot count" assert "counts" in response, "Missing counts in response" print(f"✓ Circuit executed successfully") print(f" Shots: {response['shots']}, Qubits: {response['num_qubits']}") # Test 2: Compute quantum kernel print("\n[Test 2] Computing quantum kernel...") result = await call_tool("compute_quantum_kernel", { "train_data": [[0.1, 0.2], [0.3, 0.4], [0.5, 0.6]] }) response = json.loads(result[0].text) assert response["success"] is True, "Kernel computation failed" assert len(response["kernel_matrix"]) == 3, "Incorrect kernel matrix size" print(f"✓ Kernel computed successfully") print(f" Feature dimension: {response['feature_dimension']}") # Test 3: Train VQC print("\n[Test 3] Training VQC classifier...") result = await call_tool("train_vqc", { "X_train": [[0.1, 0.2], [0.2, 0.3], [0.8, 0.9], [0.9, 0.8]], "y_train": [0, 0, 1, 1], "max_iter": 20 }) response = json.loads(result[0].text) assert response["success"] is True, "VQC training failed" assert "model" in response, "Missing model in response" assert response["train_score"] >= 0, "Invalid train score" trained_model = response["model"] print(f"✓ VQC trained successfully") print(f" Train score: {response['train_score']:.4f}") print(f" Samples: {response['num_samples']}") # Test 4: Evaluate model print("\n[Test 4] Evaluating trained model...") result = await call_tool("evaluate_model", { "model": trained_model, "X_test": [[0.15, 0.25], [0.85, 0.95]], "y_test": [0, 1] }) response = json.loads(result[0].text) assert response["success"] is True, "Model evaluation failed" assert len(response["predictions"]) == 2, "Incorrect number of predictions" assert "accuracy" in response, "Missing accuracy in response" print(f"✓ Model evaluated successfully") print(f" Predictions: {response['predictions']}") print(f" Accuracy: {response['accuracy']:.4f}") # Test 5: Error handling print("\n[Test 5] Testing error handling...") result = await call_tool("run_quantum_circuit", { "qasm": "INVALID QASM CODE", "shots": 100 }) response = json.loads(result[0].text) assert response["success"] is False, "Error handling failed" assert "error" in response, "Missing error message" print(f"✓ Error handling works correctly") print(f" Error type: {response['error_type']}") # Test 6: Safety limits print("\n[Test 6] Testing safety limits...") result = await call_tool("run_quantum_circuit", { "qasm": qasm, "shots": 200000 # Exceeds max_shots }) response = json.loads(result[0].text) assert response["success"] is False, "Safety limit not enforced" assert "exceeds maximum" in response["error"], "Wrong error message" print(f"✓ Safety limits enforced correctly") print("\n" + "=" * 70) print("All integration tests passed! ✓") print("=" * 70 + "\n") return True async def main(): """Run integration tests.""" try: success = await test_full_workflow() return 0 if success else 1 except Exception as e: print(f"\n✗ Integration test failed: {e}") import traceback traceback.print_exc() return 1 if __name__ == "__main__": exit_code = asyncio.run(main()) sys.exit(exit_code)