#!/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()