File size: 4,084 Bytes
02d2610 | 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 | """
Test script to validate data layer integration with actual JSON report.
"""
from results_loader import (
find_latest_full_report,
load_full_results,
summarize_users,
get_layer_details,
get_report_metadata
)
from alert_config import (
load_alert_config,
evaluate_alerts,
format_alert_summary
)
def test_full_integration():
"""Test complete data layer with actual report."""
print("="*70)
print("DATA LAYER INTEGRATION TEST")
print("="*70)
# Step 1: Find and load report
print("\n1. Loading Report...")
report_path = find_latest_full_report()
if not report_path:
print("❌ No report found")
return False
print(f" ✅ Found: {report_path.name}")
results = load_full_results(report_path)
print(f" ✅ Loaded {len(results)} sections")
# Step 2: Get metadata
print("\n2. Report Metadata...")
metadata = get_report_metadata(report_path, results)
print(f" • File size: {metadata['file_size_mb']:.1f} MB")
print(f" • Users: {metadata['user_count']}")
print(f" • Modified: {metadata['modified_time']}")
# Step 3: Summarize users
print("\n3. User Summaries...")
summaries = summarize_users(results)
print(f" ✅ Processed {len(summaries)} users")
# Show quality distribution
quality_bands = {'Excellent (>0.9)': 0, 'Good (0.8-0.9)': 0,
'Fair (0.7-0.8)': 0, 'Poor (<0.7)': 0}
for summary in summaries:
score = summary['overall_mean_score']
if score:
if score > 0.9:
quality_bands['Excellent (>0.9)'] += 1
elif score > 0.8:
quality_bands['Good (0.8-0.9)'] += 1
elif score > 0.7:
quality_bands['Fair (0.7-0.8)'] += 1
else:
quality_bands['Poor (<0.7)'] += 1
print("\n Quality Distribution:")
for band, count in quality_bands.items():
print(f" • {band}: {count} users")
# Step 4: Load alert config
print("\n4. Alert Configuration...")
config = load_alert_config()
print(f" ✅ Loaded thresholds for {len(config)} sections")
# Step 5: Evaluate alerts for all users
print("\n5. Alert Evaluation...")
all_alerts = []
# Get layer details once
layer1_details = get_layer_details(results, 1)
layer2_details = get_layer_details(results, 2)
layer3_details = get_layer_details(results, 3)
combined_details = {
'layer1': layer1_details,
'layer2': layer2_details,
'layer3': layer3_details
}
for summary in summaries:
user_alerts = evaluate_alerts(summary, combined_details, config)
all_alerts.extend(user_alerts)
print(f" ✅ Evaluated {len(summaries)} users")
print(f" ⚠️ Found {len(all_alerts)} total alerts")
# Show alert breakdown
if all_alerts:
severity_counts = {'HIGH': 0, 'MEDIUM': 0, 'LOW': 0}
layer_counts = {'Global': 0, 'Layer 1': 0, 'Layer 2': 0, 'Layer 3': 0}
for alert in all_alerts:
severity_counts[alert['severity']] += 1
layer_counts[alert['layer']] += 1
print("\n Alert Severity:")
for severity, count in severity_counts.items():
if count > 0:
print(f" • {severity}: {count} alerts")
print("\n Alerts by Layer:")
for layer, count in layer_counts.items():
if count > 0:
print(f" • {layer}: {count} alerts")
# Step 6: Sample alert output
if all_alerts:
print("\n6. Sample Alerts (first 3)...")
for alert in all_alerts[:3]:
print(f" • [{alert['severity']}] {alert['user_id']}: {alert['message']}")
print("\n" + "="*70)
print("✅ DATA LAYER TEST COMPLETE - ALL SYSTEMS OPERATIONAL")
print("="*70)
return True
if __name__ == "__main__":
success = test_full_integration()
exit(0 if success else 1)
|