Coverage for tests/threads/test_dl_worker.py: 100%

1import os 

2import sys 

3import pytest 

4from unittest.mock import Mock, patch, MagicMock, call 

5from PyQt6.QtCore import QProcess 

6 

7from main.threads.dl_worker import DlWorker 

8import main.threads.dl_worker 

9 

10 

11@pytest.fixture 

12def test_input_files(temp_workspace): 

13 """Create test input files.""" 

14 files = [] 

15 for i in range(3): 

16 filepath = os.path.join(temp_workspace, f"test_{i}.nii.gz") 

17 with open(filepath, "w") as f: 

18 f.write("nifti data") 

19 files.append(filepath) 

20 return files 

21 

22 

23@pytest.fixture(autouse=True) 

24def mock_external_dependencies(mocker): 

25 """ 

26 This fixture is executed automatically for each test (autouse=True). 

27 It mocks the functions in utils.py that search for external dependencies 

28 during DlWorker's __init__. 

29 """ 

30 

31 # 1. Mock get_dl_python_executable 

32 mocker.patch( 

33 "main.threads.dl_worker.get_dl_python_executable", 

34 return_value=sys.executable 

35 ) 

36 

37 # 2. Mock get_bin_path 

38 mocker.patch( 

39 "main.threads.dl_worker.get_bin_path", 

40 return_value="/fake/path/to/binary" 

41 ) 

42 

43 # 3. Mock QProcess completely (REPLACEMENT) 

44 mock_qprocess_class = mocker.patch("main.threads.dl_worker.QProcess") 

45 

46 # Configure the mock instance that will be returned 

47 process_instance = MagicMock() 

48 # We use MagicMock for signals, specifying 'connect' to avoid errors 

49 process_instance.finished = MagicMock(spec_set=['connect']) 

50 process_instance.errorOccurred = MagicMock(spec_set=['connect']) 

51 process_instance.readyReadStandardOutput = MagicMock(spec_set=['connect']) 

52 process_instance.readyReadStandardError = MagicMock(spec_set=['connect']) 

53 process_instance.start = Mock() 

54 process_instance.state = Mock(return_value=QProcess.ProcessState.NotRunning) 

55 process_instance.waitForFinished = Mock(return_value=True) 

56 process_instance.terminate = Mock() 

57 process_instance.kill = Mock() 

58 

59 # Make QProcess() return our configured instance 

60 mock_qprocess_class.return_value = process_instance 

61 

62 # === THE KEY FIX === 

63 # Expose the REAL enums on the mocked class. 

64 # Now QProcess.ExitStatus in the code will point to the real one. 

65 mock_qprocess_class.ProcessState = QProcess.ProcessState 

66 mock_qprocess_class.ExitStatus = QProcess.ExitStatus 

67 mock_qprocess_class.ProcessError = QProcess.ProcessError 

68 

69 

70class TestDlWorkerInitialization: 

71 """Tests for DlWorker initialization.""" 

72 

73 def test_initialization_basic(self, qtbot, test_input_files, temp_workspace): 

74 """Test basic initialization.""" 

75 worker = DlWorker(test_input_files, temp_workspace, False) 

76 

77 assert worker.input_files == test_input_files 

78 assert worker.workspace_path == temp_workspace 

79 assert worker.total_files is None 

80 assert worker.processed_files is None 

81 assert worker.failed_files is None 

82 assert worker.is_cancelled is False 

83 assert worker.total_phases == 6 

84 assert worker.current_file_index == 0 

85 assert worker.current_phase == 0 

86 

87 def test_initialization_signals_exist(self, qtbot, test_input_files, temp_workspace): 

88 """Test that all signals exist.""" 

89 worker = DlWorker(test_input_files, temp_workspace, False) 

90 

91 assert hasattr(worker, 'progressbar_update') 

92 assert hasattr(worker, 'file_update') 

93 assert hasattr(worker, 'log_update') 

94 assert hasattr(worker, 'finished') 

95 assert hasattr(worker, 'cancel_requested') 

96 

97 def test_initialization_process_instances_none(self, qtbot, test_input_files, temp_workspace): 

98 """Test that process instances are None initially.""" 

99 worker = DlWorker(test_input_files, temp_workspace, False) 

100 

101 assert worker.synthstrip_process is None 

102 assert worker.coregistration_process is None 

103 assert worker.reorientation_process is None 

104 assert worker.dl_preprocess is None 

105 assert worker.dl_process is None 

106 assert worker.dl_postprocess is None 

107 

108 

109class TestUpdateProgress: 

110 """Tests for the update_progress method.""" 

111 

112 def test_update_progress_basic(self, qtbot, test_input_files, temp_workspace): 

113 """Test basic progress update.""" 

114 worker = DlWorker(test_input_files, temp_workspace, False) 

115 

116 worker.total_files = 3 

117 worker.current_file_index = 0 

118 worker.current_phase = 1 

119 

120 with qtbot.waitSignal(worker.progressbar_update) as blocker: 

121 worker.update_progress() 

122 

123 # Signal should be emitted with a value 

124 assert blocker.signal_triggered 

125 assert isinstance(blocker.args[0], int) 

126 

127 def test_update_progress_zero_files(self, qtbot, test_input_files, temp_workspace): 

128 """Test with zero files.""" 

129 worker = DlWorker(test_input_files, temp_workspace, False) 

130 

131 worker.total_files = 0 

132 

133 # Should not emit signal 

134 with patch.object(worker, 'progressbar_update') as mock_signal: 

135 worker.update_progress() 

136 mock_signal.emit.assert_not_called() 

137 

138 def test_update_progress_calculation(self, qtbot, test_input_files, temp_workspace): 

139 """Test progress calculation.""" 

140 worker = DlWorker(test_input_files, temp_workspace, False) 

141 

142 worker.total_files = 2 

143 worker.total_phases = 6 

144 worker.current_file_index = 0 

145 worker.current_phase = 3 # Halfway through the first file 

146 

147 with qtbot.waitSignal(worker.progressbar_update) as blocker: 

148 worker.update_progress() 

149 

150 # Should be approx 25% (3 phases out of 12 total) 

151 progress = blocker.args[0] 

152 assert 20 <= progress <= 30 

153 

154 def test_update_progress_max_100(self, qtbot, test_input_files, temp_workspace): 

155 """Test that progress does not exceed 100.""" 

156 worker = DlWorker(test_input_files, temp_workspace, False) 

157 

158 worker.total_files = 1 

159 worker.current_file_index = 10 # Out of range value 

160 worker.current_phase = 10 

161 

162 with qtbot.waitSignal(worker.progressbar_update) as blocker: 

163 worker.update_progress() 

164 

165 assert blocker.args[0] <= 100 

166 

167 

168class TestStart: 

169 """Tests for the start method.""" 

170 

171 def test_start_initializes_counters(self, qtbot, test_input_files, temp_workspace): 

172 """Test that start initializes counters.""" 

173 worker = DlWorker(test_input_files, temp_workspace, False) 

174 

175 with patch.object(worker, 'process_single_file'): 

176 worker.start() 

177 

178 assert worker.total_files == len(test_input_files) 

179 assert worker.processed_files == 0 

180 assert worker.failed_files == [] 

181 assert worker.current_file_index == 0 

182 assert worker.current_phase == 0 

183 

184 def test_start_calls_process_single_file(self, qtbot, test_input_files, temp_workspace): 

185 """Test that start calls process_single_file.""" 

186 worker = DlWorker(test_input_files, temp_workspace, False) 

187 

188 with patch.object(worker, 'process_single_file') as mock_process: 

189 worker.start() 

190 

191 mock_process.assert_called_once() 

192 

193 def test_start_emits_initial_progress(self, qtbot, test_input_files, temp_workspace): 

194 """Test that start emits initial progress.""" 

195 worker = DlWorker(test_input_files, temp_workspace, False) 

196 

197 with patch.object(worker, 'process_single_file'): 

198 with qtbot.waitSignal(worker.progressbar_update): 

199 worker.start() 

200 

201 

202class TestProcessSingleFile: 

203 """Tests for the process_single_file method.""" 

204 

205 def test_process_single_file_creates_temp_dir(self, qtbot, test_input_files, temp_workspace): 

206 """Test that it creates a temporary directory.""" 

207 worker = DlWorker(test_input_files, temp_workspace, False) 

208 

209 worker.total_files = len(test_input_files) 

210 

211 with patch.object(worker, 'run_synthstrip'): 

212 worker.process_single_file() 

213 

214 assert worker.output_dir is not None 

215 assert os.path.exists(worker.output_dir) 

216 

217 def test_process_single_file_sets_current_file(self, qtbot, test_input_files, temp_workspace): 

218 """Test that it sets the current file.""" 

219 worker = DlWorker(test_input_files, temp_workspace, False) 

220 

221 worker.total_files = len(test_input_files) 

222 worker.current_file_index = 1 

223 

224 with patch.object(worker, 'run_synthstrip'): 

225 worker.process_single_file() 

226 

227 assert worker.current_input_file == test_input_files[1] 

228 assert worker.current_input_file_basename == os.path.basename(test_input_files[1]) 

229 

230 def test_process_single_file_calls_run_synthstrip(self, qtbot, test_input_files, temp_workspace): 

231 """Test that it calls run_synthstrip.""" 

232 worker = DlWorker(test_input_files, temp_workspace, False) 

233 

234 worker.total_files = len(test_input_files) 

235 

236 with patch.object(worker, 'run_synthstrip') as mock_synthstrip: 

237 worker.process_single_file() 

238 

239 mock_synthstrip.assert_called_once() 

240 

241 def test_process_single_file_emits_log(self, qtbot, test_input_files, temp_workspace): 

242 """Test that it emits a log.""" 

243 worker = DlWorker(test_input_files, temp_workspace, False) 

244 

245 worker.total_files = len(test_input_files) 

246 

247 with patch.object(worker, 'run_synthstrip'): 

248 with qtbot.waitSignal(worker.log_update): 

249 worker.process_single_file() 

250 

251 

252class TestRunSynthstrip: 

253 """Tests for the run_synthstrip method.""" 

254 

255 def test_run_synthstrip_creates_process(self, qtbot, test_input_files, temp_workspace): 

256 """Test that it creates the process.""" 

257 worker = DlWorker(test_input_files, temp_workspace, False) 

258 

259 worker.current_input_file = test_input_files[0] 

260 worker.current_input_file_basename = os.path.basename(test_input_files[0]) 

261 worker.output_dir = temp_workspace 

262 

263 worker.run_synthstrip() 

264 

265 assert worker.synthstrip_process is not None 

266 

267 def test_run_synthstrip_connects_signals(self, qtbot, test_input_files, temp_workspace): 

268 """Test that it connects signals.""" 

269 worker = DlWorker(test_input_files, temp_workspace, False) 

270 

271 worker.current_input_file = test_input_files[0] 

272 worker.current_input_file_basename = os.path.basename(test_input_files[0]) 

273 worker.output_dir = temp_workspace 

274 

275 worker.run_synthstrip() 

276 

277 assert worker.synthstrip_process.finished.connect.called 

278 assert worker.synthstrip_process.errorOccurred.connect.called 

279 

280 def test_run_synthstrip_starts_process(self, qtbot, test_input_files, temp_workspace): 

281 """Test that it starts the process.""" 

282 worker = DlWorker(test_input_files, temp_workspace, False) 

283 

284 worker.current_input_file = test_input_files[0] 

285 worker.current_input_file_basename = os.path.basename(test_input_files[0]) 

286 worker.output_dir = temp_workspace 

287 

288 worker.run_synthstrip() 

289 

290 worker.synthstrip_process.start.assert_called_once() 

291 

292 def test_run_synthstrip_emits_signals(self, qtbot, test_input_files, temp_workspace): 

293 """Test that it emits signals.""" 

294 worker = DlWorker(test_input_files, temp_workspace, False) 

295 

296 worker.current_input_file = test_input_files[0] 

297 worker.current_input_file_basename = os.path.basename(test_input_files[0]) 

298 worker.output_dir = temp_workspace 

299 

300 with qtbot.waitSignal(worker.file_update): 

301 worker.run_synthstrip() 

302 

303 

304class TestOnSynthstripFinished: 

305 """Tests for the on_synthstrip_finished callback.""" 

306 

307 def test_on_synthstrip_finished_success(self, qtbot, test_input_files, temp_workspace): 

308 """Test successful completion.""" 

309 worker = DlWorker(test_input_files, temp_workspace, False) 

310 

311 worker.total_files = len(test_input_files) 

312 worker.current_file_index = 0 

313 

314 worker.current_input_file_basename = "test.nii.gz" 

315 

316 with patch.object(worker, 'run_coregistration') as mock_coreg: 

317 worker.on_synthstrip_finished(0, QProcess.ExitStatus.NormalExit) 

318 

319 mock_coreg.assert_called_once() 

320 assert worker.current_phase == 2 

321 

322 def test_on_synthstrip_finished_failure(self, qtbot, test_input_files, temp_workspace): 

323 """Test completion with failure.""" 

324 worker = DlWorker(test_input_files, temp_workspace, False) 

325 

326 worker.total_files = 2 

327 worker.current_file_index = 0 

328 worker.current_input_file_basename = "test.nii.gz" 

329 

330 with patch.object(worker, 'process_single_file') as mock_process: 

331 worker.on_synthstrip_finished(1, QProcess.ExitStatus.NormalExit) 

332 

333 assert worker.current_file_index == 1 

334 

335 def test_on_synthstrip_finished_cancelled(self, qtbot, test_input_files, temp_workspace): 

336 """Test when cancelled.""" 

337 worker = DlWorker(test_input_files, temp_workspace, False) 

338 

339 worker.is_cancelled = True 

340 

341 with patch.object(worker, 'run_coregistration') as mock_coreg: 

342 worker.on_synthstrip_finished(0, QProcess.ExitStatus.NormalExit) 

343 

344 mock_coreg.assert_not_called() 

345 

346 

347class TestCancel: 

348 """Tests for the cancel method.""" 

349 

350 def test_cancel_sets_flag(self, qtbot, test_input_files, temp_workspace): 

351 """Test that it sets the flag.""" 

352 worker = DlWorker(test_input_files, temp_workspace, False) 

353 

354 worker.cancel() 

355 

356 assert worker.is_cancelled is True 

357 

358 def test_cancel_terminates_running_processes(self, qtbot, test_input_files, temp_workspace): 

359 """Test that it terminates running processes.""" 

360 worker = DlWorker(test_input_files, temp_workspace, False) 

361 

362 # Simulate running process 

363 mock_process = main.threads.dl_worker.QProcess() 

364 worker.synthstrip_process = mock_process 

365 mock_process.state = Mock(return_value=QProcess.ProcessState.Running) 

366 

367 worker.cancel() 

368 

369 worker.synthstrip_process.terminate.assert_called_once() 

370 

371 def test_cancel_kills_if_not_terminated(self, qtbot, test_input_files, temp_workspace): 

372 """Test that it kills the process if it doesn't terminate.""" 

373 worker = DlWorker(test_input_files, temp_workspace, False) 

374 

375 mock_process = main.threads.dl_worker.QProcess() 

376 

377 worker.synthstrip_process = mock_process 

378 mock_process.state = Mock(return_value=QProcess.ProcessState.Running) 

379 mock_process.waitForFinished = Mock(return_value=False) 

380 

381 worker.cancel() 

382 

383 mock_process.kill.assert_called_once() 

384 

385 def test_cancel_emits_finished_signal(self, qtbot, test_input_files, temp_workspace): 

386 """Test that it emits the finished signal.""" 

387 worker = DlWorker(test_input_files, temp_workspace, False) 

388 

389 with qtbot.waitSignal(worker.finished) as blocker: 

390 worker.cancel() 

391 

392 assert blocker.args[0] is False # success=False 

393 

394 

395class TestOnStdout: 

396 """Tests for the on_stdout method.""" 

397 

398 def test_on_stdout_basic(self, qtbot, test_input_files, temp_workspace): 

399 """Test basic stdout handling.""" 

400 worker = DlWorker(test_input_files, temp_workspace, False) 

401 

402 mock_data = Mock() 

403 mock_data.data = Mock(return_value=b"Test output line") 

404 

405 with qtbot.waitSignal(worker.log_update): 

406 worker.on_stdout("TestPhase", mock_data) 

407 

408 def test_on_stdout_multiline(self, qtbot, test_input_files, temp_workspace): 

409 """Test with multiline output.""" 

410 worker = DlWorker(test_input_files, temp_workspace, False) 

411 

412 mock_data = Mock() 

413 mock_data.data = Mock(return_value=b"Line 1\nLine 2\nLine 3") 

414 

415 log_calls = [] 

416 worker.log_update.connect(lambda msg, level: log_calls.append(msg)) 

417 

418 worker.on_stdout("TestPhase", mock_data) 

419 

420 # Should emit 3 logs 

421 assert len(log_calls) == 3 

422 

423 

424class TestOnStderr: 

425 """Tests for the on_stderr method.""" 

426 

427 def test_on_stderr_basic(self, qtbot, test_input_files, temp_workspace): 

428 """Test basic stderr handling.""" 

429 worker = DlWorker(test_input_files, temp_workspace, False) 

430 

431 mock_data = Mock() 

432 mock_data.data = Mock(return_value=b"Error message") 

433 

434 with qtbot.waitSignal(worker.log_update): 

435 worker.on_stderr("TestPhase", mock_data) 

436 

437 

438class TestOnError: 

439 """Tests for the on_error method.""" 

440 

441 def test_on_error_failed_to_start(self, qtbot, test_input_files, temp_workspace): 

442 """Test FailedToStart error.""" 

443 worker = DlWorker(test_input_files, temp_workspace, False) 

444 

445 with qtbot.waitSignal(worker.log_update): 

446 worker.on_error("TestPhase", QProcess.ProcessError.FailedToStart) 

447 

448 def test_on_error_crashed(self, qtbot, test_input_files, temp_workspace): 

449 """Test Crashed error.""" 

450 worker = DlWorker(test_input_files, temp_workspace, False) 

451 

452 with qtbot.waitSignal(worker.log_update): 

453 worker.on_error("TestPhase", QProcess.ProcessError.Crashed) 

454 

455 

456class TestPhaseTransitions: 

457 """Tests for phase transitions.""" 

458 

459 def test_synthstrip_to_coregistration(self, qtbot, test_input_files, temp_workspace): 

460 """Test synthstrip -> coregistration transition.""" 

461 worker = DlWorker(test_input_files, temp_workspace, False) 

462 

463 worker.total_files = len(test_input_files) 

464 worker.current_file_index = 0 

465 

466 worker.current_input_file_basename = "test.nii.gz" 

467 

468 with patch.object(worker, 'run_coregistration') as mock_coreg: 

469 worker.on_synthstrip_finished(0, QProcess.ExitStatus.NormalExit) 

470 

471 assert worker.current_phase == 2 

472 mock_coreg.assert_called_once() 

473 

474 def test_coregistration_to_reorientation(self, qtbot, test_input_files, temp_workspace): 

475 """Test coregistration -> reorientation transition.""" 

476 worker = DlWorker(test_input_files, temp_workspace, False) 

477 

478 worker.total_files = len(test_input_files) 

479 worker.current_file_index = 0 

480 

481 worker.current_input_file_basename = "test.nii.gz" 

482 

483 with patch.object(worker, 'run_reorientation') as mock_reorient: 

484 worker.on_coregistration_finished(0, QProcess.ExitStatus.NormalExit) 

485 

486 assert worker.current_phase == 3 

487 mock_reorient.assert_called_once() 

488 

489 

490class TestMultipleFiles: 

491 """Tests for multiple file processing.""" 

492 

493 def test_process_multiple_files_sequentially(self, qtbot, test_input_files, temp_workspace): 

494 """Test sequential file processing.""" 

495 worker = DlWorker(test_input_files, temp_workspace, False) 

496 

497 worker.total_files = 3 

498 worker.current_file_index = 0 

499 

500 with patch.object(worker, 'run_synthstrip'): 

501 # Simulate completion of the first file 

502 worker.current_phase = 6 

503 worker.on_postprocess_finished(0, QProcess.ExitStatus.NormalExit) 

504 

505 assert worker.current_file_index == 1 

506 

507 def test_finish_when_all_files_processed(self, qtbot, test_input_files, temp_workspace): 

508 """Test finish when all files are processed.""" 

509 worker = DlWorker(test_input_files, temp_workspace, False) 

510 

511 worker.total_files = 3 

512 worker.current_file_index = 2 # Last file 

513 worker.current_phase = 6 

514 

515 with qtbot.waitSignal(worker.finished) as blocker: 

516 worker.on_postprocess_finished(0, QProcess.ExitStatus.NormalExit) 

517 

518 assert blocker.args[0] is True # success=True 

519 

520 

521class TestEdgeCases: 

522 """Tests for edge cases.""" 

523 

524 def test_empty_file_list(self, qtbot, temp_workspace): 

525 """Test with empty file list.""" 

526 worker = DlWorker([], temp_workspace, False) 

527 

528 with patch.object(worker, 'process_single_file') as mock_process: 

529 worker.start() 

530 

531 assert worker.total_files == 0 

532 

533 def test_unicode_in_filename(self, qtbot, temp_workspace): 

534 """Test with unicode in filename.""" 

535 unicode_file = os.path.join(temp_workspace, "файл_文件.nii.gz") 

536 with open(unicode_file, "w") as f: 

537 f.write("data") 

538 

539 worker = DlWorker([unicode_file], temp_workspace, False) 

540 

541 with patch.object(worker, 'run_synthstrip'): 

542 worker.process_single_file() 

543 

544 assert worker.current_input_file == unicode_file 

545 

546 

547class TestIntegration: 

548 """Integration tests.""" 

549 

550 def test_full_workflow_single_file(self, qtbot, test_input_files, temp_workspace): 

551 """Test full workflow single file.""" 

552 worker = DlWorker([test_input_files[0]], temp_workspace, False) 

553 

554 # Start 

555 with patch.object(worker, 'run_synthstrip'): 

556 worker.start() 

557 

558 assert worker.total_files == 1 

559 assert worker.current_file_index == 0 

560 

561 

562class TestMemoryAndPerformance: 

563 """Tests for memory and performance.""" 

564 

565 def test_many_files_initialization(self, qtbot, temp_workspace): 

566 """Test with many files.""" 

567 many_files = [os.path.join(temp_workspace, f"file{i}.nii") for i in range(100)] 

568 

569 worker = DlWorker(many_files, temp_workspace, False) 

570 

571 assert worker.input_files == many_files 

572 

573 

574class TestAccessibility: 

575 """Tests for accessibility.""" 

576 

577 def test_signals_have_docstrings(self, qtbot, test_input_files, temp_workspace): 

578 """Test that signals have docstrings.""" 

579 worker = DlWorker(test_input_files, temp_workspace, False) 

580 

581 # Verify that the signal objects exist 

582 assert hasattr(worker, 'progressbar_update') 

583 assert hasattr(worker, 'file_update') 

584 assert hasattr(worker, 'log_update') 

585 assert hasattr(worker, 'finished') 

586 assert hasattr(worker, 'cancel_requested')