test_standalone.py•3.7 kB
#!/usr/bin/env python3
"""
Standalone test script for EdgeLake client
Tests without importing the full edgelake_mcp package to avoid MCP dependencies.
"""
import asyncio
import sys
import os
# Direct imports
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
# Import client and config modules directly without going through __init__.py
import edgelake_mcp.client as client_module
import edgelake_mcp.config as config_module
async def test_edgelake():
"""Test EdgeLake client directly"""
print("=" * 80)
print("EdgeLake Client - Standalone Test")
print("=" * 80)
# Load configuration
config = config_module.Config.from_env()
print(f"\nConfiguration:")
print(f" Host: {config.edgelake_host}")
print(f" Port: {config.edgelake_port}")
print(f" Timeout: {config.request_timeout}s")
# Create client
client = client_module.EdgeLakeClient(
host=config.edgelake_host,
port=config.edgelake_port,
timeout=config.request_timeout
)
try:
# Test 1: List databases
print("\n" + "=" * 80)
print("TEST 1: List Databases")
print("=" * 80)
databases = await client.get_databases()
print(f"✓ Found {len(databases)} databases:")
for db in databases:
print(f" - {db}")
if not databases:
print("⚠ No databases found. Stopping tests.")
return
# Use first database for remaining tests
test_db = databases[0]
# Test 2: List tables
print("\n" + "=" * 80)
print(f"TEST 2: List Tables in '{test_db}'")
print("=" * 80)
tables = await client.get_tables(test_db)
print(f"✓ Found {len(tables)} tables:")
for table in tables:
print(f" - {table}")
if not tables:
print("⚠ No tables found. Stopping tests.")
return
# Use first table for remaining tests
test_table = tables[0]
# Test 3: Get schema
print("\n" + "=" * 80)
print(f"TEST 3: Get Schema for '{test_db}.{test_table}'")
print("=" * 80)
schema = await client.get_table_schema(test_db, test_table)
print("✓ Schema retrieved:")
print(schema[:500] + "..." if len(schema) > 500 else schema)
# Test 4: Execute query
print("\n" + "=" * 80)
print(f"TEST 4: Query '{test_db}.{test_table}' (LIMIT 10)")
print("=" * 80)
query = f"SELECT * FROM {test_table} LIMIT 10"
print(f"SQL: {query}")
result = await client.execute_query(
database=test_db,
query=query,
output_format="json"
)
print("✓ Query executed successfully!")
print("Result preview:")
print(result[:500] + "..." if len(result) > 500 else result)
# Test 5: Node status
print("\n" + "=" * 80)
print("TEST 5: Get Node Status")
print("=" * 80)
status = await client.get_node_status()
print("✓ Status retrieved:")
print(status[:500] + "..." if len(status) > 500 else status)
print("\n" + "=" * 80)
print("✅ ALL TESTS PASSED!")
print("=" * 80)
except Exception as e:
print("\n" + "=" * 80)
print(f"❌ TEST FAILED: {e}")
print("=" * 80)
import traceback
traceback.print_exc()
sys.exit(1)
finally:
client.close()
def main():
"""Main entry point"""
try:
asyncio.run(test_edgelake())
except KeyboardInterrupt:
print("\n\nTest interrupted by user")
sys.exit(0)
if __name__ == "__main__":
main()