""" Integration tests for Smart Scheduling API endpoints (Phase 10) """ import json from datetime import datetime, timedelta from django.test import TestCase, Client from django.urls import reverse from django.utils import timezone from django.contrib.auth import get_user_model from appointments.models import ( AppointmentTemplate, SchedulingPreference, AppointmentPriorityRule ) from patients.models import PatientProfile from core.models import Tenant User = get_user_model() class SchedulingAPITestCase(TestCase): """Test cases for scheduling API endpoints""" 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 user self.user = User.objects.create_user( email="test@test.com", password="testpass123", tenant=self.tenant, role="PHYSICIAN" ) # Create provider self.provider = User.objects.create_user( email="provider@test.com", first_name="Test", last_name="Provider", tenant=self.tenant, role="PHYSICIAN" ) # Create patient self.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" ) # Create appointment template self.appointment_type = AppointmentTemplate.objects.create( tenant=self.tenant, name="General Consultation", appointment_type="CONSULTATION", specialty="FAMILY_MEDICINE", duration_minutes=30, is_active=True ) # Create client and login self.client = Client() self.client.login(email="test@test.com", password="testpass123") def test_find_optimal_slots_endpoint(self): """Test find optimal slots API endpoint""" url = reverse('appointments:api:find_optimal_slots') data = { 'patient_id': self.patient.id, 'provider_id': self.provider.id, 'appointment_type_id': self.appointment_type.id, 'start_date': (timezone.now() + timedelta(days=1)).date().isoformat(), 'end_date': (timezone.now() + timedelta(days=7)).date().isoformat(), 'duration_minutes': 30 } response = self.client.post( url, data=json.dumps(data), content_type='application/json' ) # Should return 200 OK self.assertEqual(response.status_code, 200) # Response should contain slots response_data = response.json() self.assertIn('slots', response_data) self.assertIsInstance(response_data['slots'], list) def test_find_optimal_slots_missing_params(self): """Test find optimal slots with missing parameters""" url = reverse('appointments:api:find_optimal_slots') data = { 'patient_id': self.patient.id, # Missing provider_id and other required fields } response = self.client.post( url, data=json.dumps(data), content_type='application/json' ) # Should return 400 Bad Request self.assertEqual(response.status_code, 400) def test_check_conflicts_endpoint(self): """Test check conflicts API endpoint""" url = reverse('appointments:api:check_conflicts') start_time = timezone.now() + timedelta(days=1, hours=10) end_time = start_time + timedelta(minutes=30) data = { 'provider_id': self.provider.id, 'start_time': start_time.isoformat(), 'end_time': end_time.isoformat() } response = self.client.post( url, data=json.dumps(data), content_type='application/json' ) # Should return 200 OK self.assertEqual(response.status_code, 200) # Response should contain conflict status response_data = response.json() self.assertIn('has_conflict', response_data) self.assertIsInstance(response_data['has_conflict'], bool) def test_provider_analytics_endpoint(self): """Test provider analytics API endpoint""" url = reverse('appointments:api:provider_analytics', kwargs={'provider_id': self.provider.id}) response = self.client.get(url) # Should return 200 OK self.assertEqual(response.status_code, 200) # Response should contain analytics data response_data = response.json() self.assertIn('utilization_rate', response_data) self.assertIn('no_show_rate', response_data) def test_update_patient_preferences_endpoint(self): """Test update patient preferences API endpoint""" url = reverse('appointments:api:update_patient_preferences', kwargs={'patient_id': self.patient.id}) response = self.client.post(url) # Should return 200 OK self.assertEqual(response.status_code, 200) # Should create or update preferences self.assertTrue( SchedulingPreference.objects.filter(patient=self.patient).exists() ) def test_api_authentication_required(self): """Test that API endpoints require authentication""" # Logout self.client.logout() url = reverse('appointments:api:find_optimal_slots') data = { 'patient_id': self.patient.id, 'provider_id': self.provider.id, } response = self.client.post( url, data=json.dumps(data), content_type='application/json' ) # Should return 401 or 403 self.assertIn(response.status_code, [401, 403]) def test_api_response_format(self): """Test API response format consistency""" url = reverse('appointments:api:find_optimal_slots') data = { 'patient_id': self.patient.id, 'provider_id': self.provider.id, 'appointment_type_id': self.appointment_type.id, 'start_date': (timezone.now() + timedelta(days=1)).date().isoformat(), 'end_date': (timezone.now() + timedelta(days=7)).date().isoformat(), 'duration_minutes': 30 } response = self.client.post( url, data=json.dumps(data), content_type='application/json' ) # Response should be JSON self.assertEqual(response['Content-Type'], 'application/json') # Should be valid JSON response_data = response.json() self.assertIsInstance(response_data, dict) def test_concurrent_requests(self): """Test handling of concurrent API requests""" import threading url = reverse('appointments:api:find_optimal_slots') data = { 'patient_id': self.patient.id, 'provider_id': self.provider.id, 'appointment_type_id': self.appointment_type.id, 'start_date': (timezone.now() + timedelta(days=1)).date().isoformat(), 'end_date': (timezone.now() + timedelta(days=7)).date().isoformat(), 'duration_minutes': 30 } results = [] def make_request(): response = self.client.post( url, data=json.dumps(data), content_type='application/json' ) results.append(response.status_code) # Create 10 concurrent requests threads = [] for _ in range(10): thread = threading.Thread(target=make_request) threads.append(thread) thread.start() # Wait for all threads to complete for thread in threads: thread.join() # All requests should succeed self.assertEqual(len(results), 10) for status_code in results: self.assertEqual(status_code, 200) class SchedulingAPIPerformanceTest(TestCase): """Performance tests for scheduling API""" 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.user = User.objects.create_user( email="test@test.com", password="testpass123", tenant=self.tenant, role="PHYSICIAN" ) self.provider = User.objects.create_user( email="provider@test.com", tenant=self.tenant, role="PHYSICIAN" ) self.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" ) self.appointment_type = AppointmentTemplate.objects.create( tenant=self.tenant, name="General Consultation", appointment_type="CONSULTATION", specialty="FAMILY_MEDICINE", duration_minutes=30, is_active=True ) self.client = Client() self.client.login(email="test@test.com", password="testpass123") def test_api_response_time(self): """Test API response time for slot finding""" import time url = reverse('appointments:api:find_optimal_slots') data = { 'patient_id': self.patient.id, 'provider_id': self.provider.id, 'appointment_type_id': self.appointment_type.id, 'start_date': (timezone.now() + timedelta(days=1)).date().isoformat(), 'end_date': (timezone.now() + timedelta(days=30)).date().isoformat(), 'duration_minutes': 30 } start_time = time.time() response = self.client.post( url, data=json.dumps(data), content_type='application/json' ) end_time = time.time() response_time = end_time - start_time # Should respond in less than 3 seconds self.assertLess(response_time, 3.0) self.assertEqual(response.status_code, 200) print(f"\nAPI response time: {response_time:.3f} seconds") def test_api_throughput(self): """Test API throughput with multiple sequential requests""" import time url = reverse('appointments:api:find_optimal_slots') data = { 'patient_id': self.patient.id, 'provider_id': self.provider.id, 'appointment_type_id': self.appointment_type.id, 'start_date': (timezone.now() + timedelta(days=1)).date().isoformat(), 'end_date': (timezone.now() + timedelta(days=7)).date().isoformat(), 'duration_minutes': 30 } num_requests = 20 start_time = time.time() for _ in range(num_requests): response = self.client.post( url, data=json.dumps(data), content_type='application/json' ) self.assertEqual(response.status_code, 200) end_time = time.time() total_time = end_time - start_time avg_time = total_time / num_requests # Average response time should be less than 1 second self.assertLess(avg_time, 1.0) print(f"\nProcessed {num_requests} requests in {total_time:.3f} seconds") print(f"Average response time: {avg_time:.3f} seconds")