Coverage for tests/ui/test_pipeline_patient_selection.py: 99%

1import pytest 

2import os 

3import json 

4import tempfile 

5import shutil 

6from unittest.mock import Mock, patch, MagicMock 

7 

8from main.ui.pipeline_patient_selection_page import PipelinePatientSelectionPage 

9 

10class TestPipelinePatientSelectionPageSetup: 

11 """Tests for initialization""" 

12 

13 @pytest.fixture 

14 def temp_workspace(self): 

15 temp_dir = tempfile.mkdtemp() 

16 yield temp_dir 

17 shutil.rmtree(temp_dir, ignore_errors=True) 

18 

19 @pytest.fixture 

20 def pipeline_page(self, qtbot, mock_context): 

21 page = PipelinePatientSelectionPage(mock_context, Mock()) 

22 qtbot.addWidget(page) 

23 return page 

24 

25 def test_page_initialization(self, pipeline_page): 

26 """Verify correct initialization""" 

27 assert pipeline_page.context is not None 

28 assert pipeline_page.workspace_path is not None 

29 assert pipeline_page.selected_patients == set() 

30 assert pipeline_page.patient_status == {} 

31 

32 def test_title_created(self, pipeline_page): 

33 """Verify title creation""" 

34 assert pipeline_page.title is not None 

35 assert pipeline_page.title.text() != "" 

36 

37 def test_buttons_created(self, pipeline_page): 

38 """Verify button creation""" 

39 assert pipeline_page.select_eligible_btn is not None 

40 assert pipeline_page.deselect_all_btn is not None 

41 assert pipeline_page.refresh_btn is not None 

42 

43 def test_summary_widget_created(self, pipeline_page): 

44 """Verify summary widget creation""" 

45 assert pipeline_page.summary_widget is not None 

46 assert pipeline_page.total_label is not None 

47 assert pipeline_page.eligible_label is not None 

48 assert pipeline_page.not_eligible_label is not None 

49 

50 

51class TestPipelinePatientRequirements: 

52 """Tests for verifying patient requirements""" 

53 

54 @pytest.fixture 

55 def temp_workspace(self): 

56 temp_dir = tempfile.mkdtemp() 

57 # Create eligible patient structure 

58 patient_dir = os.path.join(temp_dir, "sub-01") 

59 os.makedirs(os.path.join(patient_dir, "anat")) 

60 

61 # FLAIR 

62 with open(os.path.join(patient_dir, "anat", "sub-01_flair.nii.gz"), "w") as f: 

63 f.write("flair") 

64 

65 # Skull stripping 

66 skull_dir = os.path.join(temp_dir, "derivatives", "skullstrips", "sub-01", "anat") 

67 os.makedirs(skull_dir) 

68 with open(os.path.join(skull_dir, "sub-01_brain.nii.gz"), "w") as f: 

69 f.write("brain") 

70 

71 # Segmentation 

72 seg_dir = os.path.join(temp_dir, "derivatives", "manual_masks", "sub-01", "anat") 

73 os.makedirs(seg_dir) 

74 with open(os.path.join(seg_dir, "sub-01_mask.nii.gz"), "w") as f: 

75 f.write("mask") 

76 

77 yield temp_dir 

78 shutil.rmtree(temp_dir, ignore_errors=True) 

79 

80 @pytest.fixture 

81 def pipeline_page(self, qtbot, mock_context): 

82 page = PipelinePatientSelectionPage(mock_context, Mock()) 

83 qtbot.addWidget(page) 

84 return page 

85 

86 def test_check_patient_eligible(self, pipeline_page, temp_workspace): 

87 """Verify patient with all requirements""" 

88 patient_path = os.path.join(temp_workspace, "sub-01") 

89 

90 status = pipeline_page._check_patient_requirements(patient_path, "sub-01") 

91 

92 assert status['eligible'] == True 

93 assert status['requirements']['flair'] == True 

94 assert status['requirements']['skull_stripping'] == True 

95 assert status['requirements']['segmentation'] == True 

96 

97 def test_check_patient_missing_flair(self, pipeline_page, temp_workspace): 

98 """Verify patient without FLAIR""" 

99 # Create patient without FLAIR 

100 patient_dir = os.path.join(temp_workspace, "sub-02") 

101 os.makedirs(os.path.join(patient_dir, "anat")) 

102 

103 status = pipeline_page._check_patient_requirements(patient_dir, "sub-02") 

104 

105 assert status['eligible'] == False 

106 assert status['requirements']['flair'] == False 

107 assert "FLAIR" in str(status['missing_files']) 

108 

109 def test_check_patient_missing_skull_stripping(self, pipeline_page, temp_workspace): 

110 """Verify patient without skull stripping""" 

111 patient_dir = os.path.join(temp_workspace, "sub-03") 

112 os.makedirs(os.path.join(patient_dir, "anat")) 

113 with open(os.path.join(patient_dir, "anat", "sub-03_flair.nii.gz"), "w") as f: 

114 f.write("flair") 

115 

116 status = pipeline_page._check_patient_requirements(patient_dir, "sub-03") 

117 

118 assert status['eligible'] == False 

119 assert status['requirements']['skull_stripping'] == False 

120 

121 def test_check_patient_missing_segmentation(self, pipeline_page, temp_workspace): 

122 """Verify patient without segmentation""" 

123 patient_dir = os.path.join(temp_workspace, "sub-04") 

124 os.makedirs(os.path.join(patient_dir, "anat")) 

125 with open(os.path.join(patient_dir, "anat", "sub-04_flair.nii.gz"), "w") as f: 

126 f.write("flair") 

127 

128 skull_dir = os.path.join(temp_workspace, "derivatives", "skullstrips", "sub-04", "anat") 

129 os.makedirs(skull_dir) 

130 with open(os.path.join(skull_dir, "sub-04_brain.nii.gz"), "w") as f: 

131 f.write("brain") 

132 

133 status = pipeline_page._check_patient_requirements(patient_dir, "sub-04") 

134 

135 assert status['eligible'] == False 

136 assert status['requirements']['segmentation'] == False 

137 

138 

139class TestPipelinePatientSelection: 

140 """Tests for patient selection""" 

141 

142 @pytest.fixture 

143 def temp_workspace(self): 

144 temp_dir = tempfile.mkdtemp() 

145 # Create 2 eligible patients 

146 for i in range(1, 3): 

147 patient_id = f"sub-{i:02d}" 

148 patient_dir = os.path.join(temp_dir, patient_id) 

149 os.makedirs(os.path.join(patient_dir, "anat")) 

150 

151 with open(os.path.join(patient_dir, "anat", f"{patient_id}_flair.nii.gz"), "w") as f: 

152 f.write("flair") 

153 

154 skull_dir = os.path.join(temp_dir, "derivatives", "skullstrips", patient_id, "anat") 

155 os.makedirs(skull_dir) 

156 with open(os.path.join(skull_dir, f"{patient_id}_brain.nii.gz"), "w") as f: 

157 f.write("brain") 

158 

159 seg_dir = os.path.join(temp_dir, "derivatives", "manual_masks", patient_id, "anat") 

160 os.makedirs(seg_dir) 

161 with open(os.path.join(seg_dir, f"{patient_id}_mask.nii.gz"), "w") as f: 

162 f.write("mask") 

163 

164 yield temp_dir 

165 shutil.rmtree(temp_dir, ignore_errors=True) 

166 

167 @pytest.fixture 

168 def pipeline_page(self, qtbot, mock_context): 

169 page = PipelinePatientSelectionPage(mock_context, Mock()) 

170 qtbot.addWidget(page) 

171 return page 

172 

173 def test_select_all_eligible(self, pipeline_page): 

174 """Verify selection of all eligible patients""" 

175 pipeline_page._select_all_eligible_patients() 

176 

177 assert len(pipeline_page.selected_patients) == 2 

178 assert "sub-01" in pipeline_page.selected_patients 

179 assert "sub-02" in pipeline_page.selected_patients 

180 

181 def test_deselect_all(self, pipeline_page): 

182 """Verify deselection of all patients""" 

183 pipeline_page._select_all_eligible_patients() 

184 assert len(pipeline_page.selected_patients) == 2 

185 

186 pipeline_page._deselect_all_patients() 

187 

188 assert len(pipeline_page.selected_patients) == 0 

189 

190 def test_toggle_patient_select(self, pipeline_page): 

191 """Verify single patient selection""" 

192 button = Mock() 

193 pipeline_page._toggle_patient("sub-01", True, button) 

194 

195 assert "sub-01" in pipeline_page.selected_patients 

196 

197 def test_toggle_patient_deselect(self, pipeline_page): 

198 """Verify single patient deselection""" 

199 pipeline_page.selected_patients.add("sub-01") 

200 button = Mock() 

201 

202 pipeline_page._toggle_patient("sub-01", False, button) 

203 

204 assert "sub-01" not in pipeline_page.selected_patients 

205 

206class TestPipelinePatientLoading: 

207 """Tests for patient loading""" 

208 

209 @pytest.fixture 

210 def pipeline_page(self, qtbot, mock_context): 

211 page = PipelinePatientSelectionPage(mock_context, Mock()) 

212 qtbot.addWidget(page) 

213 return page 

214 

215 def test_find_patient_dirs(self, pipeline_page): 

216 """Verify patient directory search""" 

217 patient_dirs = pipeline_page._find_patient_dirs() 

218 

219 assert len(patient_dirs) == 2 

220 patient_ids = [os.path.basename(d) for d in patient_dirs] 

221 assert "sub-01" in patient_ids 

222 assert "sub-02" in patient_ids 

223 

224 def test_load_patients_creates_buttons(self, pipeline_page): 

225 """Verify patient button creation""" 

226 assert len(pipeline_page.patient_buttons) == 2 

227 

228 def test_load_patients_updates_status(self, pipeline_page): 

229 """Verify patient status update""" 

230 assert len(pipeline_page.patient_status) == 2 

231 

232 

233class TestPipelineConfigGeneration: 

234 """Tests for pipeline configuration generation""" 

235 

236 @pytest.fixture 

237 def temp_workspace(self): 

238 temp_dir = tempfile.mkdtemp() 

239 patient_dir = os.path.join(temp_dir, "sub-01") 

240 os.makedirs(os.path.join(patient_dir, "anat")) 

241 os.makedirs(os.path.join(patient_dir, "ses-01", "pet")) 

242 

243 # Necessary files 

244 with open(os.path.join(patient_dir, "anat", "sub-01_flair.nii.gz"), "w") as f: 

245 f.write("flair") 

246 with open(os.path.join(patient_dir, "ses-01", "pet", "sub-01_pet.nii.gz"), "w") as f: 

247 f.write("pet") 

248 

249 skull_dir = os.path.join(temp_dir, "derivatives", "skullstrips", "sub-01", "anat") 

250 os.makedirs(skull_dir) 

251 with open(os.path.join(skull_dir, "sub-01_brain.nii.gz"), "w") as f: 

252 f.write("brain") 

253 

254 seg_dir = os.path.join(temp_dir, "derivatives", "manual_masks", "sub-01", "anat") 

255 os.makedirs(seg_dir) 

256 with open(os.path.join(seg_dir, "sub-01_mask.nii.gz"), "w") as f: 

257 f.write("mask") 

258 

259 yield temp_dir 

260 shutil.rmtree(temp_dir, ignore_errors=True) 

261 

262 @pytest.fixture 

263 def pipeline_page(self, qtbot, mock_context): 

264 page = PipelinePatientSelectionPage(mock_context, Mock()) 

265 qtbot.addWidget(page) 

266 page.selected_patients = {"sub-01"} 

267 return page 

268 

269 def test_build_pipeline_config_creates_file(self, pipeline_page, temp_workspace): 

270 """Verify configuration file creation""" 

271 config_path = pipeline_page._build_pipeline_config() 

272 

273 assert os.path.exists(config_path) 

274 assert config_path.endswith("_config.json") 

275 

276 def test_build_pipeline_config_content(self, pipeline_page): 

277 """Verify configuration content""" 

278 config_path = pipeline_page._build_pipeline_config() 

279 

280 with open(config_path, 'r') as f: 

281 config = json.load(f) 

282 

283 assert "sub-01" in config 

284 assert "mri" in config["sub-01"] 

285 assert "mri_str" in config["sub-01"] 

286 assert "tumor_mri" in config["sub-01"] 

287 

288 def test_get_next_config_id_first(self, pipeline_page, temp_workspace): 

289 """Verify configuration ID first time""" 

290 pipeline_dir = os.path.join(temp_workspace, "pipeline") 

291 os.makedirs(pipeline_dir, exist_ok=True) 

292 

293 config_id = pipeline_page._get_next_config_id(pipeline_dir) 

294 

295 assert config_id == 1 

296 

297 def test_get_next_config_id_existing(self, pipeline_page, temp_workspace): 

298 """Verify configuration ID with existing files""" 

299 pipeline_dir = os.path.join(temp_workspace, "pipeline") 

300 os.makedirs(pipeline_dir, exist_ok=True) 

301 

302 # Create existing files 

303 with open(os.path.join(pipeline_dir, "01_config.json"), "w") as f: 

304 f.write("{}") 

305 with open(os.path.join(pipeline_dir, "02_config.json"), "w") as f: 

306 f.write("{}") 

307 

308 config_id = pipeline_page._get_next_config_id(pipeline_dir) 

309 

310 assert config_id == 3 

311 

312class TestPipelinePatientRefresh: 

313 """Tests for refreshing patient status""" 

314 

315 @pytest.fixture 

316 def pipeline_page(self, qtbot, mock_context): 

317 page = PipelinePatientSelectionPage(mock_context, Mock()) 

318 qtbot.addWidget(page) 

319 return page 

320 

321 def test_refresh_maintains_valid_selections(self, pipeline_page): 

322 """Verify that refresh maintains valid selections""" 

323 # Simulate selection 

324 pipeline_page.selected_patients.add("sub-01") 

325 

326 pipeline_page._refresh_patient_status() 

327 

328 # Selections should be validated 

329 assert isinstance(pipeline_page.selected_patients, set) 

330 

331 

332class TestPipelinePatientNavigation: 

333 """Tests for navigation""" 

334 

335 @pytest.fixture 

336 def pipeline_page(self, qtbot, mock_context): 

337 previous_page = Mock() 

338 page = PipelinePatientSelectionPage(mock_context, previous_page) 

339 qtbot.addWidget(page) 

340 return page 

341 

342 def test_is_ready_to_advance_with_selection(self, pipeline_page): 

343 """Verify ready to advance with selections""" 

344 pipeline_page.selected_patients.add("sub-01") 

345 assert pipeline_page.is_ready_to_advance() 

346 

347 def test_is_ready_to_advance_without_selection(self, pipeline_page): 

348 """Verify not ready without selections""" 

349 assert not pipeline_page.is_ready_to_advance() 

350 

351 def test_is_ready_to_go_back(self, pipeline_page): 

352 """Verify can go back""" 

353 assert pipeline_page.is_ready_to_go_back() 

354 

355 def test_back_returns_previous_page(self, pipeline_page): 

356 """Verify return to previous page""" 

357 result = pipeline_page.back() 

358 assert result == pipeline_page.previous_page 

359 pipeline_page.previous_page.on_enter.assert_called_once() 

360 

361 @patch('main.ui.pipeline_patient_selection_page.PipelineReviewPage') 

362 def test_next_creates_review_page(self, MockPage, pipeline_page): 

363 """Verify review page creation""" 

364 mock_page = Mock() 

365 MockPage.return_value = mock_page 

366 pipeline_page.selected_patients.add("sub-01") 

367 

368 result = pipeline_page.next(pipeline_page.context) 

369 

370 assert result == mock_page 

371 mock_page.on_enter.assert_called_once() 

372 

373class TestPipelinePatientSummary: 

374 """Tests for summary functions""" 

375 

376 @pytest.fixture 

377 def pipeline_page(self, qtbot, mock_context): 

378 page = PipelinePatientSelectionPage(mock_context, Mock()) 

379 qtbot.addWidget(page) 

380 return page 

381 

382 def test_get_selected_patients(self, pipeline_page): 

383 """Verify retrieval of selected patients""" 

384 pipeline_page.selected_patients = {"sub-01", "sub-02"} 

385 

386 selected = pipeline_page.get_selected_patients() 

387 

388 assert isinstance(selected, list) 

389 assert len(selected) == 2 

390 

391 def test_get_eligible_patients(self, pipeline_page): 

392 """Verify retrieval of eligible patients""" 

393 pipeline_page.patient_status = { 

394 "sub-01": {"eligible": True}, 

395 "sub-02": {"eligible": False} 

396 } 

397 

398 eligible = pipeline_page.get_eligible_patients() 

399 

400 assert len(eligible) == 1 

401 assert "sub-01" in eligible 

402 

403 def test_get_patient_status_summary(self, pipeline_page): 

404 """Verify status summary""" 

405 pipeline_page.patient_status = { 

406 "sub-01": {"eligible": True}, 

407 "sub-02": {"eligible": False} 

408 } 

409 pipeline_page.selected_patients = {"sub-01"} 

410 

411 summary = pipeline_page.get_patient_status_summary() 

412 

413 assert summary['total'] == 2 

414 assert summary['eligible'] == 1 

415 assert summary['selected'] == 1 

416 assert summary['not_eligible'] == 1 

417 

418 

419class TestPipelinePatientReset: 

420 """Tests for page reset""" 

421 

422 @pytest.fixture 

423 def pipeline_page(self, qtbot, mock_context): 

424 page = PipelinePatientSelectionPage(mock_context, Mock()) 

425 qtbot.addWidget(page) 

426 return page 

427 

428 def test_reset_clears_selections(self, pipeline_page): 

429 """Verify that reset clears selections""" 

430 pipeline_page.selected_patients = {"sub-01", "sub-02"} 

431 

432 pipeline_page.reset_page() 

433 

434 assert len(pipeline_page.selected_patients) == 0 

435 

436 def test_reset_clears_status(self, pipeline_page): 

437 """Verify that reset clears status""" 

438 pipeline_page.patient_status = {"sub-01": {"eligible": True}} 

439 

440 pipeline_page.reset_page() 

441 

442 # Status should be reloaded 

443 assert isinstance(pipeline_page.patient_status, dict) 

444 

445# Integration tests 

446class TestPipelinePatientIntegration: 

447 """Integration tests""" 

448 

449 @pytest.fixture 

450 def temp_workspace(self): 

451 temp_dir = tempfile.mkdtemp() 

452 # Create complete patient 

453 patient_dir = os.path.join(temp_dir, "sub-01") 

454 os.makedirs(os.path.join(patient_dir, "anat")) 

455 os.makedirs(os.path.join(patient_dir, "ses-01", "pet")) 

456 

457 with open(os.path.join(patient_dir, "anat", "sub-01_flair.nii.gz"), "w") as f: 

458 f.write("flair") 

459 

460 skull_dir = os.path.join(temp_dir, "derivatives", "skullstrips", "sub-01", "anat") 

461 os.makedirs(skull_dir) 

462 with open(os.path.join(skull_dir, "sub-01_brain.nii.gz"), "w") as f: 

463 f.write("brain") 

464 

465 seg_dir = os.path.join(temp_dir, "derivatives", "manual_masks", "sub-01", "anat") 

466 os.makedirs(seg_dir) 

467 with open(os.path.join(seg_dir, "sub-01_mask.nii.gz"), "w") as f: 

468 f.write("mask") 

469 

470 yield temp_dir 

471 shutil.rmtree(temp_dir, ignore_errors=True) 

472 

473 @pytest.fixture 

474 def pipeline_page(self, qtbot, mock_context): 

475 page = PipelinePatientSelectionPage(mock_context, Mock()) 

476 qtbot.addWidget(page) 

477 return page 

478 

479 def test_full_workflow(self, pipeline_page, temp_workspace): 

480 """Test full workflow""" 

481 # Verify loading 

482 assert len(pipeline_page.patient_status) == 1 

483 

484 # Patient should be eligible 

485 assert pipeline_page.patient_status["sub-01"]["eligible"] 

486 

487 # Select 

488 pipeline_page._select_all_eligible_patients() 

489 assert len(pipeline_page.selected_patients) == 1 

490 

491 # Generate config 

492 config_path = pipeline_page._build_pipeline_config() 

493 assert os.path.exists(config_path) 

494 

495 # Verify config 

496 with open(config_path, 'r') as f: 

497 config = json.load(f) 

498 assert "sub-01" in config 

499 

500 

501if __name__ == "__main__": 

502 pytest.main([__file__, "-v", "--tb=short"])