Spaces:
Sleeping
Sleeping
File size: 6,124 Bytes
81e3673 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 | #!/usr/bin/env python3
"""
Analytics Dashboard Test Script
Run this to verify the analytics dashboard is working correctly.
"""
from pathlib import Path
import sys
# Add backend to path
sys.path.insert(0, str(Path(__file__).parent))
def test_imports():
"""Test that all required modules can be imported."""
print("Testing imports...")
try:
from api.analytics_dashboard_endpoints import router
from core.workflow_analytics_engine import WorkflowAnalyticsEngine
print(" ✓ All imports successful")
return True, router
except Exception as e:
print(f" ✗ Import failed: {e}")
return False, None
def test_engine_initialization():
"""Test analytics engine initialization."""
print("\nTesting engine initialization...")
try:
from core.workflow_analytics_engine import WorkflowAnalyticsEngine
engine = WorkflowAnalyticsEngine()
print(" ✓ Analytics engine initialized")
return True, engine
except Exception as e:
print(f" ✗ Engine initialization failed: {e}")
return False, None
def test_helper_methods(engine):
"""Test all helper methods."""
print("\nTesting helper methods...")
tests_passed = 0
tests_failed = 0
# Test 1: get_unique_workflow_count
try:
count = engine.get_unique_workflow_count('24h')
print(f" ✓ get_unique_workflow_count: {count}")
tests_passed += 1
except Exception as e:
print(f" ✗ get_unique_workflow_count failed: {e}")
tests_failed += 1
# Test 2: get_all_workflow_ids
try:
workflows = engine.get_all_workflow_ids('24h')
print(f" ✓ get_all_workflow_ids: {len(workflows)} workflows")
tests_passed += 1
except Exception as e:
print(f" ✗ get_all_workflow_ids failed: {e}")
tests_failed += 1
# Test 3: get_recent_events
try:
events = engine.get_recent_events(limit=5)
print(f" ✓ get_recent_events: {len(events)} events")
tests_passed += 1
except Exception as e:
print(f" ✗ get_recent_events failed: {e}")
tests_failed += 1
# Test 4: get_all_alerts
try:
alerts = engine.get_all_alerts()
print(f" ✓ get_all_alerts: {len(alerts)} alerts")
tests_passed += 1
except Exception as e:
print(f" ✗ get_all_alerts failed: {e}")
tests_failed += 1
# Test 5: get_performance_metrics
try:
kpis = engine.get_performance_metrics('*', '24h')
print(f" ✓ get_performance_metrics: {kpis.total_executions} executions")
tests_passed += 1
except Exception as e:
print(f" ✗ get_performance_metrics failed: {e}")
tests_failed += 1
# Test 6: get_execution_timeline
try:
timeline = engine.get_execution_timeline('*', '24h', '1h')
print(f" ✓ get_execution_timeline: {len(timeline)} data points")
tests_passed += 1
except Exception as e:
print(f" ✗ get_execution_timeline failed: {e}")
tests_failed += 1
# Test 7: get_error_breakdown
try:
errors = engine.get_error_breakdown('*', '24h')
print(f" ✓ get_error_breakdown: {len(errors.get('recent_errors', []))} errors")
tests_passed += 1
except Exception as e:
print(f" ✗ get_error_breakdown failed: {e}")
tests_failed += 1
print(f"\n Helper methods: {tests_passed} passed, {tests_failed} failed")
return tests_failed == 0
def test_alert_management(engine):
"""Test alert creation, update, and delete."""
print("\nTesting alert management...")
from datetime import datetime
from core.workflow_analytics_engine import Alert, AlertSeverity
try:
# Create alert
alert = Alert(
alert_id='test-alert-validation',
name='Test Alert',
description='Test alert for validation',
severity=AlertSeverity.LOW,
condition='error_rate > 0',
threshold_value=0.0,
metric_name='error_rate',
workflow_id=None,
enabled=True,
created_at=datetime.now(),
notification_channels=[]
)
engine.create_alert(alert)
print(" ✓ Created test alert")
# Update alert
engine.update_alert(alert.alert_id, enabled=False)
print(" ✓ Updated alert")
# Delete alert
engine.delete_alert(alert.alert_id)
print(" ✓ Deleted alert")
return True
except Exception as e:
print(f" ✗ Alert management failed: {e}")
return False
def main():
"""Run all tests."""
print("=" * 70)
print("Analytics Dashboard Validation Test")
print("=" * 70)
# Test imports
success, router = test_imports()
if not success:
print("\n❌ FAILED: Cannot import required modules")
return 1
# Test router
print(f"\nRouter has {len(router.routes)} routes:")
for route in router.routes[:5]:
if hasattr(route, 'path'):
print(f" - {route.path}")
if len(router.routes) > 5:
print(f" ... and {len(router.routes) - 5} more")
# Test engine
success, engine = test_engine_initialization()
if not success:
print("\n❌ FAILED: Cannot initialize analytics engine")
return 1
# Test helper methods
if not test_helper_methods(engine):
print("\n⚠️ WARNING: Some helper methods failed")
# Test alert management
if not test_alert_management(engine):
print("\n⚠️ WARNING: Alert management failed")
print("\n" + "=" * 70)
print("✅ Analytics Dashboard Validation Complete")
print("=" * 70)
print("\nThe analytics dashboard is ready to use!")
print("\nNext steps:")
print("1. Start the backend server: python3 main_api_app.py")
print("2. Access endpoints at: http://localhost:8000/api/analytics/dashboard/*")
print("3. Integrate frontend component in Next.js app")
return 0
if __name__ == '__main__':
sys.exit(main())
|