557 lines
18 KiB
Python
557 lines
18 KiB
Python
#!/usr/bin/env python
|
||
"""
|
||
Comprehensive SLA System Test
|
||
|
||
This script tests the complete SLA (Service Level Agreement) system:
|
||
1. SLA Configuration
|
||
2. SLA Deadline Calculation
|
||
3. Overdue Detection
|
||
4. Escalation Rules
|
||
5. SLA Reminders
|
||
6. Explanation SLA
|
||
"""
|
||
import os
|
||
import django
|
||
from datetime import timedelta
|
||
|
||
# Setup Django
|
||
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.dev')
|
||
django.setup()
|
||
|
||
from apps.complaints.models import (
|
||
Complaint, ComplaintSLAConfig, EscalationRule,
|
||
ComplaintThreshold, ExplanationSLAConfig, ComplaintStatus
|
||
)
|
||
from apps.organizations.models import Hospital, Department, Staff, Patient
|
||
from apps.accounts.models import User
|
||
from django.utils import timezone
|
||
|
||
|
||
def setup_test_data():
|
||
"""Setup test data for SLA testing."""
|
||
print("\n" + "=" * 80)
|
||
print("🔧 SETTING UP TEST DATA")
|
||
print("=" * 80)
|
||
|
||
# Get or create hospital
|
||
hospital = Hospital.objects.first()
|
||
if not hospital:
|
||
print("❌ No hospital found. Please run seed commands first.")
|
||
return None, None, None
|
||
|
||
print(f"✅ Hospital: {hospital.name}")
|
||
|
||
# Get department
|
||
department = Department.objects.filter(hospital=hospital).first()
|
||
if department:
|
||
print(f"✅ Department: {department.name}")
|
||
else:
|
||
print("⚠️ No department found")
|
||
|
||
# Get user for assignment
|
||
user = User.objects.filter(hospital=hospital).first()
|
||
if user:
|
||
print(f"✅ User: {user.get_full_name()}")
|
||
else:
|
||
print("⚠️ No user found")
|
||
|
||
return hospital, department, user
|
||
|
||
|
||
def test_sla_configuration(hospital):
|
||
"""Test SLA configuration creation and retrieval."""
|
||
print("\n" + "=" * 80)
|
||
print("📋 TEST 1: SLA CONFIGURATION")
|
||
print("=" * 80)
|
||
|
||
# Clean up existing configs
|
||
ComplaintSLAConfig.objects.filter(hospital=hospital).delete()
|
||
|
||
# Create SLA configs for different severities and priorities
|
||
sla_configs = [
|
||
# Low severity
|
||
{
|
||
'severity': 'low',
|
||
'priority': 'low',
|
||
'sla_hours': 72,
|
||
'reminder_hours_before': 24,
|
||
},
|
||
{
|
||
'severity': 'low',
|
||
'priority': 'medium',
|
||
'sla_hours': 48,
|
||
'reminder_hours_before': 24,
|
||
},
|
||
{
|
||
'severity': 'low',
|
||
'priority': 'high',
|
||
'sla_hours': 24,
|
||
'reminder_hours_before': 12,
|
||
},
|
||
# Medium severity
|
||
{
|
||
'severity': 'medium',
|
||
'priority': 'low',
|
||
'sla_hours': 48,
|
||
'reminder_hours_before': 24,
|
||
},
|
||
{
|
||
'severity': 'medium',
|
||
'priority': 'medium',
|
||
'sla_hours': 24,
|
||
'reminder_hours_before': 12,
|
||
},
|
||
{
|
||
'severity': 'medium',
|
||
'priority': 'high',
|
||
'sla_hours': 12,
|
||
'reminder_hours_before': 6,
|
||
},
|
||
# High severity
|
||
{
|
||
'severity': 'high',
|
||
'priority': 'low',
|
||
'sla_hours': 24,
|
||
'reminder_hours_before': 12,
|
||
},
|
||
{
|
||
'severity': 'high',
|
||
'priority': 'medium',
|
||
'sla_hours': 12,
|
||
'reminder_hours_before': 6,
|
||
},
|
||
{
|
||
'severity': 'high',
|
||
'priority': 'high',
|
||
'sla_hours': 6,
|
||
'reminder_hours_before': 3,
|
||
},
|
||
# Critical severity
|
||
{
|
||
'severity': 'critical',
|
||
'priority': 'low',
|
||
'sla_hours': 12,
|
||
'reminder_hours_before': 6,
|
||
},
|
||
{
|
||
'severity': 'critical',
|
||
'priority': 'medium',
|
||
'sla_hours': 6,
|
||
'reminder_hours_before': 3,
|
||
},
|
||
{
|
||
'severity': 'critical',
|
||
'priority': 'high',
|
||
'sla_hours': 4,
|
||
'reminder_hours_before': 2,
|
||
'second_reminder_enabled': True,
|
||
'second_reminder_hours_before': 1,
|
||
},
|
||
]
|
||
|
||
created_count = 0
|
||
for config in sla_configs:
|
||
sla_config, created = ComplaintSLAConfig.objects.get_or_create(
|
||
hospital=hospital,
|
||
severity=config['severity'],
|
||
priority=config['priority'],
|
||
defaults=config
|
||
)
|
||
if created:
|
||
created_count += 1
|
||
print(f" ✅ Created: {sla_config.severity}/{sla_config.priority} - {sla_config.sla_hours}h")
|
||
else:
|
||
print(f" ℹ️ Existing: {sla_config.severity}/{sla_config.priority} - {sla_config.sla_hours}h")
|
||
|
||
print(f"\n📊 Total SLA configs: {ComplaintSLAConfig.objects.filter(hospital=hospital).count()}")
|
||
print(f"📝 Created: {created_count} new configs")
|
||
|
||
return True
|
||
|
||
|
||
def test_sla_deadline_calculation(hospital, department, user):
|
||
"""Test SLA deadline calculation for complaints."""
|
||
print("\n" + "=" * 80)
|
||
print("⏰ TEST 2: SLA DEADLINE CALCULATION")
|
||
print("=" * 80)
|
||
|
||
test_cases = [
|
||
{'severity': 'low', 'priority': 'low', 'expected_hours': 72},
|
||
{'severity': 'medium', 'priority': 'medium', 'expected_hours': 24},
|
||
{'severity': 'high', 'priority': 'high', 'expected_hours': 6},
|
||
{'severity': 'critical', 'priority': 'high', 'expected_hours': 4},
|
||
]
|
||
|
||
for i, test_case in enumerate(test_cases, 1):
|
||
print(f"\n📝 Test case {i}: {test_case['severity']}/{test_case['priority']}")
|
||
|
||
# Get SLA config
|
||
try:
|
||
sla_config = ComplaintSLAConfig.objects.get(
|
||
hospital=hospital,
|
||
severity=test_case['severity'],
|
||
priority=test_case['priority'],
|
||
is_active=True
|
||
)
|
||
sla_hours = sla_config.sla_hours
|
||
except ComplaintSLAConfig.DoesNotExist:
|
||
sla_hours = test_case['expected_hours'] # Fallback
|
||
|
||
print(f" SLA hours: {sla_hours}h")
|
||
|
||
# Create complaint
|
||
complaint = Complaint.objects.create(
|
||
hospital=hospital,
|
||
department=department,
|
||
title=f"Test SLA Complaint {i}",
|
||
description=f"Testing SLA deadline calculation for {test_case['severity']}/{test_case['priority']}",
|
||
severity=test_case['severity'],
|
||
priority=test_case['priority'],
|
||
status=ComplaintStatus.OPEN,
|
||
assigned_to=user
|
||
)
|
||
|
||
# Check deadline
|
||
deadline = complaint.due_at
|
||
now = timezone.now()
|
||
hours_until_deadline = (deadline - now).total_seconds() / 3600
|
||
|
||
print(f" Created at: {complaint.created_at.strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f" Due at: {deadline.strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f" Hours until deadline: {hours_until_deadline:.2f}h")
|
||
print(f" Expected: {test_case['expected_hours']}h")
|
||
|
||
# Verify deadline is correct (within 1 minute tolerance)
|
||
expected_deadline = complaint.created_at + timedelta(hours=sla_hours)
|
||
time_diff = abs((deadline - expected_deadline).total_seconds())
|
||
|
||
if time_diff < 60: # Within 1 minute
|
||
print(f" ✅ Deadline calculation CORRECT")
|
||
else:
|
||
print(f" ❌ Deadline calculation INCORRECT (diff: {time_diff:.2f}s)")
|
||
|
||
# Summary
|
||
total_complaints = Complaint.objects.filter(title__contains="Test SLA Complaint").count()
|
||
print(f"\n📊 Total test complaints created: {total_complaints}")
|
||
|
||
return True
|
||
|
||
|
||
def test_overdue_detection():
|
||
"""Test overdue complaint detection."""
|
||
print("\n" + "=" * 80)
|
||
print("⚠️ TEST 3: OVERDETECTION")
|
||
print("=" * 80)
|
||
|
||
# Get all open complaints
|
||
open_complaints = Complaint.objects.filter(
|
||
status=ComplaintStatus.OPEN,
|
||
title__contains="Test SLA Complaint"
|
||
)
|
||
|
||
print(f"📊 Open complaints: {open_complaints.count()}")
|
||
|
||
overdue_count = 0
|
||
not_overdue_count = 0
|
||
|
||
for complaint in open_complaints:
|
||
is_overdue = complaint.check_overdue()
|
||
hours_until = (complaint.due_at - timezone.now()).total_seconds() / 3600
|
||
|
||
if is_overdue:
|
||
overdue_count += 1
|
||
print(f"\n ❌ OVERDUE: {complaint.title}")
|
||
print(f" Due at: {complaint.due_at.strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f" Hours overdue: {-hours_until:.2f}h")
|
||
else:
|
||
not_overdue_count += 1
|
||
print(f"\n ✅ NOT OVERDUE: {complaint.title}")
|
||
print(f" Due at: {complaint.due_at.strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f" Hours remaining: {hours_until:.2f}h")
|
||
|
||
print(f"\n📊 Summary: {overdue_count} overdue, {not_overdue_count} not overdue")
|
||
|
||
return True
|
||
|
||
|
||
def test_escalation_rules(hospital, user):
|
||
"""Test escalation rule configuration and application."""
|
||
print("\n" + "=" * 80)
|
||
print("📈 TEST 4: ESCALATION RULES")
|
||
print("=" * 80)
|
||
|
||
# Clean up existing rules
|
||
EscalationRule.objects.filter(hospital=hospital).delete()
|
||
|
||
# Create escalation rules
|
||
escalation_rules = [
|
||
{
|
||
'name': 'Level 1 - Department Manager',
|
||
'description': 'Escalate to department manager',
|
||
'escalation_level': 1,
|
||
'max_escalation_level': 3,
|
||
'trigger_hours_overdue': 0, # Immediate
|
||
'escalate_to_role': 'department_manager',
|
||
'severity_filter': 'high',
|
||
'order': 1,
|
||
},
|
||
{
|
||
'name': 'Level 2 - Hospital Admin',
|
||
'description': 'Escalate to hospital admin',
|
||
'escalation_level': 2,
|
||
'max_escalation_level': 3,
|
||
'trigger_hours_overdue': 4, # 4 hours after overdue
|
||
'escalate_to_role': 'hospital_admin',
|
||
'severity_filter': 'high',
|
||
'order': 2,
|
||
},
|
||
{
|
||
'name': 'Level 3 - CEO',
|
||
'description': 'Escalate to CEO',
|
||
'escalation_level': 3,
|
||
'max_escalation_level': 3,
|
||
'trigger_hours_overdue': 8, # 8 hours after overdue
|
||
'escalate_to_role': 'ceo',
|
||
'severity_filter': 'critical',
|
||
'order': 3,
|
||
},
|
||
]
|
||
|
||
created_count = 0
|
||
for rule_data in escalation_rules:
|
||
rule, created = EscalationRule.objects.get_or_create(
|
||
hospital=hospital,
|
||
name=rule_data['name'],
|
||
defaults=rule_data
|
||
)
|
||
if created:
|
||
created_count += 1
|
||
print(f" ✅ Created: {rule.name} (Level {rule.escalation_level})")
|
||
print(f" Role: {rule.escalate_to_role}")
|
||
print(f" Trigger: {rule.trigger_hours_overdue}h after overdue")
|
||
else:
|
||
print(f" ℹ️ Existing: {rule.name}")
|
||
|
||
print(f"\n📊 Total escalation rules: {EscalationRule.objects.filter(hospital=hospital).count()}")
|
||
print(f"📝 Created: {created_count} new rules")
|
||
|
||
# Test escalation logic on a complaint
|
||
print("\n🧪 Testing escalation on a high-severity complaint...")
|
||
|
||
# Create a high-severity complaint
|
||
complaint = Complaint.objects.create(
|
||
hospital=hospital,
|
||
title="Test Escalation Complaint",
|
||
description="Testing escalation rules",
|
||
severity='high',
|
||
priority='high',
|
||
status=ComplaintStatus.OPEN,
|
||
assigned_to=user
|
||
)
|
||
|
||
# Manually set it to be overdue
|
||
complaint.due_at = timezone.now() - timedelta(hours=2) # 2 hours ago
|
||
complaint.save(update_fields=['due_at'])
|
||
|
||
print(f" Complaint: {complaint.title}")
|
||
print(f" Severity: {complaint.severity}")
|
||
print(f" Due at: {complaint.due_at.strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f" Hours overdue: 2h")
|
||
|
||
# Check applicable rules
|
||
applicable_rules = EscalationRule.objects.filter(
|
||
hospital=hospital,
|
||
is_active=True,
|
||
trigger_on_overdue=True,
|
||
escalation_level=1,
|
||
severity_filter=complaint.severity
|
||
).order_by('order')
|
||
|
||
print(f"\n Applicable rules: {applicable_rules.count()}")
|
||
for rule in applicable_rules:
|
||
print(f" - {rule.name}")
|
||
print(f" Level: {rule.escalation_level}")
|
||
print(f" Trigger: {rule.trigger_hours_overdue}h after overdue")
|
||
print(f" Should trigger: {2 >= rule.trigger_hours_overdue}")
|
||
|
||
return True
|
||
|
||
|
||
def test_explanation_sla(hospital):
|
||
"""Test explanation SLA configuration."""
|
||
print("\n" + "=" * 80)
|
||
print("💬 TEST 5: EXPLANATION SLA")
|
||
print("=" * 80)
|
||
|
||
# Clean up existing configs
|
||
ExplanationSLAConfig.objects.filter(hospital=hospital).delete()
|
||
|
||
# Create explanation SLA config
|
||
config_data = {
|
||
'response_hours': 48,
|
||
'reminder_hours_before': 12,
|
||
'auto_escalate_enabled': True,
|
||
'escalation_hours_overdue': 24,
|
||
'max_escalation_levels': 3,
|
||
}
|
||
|
||
sla_config, created = ExplanationSLAConfig.objects.get_or_create(
|
||
hospital=hospital,
|
||
defaults=config_data
|
||
)
|
||
|
||
if created:
|
||
print(f" ✅ Created Explanation SLA config")
|
||
print(f" Response time: {sla_config.response_hours}h")
|
||
print(f" Reminder: {sla_config.reminder_hours_before}h before deadline")
|
||
print(f" Auto-escalate: {sla_config.auto_escalate_enabled}")
|
||
print(f" Escalate after: {sla_config.escalation_hours_overdue}h overdue")
|
||
print(f" Max levels: {sla_config.max_escalation_levels}")
|
||
else:
|
||
print(f" ℹ️ Existing Explanation SLA config found")
|
||
|
||
return True
|
||
|
||
|
||
def test_sla_threshold(hospital):
|
||
"""Test complaint threshold configuration."""
|
||
print("\n" + "=" * 80)
|
||
print("🎯 TEST 6: COMPLAINT THRESHOLDS")
|
||
print("=" * 80)
|
||
|
||
# Clean up existing thresholds
|
||
ComplaintThreshold.objects.filter(hospital=hospital).delete()
|
||
|
||
# Create thresholds
|
||
thresholds = [
|
||
{
|
||
'threshold_type': 'resolution_survey_score',
|
||
'threshold_value': 50,
|
||
'comparison_operator': 'lt', # Less than 50%
|
||
'action_type': 'create_px_action',
|
||
},
|
||
]
|
||
|
||
created_count = 0
|
||
for threshold_data in thresholds:
|
||
threshold, created = ComplaintThreshold.objects.get_or_create(
|
||
hospital=hospital,
|
||
threshold_type=threshold_data['threshold_type'],
|
||
defaults=threshold_data
|
||
)
|
||
if created:
|
||
created_count += 1
|
||
print(f" ✅ Created threshold")
|
||
print(f" Type: {threshold.threshold_type}")
|
||
print(f" Condition: {threshold.comparison_operator} {threshold.threshold_value}")
|
||
print(f" Action: {threshold.action_type}")
|
||
|
||
# Test threshold checking
|
||
test_values = [30, 50, 70]
|
||
for value in test_values:
|
||
result = threshold.check_threshold(value)
|
||
print(f" Value {value}: {'✅ BREACHED' if result else '❌ NOT breached'}")
|
||
else:
|
||
print(f" ℹ️ Existing threshold found")
|
||
|
||
print(f"\n📊 Total thresholds: {ComplaintThreshold.objects.filter(hospital=hospital).count()}")
|
||
|
||
return True
|
||
|
||
|
||
def test_sla_summary():
|
||
"""Print SLA system summary."""
|
||
print("\n" + "=" * 80)
|
||
print("📊 SLA SYSTEM SUMMARY")
|
||
print("=" * 80)
|
||
|
||
hospital = Hospital.objects.first()
|
||
if not hospital:
|
||
return
|
||
|
||
# Count configurations
|
||
sla_configs = ComplaintSLAConfig.objects.filter(hospital=hospital, is_active=True)
|
||
escalation_rules = EscalationRule.objects.filter(hospital=hospital, is_active=True)
|
||
thresholds = ComplaintThreshold.objects.filter(hospital=hospital, is_active=True)
|
||
explanation_configs = ExplanationSLAConfig.objects.filter(hospital=hospital, is_active=True)
|
||
|
||
print(f"\n Hospital: {hospital.name}")
|
||
print(f"\n SLA Configurations: {sla_configs.count()}")
|
||
for config in sla_configs:
|
||
print(f" - {config.severity}/{config.priority}: {config.sla_hours}h")
|
||
|
||
print(f"\n Escalation Rules: {escalation_rules.count()}")
|
||
for rule in escalation_rules:
|
||
print(f" - {rule.name} (Level {rule.escalation_level})")
|
||
|
||
print(f"\n Thresholds: {thresholds.count()}")
|
||
for threshold in thresholds:
|
||
print(f" - {threshold.threshold_type}: {threshold.comparison_operator} {threshold.threshold_value}")
|
||
|
||
print(f"\n Explanation SLA: {explanation_configs.count()}")
|
||
for config in explanation_configs:
|
||
print(f" - {config.response_hours}h to respond")
|
||
|
||
# Count complaints by status
|
||
print(f"\n Complaints by status:")
|
||
complaints = Complaint.objects.filter(hospital=hospital)
|
||
for status in ComplaintStatus.choices:
|
||
count = complaints.filter(status=status[0]).count()
|
||
print(f" - {status[1]}: {count}")
|
||
|
||
# Count overdue complaints
|
||
overdue_count = complaints.filter(is_overdue=True).count()
|
||
print(f"\n Overdue complaints: {overdue_count}")
|
||
|
||
return True
|
||
|
||
|
||
def main():
|
||
"""Main test function."""
|
||
print("\n" + "=" * 80)
|
||
print("🚀 COMPREHENSIVE SLA SYSTEM TEST")
|
||
print("=" * 80)
|
||
print(f"📅 Started at: {timezone.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
|
||
# Setup test data
|
||
hospital, department, user = setup_test_data()
|
||
if not hospital:
|
||
return
|
||
|
||
# Run tests
|
||
try:
|
||
# Test 1: SLA Configuration
|
||
test_sla_configuration(hospital)
|
||
|
||
# Test 2: SLA Deadline Calculation
|
||
test_sla_deadline_calculation(hospital, department, user)
|
||
|
||
# Test 3: Overdue Detection
|
||
test_overdue_detection()
|
||
|
||
# Test 4: Escalation Rules
|
||
test_escalation_rules(hospital, user)
|
||
|
||
# Test 5: Explanation SLA
|
||
test_explanation_sla(hospital)
|
||
|
||
# Test 6: SLA Thresholds
|
||
test_sla_threshold(hospital)
|
||
|
||
# Summary
|
||
test_sla_summary()
|
||
|
||
print("\n" + "=" * 80)
|
||
print("✅ ALL TESTS COMPLETED")
|
||
print("=" * 80)
|
||
print(f"📅 Completed at: {timezone.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ ERROR: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|