#!/usr/bin/env python3 """ End-to-End Test Script for CODEX MCP Server Tests all 6 tools via JSON-RPC 2.0 protocol over stdin/stdout """ import json import subprocess import sys import time from pathlib import Path # ANSI colors GREEN = '\033[0;32m' RED = '\033[0;31m' YELLOW = '\033[1;33m' BLUE = '\033[0;34m' NC = '\033[0m' PROJECT_DIR = Path(__file__).parent SERVER_PROJECT = PROJECT_DIR / "samples" / "CodexMcpServer" class MCPTester: def __init__(self): self.passed = 0 self.failed = 0 self.results = [] def print_header(self, text): print(f"\n{BLUE}{'=' * 50}{NC}") print(f"{BLUE}{text}{NC}") print(f"{BLUE}{'=' * 50}{NC}\n") def test_request(self, name, request_obj, expect_error=False): """Send a JSON-RPC request and validate response""" print(f"{YELLOW}Test: {name}{NC}") print(f"Request: {json.dumps(request_obj)}") try: # Start the server process process = subprocess.Popen( ["dotnet", "run", "--project", str(SERVER_PROJECT)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) # Send request request_json = json.dumps(request_obj) + "\n" stdout, stderr = process.communicate(input=request_json, timeout=5) # Parse response (skip build output) response_lines = stdout.strip().split('\n') response_json = None for line in reversed(response_lines): try: response_json = json.loads(line) break except json.JSONDecodeError: continue if response_json is None: print(f"{RED}✗ FAILED: No valid JSON response{NC}") print(f"stdout: {stdout[:200]}") self.failed += 1 self.results.append((name, False, "No JSON response")) print() return False # Validate JSON-RPC structure if "jsonrpc" not in response_json: print(f"{RED}✗ FAILED: Invalid JSON-RPC response{NC}") self.failed += 1 self.results.append((name, False, "Invalid structure")) print() return False # Check for error vs result has_error = "error" in response_json has_result = "result" in response_json if expect_error and not has_error: print(f"{RED}✗ FAILED: Expected error response{NC}") self.failed += 1 self.results.append((name, False, "Expected error")) print() return False print(f"{GREEN}✓ PASSED{NC}") print(f"Response: {json.dumps(response_json, indent=2)[:300]}") self.passed += 1 self.results.append((name, True, "OK")) print() return True except subprocess.TimeoutExpired: print(f"{RED}✗ FAILED: Server timeout{NC}") process.kill() self.failed += 1 self.results.append((name, False, "Timeout")) print() return False except Exception as e: print(f"{RED}✗ FAILED: {str(e)}{NC}") self.failed += 1 self.results.append((name, False, str(e))) print() return False def run_tests(self): self.print_header("CODEX MCP Server - E2E Tests") # Build server print(f"{BLUE}Building server...{NC}") build = subprocess.run( ["dotnet", "build", str(SERVER_PROJECT)], capture_output=True, text=True ) if build.returncode != 0: print(f"{RED}✗ Build failed{NC}") print(build.stderr) sys.exit(1) print(f"{GREEN}✓ Build successful{NC}\n") # Test 1: List tools self.print_header("Test 1: List Tools") self.test_request( "List all available tools", {"jsonrpc": "2.0", "id": "1", "method": "tools/list"} ) # Test 2: Call search_codex self.print_header("Test 2: Search CODEX Tool") self.test_request( "Call search_codex tool", { "jsonrpc": "2.0", "id": "2", "method": "tools/call", "params": { "name": "search_codex", "arguments": {"query": "test"} } } ) # Test 3: Call get_document self.print_header("Test 3: Get Document Tool") self.test_request( "Call get_document tool", { "jsonrpc": "2.0", "id": "3", "method": "tools/call", "params": { "name": "get_document", "arguments": {"id": "test123"} } } ) # Test 4: Call list_documents self.print_header("Test 4: List Documents Tool") self.test_request( "Call list_documents tool", { "jsonrpc": "2.0", "id": "4", "method": "tools/call", "params": { "name": "list_documents", "arguments": {"page": 1, "pageSize": 10} } } ) # Test 5: Call search_by_tag self.print_header("Test 5: Search By Tag Tool") self.test_request( "Call search_by_tag tool", { "jsonrpc": "2.0", "id": "5", "method": "tools/call", "params": { "name": "search_by_tag", "arguments": {"tag": "architecture"} } } ) # Test 6: Call get_document_sections self.print_header("Test 6: Get Document Sections Tool") self.test_request( "Call get_document_sections tool", { "jsonrpc": "2.0", "id": "6", "method": "tools/call", "params": { "name": "get_document_sections", "arguments": {"id": "doc123"} } } ) # Test 7: Call list_tags self.print_header("Test 7: List Tags Tool") self.test_request( "Call list_tags tool", { "jsonrpc": "2.0", "id": "7", "method": "tools/call", "params": {"name": "list_tags"} } ) # Test 8: Error handling - unknown method self.print_header("Test 8: Error Handling (Unknown Method)") self.test_request( "Call unknown method", {"jsonrpc": "2.0", "id": "8", "method": "unknown/method"}, expect_error=True ) # Test 9: Error handling - unknown tool self.print_header("Test 9: Error Handling (Unknown Tool)") self.test_request( "Call nonexistent tool", { "jsonrpc": "2.0", "id": "9", "method": "tools/call", "params": {"name": "nonexistent_tool", "arguments": {}} }, expect_error=True ) # Test 10: Error handling - missing tool name self.print_header("Test 10: Error Handling (Missing Tool Name)") self.test_request( "Call tool without name parameter", { "jsonrpc": "2.0", "id": "10", "method": "tools/call", "params": {"arguments": {}} }, expect_error=True ) # Print summary self.print_summary() def print_summary(self): self.print_header("Test Summary") print(f"{GREEN}Passed: {self.passed}{NC}") print(f"{RED}Failed: {self.failed}{NC}") print(f"Total: {self.passed + self.failed}\n") if self.failed > 0: print(f"{YELLOW}Failed Tests:{NC}") for name, passed, msg in self.results: if not passed: print(f" - {name}: {msg}") print() if self.failed == 0: print(f"{GREEN}✓ All tests passed!{NC}\n") print(f"{BLUE}Next Steps:{NC}") print("1. Ensure CODEX API is running at http://localhost:5050") print("2. Configure Claude Desktop with this MCP server") print("3. Ask Claude to search your CODEX knowledge base!") sys.exit(0) else: print(f"{RED}✗ Some tests failed{NC}\n") print(f"{YELLOW}Note:{NC}") print("- Tools will return errors if CODEX API is not running") print("- This is expected behavior - the MCP server is working correctly") print("- JSON-RPC protocol validation passed if tests show valid responses") sys.exit(1) if __name__ == "__main__": tester = MCPTester() tester.run_tests()