#!/usr/bin/env python3
"""
Test script for Azure SQL MCP Server
This script tests the MCP server functionality without requiring
a full MCP client setup.
"""
import os
import sys
import json
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Import the server module
sys.path.insert(0, os.path.dirname(__file__))
import azure_sql_mcp
def test_connection():
"""Test database connection."""
print("Testing database connection...")
try:
conn = azure_sql_mcp.get_db_connection()
print("✓ Connection successful!")
return True
except Exception as e:
print(f"✗ Connection failed: {e}")
return False
def test_query():
"""Test a simple query."""
print("\nTesting simple query...")
try:
results = azure_sql_mcp.execute_query(
"SELECT DB_NAME() as database_name, @@VERSION as version"
)
print("✓ Query successful!")
print(f" Database: {results[0].get('database_name')}")
version = results[0].get('version', '').split('\n')[0]
print(f" Version: {version}")
return True
except Exception as e:
print(f"✗ Query failed: {e}")
return False
def test_list_tables():
"""Test listing tables."""
print("\nTesting list tables...")
try:
results = azure_sql_mcp.execute_query("""
SELECT TABLE_SCHEMA, TABLE_NAME, TABLE_TYPE
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_TYPE = 'BASE TABLE'
ORDER BY TABLE_NAME
""")
print(f"✓ Found {len(results)} tables:")
for table in results[:5]: # Show first 5
schema = table.get('TABLE_SCHEMA', 'dbo')
name = table.get('TABLE_NAME', '')
print(f" - {schema}.{name}")
if len(results) > 5:
print(f" ... and {len(results) - 5} more")
return True
except Exception as e:
print(f"✗ Failed to list tables: {e}")
return False
def test_tool_inputs():
"""Test tool input validation."""
print("\nTesting tool input validation...")
# Test QueryInput
from azure_sql_mcp import QueryInput, ResponseFormat
try:
# Valid input
valid_input = QueryInput(
query="SELECT * FROM customers",
response_format=ResponseFormat.JSON
)
print("✓ Valid input accepted")
# Test validation (should fail - empty query)
try:
invalid_input = QueryInput(query=" ")
print("✗ Empty query should have been rejected")
except ValueError:
print("✓ Empty query correctly rejected")
# Test validation (should fail - multiple statements)
try:
invalid_input = QueryInput(query="SELECT 1; DROP TABLE users")
print("✗ Multiple statements should have been rejected")
except ValueError:
print("✓ Multiple statements correctly rejected")
return True
except Exception as e:
print(f"✗ Input validation test failed: {e}")
return False
def main():
"""Run all tests."""
print("=" * 60)
print("Azure SQL MCP Server - Test Suite")
print("=" * 60)
# Check environment variables
print("\nChecking configuration...")
required_vars = [
"AZURE_SQL_SERVER",
"AZURE_SQL_DATABASE",
"AZURE_SQL_USERNAME",
"AZURE_SQL_PASSWORD"
]
missing_vars = []
for var in required_vars:
value = os.getenv(var)
if not value:
missing_vars.append(var)
print(f"✗ {var}: NOT SET")
else:
# Mask password
if "PASSWORD" in var:
display_value = "*" * len(value)
else:
display_value = value
print(f"✓ {var}: {display_value}")
if missing_vars:
print(f"\n✗ Missing required environment variables: {', '.join(missing_vars)}")
print("Please set them in your .env file")
return False
# Run tests
tests = [
("Database Connection", test_connection),
("Simple Query", test_query),
("List Tables", test_list_tables),
("Input Validation", test_tool_inputs),
]
results = []
for test_name, test_func in tests:
try:
result = test_func()
results.append((test_name, result))
except Exception as e:
print(f"\n✗ {test_name} crashed: {e}")
results.append((test_name, False))
# Summary
print("\n" + "=" * 60)
print("Test Summary")
print("=" * 60)
passed = sum(1 for _, result in results if result)
total = len(results)
for test_name, result in results:
status = "✓ PASS" if result else "✗ FAIL"
print(f"{status}: {test_name}")
print(f"\nTotal: {passed}/{total} tests passed")
if passed == total:
print("\n🎉 All tests passed! Your MCP server is ready to use.")
return True
else:
print(f"\n⚠️ {total - passed} test(s) failed. Please fix the issues above.")
return False
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)
