#!/usr/bin/env python3 """ Integration tests for enhanced verification system. Tests the integration between enhanced display formats, verification system, and CSV export functionality. Requirements: 8.2, 8.3, 8.4, 8.5 """ import pytest import tempfile import os from datetime import datetime from unittest.mock import Mock from src.core.conversation_logger import ConversationLogger, ConversationEntry from src.core.conversation_verification import ( EnhancedConversationVerificationManager, EnhancedVerificationSession, EnhancedVerificationRecord, VerificationFeedback ) from src.core.verification_exporter import EnhancedVerificationExporter from src.core.spiritual_state import SpiritualState, SpiritualAssessment from src.core.provider_summary_generator import ProviderSummary class TestEnhancedVerificationIntegration: """Test enhanced verification system integration.""" def setup_method(self): """Set up test environment.""" self.temp_dir = tempfile.mkdtemp() self.verification_manager = EnhancedConversationVerificationManager(self.temp_dir) self.exporter = EnhancedVerificationExporter(self.temp_dir) def teardown_method(self): """Clean up test environment.""" import shutil shutil.rmtree(self.temp_dir, ignore_errors=True) def create_test_conversation_logger(self) -> ConversationLogger: """Create a test conversation logger with sample data.""" logger = ConversationLogger(patient_name="Test Patient") # Add sample conversation entries assessments = [ SpiritualAssessment( state=SpiritualState.RED, confidence=0.85, indicators=["loss of meaning", "spiritual distress", "questioning faith"], reasoning="Patient expressing significant spiritual concerns and loss of meaning" ), SpiritualAssessment( state=SpiritualState.YELLOW, confidence=0.65, indicators=["mild anxiety", "uncertainty"], reasoning="Patient showing some concern but not severe distress" ), SpiritualAssessment( state=SpiritualState.GREEN, confidence=0.90, indicators=[], reasoning="Patient appears stable and content" ) ] messages = [ ("I've been struggling with the meaning of my illness", "I understand this is a difficult time..."), ("Sometimes I wonder if there's any point to all this", "Your feelings are completely valid..."), ("I'm feeling a bit better today", "That's wonderful to hear...") ] for (user_msg, assistant_msg), assessment in zip(messages, assessments): logger.log_exchange(user_msg, assistant_msg, assessment) return logger def test_enhanced_verification_session_creation(self): """Test creating enhanced verification session with new formats.""" # Create test conversation logger = self.create_test_conversation_logger() # Create enhanced verification session session = self.verification_manager.create_verification_session( logger, verifier_name="Test Verifier", enable_enhanced_formats=True ) # Verify session properties assert isinstance(session, EnhancedVerificationSession) assert session.enhanced_format_enabled is True assert len(session.verification_records) == 3 # Verify enhanced records for record in session.verification_records: assert isinstance(record, EnhancedVerificationRecord) assert record.enhanced_display_format is not None assert record.visual_sections is not None assert len(record.visual_sections) >= 2 # AI analysis + patient message def test_enhanced_csv_export_with_new_data(self): """Test CSV export includes enhanced format data.""" # Create and verify session logger = self.create_test_conversation_logger() session = self.verification_manager.create_verification_session( logger, enable_enhanced_formats=True ) # Add some verification feedback feedback = VerificationFeedback( exchange_id=session.verification_records[0].exchange_id, is_correct=True ) self.verification_manager.submit_exchange_verification( session.session_id, session.verification_records[0].exchange_id, feedback ) # Export to CSV with enhanced data csv_path = self.exporter.export_session_to_csv(session, include_enhanced_data=True) # Verify CSV file exists and contains enhanced data assert os.path.exists(csv_path) with open(csv_path, 'r', encoding='utf-8') as f: content = f.read() # Check for enhanced format columns assert 'has_enhanced_display' in content assert 'visual_sections_count' in content assert 'enhanced_indicators_count' in content assert 'enhanced_display_preview' in content # Check for enhanced session metadata assert 'Enhanced Format: True' in content def test_enhanced_summary_report_generation(self): """Test enhanced summary report includes new format statistics.""" # Create session with enhanced formats logger = self.create_test_conversation_logger() session = self.verification_manager.create_verification_session( logger, enable_enhanced_formats=True ) # Generate enhanced summary report report_path = self.exporter.export_enhanced_summary_report(session) # Verify report file exists assert os.path.exists(report_path) with open(report_path, 'r', encoding='utf-8') as f: content = f.read() # Check for enhanced format statistics assert 'ENHANCED FORMAT STATISTICS' in content assert 'Records with Enhanced Display:' in content assert 'Enhanced Display Coverage:' in content assert 'Enhanced Format Enabled: True' in content def test_verification_compatibility_with_existing_system(self): """Test that enhanced verification is compatible with existing verification system.""" # Create session without enhanced formats (legacy mode) logger = self.create_test_conversation_logger() legacy_session = self.verification_manager.create_verification_session( logger, enable_enhanced_formats=False ) # Verify it still works with basic verification feedback = VerificationFeedback( exchange_id=legacy_session.verification_records[0].exchange_id, is_correct=False, correct_classification="YELLOW", correction_reason="Should be yellow not red" ) success = self.verification_manager.submit_exchange_verification( legacy_session.session_id, legacy_session.verification_records[0].exchange_id, feedback ) assert success is True # Verify statistics still work stats = self.verification_manager.get_session_statistics(legacy_session.session_id) assert stats is not None assert 'enhanced_format_enabled' in stats assert stats['enhanced_format_enabled'] is False def test_provider_summary_integration_with_verification(self): """Test that provider summaries are properly integrated with verification.""" # Create session with provider summary data logger = self.create_test_conversation_logger() session = self.verification_manager.create_verification_session( logger, enable_enhanced_formats=True ) # Add provider summary to a record test_summary = ProviderSummary( patient_name="Test Patient", patient_phone="555-0123", classification="RED", confidence=0.85, indicators=["loss of meaning", "spiritual distress"], reasoning="Significant spiritual concerns detected", urgency_level="IMMEDIATE", severity_level="HIGH" ) # Set provider summary on first record session.verification_records[0].set_enhanced_formats( provider_summary=test_summary ) # Save updated session self.verification_manager.save_session(session) # Export with enhanced data export_data = self.verification_manager.export_session_with_enhanced_data(session.session_id) # Verify provider summary data is included assert export_data is not None record_data = export_data['records'][0] assert record_data['provider_summary'] is not None assert record_data['provider_summary']['urgency_level'] == 'IMMEDIATE' assert record_data['provider_summary']['severity_level'] == 'HIGH' def test_coherent_paragraph_format_in_verification(self): """Test that coherent paragraph format is properly handled in verification.""" # Create session logger = self.create_test_conversation_logger() session = self.verification_manager.create_verification_session( logger, enable_enhanced_formats=True ) # Add coherent paragraph to a record coherent_paragraph = "Test Patient is a 45-year-old individual with clinical history of chronic illness. The patient expressed loss of meaning and spiritual distress, which may indicate significant spiritual concerns, resulting in generation of a RED FLAG. The patient has been identified for spiritual care team contact. The preferred contact number is 555-0123." session.verification_records[0].coherent_summary_paragraph = coherent_paragraph # Save and export self.verification_manager.save_session(session) csv_path = self.exporter.export_session_to_csv(session, include_enhanced_data=True) # Verify coherent paragraph data is in CSV with open(csv_path, 'r', encoding='utf-8') as f: content = f.read() # Should have coherent paragraph length > 0 lines = content.split('\n') data_lines = [line for line in lines if line and not line.startswith('#') and 'coherent_paragraph_length' not in line] if len(data_lines) > 2: # Header + data rows # Find the data row for the first record for line in data_lines[2:]: # Skip header and metadata if line.strip(): fields = line.split(',') if len(fields) > 15: # Should have enhanced fields coherent_length_field = fields[-5] # coherent_paragraph_length field if coherent_length_field.isdigit(): assert int(coherent_length_field) > 0 break if __name__ == "__main__": pytest.main([__file__])