Files
multitenetsaas/backend/tests/integration/test_healthcare_operations.py
AHMET YILMAZ b3fff546e9
Some checks failed
System Monitoring / Health Checks (push) Has been cancelled
System Monitoring / Performance Monitoring (push) Has been cancelled
System Monitoring / Database Monitoring (push) Has been cancelled
System Monitoring / Cache Monitoring (push) Has been cancelled
System Monitoring / Log Monitoring (push) Has been cancelled
System Monitoring / Resource Monitoring (push) Has been cancelled
System Monitoring / Uptime Monitoring (push) Has been cancelled
System Monitoring / Backup Monitoring (push) Has been cancelled
System Monitoring / Security Monitoring (push) Has been cancelled
System Monitoring / Monitoring Dashboard (push) Has been cancelled
System Monitoring / Alerting (push) Has been cancelled
Security Scanning / Dependency Scanning (push) Has been cancelled
Security Scanning / Code Security Scanning (push) Has been cancelled
Security Scanning / Secrets Scanning (push) Has been cancelled
Security Scanning / Container Security Scanning (push) Has been cancelled
Security Scanning / Compliance Checking (push) Has been cancelled
Security Scanning / Security Dashboard (push) Has been cancelled
Security Scanning / Security Remediation (push) Has been cancelled
project initialization
2025-10-05 02:37:33 +08:00

626 lines
22 KiB
Python

"""
Integration test for healthcare module operations.
This test MUST fail before implementation.
"""
import pytest
from django.test import TestCase
from django.urls import reverse
from rest_framework.test import APIClient
from rest_framework import status
import json
from datetime import datetime, timedelta
class HealthcareOperationsIntegrationTest(TestCase):
def setUp(self):
self.client = APIClient()
# Tenant authentication header
self.tenant_auth = {'HTTP_AUTHORIZATION': 'Bearer tenant_token'}
# Test patient data
self.patient_data = {
'ic_number': '900101-10-1234',
'name': 'Ahmad bin Hassan',
'gender': 'MALE',
'date_of_birth': '1990-01-01',
'phone': '+60123456789',
'email': 'ahmad.hassan@example.com',
'address': {
'street': '123 Jalan Healthcare',
'city': 'Kuala Lumpur',
'state': 'Wilayah Persekutuan',
'postal_code': '50400',
'country': 'Malaysia'
},
'blood_type': 'O+',
'allergies': ['Penicillin'],
'medications': ['Metformin 500mg']
}
# Test doctor data
self.doctor_data = {
'name': 'Dr. Sarah Johnson',
'specialization': 'General Practitioner',
'license_number': 'L12345',
'department': 'Primary Care',
'phone': '+60312345678',
'email': 'sarah.johnson@hospital.com'
}
def test_complete_patient_workflow(self):
"""Test complete patient workflow from registration to treatment."""
# Step 1: Patient registration (should fail before implementation)
patient_response = self.client.post(
'/api/v1/healthcare/patients/',
data=json.dumps(self.patient_data),
content_type='application/json',
**self.tenant_auth
)
assert patient_response.status_code == status.HTTP_201_CREATED
patient_data = patient_response.json()
# Verify patient structure
assert 'id' in patient_data
assert patient_data['ic_number'] == self.patient_data['ic_number']
assert patient_data['name'] == self.patient_data['name']
assert patient_data['age'] == 34 # Calculated from DOB
assert patient_data['status'] == 'ACTIVE'
# Step 2: Create doctor
doctor_response = self.client.post(
'/api/v1/healthcare/doctors/',
data=json.dumps(self.doctor_data),
content_type='application/json',
**self.tenant_auth
)
assert doctor_response.status_code == status.HTTP_201_CREATED
doctor_data = doctor_response.json()
# Step 3: Schedule appointment
appointment_data = {
'patient_id': patient_data['id'],
'doctor_id': doctor_data['id'],
'appointment_datetime': '2024-02-15T14:30:00+08:00',
'duration': 30,
'type': 'CONSULTATION',
'reason': 'Regular checkup for diabetes management',
'priority': 'NORMAL'
}
appointment_response = self.client.post(
'/api/v1/healthcare/appointments/',
data=json.dumps(appointment_data),
content_type='application/json',
**self.tenant_auth
)
assert appointment_response.status_code == status.HTTP_201_CREATED
appointment_data = appointment_response.json()
assert appointment_data['status'] == 'SCHEDULED'
# Step 4: Update appointment status to in-progress
status_update_response = self.client.put(
f'/api/v1/healthcare/appointments/{appointment_data["id"]}/status/',
data=json.dumps({'status': 'IN_PROGRESS'}),
content_type='application/json',
**self.tenant_auth
)
assert status_update_response.status_code == status.HTTP_200_OK
# Step 5: Create medical record
medical_record_data = {
'patient_id': patient_data['id'],
'appointment_id': appointment_data['id'],
'doctor_id': doctor_data['id'],
'diagnosis': 'Type 2 Diabetes - well controlled',
'treatment': 'Continue current medication regimen',
'prescriptions': [
{
'medication': 'Metformin',
'dosage': '500mg',
'frequency': 'Twice daily',
'duration': '30 days',
'instructions': 'Take with meals'
}
],
'vitals': {
'blood_pressure': '120/80',
'heart_rate': 72,
'temperature': 36.5,
'weight': 75.5,
'height': 175.0
},
'notes': 'Patient reports good compliance with medication. Blood sugar levels well controlled.'
}
record_response = self.client.post(
'/api/v1/healthcare/medical-records/',
data=json.dumps(medical_record_data),
content_type='application/json',
**self.tenant_auth
)
assert record_response.status_code == status.HTTP_201_CREATED
record_data = record_response.json()
# Step 6: Complete appointment
complete_response = self.client.put(
f'/api/v1/healthcare/appointments/{appointment_data["id"]}/status/',
data=json.dumps({'status': 'COMPLETED'}),
content_type='application/json',
**self.tenant_auth
)
assert complete_response.status_code == status.HTTP_200_OK
# Step 7: Schedule follow-up appointment
follow_up_data = {
'patient_id': patient_data['id'],
'doctor_id': doctor_data['id'],
'appointment_datetime': '2024-03-15T14:30:00+08:00',
'duration': 20,
'type': 'FOLLOW_UP',
'reason': 'Diabetes follow-up'
}
follow_up_response = self.client.post(
'/api/v1/healthcare/appointments/',
data=json.dumps(follow_up_data),
content_type='application/json',
**self.tenant_auth
)
assert follow_up_response.status_code == status.HTTP_201_CREATED
def test_medical_records_management(self):
"""Test medical records management and history."""
# Create patient first
patient_response = self.client.post(
'/api/v1/healthcare/patients/',
data=json.dumps(self.patient_data),
content_type='application/json',
**self.tenant_auth
)
assert patient_response.status_code == status.HTTP_201_CREATED
patient_data = patient_response.json()
# Create multiple medical records over time
records_data = [
{
'diagnosis': 'Hypertension',
'treatment': 'Lifestyle modifications',
'prescriptions': [
{
'medication': 'Lisinopril',
'dosage': '10mg',
'frequency': 'Once daily'
}
]
},
{
'diagnosis': 'Annual checkup - normal',
'treatment': 'Continue healthy lifestyle',
'vitals': {
'blood_pressure': '118/76',
'heart_rate': 68,
'cholesterol': 180
}
}
]
created_records = []
for record_data in records_data:
full_record_data = {
'patient_id': patient_data['id'],
'doctor_id': 'doctor-001',
'diagnosis': record_data['diagnosis'],
'treatment': record_data['treatment'],
**{k: v for k, v in record_data.items() if k not in ['diagnosis', 'treatment']}
}
record_response = self.client.post(
'/api/v1/healthcare/medical-records/',
data=json.dumps(full_record_data),
content_type='application/json',
**self.tenant_auth
)
assert record_response.status_code == status.HTTP_201_CREATED
created_records.append(record_response.json())
# Test medical history retrieval
history_response = self.client.get(
f'/api/v1/healthcare/patients/{patient_data["id"]}/medical-history/',
**self.tenant_auth
)
assert history_response.status_code == status.HTTP_200_OK
history_data = history_response.json()
assert 'medical_records' in history_data
assert 'conditions' in history_data
assert 'medications' in history_data
assert 'allergies' in history_data
# Verify records are chronological
records = history_data['medical_records']
assert len(records) == len(created_records)
# Test record search and filtering
search_response = self.client.get(
f'/api/v1/healthcare/medical-records/',
data={'patient_id': patient_data['id'], 'diagnosis': 'Hypertension'},
**self.tenant_auth
)
assert search_response.status_code == status.HTTP_200_OK
search_results = search_response.json()['records']
assert len(search_results) > 0
assert any('Hypertension' in record['diagnosis'] for record in search_results)
def test_prescription_management(self):
"""Test prescription management and dispensing."""
# Create patient
patient_response = self.client.post(
'/api/v1/healthcare/patients/',
data=json.dumps(self.patient_data),
content_type='application/json',
**self.tenant_auth
)
assert patient_response.status_code == status.HTTP_201_CREATED
patient_data = patient_response.json()
# Create prescription
prescription_data = {
'patient_id': patient_data['id'],
'doctor_id': 'doctor-001',
'medications': [
{
'name': 'Amoxicillin',
'dosage': '500mg',
'frequency': 'Three times daily',
'duration': '7 days',
'quantity': 21,
'instructions': 'Take after meals',
'refills_allowed': 0
},
{
'name': 'Ibuprofen',
'dosage': '400mg',
'frequency': 'As needed for pain',
'duration': '3 days',
'quantity': 9,
'instructions': 'Take with food',
'refills_allowed': 1
}
],
'diagnosis': 'Bacterial infection',
'notes': 'Complete full course of antibiotics'
}
prescription_response = self.client.post(
'/api/v1/healthcare/prescriptions/',
data=json.dumps(prescription_data),
content_type='application/json',
**self.tenant_auth
)
assert prescription_response.status_code == status.HTTP_201_CREATED
prescription_data = prescription_response.json()
# Test prescription status management
dispense_data = {
'dispensed_by': 'pharmacist-001',
'dispensed_at': datetime.now().isoformat(),
'notes': 'Patient counseled on medication use'
}
dispense_response = self.client.post(
f'/api/v1/healthcare/prescriptions/{prescription_data["id"]}/dispense/',
data=json.dumps(dispense_data),
content_type='application/json',
**self.tenant_auth
)
assert dispense_response.status_code == status.HTTP_200_OK
# Test refill request
refill_response = self.client.post(
f'/api/v1/healthcare/prescriptions/{prescription_data["id"]}/refill/',
data=json.dumps({}),
content_type='application/json',
**self.tenant_auth
)
assert refill_response.status_code == status.HTTP_200_OK
def test_laboratory_and_imaging_orders(self):
"""Test laboratory and imaging order management."""
# Create patient
patient_response = self.client.post(
'/api/v1/healthcare/patients/',
data=json.dumps(self.patient_data),
content_type='application/json',
**self.tenant_auth
)
assert patient_response.status_code == status.HTTP_201_CREATED
patient_data = patient_response.json()
# Create lab order
lab_order_data = {
'patient_id': patient_data['id'],
'doctor_id': 'doctor-001',
'tests': [
{
'test_code': 'CBC',
'test_name': 'Complete Blood Count',
'priority': 'ROUTINE',
'clinical_indication': 'Annual checkup'
},
{
'test_code': 'HBA1C',
'test_name': 'Hemoglobin A1C',
'priority': 'ROUTINE',
'clinical_indication': 'Diabetes monitoring'
}
],
'notes': 'Patient fasting for 12 hours'
}
lab_order_response = self.client.post(
'/api/v1/healthcare/laboratory-orders/',
data=json.dumps(lab_order_data),
content_type='application/json',
**self.tenant_auth
)
assert lab_order_response.status_code == status.HTTP_201_CREATED
lab_order = lab_order_response.json()
# Update lab results
results_data = {
'results': [
{
'test_code': 'CBC',
'result_value': 'Normal',
'reference_range': '4.5-5.5 x 10^12/L',
'units': 'x 10^12/L',
'status': 'NORMAL'
},
{
'test_code': 'HBA1C',
'result_value': '6.2',
'reference_range': '< 5.7%',
'units': '%',
'status': 'ABNORMAL',
'notes': 'Slightly elevated - monitor'
}
],
'interpreted_by': 'Dr. Lab Specialist',
'interpretation': 'HbA1c shows prediabetes range'
}
results_response = self.client.post(
f'/api/v1/healthcare/laboratory-orders/{lab_order["id"]}/results/',
data=json.dumps(results_data),
content_type='application/json',
**self.tenant_auth
)
assert results_response.status_code == status.HTTP_200_OK
def test_billing_and_insurance_integration(self):
"""Test billing and insurance claim processing."""
# Create patient with insurance
patient_with_insurance = self.patient_data.copy()
patient_with_insurance['insurance'] = {
'provider': 'Malaysia National Insurance',
'policy_number': 'MNI-123456789',
'coverage_details': 'Full coverage',
'expiry_date': '2024-12-31'
}
patient_response = self.client.post(
'/api/v1/healthcare/patients/',
data=json.dumps(patient_with_insurance),
content_type='application/json',
**self.tenant_auth
)
assert patient_response.status_code == status.HTTP_201_CREATED
patient_data = patient_response.json()
# Create consultation and generate bill
billing_data = {
'patient_id': patient_data['id'],
'services': [
{
'service_code': 'CONSULT_GP',
'description': 'General Practitioner Consultation',
'amount': 150.00,
'quantity': 1
},
{
'service_code': 'LAB_CBC',
'description': 'Complete Blood Count',
'amount': 50.00,
'quantity': 1
}
],
'insurance_claim': {
'provider': patient_data['insurance']['provider'],
'policy_number': patient_data['insurance']['policy_number'],
'pre_authorization_code': 'PA-2024-001'
}
}
billing_response = self.client.post(
'/api/v1/healthcare/billing/',
data=json.dumps(billing_data),
content_type='application/json',
**self.tenant_auth
)
assert billing_response.status_code == status.HTTP_201_CREATED
billing_data = billing_response.json()
# Verify insurance claim processing
assert 'insurance_coverage' in billing_data
assert 'patient_responsibility' in billing_data
assert 'claim_status' in billing_data
def test_healthcare_compliance_and_reporting(self):
"""Test healthcare compliance and reporting features."""
# Test PDPA compliance (Personal Data Protection Act)
compliance_response = self.client.get(
'/api/v1/healthcare/compliance/data-protection/',
**self.tenant_auth
)
assert compliance_response.status_code == status.HTTP_200_OK
compliance_data = compliance_response.json()
assert 'consent_records' in compliance_data
assert 'data_access_logs' in compliance_data
assert 'retention_policies' in compliance_data
# Test clinical reporting
clinical_report_response = self.client.get(
'/api/v1/healthcare/reports/clinical/',
data={
'period': 'monthly',
'year': 2024,
'month': 1
},
**self.tenant_auth
)
assert clinical_report_response.status_code == status.HTTP_200_OK
clinical_report = clinical_report_response.json()
assert 'patient_visits' in clinical_report
assert 'common_diagnoses' in clinical_report
assert 'prescription_trends' in clinical_report
# Test adverse event reporting
adverse_event_data = {
'patient_id': 'patient-001',
'event_type': 'MEDICATION_ERROR',
'description': 'Wrong dosage administered',
'severity': 'MINOR',
'date_occurred': datetime.now().isoformat(),
'reported_by': 'nurse-001',
'actions_taken': 'Corrected dosage, patient monitored'
}
adverse_response = self.client.post(
'/api/v1/healthcare/adverse-events/',
data=json.dumps(adverse_event_data),
content_type='application/json',
**self.tenant_auth
)
assert adverse_response.status_code == status.HTTP_201_CREATED
def test_telemedicine_integration(self):
"""Test telemedicine and virtual consultation features."""
# Create virtual appointment
virtual_appointment_data = {
'patient_id': 'patient-001',
'doctor_id': 'doctor-001',
'appointment_datetime': '2024-02-15T15:00:00+08:00',
'duration': 20,
'type': 'CONSULTATION',
'is_virtual': True,
'virtual_consultation': {
'platform': 'ZOOM',
'link': 'https://zoom.us/j/123456789',
'instructions': 'Join 5 minutes early, test audio/video',
'meeting_id': '123456789',
'password': 'health2024'
},
'reason': 'Follow-up consultation'
}
virtual_response = self.client.post(
'/api/v1/healthcare/appointments/',
data=json.dumps(virtual_appointment_data),
content_type='application/json',
**self.tenant_auth
)
assert virtual_response.status_code == status.HTTP_201_CREATED
virtual_appointment = virtual_response.json()
assert virtual_appointment['is_virtual'] is True
assert 'virtual_consultation' in virtual_appointment
# Test telemedicine session logging
session_log_data = {
'appointment_id': virtual_appointment['id'],
'start_time': '2024-02-15T15:00:00Z',
'end_time': '2024-02-15T15:18:00Z',
'duration_minutes': 18,
'connection_quality': 'GOOD',
'technical_issues': None,
'notes': 'Successful virtual consultation'
}
session_log_response = self.client.post(
'/api/v1/healthcare/telemedicine/session-logs/',
data=json.dumps(session_log_data),
content_type='application/json',
**self.tenant_auth
)
assert session_log_response.status_code == status.HTTP_201_CREATED
def test_emergency_management(self):
"""Test emergency case management and triage."""
# Create emergency appointment
emergency_data = {
'patient_id': 'patient-001',
'doctor_id': 'doctor-emergency',
'appointment_datetime': datetime.now().isoformat(),
'duration': 60,
'type': 'EMERGENCY',
'priority': 'URGENT',
'reason': 'Chest pain and shortness of breath',
'triage_level': 'YELLOW'
}
emergency_response = self.client.post(
'/api/v1/healthcare/appointments/',
data=json.dumps(emergency_data),
content_type='application/json',
**self.tenant_auth
)
assert emergency_response.status_code == status.HTTP_201_CREATED
emergency_appointment = emergency_response.json()
assert emergency_appointment['type'] == 'EMERGENCY'
assert emergency_appointment['priority'] == 'URGENT'
# Test emergency response protocol
protocol_response = self.client.get(
f'/api/v1/healthcare/emergency/protocols/{emergency_appointment["triage_level"]}/',
**self.tenant_auth
)
assert protocol_response.status_code == status.HTTP_200_OK
protocol_data = protocol_response.json()
assert 'response_time_target' in protocol_data
assert 'required_actions' in protocol_data
assert 'staffing_requirements' in protocol_data