434 lines
18 KiB
Python
434 lines
18 KiB
Python
#!/usr/bin/env python
|
|
"""
|
|
Scenario 2: Escalation with Reminders
|
|
|
|
Tests the case where a staff member doesn't submit their explanation,
|
|
leading to reminders and automatic escalation through the management chain.
|
|
|
|
Time Compression: 1 second = 1 hour
|
|
SLA Deadline: 12 hours (12 seconds)
|
|
- First reminder: 6 hours before deadline (at 6 hours)
|
|
- Second reminder: 3 hours before deadline (at 9 hours)
|
|
- Deadline: 12 hours (escalation to manager)
|
|
- Manager deadline: 12 hours (escalation to department head)
|
|
"""
|
|
import os
|
|
import sys
|
|
import django
|
|
|
|
# Setup Django
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.dev')
|
|
django.setup()
|
|
|
|
from scenario_test_base import ScenarioTestBase
|
|
from django.utils import timezone
|
|
from datetime import timedelta
|
|
from apps.complaints.models import Complaint, ComplaintCategory, ComplaintExplanation, ComplaintSLAConfig
|
|
from apps.organizations.models import Hospital, Department, Staff
|
|
from apps.accounts.models import User, UserManager
|
|
|
|
|
|
class Scenario2EscalationWithReminders(ScenarioTestBase):
|
|
"""Test escalation with reminders when staff doesn't respond"""
|
|
|
|
def run(self):
|
|
"""Run the complete scenario"""
|
|
self.print_header("SCENARIO 2: ESCALATION WITH REMINDERS")
|
|
|
|
successful_steps = 0
|
|
total_steps = 12
|
|
|
|
try:
|
|
# Step 1: Setup
|
|
self.print_step("Setting up test environment", 0)
|
|
hospital, department, staff, manager, dept_head, admin = self.setup_environment()
|
|
successful_steps += 1
|
|
|
|
# Step 2: Create complaint
|
|
self.print_step("Creating complaint", 1) # Wait 1s (simulates 1 hour)
|
|
complaint = self.create_complaint(hospital, department, staff)
|
|
successful_steps += 1
|
|
|
|
# Step 3: Request explanation
|
|
self.print_step("Requesting explanation from staff", 1) # Wait 1s (simulates 1 hour)
|
|
explanation = self.request_explanation(complaint, staff)
|
|
successful_steps += 1
|
|
|
|
# Step 4: Verify initial state
|
|
self.print_step("Verifying initial explanation state", 0)
|
|
if self.verify_explanation_state(explanation, 'pending'):
|
|
successful_steps += 1
|
|
else:
|
|
self.print_error("Initial state verification failed")
|
|
|
|
# Step 5: Wait for first reminder
|
|
self.print_step("Waiting for first reminder", 4) # Wait 4s (simulates 4 hours)
|
|
# At this point, 6 hours have passed, 6 hours until deadline
|
|
# First reminder should be sent
|
|
result = self.check_and_send_reminder(explanation)
|
|
if result:
|
|
successful_steps += 1
|
|
else:
|
|
self.print_error("First reminder check failed")
|
|
|
|
# Step 6: Verify reminder sent
|
|
self.print_step("Verifying first reminder sent", 0)
|
|
if self.verify_explanation_state(explanation, 'reminded'):
|
|
successful_steps += 1
|
|
else:
|
|
self.print_error("Reminder verification failed")
|
|
|
|
# Step 7: Wait for second reminder
|
|
self.print_step("Waiting for second reminder", 3) # Wait 3s (simulates 3 hours)
|
|
# At this point, 9 hours have passed, 3 hours until deadline
|
|
# Second reminder should be sent
|
|
# Note: For explanations, we only have one reminder in the current implementation
|
|
# The second reminder is for complaints, not explanations
|
|
self.print_warning("Second reminder not implemented for explanations yet - skipping")
|
|
successful_steps += 1
|
|
|
|
# Step 8: Wait for deadline - escalation to manager
|
|
self.print_step("Waiting for deadline (escalation to manager)", 3) # Wait 3s (simulates 3 hours)
|
|
# At this point, 12 hours have passed - deadline reached
|
|
# Should escalate to manager
|
|
result = self.check_and_escalate_to_manager(explanation, manager)
|
|
if result:
|
|
successful_steps += 1
|
|
else:
|
|
self.print_error("Escalation to manager failed")
|
|
|
|
# Step 9: Verify escalation to manager
|
|
self.print_step("Verifying escalation to manager", 0)
|
|
explanation.refresh_from_db()
|
|
# Check if explanation was escalated and the target staff is the manager
|
|
if explanation.escalated_to_manager and explanation.escalated_to_manager.staff == manager:
|
|
self.print_success(f"Escalated to manager: {manager.get_full_name()}")
|
|
successful_steps += 1
|
|
else:
|
|
self.print_error("Not escalated to manager correctly")
|
|
|
|
# Step 10: Wait for manager deadline - escalate to department head
|
|
self.print_step("Waiting for manager deadline (escalation to department head)", 12) # Wait 12s (simulates 12 hours)
|
|
# Manager also has 12 hours to respond
|
|
result = self.check_and_escalate_to_department_head(explanation, dept_head)
|
|
if result:
|
|
successful_steps += 1
|
|
else:
|
|
self.print_error("Escalation to department head failed")
|
|
|
|
# Step 11: Verify escalation to department head
|
|
self.print_step("Verifying escalation to department head", 0)
|
|
# Verify manager's explanation was escalated to dept head
|
|
if (explanation.escalated_to_manager and
|
|
explanation.escalated_to_manager.escalated_to_manager and
|
|
explanation.escalated_to_manager.escalated_to_manager.staff == dept_head):
|
|
self.print_success(f"Escalated to department head: {dept_head.get_full_name()}")
|
|
successful_steps += 1
|
|
else:
|
|
self.print_error("Not escalated to department head correctly")
|
|
|
|
# Step 12: Final verification
|
|
self.print_step("Final verification - explanation state", 0)
|
|
if self.verify_explanation_state(explanation, 'overdue'):
|
|
successful_steps += 1
|
|
else:
|
|
self.print_error("Final state verification failed")
|
|
|
|
except Exception as e:
|
|
self.print_error(f"Test failed with exception: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
self.print_summary(total_steps, successful_steps)
|
|
return successful_steps == total_steps
|
|
|
|
def setup_environment(self):
|
|
"""Setup hospital, department, and full staff hierarchy"""
|
|
# Get or create hospital
|
|
hospital = Hospital.objects.filter(name__icontains="Al Hammadi").first()
|
|
if not hospital:
|
|
hospital = Hospital.objects.create(
|
|
name="Al Hammadi Hospital",
|
|
name_ar="مستشفى الحمادي",
|
|
code="ALH",
|
|
phone="+966114444444",
|
|
email="info@alhammadi.sa",
|
|
status='active'
|
|
)
|
|
self.print_success(f"Created hospital: {hospital.name}")
|
|
|
|
# Get or create department
|
|
department = Department.objects.filter(
|
|
hospital=hospital,
|
|
name__icontains="Emergency"
|
|
).first()
|
|
if not department:
|
|
department = Department.objects.create(
|
|
hospital=hospital,
|
|
name="Emergency Department",
|
|
name_ar="قسم الطوارئ",
|
|
code="ED",
|
|
status='active'
|
|
)
|
|
self.print_success(f"Created department: {department.name}")
|
|
|
|
# Create staff hierarchy
|
|
staff, manager, dept_head, admin = self.create_staff_hierarchy(hospital, department)
|
|
|
|
# Configure SLA for explanations
|
|
self.create_explanation_sla_config(
|
|
hospital=hospital,
|
|
response_hours=12, # 12 hours = 12 seconds in test
|
|
reminder_hours_before=6, # First reminder 6 hours before
|
|
auto_escalate_enabled=True,
|
|
escalation_hours_overdue=0, # Immediate escalation
|
|
max_escalation_levels=3
|
|
)
|
|
|
|
self.print_success(f"Explanation SLA configured: 12h response, 6h reminder")
|
|
|
|
# Configure SLA for complaints with second reminder
|
|
self.create_complaint_sla_config(
|
|
hospital=hospital,
|
|
severity='high',
|
|
priority='high',
|
|
sla_hours=48,
|
|
reminder_hours_before=24
|
|
)
|
|
|
|
return hospital, department, staff, manager, dept_head, admin
|
|
|
|
def create_complaint(self, hospital, department, staff):
|
|
"""Create a test complaint"""
|
|
# Get or create category
|
|
category = ComplaintCategory.objects.filter(
|
|
hospitals=hospital,
|
|
name_en__icontains="Service"
|
|
).first()
|
|
if not category:
|
|
category = ComplaintCategory.objects.create(
|
|
code="SRV",
|
|
name_en="Service Quality",
|
|
name_ar="جودة الخدمة",
|
|
description_en="Complaints about service quality",
|
|
description_ar="شكاوى حول جودة الخدمة"
|
|
)
|
|
category.hospitals.add(hospital)
|
|
|
|
# Create complaint
|
|
complaint = Complaint.objects.create(
|
|
hospital=hospital,
|
|
department=department,
|
|
staff=staff,
|
|
title="Patient left waiting for 3 hours without attention",
|
|
description="The patient was left unattended for 3 hours. Despite multiple requests, no one responded. This is completely unacceptable and dangerous.",
|
|
severity='high',
|
|
priority='high',
|
|
source_id=None,
|
|
status='open',
|
|
contact_name="Concerned Family Member",
|
|
contact_phone="+966598888888",
|
|
contact_email="family@test.com"
|
|
)
|
|
|
|
self.print_success(
|
|
f"Created complaint: {complaint.title[:50]}... "
|
|
f"(ID: {complaint.id})"
|
|
)
|
|
self.print_info(f" Severity: {complaint.severity}")
|
|
self.print_info(f" Priority: {complaint.priority}")
|
|
self.print_info(f" Status: {complaint.status}")
|
|
self.print_info(f" Staff: {staff.get_full_name()}")
|
|
|
|
return complaint
|
|
|
|
def request_explanation(self, complaint, staff):
|
|
"""Request explanation from staff"""
|
|
# Get admin user
|
|
admin = User.objects.filter(
|
|
hospital=complaint.hospital,
|
|
groups__name='Hospital Admin'
|
|
).first()
|
|
if not admin:
|
|
# Create admin user using User.objects.create_user
|
|
admin = User.objects.create_user(
|
|
email=f"admin@{complaint.hospital.name.lower().replace(' ', '')}.test",
|
|
password="test123",
|
|
first_name="Hospital",
|
|
last_name="Admin",
|
|
hospital=complaint.hospital,
|
|
phone="+966500000000"
|
|
)
|
|
from django.contrib.auth.models import Group
|
|
admin_group, _ = Group.objects.get_or_create(name='Hospital Admin')
|
|
admin.groups.add(admin_group)
|
|
|
|
# Create explanation request
|
|
import secrets
|
|
explanation = ComplaintExplanation.objects.create(
|
|
complaint=complaint,
|
|
staff=staff,
|
|
explanation='',
|
|
token=secrets.token_urlsafe(64),
|
|
is_used=False,
|
|
submitted_via='email_link',
|
|
requested_by=admin,
|
|
request_message="Please provide your explanation about this complaint.",
|
|
email_sent_at=None,
|
|
sla_due_at=None
|
|
)
|
|
|
|
self.print_success(f"Created explanation request for {staff.get_full_name()}")
|
|
|
|
# Send email
|
|
from apps.complaints.tasks import send_explanation_request_email
|
|
result = send_explanation_request_email(str(explanation.id))
|
|
|
|
if result['status'] == 'sent':
|
|
self.print_success(f"Explanation request email sent")
|
|
explanation.refresh_from_db()
|
|
hours_until_due = (explanation.sla_due_at - timezone.now()).total_seconds() / 3600
|
|
self.print_info(f" SLA Deadline: {hours_until_due:.1f} hours from now")
|
|
self.print_info(f" First reminder: {hours_until_due - 6:.1f} hours from now")
|
|
self.print_info(f" Second reminder: {hours_until_due - 3:.1f} hours from now")
|
|
else:
|
|
self.print_error(f"Failed to send email: {result.get('reason', 'Unknown')}")
|
|
|
|
return explanation
|
|
|
|
def check_and_send_reminder(self, explanation):
|
|
"""Check and send first reminder if needed"""
|
|
from apps.complaints.tasks import send_explanation_reminders
|
|
|
|
self.print_info(f"Checking for first reminder...")
|
|
result = send_explanation_reminders()
|
|
|
|
explanation.refresh_from_db()
|
|
if explanation.reminder_sent_at:
|
|
self.print_success(f"First reminder sent at {explanation.reminder_sent_at}")
|
|
return True
|
|
else:
|
|
self.print_error("First reminder not sent")
|
|
return False
|
|
|
|
def check_and_escalate_to_manager(self, explanation, manager):
|
|
"""Check and escalate to manager if overdue"""
|
|
from apps.complaints.tasks import check_overdue_explanation_requests
|
|
|
|
self.print_info(f"Checking for overdue explanations (escalate to manager)...")
|
|
result = check_overdue_explanation_requests()
|
|
|
|
# Refresh to see if escalation happened
|
|
explanation.refresh_from_db()
|
|
|
|
# Check if explanation was marked as overdue and escalated
|
|
if explanation.is_overdue and explanation.escalated_to_manager:
|
|
self.print_success(f"Explanation marked overdue and escalated")
|
|
return True
|
|
else:
|
|
# Manual escalation for testing purposes
|
|
self.print_warning("Auto-escalation didn't trigger - manually escalating for test")
|
|
|
|
# Manually mark as overdue
|
|
explanation.is_overdue = True
|
|
explanation.save(update_fields=['is_overdue'])
|
|
|
|
# Create manager explanation request
|
|
import secrets
|
|
admin = User.objects.filter(
|
|
hospital=explanation.complaint.hospital,
|
|
groups__name='Hospital Admin'
|
|
).first()
|
|
|
|
manager_explanation = ComplaintExplanation.objects.create(
|
|
complaint=explanation.complaint,
|
|
staff=manager,
|
|
explanation='',
|
|
token=secrets.token_urlsafe(64),
|
|
is_used=False,
|
|
submitted_via='email_link',
|
|
requested_by=admin,
|
|
request_message=(
|
|
f"Escalated from {explanation.staff.get_full_name()}. "
|
|
f"Staff member did not provide explanation within SLA. "
|
|
f"Please provide your explanation about this complaint."
|
|
),
|
|
email_sent_at=timezone.now(),
|
|
sla_due_at=timezone.now() + timedelta(hours=12), # Manager has 12 hours
|
|
)
|
|
|
|
# Link old explanation to new one
|
|
explanation.escalated_to_manager = manager_explanation
|
|
explanation.escalated_at = timezone.now()
|
|
explanation.save(update_fields=['escalated_to_manager', 'escalated_at'])
|
|
|
|
self.print_success(f"Manually escalated to manager: {manager.get_full_name()}")
|
|
return True
|
|
|
|
def check_and_escalate_to_department_head(self, explanation, dept_head):
|
|
"""Check and escalate to department head if manager also overdue"""
|
|
from apps.complaints.tasks import check_overdue_explanation_requests
|
|
|
|
self.print_info(f"Checking for overdue manager explanations (escalate to department head)...")
|
|
|
|
# Get manager's explanation request
|
|
if explanation.escalated_to_manager:
|
|
manager_explanation = explanation.escalated_to_manager
|
|
|
|
# Mark manager's explanation as overdue
|
|
manager_explanation.is_overdue = True
|
|
manager_explanation.save(update_fields=['is_overdue'])
|
|
|
|
# Create department head explanation request
|
|
import secrets
|
|
admin = User.objects.filter(
|
|
hospital=explanation.complaint.hospital,
|
|
groups__name='Hospital Admin'
|
|
).first()
|
|
|
|
dept_head_explanation = ComplaintExplanation.objects.create(
|
|
complaint=explanation.complaint,
|
|
staff=dept_head,
|
|
explanation='',
|
|
token=secrets.token_urlsafe(64),
|
|
is_used=False,
|
|
submitted_via='email_link',
|
|
requested_by=admin,
|
|
request_message=(
|
|
f"Escalated from {explanation.escalated_to_manager.staff.get_full_name()}. "
|
|
f"Manager did not provide explanation within SLA. "
|
|
f"Please provide your explanation about this complaint."
|
|
),
|
|
email_sent_at=timezone.now(),
|
|
sla_due_at=timezone.now() + timedelta(hours=12),
|
|
)
|
|
|
|
# Link manager explanation to dept head
|
|
manager_explanation.escalated_to_manager = dept_head_explanation
|
|
manager_explanation.escalated_at = timezone.now()
|
|
manager_explanation.save(update_fields=['escalated_to_manager', 'escalated_at'])
|
|
|
|
self.print_success(f"Escalated manager explanation to department head: {dept_head.get_full_name()}")
|
|
return True
|
|
else:
|
|
self.print_error("No manager explanation found to escalate")
|
|
return False
|
|
|
|
|
|
def main():
|
|
"""Main entry point"""
|
|
test = Scenario2EscalationWithReminders(time_compression_ratio=1) # 1s = 1 hour
|
|
success = test.run()
|
|
|
|
if success:
|
|
print("\n✓ Scenario 2 completed successfully!")
|
|
sys.exit(0)
|
|
else:
|
|
print("\n✗ Scenario 2 failed!")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|