385 lines
12 KiB
Python
385 lines
12 KiB
Python
"""
|
|
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")
|