HH/test_mandatory_px_action.py

389 lines
14 KiB
Python

#!/usr/bin/env python
"""
Test script to verify MANDATORY PX Action creation for all complaints.
This script tests that:
1. Every complaint automatically creates a PX Action (no hospital config needed)
2. AI intelligently selects PX Action category from available options
3. PX Action contains AI-generated title, description, and classification
"""
import os
import sys
import django
# Setup Django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'PX360.settings')
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
django.setup()
from apps.complaints.models import Complaint, ComplaintCategory
from apps.organizations.models import Hospital, Department, Patient
from apps.px_action_center.models import PXAction, PXActionLog
from apps.accounts.models import User
from apps.complaints.tasks import analyze_complaint_with_ai
from django.contrib.contenttypes.models import ContentType
def print_section(title):
"""Print a formatted section header"""
print("\n" + "="*80)
print(f" {title}")
print("="*80 + "\n")
def test_mandatory_px_action():
"""Test that PX Action creation is mandatory for all complaints"""
print_section("TEST: MANDATORY PX ACTION CREATION")
# Get or create test data
try:
hospital = Hospital.objects.first()
if not hospital:
print("❌ ERROR: No hospital found in database")
return False
print(f"✓ Using hospital: {hospital.name_en}")
# Get first active department
department = Department.objects.filter(
hospital=hospital,
status='active'
).first()
if not department:
print("❌ ERROR: No active department found")
return False
print(f"✓ Using department: {department.name}")
# Get or create a patient
patient = Patient.objects.filter(hospital=hospital).first()
if not patient:
print("❌ ERROR: No patient found")
return False
print(f"✓ Using patient: {patient.first_name} {patient.last_name}")
# Get a complaint category
category = ComplaintCategory.objects.filter(parent__isnull=True).first()
if not category:
print("❌ ERROR: No complaint category found")
return False
print(f"✓ Using category: {category.name_en}")
except Exception as e:
print(f"❌ ERROR setting up test data: {str(e)}")
return False
# Test Case 1: Clinical Quality Complaint
print_section("TEST CASE 1: Clinical Quality Complaint")
complaint1 = Complaint.objects.create(
title="Incorrect medication prescribed",
description=(
"My father was given the wrong medication dosage. "
"The nurse didn't double-check before administration. "
"This is a serious safety concern that could have harmed him."
),
patient=patient,
hospital=hospital,
department=department,
category=category,
status='open',
severity='medium',
priority='medium'
)
print(f"✓ Created complaint: {complaint1.title}")
print(f" ID: {complaint1.id}")
print(f" Description: {complaint1.description[:100]}...")
# Run AI analysis task synchronously (not async)
print("\n→ Running AI analysis...")
result = analyze_complaint_with_ai(str(complaint1.id))
# Refresh complaint from DB
complaint1.refresh_from_db()
print(f"\n✓ AI Analysis Result:")
print(f" Status: {result.get('status')}")
print(f" Severity: {result.get('severity')}{complaint1.severity}")
print(f" Priority: {result.get('priority')}{complaint1.priority}")
print(f" Category: {result.get('category')}")
print(f" Department: {result.get('department')}")
print(f" PX Action Created: {result.get('px_action_auto_created')}")
# Check if PX Action was created
px_actions = PXAction.objects.filter(
source_type='complaint',
object_id=complaint1.id
)
if px_actions.exists():
action1 = px_actions.first()
print(f"\n✓ PX Action Created (MANDATORY):")
print(f" Action ID: {action1.id}")
print(f" Title: {action1.title}")
print(f" Description: {action1.description[:150]}...")
print(f" Category: {action1.category}")
print(f" Severity: {action1.severity}")
print(f" Priority: {action1.priority}")
print(f" Status: {action1.status}")
print(f" AI Generated: {action1.metadata.get('ai_generated')}")
# Check action logs
logs = PXActionLog.objects.filter(action=action1)
print(f"\n✓ Action Logs ({logs.count()})")
for log in logs:
print(f" - {log.log_type}: {log.message[:80]}...")
# Verify AI selected appropriate category
valid_categories = [
'clinical_quality', 'patient_safety', 'service_quality',
'staff_behavior', 'facility', 'process_improvement', 'other'
]
if action1.category in valid_categories:
print(f"\n✓ Valid AI category selection: {action1.category}")
else:
print(f"\n⚠ WARNING: Invalid category: {action1.category}")
else:
print(f"\n❌ ERROR: No PX Action created for complaint {complaint1.id}")
print("PX Action creation should be MANDATORY for all complaints!")
return False
# Test Case 2: Service Quality Complaint
print_section("TEST CASE 2: Service Quality Complaint")
complaint2 = Complaint.objects.create(
title="Long wait time in emergency room",
description=(
"I waited 4 hours in the emergency room before seeing a doctor. "
"The staff was rude and didn't provide any updates. "
"The waiting area was overcrowded and uncomfortable."
),
patient=patient,
hospital=hospital,
department=department,
category=category,
status='open',
severity='high',
priority='high'
)
print(f"✓ Created complaint: {complaint2.title}")
# Run AI analysis
print("\n→ Running AI analysis...")
result2 = analyze_complaint_with_ai(str(complaint2.id))
complaint2.refresh_from_db()
print(f"\n✓ AI Analysis Result:")
print(f" Status: {result2.get('status')}")
print(f" Severity: {result2.get('severity')}{complaint2.severity}")
print(f" Priority: {result2.get('priority')}{complaint2.priority}")
print(f" PX Action Created: {result2.get('px_action_auto_created')}")
# Check PX Action
px_actions2 = PXAction.objects.filter(
source_type='complaint',
object_id=complaint2.id
)
if px_actions2.exists():
action2 = px_actions2.first()
print(f"\n✓ PX Action Created (MANDATORY):")
print(f" Action ID: {action2.id}")
print(f" Title: {action2.title}")
print(f" Category: {action2.category}")
print(f" Severity: {action2.severity}")
print(f" Priority: {action2.priority}")
else:
print(f"\n❌ ERROR: No PX Action created for complaint {complaint2.id}")
return False
# Test Case 3: Staff Behavior Complaint
print_section("TEST CASE 3: Staff Behavior Complaint")
complaint3 = Complaint.objects.create(
title="Nurse was disrespectful and unprofessional",
description=(
"The nurse on night shift, Sarah, was very rude to my elderly mother. "
"She spoke harshly and didn't provide proper care. "
"This is unacceptable behavior for healthcare professionals."
),
patient=patient,
hospital=hospital,
department=department,
category=category,
status='open',
severity='high',
priority='high'
)
print(f"✓ Created complaint: {complaint3.title}")
# Run AI analysis
print("\n→ Running AI analysis...")
result3 = analyze_complaint_with_ai(str(complaint3.id))
complaint3.refresh_from_db()
print(f"\n✓ AI Analysis Result:")
print(f" Status: {result3.get('status')}")
print(f" Severity: {result3.get('severity')}{complaint3.severity}")
print(f" Priority: {result3.get('priority')}{complaint3.priority}")
print(f" PX Action Created: {result3.get('px_action_auto_created')}")
# Check PX Action
px_actions3 = PXAction.objects.filter(
source_type='complaint',
object_id=complaint3.id
)
if px_actions3.exists():
action3 = px_actions3.first()
print(f"\n✓ PX Action Created (MANDATORY):")
print(f" Action ID: {action3.id}")
print(f" Title: {action3.title}")
print(f" Category: {action3.category}")
print(f" Severity: {action3.severity}")
print(f" Priority: {action3.priority}")
else:
print(f"\n❌ ERROR: No PX Action created for complaint {complaint3.id}")
return False
# Summary
print_section("TEST SUMMARY")
print(f"✅ All 3 complaints automatically created PX Actions (MANDATORY)")
print(f"✅ AI intelligently selected appropriate categories:")
print(f" 1. Clinical Quality: {action1.category}")
print(f" 2. Service Quality: {action2.category}")
print(f" 3. Staff Behavior: {action3.category}")
print(f"✅ No hospital configuration needed (mandatory for all)")
print(f"✅ AI generated titles and descriptions")
# Display available PX Action categories
print_section("AVAILABLE PX ACTION CATEGORIES")
print("""
The AI can select from these 7 categories:
• clinical_quality - Issues related to medical care quality, diagnosis, treatment
• patient_safety - Issues that could harm patients, safety violations, risks
• service_quality - Issues with service delivery, wait times, customer service
• staff_behavior - Issues with staff professionalism, attitude, conduct
• facility - Issues with facilities, equipment, environment, cleanliness
• process_improvement - Issues with processes, workflows, procedures
• other - General issues that don't fit specific categories
""")
return True
def test_category_selection():
"""Test AI category selection accuracy"""
print_section("TEST: AI CATEGORY SELECTION ACCURACY")
test_cases = [
{
'description': 'Doctor prescribed wrong medication',
'expected_category': 'clinical_quality',
'expected_severity': 'high'
},
{
'description': 'Nurse was rude and dismissive',
'expected_category': 'staff_behavior',
'expected_severity': 'medium'
},
{
'description': 'Long wait times in emergency room',
'expected_category': 'service_quality',
'expected_severity': 'medium'
},
{
'description': 'Hospital floors are dirty and unsanitary',
'expected_category': 'facility',
'expected_severity': 'medium'
},
{
'description': 'Patient fell due to wet floor',
'expected_category': 'patient_safety',
'expected_severity': 'critical'
}
]
from apps.core.ai_service import AIService
for i, test_case in enumerate(test_cases, 1):
print(f"\n{'='*60}")
print(f"Test Case {i}: {test_case['description']}")
print(f"{'='*60}")
try:
# Get hospital
hospital = Hospital.objects.first()
# Call AI to analyze
result = AIService.create_px_action_from_complaint(
complaint=type('MockComplaint', (), {
'title': test_case['description'][:50],
'description': test_case['description'],
'category': type('MockCategory', (), {'name_en': 'other'})(),
'severity': 'medium',
'priority': 'medium'
})()
)
print(f"✓ AI Selected Category: {result['category']}")
print(f" Expected Category: {test_case['expected_category']}")
if result['category'] == test_case['expected_category']:
print(f" ✅ CORRECT!")
else:
print(f" ⚠ DIFFERENT (AI may have different interpretation)")
print(f" Priority: {result['priority']}")
print(f" Severity: {result['severity']}")
except Exception as e:
print(f"❌ Error: {str(e)}")
print("\n" + "="*80)
print(" CATEGORY SELECTION TEST COMPLETE")
print("="*80 + "\n")
if __name__ == '__main__':
print("\n" + "#"*80)
print("# MANDATORY PX ACTION CREATION TEST")
print("# Testing: AI-powered automatic PX Action creation for ALL complaints")
print("#"*80)
try:
success = test_mandatory_px_action()
if success:
print("\n\n" + ""*40)
print("✅ ALL TESTS PASSED!")
print("✅ PX Action creation is MANDATORY for all complaints")
print("✅ AI intelligently selects categories")
print(""*40 + "\n")
# Optional: Run category selection test
response = input("\nRun category selection accuracy test? (y/n): ")
if response.lower() == 'y':
test_category_selection()
else:
print("\n\n" + ""*40)
print("❌ TESTS FAILED!")
print(""*40 + "\n")
sys.exit(1)
except Exception as e:
print(f"\n\n❌ FATAL ERROR: {str(e)}")
import traceback
traceback.print_exc()
sys.exit(1)