Spaces:
Sleeping
Sleeping
Create test_integration.py
Browse files- tests/test_integration.py +270 -0
tests/test_integration.py
ADDED
@@ -0,0 +1,270 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
1 |
+
import pytest
|
2 |
+
|
3 |
+
class TestIntegration:
|
4 |
+
"""Integration tests for the complete workflow"""
|
5 |
+
|
6 |
+
def test_complete_workflow_text_input(self, sample_medical_text):
|
7 |
+
"""Test complete workflow from text input to SOAP output"""
|
8 |
+
def mock_complete_workflow(medical_text):
|
9 |
+
# Step 1: Validate input
|
10 |
+
if not medical_text.strip():
|
11 |
+
return "β No input provided"
|
12 |
+
|
13 |
+
# Step 2: Process text (mock preprocessing)
|
14 |
+
processed_text = medical_text.strip()
|
15 |
+
|
16 |
+
# Step 3: Generate SOAP (mocked)
|
17 |
+
soap_sections = [
|
18 |
+
"SUBJECTIVE:",
|
19 |
+
"Patient reports chest pain for 2 hours.",
|
20 |
+
"",
|
21 |
+
"OBJECTIVE:",
|
22 |
+
"Vital signs show elevated blood pressure.",
|
23 |
+
"",
|
24 |
+
"ASSESSMENT:",
|
25 |
+
"Acute chest pain, rule out cardiac causes.",
|
26 |
+
"",
|
27 |
+
"PLAN:",
|
28 |
+
"Order EKG and cardiac enzymes."
|
29 |
+
]
|
30 |
+
|
31 |
+
# Step 4: Format output
|
32 |
+
result = "\n".join(soap_sections)
|
33 |
+
result += f"\n\n--- Processing Summary ---"
|
34 |
+
result += f"\nProcessed {len(medical_text)} characters of medical text."
|
35 |
+
result += f"\nGenerated SOAP note with {len(soap_sections)} sections."
|
36 |
+
|
37 |
+
return result
|
38 |
+
|
39 |
+
result = mock_complete_workflow(sample_medical_text)
|
40 |
+
|
41 |
+
assert "SUBJECTIVE:" in result
|
42 |
+
assert "OBJECTIVE:" in result
|
43 |
+
assert "ASSESSMENT:" in result
|
44 |
+
assert "PLAN:" in result
|
45 |
+
assert "Processing Summary" in result
|
46 |
+
assert "characters of medical text" in result
|
47 |
+
|
48 |
+
def test_workflow_with_image_and_text(self, sample_medical_text, sample_image):
|
49 |
+
"""Test workflow with both image and text input"""
|
50 |
+
def mock_workflow_with_image(text_input, image_input):
|
51 |
+
combined_text = text_input or ""
|
52 |
+
|
53 |
+
if image_input is not None:
|
54 |
+
# Mock OCR processing
|
55 |
+
ocr_text = "Additional findings: Patient appears anxious, diaphoretic"
|
56 |
+
if combined_text:
|
57 |
+
combined_text += f"\n\n--- Extracted from image ---\n{ocr_text}"
|
58 |
+
else:
|
59 |
+
combined_text = ocr_text
|
60 |
+
|
61 |
+
if not combined_text:
|
62 |
+
return "β No input provided"
|
63 |
+
|
64 |
+
# Mock SOAP generation
|
65 |
+
sections = ["SUBJECTIVE", "OBJECTIVE", "ASSESSMENT", "PLAN"]
|
66 |
+
soap_content = []
|
67 |
+
|
68 |
+
for section in sections:
|
69 |
+
soap_content.append(f"{section}:")
|
70 |
+
soap_content.append(f"Content for {section.lower()} section")
|
71 |
+
soap_content.append("")
|
72 |
+
|
73 |
+
result = "\n".join(soap_content)
|
74 |
+
result += f"\n--- Workflow Summary ---"
|
75 |
+
result += f"\nCombined input: {len(combined_text)} characters"
|
76 |
+
result += f"\nImage processing: {'Yes' if image_input else 'No'}"
|
77 |
+
|
78 |
+
return result
|
79 |
+
|
80 |
+
result = mock_workflow_with_image(sample_medical_text, sample_image)
|
81 |
+
assert "SUBJECTIVE:" in result
|
82 |
+
assert "OBJECTIVE:" in result
|
83 |
+
assert "ASSESSMENT:" in result
|
84 |
+
assert "PLAN:" in result
|
85 |
+
assert "Workflow Summary" in result
|
86 |
+
assert "Image processing: Yes" in result
|
87 |
+
|
88 |
+
def test_error_recovery_workflow(self):
|
89 |
+
"""Test workflow error handling and recovery"""
|
90 |
+
def mock_error_recovery_workflow(input_data, simulate_error=False):
|
91 |
+
try:
|
92 |
+
if simulate_error:
|
93 |
+
raise Exception("Simulated processing error")
|
94 |
+
|
95 |
+
if not input_data:
|
96 |
+
return "β No input provided"
|
97 |
+
|
98 |
+
# Simulate successful processing
|
99 |
+
return "β
SOAP note generated successfully"
|
100 |
+
|
101 |
+
except Exception as e:
|
102 |
+
# Error recovery
|
103 |
+
error_msg = f"β Error occurred: {str(e)}"
|
104 |
+
recovery_msg = "\nπ‘ Please try again with different input or check your image quality."
|
105 |
+
return error_msg + recovery_msg
|
106 |
+
|
107 |
+
# Test normal operation
|
108 |
+
result = mock_error_recovery_workflow("Valid input")
|
109 |
+
assert "β
SOAP note generated successfully" in result
|
110 |
+
|
111 |
+
# Test error handling
|
112 |
+
result = mock_error_recovery_workflow("Input", simulate_error=True)
|
113 |
+
assert "β Error occurred:" in result
|
114 |
+
assert "π‘ Please try again" in result
|
115 |
+
|
116 |
+
# Test empty input
|
117 |
+
result = mock_error_recovery_workflow("")
|
118 |
+
assert "β No input provided" in result
|
119 |
+
|
120 |
+
def test_performance_workflow(self, sample_medical_text):
|
121 |
+
"""Test workflow performance characteristics"""
|
122 |
+
def mock_performance_workflow(input_text):
|
123 |
+
import time
|
124 |
+
|
125 |
+
start_time = time.time()
|
126 |
+
|
127 |
+
# Simulate processing steps
|
128 |
+
steps = [
|
129 |
+
"Validating input",
|
130 |
+
"Preprocessing text",
|
131 |
+
"Extracting medical entities",
|
132 |
+
"Generating SOAP structure",
|
133 |
+
"Formatting output"
|
134 |
+
]
|
135 |
+
|
136 |
+
processing_log = []
|
137 |
+
for i, step in enumerate(steps):
|
138 |
+
step_time = time.time()
|
139 |
+
processing_log.append(f"Step {i+1}: {step} - {step_time - start_time:.3f}s")
|
140 |
+
time.sleep(0.01) # Simulate processing time
|
141 |
+
|
142 |
+
total_time = time.time() - start_time
|
143 |
+
|
144 |
+
result = "Generated SOAP Note\n\n"
|
145 |
+
result += "--- Performance Log ---\n"
|
146 |
+
result += "\n".join(processing_log)
|
147 |
+
result += f"\nTotal processing time: {total_time:.3f}s"
|
148 |
+
|
149 |
+
return result, total_time
|
150 |
+
|
151 |
+
result, processing_time = mock_performance_workflow(sample_medical_text)
|
152 |
+
|
153 |
+
assert "Generated SOAP Note" in result
|
154 |
+
assert "Performance Log" in result
|
155 |
+
assert "Total processing time:" in result
|
156 |
+
assert processing_time > 0
|
157 |
+
assert processing_time < 1.0 # Should be fast for mocked version
|
158 |
+
|
159 |
+
def test_batch_processing_workflow(self):
|
160 |
+
"""Test batch processing of multiple medical notes"""
|
161 |
+
def mock_batch_workflow(medical_notes_list):
|
162 |
+
if not medical_notes_list:
|
163 |
+
return "β No notes provided for batch processing"
|
164 |
+
|
165 |
+
results = []
|
166 |
+
|
167 |
+
for i, notes in enumerate(medical_notes_list):
|
168 |
+
if not notes.strip():
|
169 |
+
results.append(f"Note {i+1}: β Empty input")
|
170 |
+
continue
|
171 |
+
|
172 |
+
# Mock SOAP generation for each note
|
173 |
+
soap_result = f"""Note {i+1} - SOAP Generated:
|
174 |
+
SUBJECTIVE: Patient presentation from note {i+1}
|
175 |
+
OBJECTIVE: Clinical findings
|
176 |
+
ASSESSMENT: Medical diagnosis
|
177 |
+
PLAN: Treatment approach
|
178 |
+
"""
|
179 |
+
results.append(soap_result)
|
180 |
+
|
181 |
+
summary = f"\n--- Batch Summary ---"
|
182 |
+
summary += f"\nTotal notes processed: {len(medical_notes_list)}"
|
183 |
+
summary += f"\nSuccessful: {len([r for r in results if 'β' not in r])}"
|
184 |
+
summary += f"\nFailed: {len([r for r in results if 'β' in r])}"
|
185 |
+
|
186 |
+
return "\n".join(results) + summary
|
187 |
+
|
188 |
+
# Test batch processing
|
189 |
+
notes_batch = [
|
190 |
+
"Patient 1: Chest pain complaint",
|
191 |
+
"Patient 2: Diabetes follow-up",
|
192 |
+
"", # Empty note
|
193 |
+
"Patient 3: Pediatric fever case"
|
194 |
+
]
|
195 |
+
|
196 |
+
result = mock_batch_workflow(notes_batch)
|
197 |
+
|
198 |
+
assert "Note 1 - SOAP Generated:" in result
|
199 |
+
assert "Note 2 - SOAP Generated:" in result
|
200 |
+
assert "Note 3: β Empty input" in result
|
201 |
+
assert "Note 4 - SOAP Generated:" in result
|
202 |
+
assert "Batch Summary" in result
|
203 |
+
assert "Total notes processed: 4" in result
|
204 |
+
assert "Successful: 3" in result
|
205 |
+
assert "Failed: 1" in result
|
206 |
+
|
207 |
+
def test_gradio_integration(self, sample_medical_text, sample_image):
|
208 |
+
"""Test Gradio interface integration"""
|
209 |
+
def mock_gradio_integration(text_input, image_input, example_selection=None):
|
210 |
+
# Simulate Gradio interface behavior
|
211 |
+
if example_selection:
|
212 |
+
# Load example
|
213 |
+
examples = {
|
214 |
+
"chest_pain": "Example chest pain case",
|
215 |
+
"diabetes": "Example diabetes case",
|
216 |
+
"pediatric": "Example pediatric case"
|
217 |
+
}
|
218 |
+
text_input = examples.get(example_selection, text_input)
|
219 |
+
|
220 |
+
# Process inputs (same as gradio_generate_soap)
|
221 |
+
final_text = text_input or ""
|
222 |
+
|
223 |
+
if image_input is not None:
|
224 |
+
ocr_result = "OCR extracted: Patient vital signs documented"
|
225 |
+
if final_text:
|
226 |
+
final_text += f"\n\n--- From Image ---\n{ocr_result}"
|
227 |
+
else:
|
228 |
+
final_text = ocr_result
|
229 |
+
|
230 |
+
if not final_text:
|
231 |
+
return "β Please provide input"
|
232 |
+
|
233 |
+
# Generate SOAP
|
234 |
+
soap_output = f"""π SOAP NOTE GENERATED
|
235 |
+
|
236 |
+
SUBJECTIVE:
|
237 |
+
{final_text[:100]}...
|
238 |
+
|
239 |
+
OBJECTIVE:
|
240 |
+
Clinical examination findings documented.
|
241 |
+
|
242 |
+
ASSESSMENT:
|
243 |
+
Medical evaluation completed.
|
244 |
+
|
245 |
+
PLAN:
|
246 |
+
Treatment recommendations provided.
|
247 |
+
|
248 |
+
--- Generated via Gradio Interface ---
|
249 |
+
Input sources: {'Text + Image' if text_input and image_input else 'Text' if text_input else 'Image'}
|
250 |
+
Processing timestamp: 2025-01-15 10:30:00
|
251 |
+
"""
|
252 |
+
return soap_output
|
253 |
+
|
254 |
+
# Test text only
|
255 |
+
result = mock_gradio_integration(sample_medical_text, None)
|
256 |
+
assert "π SOAP NOTE GENERATED" in result
|
257 |
+
assert "Input sources: Text" in result
|
258 |
+
|
259 |
+
# Test image only
|
260 |
+
result = mock_gradio_integration("", sample_image)
|
261 |
+
assert "π SOAP NOTE GENERATED" in result
|
262 |
+
assert "From Image" in result
|
263 |
+
|
264 |
+
# Test text + image
|
265 |
+
result = mock_gradio_integration(sample_medical_text, sample_image)
|
266 |
+
assert "Input sources: Text + Image" in result
|
267 |
+
|
268 |
+
# Test example selection
|
269 |
+
result = mock_gradio_integration("", None, "chest_pain")
|
270 |
+
assert "Example chest pain case" in result
|