#!/usr/bin/env python3 """ Test script for the consent response processor. Tests Task 5.3 implementation. """ import sys import os sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'src')) from config.prompt_management.consent_response_processor import ( ConsentResponseProcessor, ProcessingAction, ReferralUrgency, ProcessingResult ) from config.prompt_management.consent_manager import ConsentInteraction, ConsentMessageType, ConsentResponse from datetime import datetime def test_consent_response_processor_initialization(): """Test that the consent response processor initializes correctly.""" print("Testing consent response processor initialization...") processor = ConsentResponseProcessor() # Verify components are initialized assert hasattr(processor, 'consent_manager') assert hasattr(processor, 'message_generator') assert hasattr(processor, 'processing_rules') assert hasattr(processor, 'medical_transition_phrases') # Verify processing rules structure assert 'clarification_attempts_limit' in processor.processing_rules assert 'follow_up_delay_hours' in processor.processing_rules assert 'urgency_indicators' in processor.processing_rules # Verify urgency indicators urgency_indicators = processor.processing_rules['urgency_indicators'] assert 'high' in urgency_indicators assert 'medium' in urgency_indicators assert 'low' in urgency_indicators # Verify medical transition phrases assert len(processor.medical_transition_phrases) > 0 for phrase in processor.medical_transition_phrases: assert isinstance(phrase, str) assert len(phrase) > 0 print("✓ Consent response processor initializes correctly") def test_acceptance_processing(): """Test processing of patient acceptance responses.""" print("Testing acceptance processing...") processor = ConsentResponseProcessor() # Test basic acceptance accept_response = "Yes, I would like to speak with someone" context = {'distress_level': 'medium', 'message_content': 'I feel overwhelmed'} result = processor.process_patient_response(accept_response, "test_session_1", context) # Verify result structure assert isinstance(result, ProcessingResult) assert result.action == ProcessingAction.PROCEED_WITH_REFERRAL assert result.generate_provider_summary == True assert result.log_referral == True assert result.referral_urgency is not None assert result.requires_follow_up == False # Verify interaction record interaction = result.interaction_record assert interaction.patient_response == accept_response assert interaction.response_classification == ConsentResponse.ACCEPT assert interaction.message_type == ConsentMessageType.CONFIRMATION # Verify next steps assert len(result.next_steps) > 0 assert any('provider summary' in step.lower() for step in result.next_steps) assert any('log referral' in step.lower() for step in result.next_steps) # Verify context updates assert result.context_updates['consent_status'] == 'accepted' assert result.context_updates['provider_contact_required'] == True print("✓ Acceptance processing works correctly") def test_decline_processing(): """Test processing of patient decline responses.""" print("Testing decline processing...") processor = ConsentResponseProcessor() # Test decline response decline_response = "No, I'm fine" context = {'distress_level': 'low'} result = processor.process_patient_response(decline_response, "test_session_2", context) # Verify result structure assert result.action == ProcessingAction.RETURN_TO_MEDICAL_DIALOGUE assert result.generate_provider_summary == False assert result.log_referral == False assert result.referral_urgency is None assert result.requires_follow_up == False # Verify message contains both acknowledgment and medical transition message = result.message assert len(message) > 0 # Should contain elements from both acknowledgment and transition assert any(word in message.lower() for word in ['understand', 'respect', 'decision']) assert any(word in message.lower() for word in ['medical', 'healthcare', 'continue']) # Verify interaction record interaction = result.interaction_record assert interaction.patient_response == decline_response assert interaction.response_classification == ConsentResponse.DECLINE assert interaction.message_type == ConsentMessageType.DECLINE_ACKNOWLEDGMENT # Verify next steps include medical dialogue return assert any('medical dialogue' in step.lower() for step in result.next_steps) assert any('healthcare' in step.lower() for step in result.next_steps) # Verify context updates assert result.context_updates['consent_status'] == 'declined' assert result.context_updates['spiritual_care_declined'] == True assert result.context_updates['return_to_medical_dialogue'] == True print("✓ Decline processing works correctly") def test_ambiguous_response_processing(): """Test processing of ambiguous patient responses.""" print("Testing ambiguous response processing...") processor = ConsentResponseProcessor() # Test ambiguous response ambiguous_response = "I don't know, what would that involve?" context = {'distress_level': 'medium'} result = processor.process_patient_response(ambiguous_response, "test_session_3", context) # Verify result structure assert result.action == ProcessingAction.REQUEST_CLARIFICATION assert result.generate_provider_summary == False assert result.log_referral == False assert result.referral_urgency is None assert result.requires_follow_up == True assert result.follow_up_delay_hours is not None # Verify interaction record interaction = result.interaction_record assert interaction.patient_response == ambiguous_response assert interaction.response_classification == ConsentResponse.AMBIGUOUS assert interaction.message_type == ConsentMessageType.CLARIFICATION assert interaction.requires_clarification == True assert interaction.clarification_attempts == 1 # Verify clarification message assert len(result.message) > 0 # Should be informative for information-seeking ambiguity assert any(word in result.message.lower() for word in ['chaplain', 'counselor', 'support', 'team']) # Verify context updates assert result.context_updates['consent_status'] == 'clarification_needed' assert result.context_updates['clarification_attempts'] == 1 assert result.context_updates['awaiting_clarification'] == True print("✓ Ambiguous response processing works correctly") def test_clarification_attempt_limits(): """Test clarification attempt limits and escalation.""" print("Testing clarification attempt limits...") processor = ConsentResponseProcessor() # Create interaction history with multiple clarification attempts interaction_history = [] for i in range(3): # 3 previous clarification attempts interaction = ConsentInteraction( interaction_id=f"test_{i}", message_type=ConsentMessageType.CLARIFICATION, message_content=f"Clarification {i}", patient_response=f"Response {i}", response_classification=ConsentResponse.AMBIGUOUS, timestamp=datetime.now(), session_id="test_session_4", requires_clarification=True, clarification_attempts=i ) interaction_history.append(interaction) # Test response that would exceed limit ambiguous_response = "I'm still not sure" result = processor.process_patient_response( ambiguous_response, "test_session_4", {}, interaction_history ) # Should escalate to human assert result.action == ProcessingAction.ESCALATE_TO_HUMAN assert result.requires_follow_up == True assert result.follow_up_delay_hours == 4 # Human review within 4 hours # Verify context updates indicate escalation assert result.context_updates['consent_status'] == 'escalated_to_human' assert result.context_updates['human_review_required'] == True assert 'escalation_reason' in result.context_updates print("✓ Clarification attempt limits work correctly") def test_referral_urgency_determination(): """Test referral urgency determination based on context.""" print("Testing referral urgency determination...") processor = ConsentResponseProcessor() # Test urgent context urgent_context = { 'message_content': 'I am in crisis and need immediate help', 'distress_level': 'high' } urgent_urgency = processor._determine_referral_urgency(urgent_context) assert urgent_urgency == ReferralUrgency.URGENT # Test high urgency context high_context = { 'message_content': 'I am struggling with significant distress', 'distress_level': 'high' } high_urgency = processor._determine_referral_urgency(high_context) assert high_urgency == ReferralUrgency.HIGH # Test medium urgency context medium_context = { 'message_content': 'I am feeling overwhelmed', 'distress_level': 'medium' } medium_urgency = processor._determine_referral_urgency(medium_context) assert medium_urgency == ReferralUrgency.MEDIUM # Test low urgency context low_context = { 'message_content': 'I could use some support', 'distress_level': 'low' } low_urgency = processor._determine_referral_urgency(low_context) assert low_urgency == ReferralUrgency.LOW print("✓ Referral urgency determination works correctly") def test_follow_up_delay_calculation(): """Test follow-up delay calculation based on attempts.""" print("Testing follow-up delay calculation...") processor = ConsentResponseProcessor() # Test first attempt delay first_delay = processor._get_follow_up_delay(0) assert first_delay == 24 # 24 hours for first attempt # Test second attempt delay second_delay = processor._get_follow_up_delay(1) assert second_delay == 72 # 72 hours for second attempt # Test final attempt delay final_delay = processor._get_follow_up_delay(2) assert final_delay == 168 # 168 hours (1 week) for final attempt # Test beyond limit beyond_delay = processor._get_follow_up_delay(5) assert beyond_delay == 168 # Should use final attempt delay print("✓ Follow-up delay calculation works correctly") def test_processing_statistics(): """Test processing statistics generation.""" print("Testing processing statistics...") processor = ConsentResponseProcessor() # Create test interaction history interactions = [ ConsentInteraction( interaction_id="stat_1", message_type=ConsentMessageType.INITIAL_REQUEST, message_content="Initial request", patient_response="Yes", response_classification=ConsentResponse.ACCEPT, timestamp=datetime.now(), session_id="stat_session", clarification_attempts=0 ), ConsentInteraction( interaction_id="stat_2", message_type=ConsentMessageType.CLARIFICATION, message_content="Clarification", patient_response="Maybe", response_classification=ConsentResponse.AMBIGUOUS, timestamp=datetime.now(), session_id="stat_session", requires_clarification=True, clarification_attempts=1 ), ConsentInteraction( interaction_id="stat_3", message_type=ConsentMessageType.DECLINE_ACKNOWLEDGMENT, message_content="Acknowledgment", patient_response="No", response_classification=ConsentResponse.DECLINE, timestamp=datetime.now(), session_id="stat_session", clarification_attempts=0 ) ] # Generate statistics stats = processor.get_processing_statistics(interactions) # Verify statistics structure required_fields = ['total_interactions', 'response_type_counts', 'message_type_counts', 'successful_resolutions', 'resolution_rate', 'clarification_rate', 'average_clarification_attempts'] for field in required_fields: assert field in stats, f"Statistics missing required field: {field}" # Verify counts assert stats['total_interactions'] == 3 assert stats['response_type_counts']['accept'] == 1 assert stats['response_type_counts']['decline'] == 1 assert stats['response_type_counts']['ambiguous'] == 1 # Verify rates assert stats['successful_resolutions'] == 2 # Accept + Decline assert stats['resolution_rate'] == 2/3 # 2 out of 3 assert stats['clarification_rate'] == 1/3 # 1 out of 3 # Test empty interactions empty_stats = processor.get_processing_statistics([]) assert empty_stats['total_interactions'] == 0 print("✓ Processing statistics work correctly") def test_unclear_response_processing(): """Test processing of unclear patient responses.""" print("Testing unclear response processing...") processor = ConsentResponseProcessor() # Test unclear response (something that doesn't match any pattern) unclear_response = "Banana elephant purple" context = {'distress_level': 'medium'} result = processor.process_patient_response(unclear_response, "test_session_5", context) # Verify result structure assert result.action == ProcessingAction.REQUEST_CLARIFICATION assert result.generate_provider_summary == False assert result.log_referral == False assert result.requires_follow_up == True # Verify interaction record interaction = result.interaction_record assert interaction.patient_response == unclear_response assert interaction.response_classification == ConsentResponse.UNCLEAR assert interaction.message_type == ConsentMessageType.CLARIFICATION assert interaction.requires_clarification == True # Verify clarification message is general assert len(result.message) > 0 assert 'understand your preferences' in result.message.lower() # Verify context updates assert result.context_updates['consent_status'] == 'unclear_response' assert result.context_updates['response_clarity_issues'] == True print("✓ Unclear response processing works correctly") def test_processing_result_serialization(): """Test ProcessingResult serialization.""" print("Testing ProcessingResult serialization...") processor = ConsentResponseProcessor() # Process a response to get a result result = processor.process_patient_response("Yes, please", "test_session_6") # Test serialization result_dict = result.to_dict() # Verify serialized structure required_fields = ['action', 'message', 'generate_provider_summary', 'log_referral', 'referral_urgency', 'requires_follow_up', 'follow_up_delay_hours', 'interaction_record', 'next_steps', 'context_updates'] for field in required_fields: assert field in result_dict, f"Serialized result missing field: {field}" # Verify data types assert isinstance(result_dict['action'], str) assert isinstance(result_dict['message'], str) assert isinstance(result_dict['generate_provider_summary'], bool) assert isinstance(result_dict['next_steps'], list) assert isinstance(result_dict['context_updates'], dict) assert isinstance(result_dict['interaction_record'], dict) print("✓ ProcessingResult serialization works correctly") def main(): """Run all consent response processor tests.""" print("=" * 60) print("CONSENT RESPONSE PROCESSOR TESTS") print("=" * 60) tests = [ test_consent_response_processor_initialization, test_acceptance_processing, test_decline_processing, test_ambiguous_response_processing, test_clarification_attempt_limits, test_referral_urgency_determination, test_follow_up_delay_calculation, test_processing_statistics, test_unclear_response_processing, test_processing_result_serialization ] passed = 0 failed = 0 for test in tests: try: print(f"\n{test.__name__.replace('_', ' ').title()}:") print("-" * 40) test() # Just call the test, don't expect return value passed += 1 print("✓ PASSED") except Exception as e: failed += 1 print(f"✗ FAILED: {str(e)}") print("\n" + "=" * 60) print(f"RESULTS: {passed} passed, {failed} failed") print("=" * 60) if failed == 0: print("🎉 All consent response processor tests passed!") print("\n**Task 5.3: Enhanced Consent Response Processing**") print("✓ COMPLETED: Patient decline handling with medical dialogue return") print("✓ COMPLETED: Acceptance processing with referral generation") print("✓ COMPLETED: Ambiguous response clarification workflow") print("✓ COMPLETED: Referral urgency determination") print("✓ COMPLETED: Clarification attempt limits and escalation") print("✓ VALIDATED: Requirements 4.2, 4.3, 4.4") return True else: print("❌ Some tests failed. Please check the implementation.") return False if __name__ == "__main__": success = main() sys.exit(0 if success else 1)