""" 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")