#!/usr/bin/env python3
"""
Test script for Azure SQL MCP Server Chart Visualization
Tests the azure_sql_visualize_data tool functionality
"""
import json
import sys
from pathlib import Path
# Add parent directory to path to import azure_sql_mcp
sys.path.insert(0, str(Path(__file__).parent))
try:
from azure_sql_mcp import (
visualize_data,
VisualizeInput,
get_db_connection,
execute_query
)
print("✓ Successfully imported MCP server modules")
except ImportError as e:
print(f"✗ Error importing modules: {e}")
print("\nMake sure azure_sql_mcp.py is in the same directory.")
sys.exit(1)
def print_section(title):
"""Print a formatted section header."""
print(f"\n{'='*70}")
print(f" {title}")
print('='*70)
def test_connection():
"""Test database connection."""
print_section("TEST 1: Database Connection")
try:
conn = get_db_connection()
print("✓ Database connection successful")
return True
except Exception as e:
print(f"✗ Connection failed: {e}")
print("\nPlease check your .env file configuration:")
print(" - AZURE_SQL_SERVER")
print(" - AZURE_SQL_DATABASE")
print(" - AZURE_SQL_USERNAME")
print(" - AZURE_SQL_PASSWORD")
return False
def test_sample_data():
"""Create sample data for testing charts."""
print_section("TEST 2: Create Sample Data")
try:
# Create sample table
create_table_query = """
IF OBJECT_ID('chart_test_data', 'U') IS NOT NULL
DROP TABLE chart_test_data;
CREATE TABLE chart_test_data (
region VARCHAR(50),
sales DECIMAL(10, 2),
category VARCHAR(50)
);
INSERT INTO chart_test_data (region, sales, category) VALUES
('North', 50000, 'Electronics'),
('South', 30000, 'Electronics'),
('East', 45000, 'Electronics'),
('West', 20000, 'Electronics'),
('North', 35000, 'Clothing'),
('South', 25000, 'Clothing'),
('East', 30000, 'Clothing'),
('West', 15000, 'Clothing');
"""
execute_query(create_table_query)
print("✓ Sample data created successfully")
# Verify data
verify_query = "SELECT COUNT(*) as count FROM chart_test_data"
result = execute_query(verify_query)
print(f"✓ Inserted {result[0]['count']} rows")
return True
except Exception as e:
print(f"✗ Failed to create sample data: {e}")
return False
async def test_bar_chart():
"""Test bar chart generation."""
print_section("TEST 3: Bar Chart - Sales by Region")
try:
params = VisualizeInput(
query="SELECT region, SUM(sales) as total_sales FROM chart_test_data GROUP BY region ORDER BY total_sales DESC",
chart_type="bar",
title="Sales by Region",
label_column="region",
value_column="total_sales",
width=800,
height=500
)
result = await visualize_data(params)
# Parse and display result
card = json.loads(result)
print("✓ Bar chart generated successfully")
print(f"\n📊 Chart Type: {card['body'][0]['text']}")
print(f"🖼️ Image URL: {card['body'][1]['url'][:80]}...")
# Display statistics
print("\n📈 Statistics:")
for fact in card['body'][2]['facts']:
print(f" {fact['title']}: {fact['value']}")
return True
except Exception as e:
print(f"✗ Bar chart test failed: {e}")
return False
async def test_pie_chart():
"""Test pie chart generation."""
print_section("TEST 4: Pie Chart - Sales by Category")
try:
params = VisualizeInput(
query="SELECT category, SUM(sales) as total FROM chart_test_data GROUP BY category",
chart_type="pie",
title="Sales by Category",
label_column="category",
value_column="total"
)
result = await visualize_data(params)
card = json.loads(result)
print("✓ Pie chart generated successfully")
print(f"📊 {card['body'][0]['text']}")
return True
except Exception as e:
print(f"✗ Pie chart test failed: {e}")
return False
async def test_line_chart():
"""Test line chart generation."""
print_section("TEST 5: Line Chart - Region Performance")
try:
params = VisualizeInput(
query="SELECT region, SUM(sales) as revenue FROM chart_test_data GROUP BY region ORDER BY region",
chart_type="line",
title="Revenue by Region",
label_column="region",
value_column="revenue"
)
result = await visualize_data(params)
card = json.loads(result)
print("✓ Line chart generated successfully")
print(f"📊 {card['body'][0]['text']}")
return True
except Exception as e:
print(f"✗ Line chart test failed: {e}")
return False
async def test_error_handling():
"""Test error handling."""
print_section("TEST 6: Error Handling")
test_passed = True
# Test 1: Invalid column name
try:
params = VisualizeInput(
query="SELECT region, SUM(sales) as total FROM chart_test_data GROUP BY region",
chart_type="bar",
title="Test",
label_column="region",
value_column="invalid_column" # This column doesn't exist
)
result = await visualize_data(params)
card = json.loads(result)
if "Column Not Found" in card['body'][0]['text']:
print("✓ Invalid column error handled correctly")
else:
print("✗ Invalid column error not detected")
test_passed = False
except Exception as e:
print(f"✗ Unexpected error: {e}")
test_passed = False
# Test 2: Empty result set
try:
params = VisualizeInput(
query="SELECT region, SUM(sales) as total FROM chart_test_data WHERE 1=0 GROUP BY region",
chart_type="bar",
title="Test",
label_column="region",
value_column="total"
)
result = await visualize_data(params)
card = json.loads(result)
if "No Data Found" in card['body'][0]['text']:
print("✓ Empty result set handled correctly")
else:
print("✗ Empty result set error not detected")
test_passed = False
except Exception as e:
print(f"✗ Unexpected error: {e}")
test_passed = False
# Test 3: Invalid chart type
try:
params = VisualizeInput(
query="SELECT region, SUM(sales) as total FROM chart_test_data GROUP BY region",
chart_type="invalid_type",
title="Test",
label_column="region",
value_column="total"
)
print("✗ Invalid chart type not caught by validation")
test_passed = False
except ValueError as e:
print("✓ Invalid chart type validation works")
except Exception as e:
print(f"✗ Unexpected error: {e}")
test_passed = False
return test_passed
def cleanup_sample_data():
"""Clean up test data."""
print_section("Cleanup")
try:
execute_query("DROP TABLE IF EXISTS chart_test_data")
print("✓ Sample data cleaned up")
except Exception as e:
print(f"✗ Cleanup failed: {e}")
async def run_all_tests():
"""Run all tests."""
print("""
╔══════════════════════════════════════════════════════════════════════╗
║ ║
║ Azure SQL MCP Server - Chart Visualization Tests ║
║ ║
╚══════════════════════════════════════════════════════════════════════╝
""")
results = []
# Test 1: Connection
results.append(("Connection", test_connection()))
if not results[-1][1]:
print("\n❌ Cannot proceed without database connection")
return
# Test 2: Sample Data
results.append(("Sample Data", test_sample_data()))
if not results[-1][1]:
print("\n❌ Cannot proceed without sample data")
return
# Test 3-5: Chart Types
results.append(("Bar Chart", await test_bar_chart()))
results.append(("Pie Chart", await test_pie_chart()))
results.append(("Line Chart", await test_line_chart()))
# Test 6: Error Handling
results.append(("Error Handling", await test_error_handling()))
# Cleanup
cleanup_sample_data()
# Summary
print_section("TEST SUMMARY")
passed = sum(1 for _, result in results if result)
total = len(results)
for name, result in results:
status = "✓ PASSED" if result else "✗ FAILED"
print(f"{status:12} - {name}")
print(f"\n{'='*70}")
print(f"Results: {passed}/{total} tests passed")
if passed == total:
print("\n🎉 All tests passed! Chart visualization is working perfectly!")
print("\nNext steps:")
print(" 1. Try creating charts in Copilot Studio")
print(" 2. Check CHART_GUIDE.md for examples")
print(" 3. See EXAMPLES.md for more use cases")
else:
print("\n⚠️ Some tests failed. Please review the errors above.")
print("\nTroubleshooting:")
print(" - Check your database connection")
print(" - Verify table and column names")
print(" - Review error messages for details")
if __name__ == "__main__":
import asyncio
try:
asyncio.run(run_all_tests())
except KeyboardInterrupt:
print("\n\n⚠️ Tests interrupted by user")
except Exception as e:
print(f"\n\n❌ Unexpected error: {e}")
import traceback
traceback.print_exc()
