Spaces:
Sleeping
Sleeping
| #!/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__]) |