#!/usr/bin/env python3
"""
MCP Test Runner
Orchestrates testing across all test categories with reporting
"""
import os
import sys
import subprocess
import time
import argparse
from pathlib import Path
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))
class TestRunner:
def __init__(self):
self.test_dir = Path(__file__).parent
self.results = []
def run_test(self, test_path, description=""):
"""Run a single test file and capture results"""
print(f"\n{'='*60}")
print(f"Running: {test_path.name}")
if description:
print(f"Purpose: {description}")
print(f"{'='*60}")
start_time = time.time()
try:
# Run the test
result = subprocess.run(
[sys.executable, str(test_path)],
cwd=self.test_dir.parent,
capture_output=True,
text=True,
timeout=300 # 5 minute timeout
)
execution_time = time.time() - start_time
if result.returncode == 0:
status = "PASS"
print(f"[PASS] in {execution_time:.2f}s")
else:
status = "FAIL"
print(f"[FAIL] in {execution_time:.2f}s")
print("STDERR:", result.stderr[:500])
# Show some output
if result.stdout:
output_lines = result.stdout.split('\n')
relevant_lines = [line for line in output_lines if any(keyword in line.lower() for keyword in ['pass', 'fail', 'success', 'error', 'complete'])]
if relevant_lines:
print("Key outputs:")
for line in relevant_lines[-5:]: # Last 5 relevant lines
print(f" {line}")
self.results.append({
'name': test_path.name,
'status': status,
'time': execution_time,
'category': test_path.parent.name,
'returncode': result.returncode
})
except subprocess.TimeoutExpired:
print(f"[TIMEOUT] after 5 minutes")
self.results.append({
'name': test_path.name,
'status': 'TIMEOUT',
'time': 300,
'category': test_path.parent.name,
'returncode': -1
})
except Exception as e:
print(f"[ERROR]: {e}")
self.results.append({
'name': test_path.name,
'status': 'ERROR',
'time': 0,
'category': test_path.parent.name,
'returncode': -1
})
def run_category_tests(self, category):
"""Run all tests in a specific category"""
category_dir = self.test_dir / category
if not category_dir.exists():
print(f"[ERROR] Category '{category}' not found")
return
test_files = list(category_dir.glob("*.py"))
if not test_files:
print(f"[WARN] No test files found in '{category}'")
return
print(f"\nRunning {len(test_files)} tests in category: {category}")
# Define test descriptions
descriptions = {
'test_all_endpoints_simple.py': 'Comprehensive API endpoint testing',
'test_core_smart_search.py': 'Core smart search functionality',
'test_smart_search_simple.py': 'Basic smart search validation',
'test_streaming.py': 'Streaming endpoint testing',
'test_conversation_chat.py': 'Conversation memory testing',
'test_complete_server.py': 'Full server integration testing',
'test_mcp.py': 'MCP protocol compliance testing'
}
for test_file in sorted(test_files):
if test_file.name.startswith('__'):
continue
description = descriptions.get(test_file.name, "")
self.run_test(test_file, description)
def run_priority_tests(self):
"""Run the most important tests first"""
priority_tests = [
('api/test_all_endpoints_simple.py', 'Comprehensive API endpoint testing'),
('core/test_core_smart_search.py', 'Core smart search functionality'),
('integration/test_complete_server.py', 'Full server integration')
]
print("\n🎯 Running Priority Tests")
print("=" * 60)
for test_path, description in priority_tests:
full_path = self.test_dir / test_path
if full_path.exists():
self.run_test(full_path, description)
else:
print(f"⚠️ Priority test not found: {test_path}")
def print_summary(self):
"""Print test execution summary"""
if not self.results:
print("\n⚠️ No tests were executed")
return
print(f"\n📊 Test Summary")
print("=" * 60)
# Overall stats
total = len(self.results)
passed = len([r for r in self.results if r['status'] == 'PASS'])
failed = len([r for r in self.results if r['status'] == 'FAIL'])
errors = len([r for r in self.results if r['status'] in ['ERROR', 'TIMEOUT']])
success_rate = (passed / total * 100) if total > 0 else 0
print(f"Total Tests: {total}")
print(f"✅ Passed: {passed}")
print(f"❌ Failed: {failed}")
print(f"💥 Errors: {errors}")
print(f"Success Rate: {success_rate:.1f}%")
# Category breakdown
categories = {}
for result in self.results:
cat = result['category']
if cat not in categories:
categories[cat] = {'total': 0, 'passed': 0}
categories[cat]['total'] += 1
if result['status'] == 'PASS':
categories[cat]['passed'] += 1
print(f"\nBy Category:")
for cat, stats in categories.items():
rate = (stats['passed'] / stats['total'] * 100) if stats['total'] > 0 else 0
print(f" {cat}: {stats['passed']}/{stats['total']} ({rate:.1f}%)")
# Failed tests
failed_tests = [r for r in self.results if r['status'] != 'PASS']
if failed_tests:
print(f"\n❌ Failed Tests:")
for test in failed_tests:
print(f" {test['name']} ({test['status']})")
# Performance
execution_times = [r['time'] for r in self.results if r['time'] > 0]
if execution_times:
avg_time = sum(execution_times) / len(execution_times)
max_time = max(execution_times)
print(f"\n⏱️ Performance:")
print(f" Average: {avg_time:.2f}s")
print(f" Slowest: {max_time:.2f}s")
# Overall status
print(f"\n🎯 Overall Status:")
if failed == 0 and errors == 0:
print("🎉 All tests passed! System is healthy.")
elif errors == 0:
print("⚠️ Some tests failed but no critical errors.")
else:
print("🚨 Critical errors detected. System needs attention.")
def main():
parser = argparse.ArgumentParser(description='MCP Test Runner')
parser.add_argument('--category', choices=['api', 'core', 'integration', 'phase_tests', 'legacy'],
help='Run tests from specific category')
parser.add_argument('--priority', action='store_true',
help='Run only priority tests')
parser.add_argument('--all', action='store_true',
help='Run all tests')
parser.add_argument('--quick', action='store_true',
help='Run quick health check tests')
args = parser.parse_args()
runner = TestRunner()
print("🧪 MCP Test Runner")
print("=" * 60)
if args.priority:
runner.run_priority_tests()
elif args.category:
runner.run_category_tests(args.category)
elif args.all:
for category in ['api', 'core', 'integration']:
runner.run_category_tests(category)
elif args.quick:
# Quick health check
quick_tests = [
('api/test_all_endpoints_simple.py', 'API health check'),
]
for test_path, description in quick_tests:
full_path = runner.test_dir / test_path
if full_path.exists():
runner.run_test(full_path, description)
else:
# Default: run priority tests
print("No specific option provided. Running priority tests...")
print("Use --help to see all options")
runner.run_priority_tests()
runner.print_summary()
if __name__ == "__main__":
main()
