test_integration.py•3.46 kB
#!/usr/bin/env python3
"""
Integration test script to verify CLI and MCP client functionality.
Run this after starting your DP-MCP server to ensure everything works.
"""
import asyncio
import subprocess
import sys
import json
from pathlib import Path
def run_command(cmd, description):
"""Run a command and return success/failure."""
print(f"\n🧪 Testing: {description}")
print(f"Command: {cmd}")
try:
result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30)
if result.returncode == 0:
print(f"✅ SUCCESS: {description}")
return True
else:
print(f"❌ FAILED: {description}")
print(f"Error: {result.stderr}")
return False
except subprocess.TimeoutExpired:
print(f"⏰ TIMEOUT: {description}")
return False
except Exception as e:
print(f"💥 ERROR: {description} - {e}")
return False
async def main():
"""Run integration tests."""
print("🚀 DP-MCP Integration Test Suite")
print("=================================")
# Check if we're in the right directory
if not Path("cli.py").exists() or not Path("mcp_client.py").exists():
print("❌ Please run this script from the DP-MCP project root directory")
sys.exit(1)
tests = []
# CLI Tests
print("\n📋 Testing CLI Tool")
print("==================")
cli_tests = [
("./dp-cli list-tables", "CLI: List database tables"),
("./dp-cli describe-table users", "CLI: Describe users table"),
("./dp-cli query 'SELECT COUNT(*) FROM users'", "CLI: Execute simple query"),
("./dp-cli list-buckets", "CLI: List MinIO buckets"),
("./dp-cli export-csv users --limit 5", "CLI: Export users table (5 rows)"),
]
for cmd, desc in cli_tests:
tests.append(run_command(cmd, desc))
# MCP Client Tests
print("\n🔌 Testing MCP Client")
print("====================")
mcp_tests = [
("uv run python mcp_client.py --ping", "MCP: Ping server"),
("uv run python mcp_client.py --list-tools", "MCP: List available tools"),
("uv run python mcp_client.py --call list_db_tables", "MCP: Call list_db_tables"),
("uv run python mcp_client.py --call describe_db_table --args '{\"table_name\": \"users\"}'", "MCP: Call describe_db_table"),
("uv run python mcp_client.py --call list_minio_buckets", "MCP: Call list_minio_buckets"),
]
for cmd, desc in mcp_tests:
tests.append(run_command(cmd, desc))
# Summary
print("\n📊 Test Results Summary")
print("=======================")
passed = sum(tests)
total = len(tests)
print(f"Total tests: {total}")
print(f"Passed: {passed}")
print(f"Failed: {total - passed}")
print(f"Success rate: {(passed/total)*100:.1f}%")
if passed == total:
print("\n🎉 All tests passed! Your DP-MCP setup is working perfectly.")
return 0
else:
print(f"\n⚠️ {total - passed} tests failed. Please check your setup:")
print("1. Make sure Docker services are running: docker-compose ps")
print("2. Ensure DP-MCP server is running: uv run python src/dp_mcp/server.py")
print("3. Check server logs for errors")
return 1
if __name__ == "__main__":
exit_code = asyncio.run(main())
sys.exit(exit_code)