295 lines
12 KiB
Python
295 lines
12 KiB
Python
#!/usr/bin/env python
|
|
"""
|
|
Scenario 1: Successful Explanation Submission
|
|
|
|
Tests the happy path where a staff member submits their explanation
|
|
before the SLA deadline. No escalation should occur.
|
|
|
|
Time Compression: 1 second = 1 hour
|
|
SLA Deadline: 10 hours (10 seconds)
|
|
"""
|
|
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 apps.complaints.models import Complaint, ComplaintCategory, ComplaintExplanation, ComplaintSLAConfig
|
|
from apps.organizations.models import Hospital, Department, Staff
|
|
from apps.accounts.models import User, UserManager
|
|
|
|
|
|
class Scenario1SuccessfulExplanation(ScenarioTestBase):
|
|
"""Test successful explanation submission before SLA deadline"""
|
|
|
|
def run(self):
|
|
"""Run the complete scenario"""
|
|
self.print_header("SCENARIO 1: SUCCESSFUL EXPLANATION SUBMISSION")
|
|
|
|
successful_steps = 0
|
|
total_steps = 7
|
|
|
|
try:
|
|
# Step 1: Setup
|
|
self.print_step("Setting up test environment", 0)
|
|
hospital, department, staff = self.setup_environment()
|
|
successful_steps += 1
|
|
|
|
# Step 2: Create complaint
|
|
self.print_step("Creating complaint", 2) # Wait 2s (simulates 2 hours)
|
|
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 (pending, no reminder, no overdue)
|
|
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: Staff submits explanation (before deadline)
|
|
self.print_step("Staff submits explanation (before deadline)", 3) # Wait 3s (simulates 3 hours)
|
|
self.submit_explanation(explanation, "This is my explanation about the complaint.")
|
|
successful_steps += 1
|
|
|
|
# Step 6: Verify explanation submitted
|
|
self.print_step("Verifying explanation submitted successfully", 0)
|
|
if self.verify_explanation_state(explanation, 'submitted'):
|
|
successful_steps += 1
|
|
else:
|
|
self.print_error("Submission verification failed")
|
|
|
|
# Step 7: Verify no escalation occurred
|
|
self.print_step("Verifying no escalation occurred", 0)
|
|
explanation.refresh_from_db()
|
|
if explanation.escalated_to_manager is None:
|
|
self.print_success("No escalation occurred (as expected)")
|
|
successful_steps += 1
|
|
else:
|
|
self.print_error("Unexpected escalation occurred")
|
|
|
|
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 staff"""
|
|
# 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 or get staff member
|
|
staff, created = Staff.objects.get_or_create(
|
|
email=f"omar.harbi@{hospital.name.lower().replace(' ', '')}.test",
|
|
defaults={
|
|
'hospital': hospital,
|
|
'department': department,
|
|
'first_name': "Omar",
|
|
'last_name': "Al-Harbi",
|
|
'first_name_ar': "عمر",
|
|
'last_name_ar': "الحربي",
|
|
'job_title': "Nurse",
|
|
'specialization': "Patient Care",
|
|
'phone': "+966513333333",
|
|
'status': 'active'
|
|
}
|
|
)
|
|
if created:
|
|
self.print_success(f"Created staff member: {staff.get_full_name()}")
|
|
else:
|
|
self.print_info(f"Using existing staff member: {staff.get_full_name()}")
|
|
|
|
# Configure SLA
|
|
self.create_explanation_sla_config(
|
|
hospital=hospital,
|
|
response_hours=10, # 10 hours = 10 seconds in test
|
|
reminder_hours_before=5, # Reminder 5 hours before
|
|
auto_escalate_enabled=True,
|
|
escalation_hours_overdue=0
|
|
)
|
|
|
|
self.create_complaint_sla_config(
|
|
hospital=hospital,
|
|
severity='medium',
|
|
priority='medium',
|
|
sla_hours=72,
|
|
reminder_hours_before=24
|
|
)
|
|
|
|
return hospital, department, staff
|
|
|
|
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="Poor response time in emergency",
|
|
description="Nurse Omar Al-Harbi took too long to respond to my emergency situation. The delay was unacceptable and caused unnecessary distress.",
|
|
severity='medium',
|
|
priority='medium',
|
|
source_id=None,
|
|
status='open',
|
|
contact_name="Test Patient",
|
|
contact_phone="+966599999999",
|
|
contact_email="patient@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='', # Will be filled by staff
|
|
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, # Will be set by email task
|
|
sla_due_at=None # Will be set by email task
|
|
)
|
|
|
|
self.print_success(f"Created explanation request for {staff.get_full_name()}")
|
|
self.print_info(f" Token: {explanation.token[:20]}...")
|
|
|
|
# Send email (simulate by calling the task directly)
|
|
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()
|
|
self.print_info(f" SLA Due At: {explanation.sla_due_at}")
|
|
hours_until_due = (explanation.sla_due_at - timezone.now()).total_seconds() / 3600
|
|
self.print_info(f" Hours until deadline: {hours_until_due:.1f}")
|
|
else:
|
|
self.print_error(f"Failed to send email: {result.get('reason', 'Unknown')}")
|
|
|
|
return explanation
|
|
|
|
def submit_explanation(self, explanation, explanation_text):
|
|
"""Simulate staff submitting explanation"""
|
|
# Update explanation
|
|
explanation.explanation = explanation_text
|
|
explanation.is_used = True
|
|
explanation.responded_at = timezone.now()
|
|
explanation.save(update_fields=['explanation', 'is_used', 'responded_at'])
|
|
|
|
self.print_success("Staff submitted explanation")
|
|
self.print_info(f" Response time: {explanation.responded_at}")
|
|
self.print_info(f" SLA deadline: {explanation.sla_due_at}")
|
|
|
|
# Check if submitted before deadline
|
|
if explanation.responded_at < explanation.sla_due_at:
|
|
hours_before_deadline = (explanation.sla_due_at - explanation.responded_at).total_seconds() / 3600
|
|
self.print_success(f"Submitted {hours_before_deadline:.1f} hours BEFORE deadline ✓")
|
|
else:
|
|
hours_after_deadline = (explanation.responded_at - explanation.sla_due_at).total_seconds() / 3600
|
|
self.print_warning(f"Submitted {hours_after_deadline:.1f} hours AFTER deadline")
|
|
|
|
return explanation
|
|
|
|
|
|
def main():
|
|
"""Main entry point"""
|
|
test = Scenario1SuccessfulExplanation(time_compression_ratio=1) # 1s = 1 hour
|
|
success = test.run()
|
|
|
|
if success:
|
|
print("\n✓ Scenario 1 completed successfully!")
|
|
sys.exit(0)
|
|
else:
|
|
print("\n✗ Scenario 1 failed!")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|