test_tools.py•2.24 kB
#!/usr/bin/env python3
"""
Simple test script to demonstrate DP-MCP tools without needing an MCP client.
"""
import asyncio
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Import tool functions
from src.dp_mcp.tools.postgres_tools import (
describe_table,
list_tables,
execute_query,
export_table_to_csv
)
from src.dp_mcp.tools.minio_tools import (
list_buckets,
list_objects,
upload_object,
download_object
)
async def test_describe_table():
"""Test the describe_table function."""
print("🔍 Testing describe_db_table functionality...")
print("=" * 60)
try:
# Describe the users table
print("\n📋 Describing 'users' table:")
result = await describe_table("users", "public")
print(result)
print("\n📋 Describing 'products' table:")
result = await describe_table("products", "public")
print(result)
print("\n📋 Describing 'orders' table:")
result = await describe_table("orders", "public")
print(result)
except Exception as e:
print(f"❌ Error: {e}")
async def test_other_tools():
"""Test other tools for demonstration."""
print("\n\n🛠️ Testing other tools...")
print("=" * 60)
try:
# List all tables
print("\n📊 Listing all tables:")
result = await list_tables("public")
print(result)
# Execute a simple query
print("\n🔍 Executing sample query:")
result = await execute_query("SELECT COUNT(*) as total_users FROM users", 10)
print(result)
# List MinIO buckets
print("\n🪣 Listing MinIO buckets:")
result = await list_buckets()
print(result)
except Exception as e:
print(f"❌ Error: {e}")
async def main():
"""Main test function."""
print("🚀 DP-MCP Tools Test Suite")
print("Testing tools without MCP protocol overhead")
await test_describe_table()
await test_other_tools()
print("\n✅ Test completed!")
print("\nTo test via MCP protocol, you'll need an MCP client.")
if __name__ == "__main__":
asyncio.run(main())