import pytest class TestIntegration: """Integration tests for the complete workflow""" def test_complete_workflow_text_input(self, sample_medical_text): """Test complete workflow from text input to SOAP output""" def mock_complete_workflow(medical_text): # Step 1: Validate input if not medical_text.strip(): return "❌ No input provided" # Step 2: Process text (mock preprocessing) processed_text = medical_text.strip() # Step 3: Generate SOAP (mocked) soap_sections = [ "SUBJECTIVE:", "Patient reports chest pain for 2 hours.", "", "OBJECTIVE:", "Vital signs show elevated blood pressure.", "", "ASSESSMENT:", "Acute chest pain, rule out cardiac causes.", "", "PLAN:", "Order EKG and cardiac enzymes." ] # Step 4: Format output result = "\n".join(soap_sections) result += f"\n\n--- Processing Summary ---" result += f"\nProcessed {len(medical_text)} characters of medical text." result += f"\nGenerated SOAP note with {len(soap_sections)} sections." return result result = mock_complete_workflow(sample_medical_text) assert "SUBJECTIVE:" in result assert "OBJECTIVE:" in result assert "ASSESSMENT:" in result assert "PLAN:" in result assert "Processing Summary" in result assert "characters of medical text" in result def test_workflow_with_image_and_text(self, sample_medical_text, sample_image): """Test workflow with both image and text input""" def mock_workflow_with_image(text_input, image_input): combined_text = text_input or "" if image_input is not None: # Mock OCR processing ocr_text = "Additional findings: Patient appears anxious, diaphoretic" if combined_text: combined_text += f"\n\n--- Extracted from image ---\n{ocr_text}" else: combined_text = ocr_text if not combined_text: return "❌ No input provided" # Mock SOAP generation sections = ["SUBJECTIVE", "OBJECTIVE", "ASSESSMENT", "PLAN"] soap_content = [] for section in sections: soap_content.append(f"{section}:") soap_content.append(f"Content for {section.lower()} section") soap_content.append("") result = "\n".join(soap_content) result += f"\n--- Workflow Summary ---" result += f"\nCombined input: {len(combined_text)} characters" result += f"\nImage processing: {'Yes' if image_input else 'No'}" return result result = mock_workflow_with_image(sample_medical_text, sample_image) assert "SUBJECTIVE:" in result assert "OBJECTIVE:" in result assert "ASSESSMENT:" in result assert "PLAN:" in result assert "Workflow Summary" in result assert "Image processing: Yes" in result def test_error_recovery_workflow(self): """Test workflow error handling and recovery""" def mock_error_recovery_workflow(input_data, simulate_error=False): try: if simulate_error: raise Exception("Simulated processing error") if not input_data: return "❌ No input provided" # Simulate successful processing return "✅ SOAP note generated successfully" except Exception as e: # Error recovery error_msg = f"❌ Error occurred: {str(e)}" recovery_msg = "\n💡 Please try again with different input or check your image quality." return error_msg + recovery_msg # Test normal operation result = mock_error_recovery_workflow("Valid input") assert "✅ SOAP note generated successfully" in result # Test error handling result = mock_error_recovery_workflow("Input", simulate_error=True) assert "❌ Error occurred:" in result assert "💡 Please try again" in result # Test empty input result = mock_error_recovery_workflow("") assert "❌ No input provided" in result def test_performance_workflow(self, sample_medical_text): """Test workflow performance characteristics""" def mock_performance_workflow(input_text): import time start_time = time.time() # Simulate processing steps steps = [ "Validating input", "Preprocessing text", "Extracting medical entities", "Generating SOAP structure", "Formatting output" ] processing_log = [] for i, step in enumerate(steps): step_time = time.time() processing_log.append(f"Step {i+1}: {step} - {step_time - start_time:.3f}s") time.sleep(0.01) # Simulate processing time total_time = time.time() - start_time result = "Generated SOAP Note\n\n" result += "--- Performance Log ---\n" result += "\n".join(processing_log) result += f"\nTotal processing time: {total_time:.3f}s" return result, total_time result, processing_time = mock_performance_workflow(sample_medical_text) assert "Generated SOAP Note" in result assert "Performance Log" in result assert "Total processing time:" in result assert processing_time > 0 assert processing_time < 1.0 # Should be fast for mocked version def test_batch_processing_workflow(self): """Test batch processing of multiple medical notes""" def mock_batch_workflow(medical_notes_list): if not medical_notes_list: return "❌ No notes provided for batch processing" results = [] for i, notes in enumerate(medical_notes_list): if not notes.strip(): results.append(f"Note {i+1}: ❌ Empty input") continue # Mock SOAP generation for each note soap_result = f"""Note {i+1} - SOAP Generated: SUBJECTIVE: Patient presentation from note {i+1} OBJECTIVE: Clinical findings ASSESSMENT: Medical diagnosis PLAN: Treatment approach """ results.append(soap_result) summary = f"\n--- Batch Summary ---" summary += f"\nTotal notes processed: {len(medical_notes_list)}" summary += f"\nSuccessful: {len([r for r in results if '❌' not in r])}" summary += f"\nFailed: {len([r for r in results if '❌' in r])}" return "\n".join(results) + summary # Test batch processing notes_batch = [ "Patient 1: Chest pain complaint", "Patient 2: Diabetes follow-up", "", # Empty note "Patient 3: Pediatric fever case" ] result = mock_batch_workflow(notes_batch) assert "Note 1 - SOAP Generated:" in result assert "Note 2 - SOAP Generated:" in result assert "Note 3: ❌ Empty input" in result assert "Note 4 - SOAP Generated:" in result assert "Batch Summary" in result assert "Total notes processed: 4" in result assert "Successful: 3" in result assert "Failed: 1" in result def test_gradio_integration(self, sample_medical_text, sample_image): """Test Gradio interface integration""" def mock_gradio_integration(text_input, image_input, example_selection=None): # Simulate Gradio interface behavior if example_selection: # Load example examples = { "chest_pain": "Example chest pain case", "diabetes": "Example diabetes case", "pediatric": "Example pediatric case" } text_input = examples.get(example_selection, text_input) # Process inputs (same as gradio_generate_soap) final_text = text_input or "" if image_input is not None: ocr_result = "OCR extracted: Patient vital signs documented" if final_text: final_text += f"\n\n--- From Image ---\n{ocr_result}" else: final_text = ocr_result if not final_text: return "❌ Please provide input" # Generate SOAP soap_output = f"""📋 SOAP NOTE GENERATED SUBJECTIVE: {final_text[:100]}... OBJECTIVE: Clinical examination findings documented. ASSESSMENT: Medical evaluation completed. PLAN: Treatment recommendations provided. --- Generated via Gradio Interface --- Input sources: {'Text + Image' if text_input and image_input else 'Text' if text_input else 'Image'} Processing timestamp: 2025-01-15 10:30:00 """ return soap_output # Test text only result = mock_gradio_integration(sample_medical_text, None) assert "📋 SOAP NOTE GENERATED" in result assert "Input sources: Text" in result # Test image only result = mock_gradio_integration("", sample_image) assert "📋 SOAP NOTE GENERATED" in result assert "From Image" in result # Test text + image result = mock_gradio_integration(sample_medical_text, sample_image) assert "Input sources: Text + Image" in result # Test example selection result = mock_gradio_integration("", None, "chest_pain") assert "Example chest pain case" in result