544 lines
18 KiB
Python
544 lines
18 KiB
Python
"""
|
|
Unit tests for Advanced Queue Engine (Phase 11)
|
|
Tests for AdvancedQueueEngine and queue management algorithms
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import datetime, timedelta
|
|
from django.test import TestCase
|
|
from django.utils import timezone
|
|
from django.contrib.auth import get_user_model
|
|
|
|
from appointments.queue.queue_engine import AdvancedQueueEngine
|
|
from appointments.models import (
|
|
WaitingQueue, QueueEntry, QueueConfiguration, QueueMetrics,
|
|
AppointmentRequest
|
|
)
|
|
from patients.models import PatientProfile
|
|
from core.models import Tenant
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class AdvancedQueueEngineTestCase(TestCase):
|
|
"""Test cases for AdvancedQueueEngine class"""
|
|
|
|
def setUp(self):
|
|
"""Set up test data"""
|
|
# Create tenant
|
|
self.tenant = Tenant.objects.create(
|
|
name="Test Hospital",
|
|
display_name="Test Hospital",
|
|
address_line1="123 Test St",
|
|
city="Test City",
|
|
state="Test State",
|
|
postal_code="12345",
|
|
phone_number="+966501234567",
|
|
email="test@hospital.com"
|
|
)
|
|
|
|
# Create queue
|
|
self.queue = WaitingQueue.objects.create(
|
|
tenant=self.tenant,
|
|
name="Test Queue",
|
|
queue_type="GENERAL",
|
|
specialty="FAMILY_MEDICINE",
|
|
max_queue_size=50,
|
|
average_service_time_minutes=15,
|
|
is_active=True
|
|
)
|
|
|
|
# Create queue configuration
|
|
self.config = QueueConfiguration.objects.create(
|
|
queue=self.queue,
|
|
use_dynamic_positioning=True,
|
|
priority_weight=0.5,
|
|
wait_time_weight=0.3,
|
|
appointment_time_weight=0.2
|
|
)
|
|
|
|
# Create patients
|
|
self.patients = []
|
|
for i in range(5):
|
|
patient = PatientProfile.objects.create(
|
|
tenant=self.tenant,
|
|
first_name=f"Patient{i}",
|
|
last_name="Test",
|
|
date_of_birth=datetime(1990, 1, 1).date(),
|
|
gender="M",
|
|
mrn=f"TEST00{i}"
|
|
)
|
|
self.patients.append(patient)
|
|
|
|
# Initialize engine
|
|
self.engine = AdvancedQueueEngine(self.queue)
|
|
|
|
def test_engine_initialization(self):
|
|
"""Test AdvancedQueueEngine initialization"""
|
|
self.assertIsNotNone(self.engine)
|
|
self.assertEqual(self.engine.queue, self.queue)
|
|
self.assertEqual(self.engine.config, self.config)
|
|
|
|
def test_add_to_queue(self):
|
|
"""Test adding patient to queue"""
|
|
patient = self.patients[0]
|
|
|
|
entry = self.engine.add_to_queue(
|
|
patient=patient,
|
|
priority_score=1.0
|
|
)
|
|
|
|
# Entry should be created
|
|
self.assertIsNotNone(entry)
|
|
self.assertEqual(entry.patient, patient)
|
|
self.assertEqual(entry.queue, self.queue)
|
|
self.assertEqual(entry.status, 'WAITING')
|
|
|
|
# Position should be assigned
|
|
self.assertIsNotNone(entry.queue_position)
|
|
|
|
def test_calculate_optimal_position(self):
|
|
"""Test optimal position calculation"""
|
|
# Add some patients first
|
|
for i, patient in enumerate(self.patients[:3]):
|
|
self.engine.add_to_queue(
|
|
patient=patient,
|
|
priority_score=float(i + 1)
|
|
)
|
|
|
|
# Calculate position for new patient with high priority
|
|
position = self.engine.calculate_optimal_position(
|
|
patient=self.patients[3],
|
|
priority_score=5.0
|
|
)
|
|
|
|
# Should be positioned near the front
|
|
self.assertLessEqual(position, 2)
|
|
|
|
def test_get_next_patient(self):
|
|
"""Test getting next patient from queue"""
|
|
# Add patients
|
|
for patient in self.patients[:3]:
|
|
self.engine.add_to_queue(patient=patient, priority_score=1.0)
|
|
|
|
# Get next patient
|
|
next_entry = self.engine.get_next_patient()
|
|
|
|
# Should return first waiting patient
|
|
self.assertIsNotNone(next_entry)
|
|
self.assertEqual(next_entry.status, 'CALLED')
|
|
self.assertIsNotNone(next_entry.called_at)
|
|
|
|
def test_reposition_queue_entries(self):
|
|
"""Test repositioning all queue entries"""
|
|
# Add patients with different priorities
|
|
priorities = [1.0, 3.0, 2.0, 5.0, 1.5]
|
|
entries = []
|
|
|
|
for patient, priority in zip(self.patients, priorities):
|
|
entry = self.engine.add_to_queue(
|
|
patient=patient,
|
|
priority_score=priority
|
|
)
|
|
entries.append(entry)
|
|
|
|
# Reposition
|
|
self.engine.reposition_queue_entries()
|
|
|
|
# Verify positions are ordered by priority
|
|
entries = QueueEntry.objects.filter(
|
|
queue=self.queue,
|
|
status='WAITING'
|
|
).order_by('queue_position')
|
|
|
|
# Higher priority should be earlier in queue
|
|
prev_priority = float('inf')
|
|
for entry in entries:
|
|
self.assertLessEqual(entry.priority_score, prev_priority)
|
|
prev_priority = entry.priority_score
|
|
|
|
def test_get_queue_status(self):
|
|
"""Test getting queue status"""
|
|
# Add some patients
|
|
for patient in self.patients[:3]:
|
|
self.engine.add_to_queue(patient=patient, priority_score=1.0)
|
|
|
|
# Get status
|
|
status = self.engine.get_queue_status()
|
|
|
|
# Should contain required fields
|
|
self.assertIn('total_waiting', status)
|
|
self.assertIn('total_in_service', status)
|
|
self.assertIn('total_completed_today', status)
|
|
self.assertIn('average_wait_time', status)
|
|
self.assertIn('estimated_wait_time', status)
|
|
self.assertIn('load_factor', status)
|
|
|
|
# Total waiting should match
|
|
self.assertEqual(status['total_waiting'], 3)
|
|
|
|
def test_calculate_load_factor(self):
|
|
"""Test load factor calculation"""
|
|
# Add patients to fill queue partially
|
|
for patient in self.patients:
|
|
self.engine.add_to_queue(patient=patient, priority_score=1.0)
|
|
|
|
# Calculate load factor
|
|
load_factor = self.engine._calculate_load_factor()
|
|
|
|
# Should be between 0 and 1
|
|
self.assertGreaterEqual(load_factor, 0.0)
|
|
self.assertLessEqual(load_factor, 1.0)
|
|
|
|
# Should reflect current queue size
|
|
expected_load = len(self.patients) / self.queue.max_queue_size
|
|
self.assertAlmostEqual(load_factor, expected_load, places=2)
|
|
|
|
def test_estimate_wait_time(self):
|
|
"""Test wait time estimation"""
|
|
# Add patients
|
|
for patient in self.patients[:3]:
|
|
self.engine.add_to_queue(patient=patient, priority_score=1.0)
|
|
|
|
# Estimate wait time for position 2
|
|
wait_time = self.engine._estimate_wait_time(position=2)
|
|
|
|
# Should be positive
|
|
self.assertGreater(wait_time, 0)
|
|
|
|
# Should be based on service time and position
|
|
expected_wait = 2 * self.queue.average_service_time_minutes
|
|
self.assertAlmostEqual(wait_time, expected_wait, delta=5)
|
|
|
|
def test_multi_factor_positioning(self):
|
|
"""Test multi-factor positioning algorithm"""
|
|
# Create appointment for one patient
|
|
provider = User.objects.create_user(
|
|
email="provider@test.com",
|
|
tenant=self.tenant,
|
|
role="PHYSICIAN"
|
|
)
|
|
|
|
appointment = AppointmentRequest.objects.create(
|
|
tenant=self.tenant,
|
|
patient=self.patients[0],
|
|
provider=provider,
|
|
appointment_type="CONSULTATION",
|
|
scheduled_datetime=timezone.now() + timedelta(hours=1),
|
|
status="CONFIRMED"
|
|
)
|
|
|
|
# Add patient with appointment
|
|
entry_with_appt = self.engine.add_to_queue(
|
|
patient=self.patients[0],
|
|
appointment=appointment,
|
|
priority_score=2.0
|
|
)
|
|
|
|
# Add patient without appointment but higher priority
|
|
entry_without_appt = self.engine.add_to_queue(
|
|
patient=self.patients[1],
|
|
priority_score=3.0
|
|
)
|
|
|
|
# Patient with appointment should be positioned considering appointment time
|
|
self.assertIsNotNone(entry_with_appt.queue_position)
|
|
self.assertIsNotNone(entry_without_appt.queue_position)
|
|
|
|
def test_get_analytics_summary(self):
|
|
"""Test analytics summary generation"""
|
|
# Add and process some patients
|
|
for patient in self.patients[:3]:
|
|
entry = self.engine.add_to_queue(patient=patient, priority_score=1.0)
|
|
entry.status = 'COMPLETED'
|
|
entry.served_at = timezone.now()
|
|
entry.save()
|
|
|
|
# Get analytics
|
|
analytics = self.engine.get_analytics_summary(days=7)
|
|
|
|
# Should contain analytics data
|
|
self.assertIn('total_entries', analytics)
|
|
self.assertIn('completed_entries', analytics)
|
|
self.assertIn('average_wait_time', analytics)
|
|
self.assertIn('utilization_rate', analytics)
|
|
|
|
def test_overflow_queue_handling(self):
|
|
"""Test overflow queue handling"""
|
|
# Enable overflow queue
|
|
self.config.enable_overflow_queue = True
|
|
self.config.overflow_threshold = 3
|
|
self.config.save()
|
|
|
|
# Add patients up to overflow threshold
|
|
for patient in self.patients:
|
|
self.engine.add_to_queue(patient=patient, priority_score=1.0)
|
|
|
|
# Check if overflow is detected
|
|
status = self.engine.get_queue_status()
|
|
load_factor = status['load_factor']
|
|
|
|
# Load factor should indicate overflow
|
|
self.assertGreater(load_factor, self.config.overflow_threshold / self.queue.max_queue_size)
|
|
|
|
|
|
class QueuePositioningAlgorithmTest(TestCase):
|
|
"""Test cases for queue positioning algorithms"""
|
|
|
|
def setUp(self):
|
|
"""Set up test data"""
|
|
self.tenant = Tenant.objects.create(
|
|
name="Test Hospital",
|
|
display_name="Test Hospital",
|
|
address_line1="123 Test St",
|
|
city="Test City",
|
|
state="Test State",
|
|
postal_code="12345",
|
|
phone_number="+966501234567",
|
|
email="test@hospital.com"
|
|
)
|
|
|
|
self.queue = WaitingQueue.objects.create(
|
|
tenant=self.tenant,
|
|
name="Test Queue",
|
|
queue_type="GENERAL",
|
|
specialty="FAMILY_MEDICINE",
|
|
max_queue_size=50,
|
|
average_service_time_minutes=15,
|
|
is_active=True
|
|
)
|
|
|
|
self.config = QueueConfiguration.objects.create(
|
|
queue=self.queue,
|
|
use_dynamic_positioning=True,
|
|
priority_weight=0.5,
|
|
wait_time_weight=0.3,
|
|
appointment_time_weight=0.2
|
|
)
|
|
|
|
self.engine = AdvancedQueueEngine(self.queue)
|
|
|
|
def test_priority_based_positioning(self):
|
|
"""Test positioning based on priority score"""
|
|
patient1 = PatientProfile.objects.create(
|
|
tenant=self.tenant,
|
|
first_name="Low",
|
|
last_name="Priority",
|
|
date_of_birth=datetime(1990, 1, 1).date(),
|
|
gender="M",
|
|
mrn="LOW001"
|
|
)
|
|
|
|
patient2 = PatientProfile.objects.create(
|
|
tenant=self.tenant,
|
|
first_name="High",
|
|
last_name="Priority",
|
|
date_of_birth=datetime(1990, 1, 1).date(),
|
|
gender="M",
|
|
mrn="HIGH001"
|
|
)
|
|
|
|
# Add low priority patient first
|
|
entry1 = self.engine.add_to_queue(patient=patient1, priority_score=1.0)
|
|
|
|
# Add high priority patient
|
|
entry2 = self.engine.add_to_queue(patient=patient2, priority_score=5.0)
|
|
|
|
# High priority should be positioned before low priority
|
|
self.assertLess(entry2.queue_position, entry1.queue_position)
|
|
|
|
def test_wait_time_consideration(self):
|
|
"""Test that wait time is considered in positioning"""
|
|
patient = PatientProfile.objects.create(
|
|
tenant=self.tenant,
|
|
first_name="Test",
|
|
last_name="Patient",
|
|
date_of_birth=datetime(1990, 1, 1).date(),
|
|
gender="M",
|
|
mrn="TEST001"
|
|
)
|
|
|
|
# Add patient and simulate wait time
|
|
entry = self.engine.add_to_queue(patient=patient, priority_score=1.0)
|
|
|
|
# Modify joined_at to simulate waiting
|
|
entry.joined_at = timezone.now() - timedelta(minutes=30)
|
|
entry.save()
|
|
|
|
# Reposition queue
|
|
self.engine.reposition_queue_entries()
|
|
|
|
# Entry should be repositioned considering wait time
|
|
entry.refresh_from_db()
|
|
self.assertIsNotNone(entry.queue_position)
|
|
|
|
def test_appointment_time_priority(self):
|
|
"""Test positioning based on appointment time"""
|
|
provider = User.objects.create_user(
|
|
email="provider@test.com",
|
|
tenant=self.tenant,
|
|
role="PHYSICIAN"
|
|
)
|
|
|
|
patient1 = PatientProfile.objects.create(
|
|
tenant=self.tenant,
|
|
first_name="Later",
|
|
last_name="Appointment",
|
|
date_of_birth=datetime(1990, 1, 1).date(),
|
|
gender="M",
|
|
mrn="LATER001"
|
|
)
|
|
|
|
patient2 = PatientProfile.objects.create(
|
|
tenant=self.tenant,
|
|
first_name="Sooner",
|
|
last_name="Appointment",
|
|
date_of_birth=datetime(1990, 1, 1).date(),
|
|
gender="M",
|
|
mrn="SOONER001"
|
|
)
|
|
|
|
# Create appointments
|
|
appt1 = AppointmentRequest.objects.create(
|
|
tenant=self.tenant,
|
|
patient=patient1,
|
|
provider=provider,
|
|
appointment_type="CONSULTATION",
|
|
scheduled_datetime=timezone.now() + timedelta(hours=2),
|
|
status="CONFIRMED"
|
|
)
|
|
|
|
appt2 = AppointmentRequest.objects.create(
|
|
tenant=self.tenant,
|
|
patient=patient2,
|
|
provider=provider,
|
|
appointment_type="CONSULTATION",
|
|
scheduled_datetime=timezone.now() + timedelta(minutes=30),
|
|
status="CONFIRMED"
|
|
)
|
|
|
|
# Add patients with appointments
|
|
entry1 = self.engine.add_to_queue(
|
|
patient=patient1,
|
|
appointment=appt1,
|
|
priority_score=1.0
|
|
)
|
|
|
|
entry2 = self.engine.add_to_queue(
|
|
patient=patient2,
|
|
appointment=appt2,
|
|
priority_score=1.0
|
|
)
|
|
|
|
# Patient with sooner appointment should be positioned earlier
|
|
self.assertLess(entry2.queue_position, entry1.queue_position)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
class QueueEnginePerformanceTest(TestCase):
|
|
"""Performance tests for AdvancedQueueEngine"""
|
|
|
|
def setUp(self):
|
|
"""Set up test data"""
|
|
self.tenant = Tenant.objects.create(
|
|
name="Test Hospital",
|
|
display_name="Test Hospital",
|
|
address_line1="123 Test St",
|
|
city="Test City",
|
|
state="Test State",
|
|
postal_code="12345",
|
|
phone_number="+966501234567",
|
|
email="test@hospital.com"
|
|
)
|
|
|
|
self.queue = WaitingQueue.objects.create(
|
|
tenant=self.tenant,
|
|
name="Test Queue",
|
|
queue_type="GENERAL",
|
|
specialty="FAMILY_MEDICINE",
|
|
max_queue_size=100,
|
|
average_service_time_minutes=15,
|
|
is_active=True
|
|
)
|
|
|
|
self.config = QueueConfiguration.objects.create(
|
|
queue=self.queue,
|
|
use_dynamic_positioning=True
|
|
)
|
|
|
|
self.engine = AdvancedQueueEngine(self.queue)
|
|
|
|
# Create many patients
|
|
self.patients = []
|
|
for i in range(50):
|
|
patient = PatientProfile.objects.create(
|
|
tenant=self.tenant,
|
|
first_name=f"Patient{i}",
|
|
last_name="Test",
|
|
date_of_birth=datetime(1990, 1, 1).date(),
|
|
gender="M",
|
|
mrn=f"TEST{i:03d}"
|
|
)
|
|
self.patients.append(patient)
|
|
|
|
def test_add_patient_performance(self):
|
|
"""Test performance of adding patients to queue"""
|
|
import time
|
|
|
|
start_time = time.time()
|
|
|
|
# Add 50 patients
|
|
for patient in self.patients:
|
|
self.engine.add_to_queue(patient=patient, priority_score=1.0)
|
|
|
|
end_time = time.time()
|
|
execution_time = end_time - start_time
|
|
|
|
# Should complete in less than 2 seconds
|
|
self.assertLess(execution_time, 2.0)
|
|
|
|
print(f"\nAdded 50 patients in {execution_time:.3f} seconds")
|
|
|
|
def test_repositioning_performance(self):
|
|
"""Test performance of queue repositioning"""
|
|
import time
|
|
|
|
# Add patients first
|
|
for patient in self.patients:
|
|
self.engine.add_to_queue(patient=patient, priority_score=1.0)
|
|
|
|
start_time = time.time()
|
|
|
|
# Reposition queue
|
|
self.engine.reposition_queue_entries()
|
|
|
|
end_time = time.time()
|
|
execution_time = end_time - start_time
|
|
|
|
# Should complete in less than 1 second
|
|
self.assertLess(execution_time, 1.0)
|
|
|
|
print(f"\nRepositioned 50 entries in {execution_time:.3f} seconds")
|
|
|
|
def test_status_calculation_performance(self):
|
|
"""Test performance of status calculation"""
|
|
import time
|
|
|
|
# Add patients
|
|
for patient in self.patients:
|
|
self.engine.add_to_queue(patient=patient, priority_score=1.0)
|
|
|
|
start_time = time.time()
|
|
|
|
# Calculate status 100 times
|
|
for _ in range(100):
|
|
self.engine.get_queue_status()
|
|
|
|
end_time = time.time()
|
|
execution_time = end_time - start_time
|
|
|
|
# Should complete in less than 1 second
|
|
self.assertLess(execution_time, 1.0)
|
|
|
|
print(f"\n100 status calculations in {execution_time:.3f} seconds")
|