Coverage for tests/threads/test_skull_strip

1""""

2test_skull_strip_thread.py - Test Suite for SkullStripThread

4This suite tests all functionalities of the skull-stripping thread:

5- Execution of FSL BET and HD-BET

6- Batch processing of multiple files

7- Progress tracking and cancellation

8- Generation of BIDS JSON metadata

9- Error handling and edge cases

10"""

12import os

13import json

14import tempfile

15import shutil

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

17import pytest

18from PyQt6.QtCore import QProcess

20from main.threads.skull_strip_thread import SkullStripThread

23class TestSkullStripThreadInitialization:

24 """Tests for the initialization of SkullStripThread"""

26 def test_init_with_bet(self, temp_workspace):

27 """Test initialization with FSL BET"""

28 files = [os.path.join(temp_workspace, "sub-01", "anat", "T1w.nii.gz")]

29 parameters = {'f_val': 0.5, 'opt_m': True}

31 thread = SkullStripThread(

32 files=files,

33 workspace_path=temp_workspace,

34 parameters=parameters,

35 has_cuda=False,

36 bet_tool="fsl-bet"

37 )

39 assert thread.files == files

40 assert thread.workspace_path == temp_workspace

41 assert thread.parameters == parameters

42 assert thread.has_cuda is False

43 assert thread.bet_tool == "fsl-bet"

44 assert thread.is_cancelled is False

45 assert thread.success_count == 0

46 assert thread.failed_files == []

48 def test_init_with_hdbet(self, temp_workspace):

49 """Test initialization with HD-BET"""

50 files = [os.path.join(temp_workspace, "test.nii")]

51 parameters = {}

53 thread = SkullStripThread(

54 files=files,

55 workspace_path=temp_workspace,

56 parameters=parameters,

57 has_cuda=True,

58 bet_tool="fsl-bet"

59 )

61 assert thread.has_cuda is True

62 assert thread.bet_tool == "fsl-bet"

64 def test_init_with_synthstrip(self, temp_workspace):

65 """Test initialization with SynthStrip"""

66 files = [os.path.join(temp_workspace, "test.nii")]

67 parameters = {}

69 thread = SkullStripThread(

70 files=files,

71 workspace_path=temp_workspace,

72 parameters=parameters,

73 has_cuda=True,

74 bet_tool="synthstrip"

75 )

77 assert thread.has_cuda is True

78 assert thread.bet_tool == "synthstrip"

80 def test_signals_exist(self, temp_workspace):

81 """Verify that the signals are correctly defined"""

82 thread = SkullStripThread([], temp_workspace, {}, False, "fsl-bet")

84 assert hasattr(thread, 'progress_updated')

85 assert hasattr(thread, 'progress_value_updated')

86 assert hasattr(thread, 'file_started')

87 assert hasattr(thread, 'file_completed')

88 assert hasattr(thread, 'all_completed')

91class TestSkullStripThreadCancellation:

92 """Tests for thread cancellation"""

94 def test_cancel_flag_set(self, temp_workspace):

95 """Test that the cancellation flag is set"""

96 thread = SkullStripThread([], temp_workspace, {}, False, "fsl-bet")

98 assert thread.is_cancelled is False

99 thread.cancel()

100 assert thread.is_cancelled is True

101

102 def test_cancel_terminates_process(self, temp_workspace):

103 """Test process termination during cancellation"""

104 thread = SkullStripThread([], temp_workspace, {}, False, "fsl-bet")

105

106 # Mock a running process

107 mock_process = Mock(spec=QProcess)

108 thread.process = mock_process

109

110 thread.cancel()

111

112 mock_process.terminate.assert_called_once()

113 mock_process.kill.assert_called_once()

114

115 def test_cancel_no_process(self, temp_workspace):

116 """Test cancellation without an active process"""

117 thread = SkullStripThread([], temp_workspace, {}, False, "fsl-bet")

118

119 # Should not raise exceptions

120 thread.cancel()

121 assert thread.is_cancelled is True

122

123 def test_cancel_process_error_handling(self, temp_workspace):

124 """Test error handling during process cancellation"""

125 thread = SkullStripThread([], temp_workspace, {}, False, "fsl-bet")

126

127 mock_process = Mock()

128 mock_process.terminate.side_effect = RuntimeError("Process error")

129 thread.process = mock_process

130

131 # Should not propagate the exception

132 thread.cancel()

133 assert thread.is_cancelled is True

134

135

136class TestBETCommandBuilding:

137 """Tests for building FSL BET commands"""

138

139 @patch('main.threads.skull_strip_thread.setup_fsl_env')

140 @patch('main.threads.skull_strip_thread.QProcess')

141 def test_bet_basic_command(self, mock_qprocess, mock_fsl_env, temp_workspace):

142 """Test basic BET command"""

143 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

144

145 # Mock process

146 mock_process_instance = Mock()

147 mock_process_instance.waitForFinished.return_value = True

148 mock_process_instance.exitCode.return_value = 0

149 mock_process_instance.readAllStandardError.return_value = b''

150 mock_process_instance.readAllStandardOutput.return_value = b''

151 mock_qprocess.return_value = mock_process_instance

152

153 # Create input file

154 input_file = os.path.join(temp_workspace, "sub-01", "anat", "T1w.nii.gz")

155 os.makedirs(os.path.dirname(input_file), exist_ok=True)

156 with open(input_file, 'w') as f:

157 f.write("nifti data")

158

159 parameters = {'f_val': 0.5}

160 thread = SkullStripThread([input_file], temp_workspace, parameters, False, "fsl-bet")

161

162 with patch('shutil.move'), patch('shutil.rmtree'), patch('os.makedirs'):

163 thread.run()

164

165 # Verify process call

166 mock_process_instance.start.assert_called_once()

167 call_args = mock_process_instance.start.call_args[0]

168

169 assert call_args[0] == "bet"

170 assert input_file in call_args[1]

171 assert "-f" in call_args[1]

172 assert "0.5" in call_args[1]

173

174 @patch('main.threads.skull_strip_thread.setup_fsl_env')

175 @patch('main.threads.skull_strip_thread.QProcess')

176 def test_bet_with_options(self, mock_qprocess, mock_fsl_env, temp_workspace):

177 """Test BET command with advanced options"""

178 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

179

180 mock_process = Mock()

181 mock_process.waitForFinished.return_value = True

182 mock_process.exitCode.return_value = 0

183 mock_process.readAllStandardError.return_value = b''

184 mock_process.readAllStandardOutput.return_value = b''

185 mock_qprocess.return_value = mock_process

186

187 input_file = os.path.join(temp_workspace, "sub-02", "anat", "brain.nii")

188 os.makedirs(os.path.dirname(input_file), exist_ok=True)

189 with open(input_file, 'w') as f:

190 f.write("data")

191

192 parameters = {

193 'f_val': 0.3,

194 'opt_m': True,

195 'opt_t': True,

196 'opt_s': False,

197 'opt_o': True

198 }

199

200 thread = SkullStripThread([input_file], temp_workspace, parameters, False, "fsl-bet")

201

202 with patch('shutil.move'), patch('shutil.rmtree'), patch('os.makedirs'):

203 thread.run()

204

205 call_args = mock_process.start.call_args[0]

206 cmd = call_args[1]

207

208 # Verify presence of options

209 assert "-m" in cmd # opt_m

210 assert "-t" in cmd # opt_t

211 assert "-o" in cmd # opt_o

212 assert "-s" not in cmd # opt_s is False

213

214 @patch('main.threads.skull_strip_thread.setup_fsl_env')

215 @patch('main.threads.skull_strip_thread.QProcess')

216 def test_bet_with_center_coordinates(self, mock_qprocess, mock_fsl_env, temp_workspace):

217 """Test BET command with center coordinates"""

218 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

219

220 mock_process = Mock()

221 mock_process.waitForFinished.return_value = True

222 mock_process.exitCode.return_value = 0

223 mock_process.readAllStandardError.return_value = b''

224 mock_process.readAllStandardOutput.return_value = b''

225 mock_qprocess.return_value = mock_process

226

227 input_file = os.path.join(temp_workspace, "sub-03", "anat", "scan.nii.gz")

228 os.makedirs(os.path.dirname(input_file), exist_ok=True)

229 with open(input_file, 'w') as f:

230 f.write("data")

231

232 parameters = {

233 'f_val': 0.5,

234 'c_x': 128,

235 'c_y': 128,

236 'c_z': 64

237 }

238

239 thread = SkullStripThread([input_file], temp_workspace, parameters, False, "fsl-bet")

240

241 with patch('shutil.move'), patch('shutil.rmtree'), patch('os.makedirs'):

242 thread.run()

243

244 call_args = mock_process.start.call_args[0]

245 cmd = call_args[1]

246

247 # Verify coordinates

248 assert "-c" in cmd

249 assert "128" in cmd or "64" in cmd

250

251 @patch('main.threads.skull_strip_thread.setup_fsl_env')

252 @patch('main.threads.skull_strip_thread.QProcess')

253 def test_bet_brain_extracted_option(self, mock_qprocess, mock_fsl_env, temp_workspace):

254 """Test -n option for brain extraction"""

255 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

256

257 mock_process = Mock()

258 mock_process.waitForFinished.return_value = True

259 mock_process.exitCode.return_value = 0

260 mock_process.readAllStandardError.return_value = b''

261 mock_process.readAllStandardOutput.return_value = b''

262 mock_qprocess.return_value = mock_process

263

264 input_file = os.path.join(temp_workspace, "sub-04", "anat", "T1.nii")

265 os.makedirs(os.path.dirname(input_file), exist_ok=True)

266 with open(input_file, 'w') as f:

267 f.write("data")

268

269 parameters = {

270 'f_val': 0.5,

271 'opt_brain_extracted': False # Disable mask extraction

272 }

273

274 thread = SkullStripThread([input_file], temp_workspace, parameters, False, "fsl-bet")

275

276 with patch('shutil.move'), patch('shutil.rmtree'), patch('os.makedirs'):

277 thread.run()

278

279 call_args = mock_process.start.call_args[0]

280 cmd = call_args[1]

281

282 assert "-n" in cmd

283

284

285class TestHDBETCommandBuilding:

286 """Tests for building HD-BET commands"""

287

288 @patch('main.threads.skull_strip_thread.get_bin_path')

289 @patch('main.threads.skull_strip_thread.QProcess')

290 def test_hdbet_basic_command(self, mock_qprocess, mock_get_bin, temp_workspace):

291 """Test basic HD-BET command"""

292 mock_get_bin.return_value = "/usr/bin/hd-bet"

293

294 mock_process = Mock()

295 mock_process.waitForFinished.return_value = True

296 mock_process.exitCode.return_value = 0

297 mock_process.readAllStandardError.return_value = b''

298 mock_process.readAllStandardOutput.return_value = b''

299 mock_qprocess.return_value = mock_process

300

301 input_file = os.path.join(temp_workspace, "sub-01", "anat", "brain.nii.gz")

302 os.makedirs(os.path.dirname(input_file), exist_ok=True)

303 with open(input_file, 'w') as f:

304 f.write("data")

305

306 thread = SkullStripThread([input_file], temp_workspace, {}, True, "hd-bet")

307

308 with patch('shutil.move'), patch('shutil.rmtree'), patch('os.makedirs'):

309 thread.run()

310

311 call_args = mock_process.start.call_args[0]

312

313 assert call_args[0] == "/usr/bin/hd-bet"

314 assert "-i" in call_args[1]

315 assert "-o" in call_args[1]

316

317 @patch('main.threads.skull_strip_thread.get_bin_path')

318 @patch('main.threads.skull_strip_thread.QProcess')

319 def test_hdbet_cpu_mode(self, mock_qprocess, mock_get_bin, temp_workspace):

320 """Test HD-BET in CPU mode (without CUDA)"""

321 mock_get_bin.return_value = "/usr/bin/hd-bet"

322

323 mock_process = Mock()

324 mock_process.waitForFinished.return_value = True

325 mock_process.exitCode.return_value = 0

326 mock_process.readAllStandardError.return_value = b''

327 mock_process.readAllStandardOutput.return_value = b''

328 mock_qprocess.return_value = mock_process

329

330 input_file = os.path.join(temp_workspace, "sub-02", "anat", "T1w.nii")

331 os.makedirs(os.path.dirname(input_file), exist_ok=True)

332 with open(input_file, 'w') as f:

333 f.write("data")

334

335 # has_cuda = False

336 thread = SkullStripThread([input_file], temp_workspace, {}, False, "hd-bet")

337

338 with patch('shutil.move'), patch('shutil.rmtree'), patch('os.makedirs'):

339 thread.run()

340

341 call_args = mock_process.start.call_args[0]

342 cmd = call_args[1]

343

344 # Verify CPU options

345 assert "-device" in cmd

346 assert "cpu" in cmd

347 assert "--disable_tta" in cmd

348

349 @patch('main.threads.skull_strip_thread.get_bin_path')

350 @patch('main.threads.skull_strip_thread.QProcess')

351 def test_hdbet_cuda_mode(self, mock_qprocess, mock_get_bin, temp_workspace):

352 """Test HD-BET with CUDA"""

353 mock_get_bin.return_value = "/usr/bin/hd-bet"

354

355 mock_process = Mock()

356 mock_process.waitForFinished.return_value = True

357 mock_process.exitCode.return_value = 0

358 mock_process.readAllStandardError.return_value = b''

359 mock_process.readAllStandardOutput.return_value = b''

360 mock_qprocess.return_value = mock_process

361

362 input_file = os.path.join(temp_workspace, "sub-03", "anat", "scan.nii.gz")

363 os.makedirs(os.path.dirname(input_file), exist_ok=True)

364 with open(input_file, 'w') as f:

365 f.write("data")

366

367 # has_cuda = True

368 thread = SkullStripThread([input_file], temp_workspace, {}, True, "hd-bet")

369

370 with patch('shutil.move'), patch('shutil.rmtree'), patch('os.makedirs'):

371 thread.run()

372

373 call_args = mock_process.start.call_args[0]

374 cmd = call_args[1]

375

376 # Should NOT have CPU options

377 assert "-device" not in cmd or "cpu" not in cmd

378

379 @patch('main.threads.skull_strip_thread.get_bin_path')

380 @patch('main.threads.skull_strip_thread.QProcess')

381 def test_sinthstrip_basic_command(self, mock_qprocess, mock_get_bin, temp_workspace):

382 """Test basic SynthStrip command"""

383 mock_get_bin.return_value = "/usr/bin/mri_sinthstrip"

384

385 mock_process = Mock()

386 mock_process.waitForFinished.return_value = True

387 mock_process.exitCode.return_value = 0

388 mock_process.readAllStandardError.return_value = b''

389 mock_process.readAllStandardOutput.return_value = b''

390 mock_qprocess.return_value = mock_process

391

392 input_file = os.path.join(temp_workspace, "sub-01", "anat", "brain.nii.gz")

393 os.makedirs(os.path.dirname(input_file), exist_ok=True)

394 with open(input_file, 'w') as f:

395 f.write("data")

396

397 thread = SkullStripThread([input_file], temp_workspace, {}, True, "synthstrip")

398

399 with patch('shutil.move'), patch('shutil.rmtree'), patch('os.makedirs'):

400 thread.run()

401

402 call_args = mock_process.start.call_args[0]

403

404 assert call_args[0] == "/usr/bin/mri_sinthstrip"

405 assert "-i" in call_args[1]

406 assert "-o" in call_args[1]

407

408class TestOutputGeneration:

409 """Tests for output and metadata generation"""

410

411 @patch('main.threads.skull_strip_thread.setup_fsl_env')

412 @patch('main.threads.skull_strip_thread.QProcess')

413 def test_output_directory_creation(self, mock_qprocess, mock_fsl_env, temp_workspace):

414 """Test output directory creation"""

415 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

416

417 mock_process = Mock()

418 mock_process.waitForFinished.return_value = True

419 mock_process.exitCode.return_value = 0

420 mock_process.readAllStandardError.return_value = b''

421 mock_process.readAllStandardOutput.return_value = b''

422 mock_qprocess.return_value = mock_process

423

424 input_file = os.path.join(temp_workspace, "sub-05", "anat", "brain.nii.gz")

425 os.makedirs(os.path.dirname(input_file), exist_ok=True)

426 with open(input_file, 'w') as f:

427 f.write("data")

428

429 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

430

431 # Mock shutil.move to simulate created file

432 def mock_move(src, dst):

433 os.makedirs(os.path.dirname(dst), exist_ok=True)

434 with open(dst, 'w') as f:

435 f.write("stripped brain")

436

437 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

438 thread.run()

439

440 # Verify derivatives directory created

441 expected_dir = os.path.join(

442 temp_workspace, 'derivatives', 'skullstrips', 'sub-05', 'anat'

443 )

444 assert os.path.exists(expected_dir)

445

446 @patch('main.threads.skull_strip_thread.setup_fsl_env')

447 @patch('main.threads.skull_strip_thread.QProcess')

448 def test_json_metadata_creation_bet(self, mock_qprocess, mock_fsl_env, temp_workspace):

449 """Test JSON metadata creation for BET"""

450 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

451

452 mock_process = Mock()

453 mock_process.waitForFinished.return_value = True

454 mock_process.exitCode.return_value = 0

455 mock_process.readAllStandardError.return_value = b''

456 mock_process.readAllStandardOutput.return_value = b''

457 mock_qprocess.return_value = mock_process

458

459 input_file = os.path.join(temp_workspace, "sub-06", "anat", "T1w.nii.gz")

460 os.makedirs(os.path.dirname(input_file), exist_ok=True)

461 with open(input_file, 'w') as f:

462 f.write("data")

463

464 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.4}, False, "fsl-bet")

465

466 def mock_move(src, dst):

467 os.makedirs(os.path.dirname(dst), exist_ok=True)

468 with open(dst, 'w') as f:

469 f.write("stripped")

470

471 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

472 thread.run()

473

474 # Verify JSON created

475 json_file = os.path.join(

476 temp_workspace, 'derivatives', 'skullstrips', 'sub-06', 'anat',

477 'T1w_f04_brain.json'

478 )

479

480 assert os.path.exists(json_file)

481

482 with open(json_file, 'r') as f:

483 metadata = json.load(f)

484

485 assert metadata["SkullStripped"] is True

486 assert metadata["Description"] == "Skull-stripped brain image"

487 assert "T1w.nii.gz" in metadata["Sources"]

488 assert metadata["SkullStrippingMethod"] == "FSL BET"

489

490 @patch('main.threads.skull_strip_thread.get_bin_path')

491 @patch('main.threads.skull_strip_thread.QProcess')

492 def test_json_metadata_creation_hdbet(self, mock_qprocess, mock_get_bin, temp_workspace):

493 """Test JSON metadata creation for HD-BET"""

494 mock_get_bin.return_value = "/usr/bin/hd-bet"

495

496 mock_process = Mock()

497 mock_process.waitForFinished.return_value = True

498 mock_process.exitCode.return_value = 0

499 mock_process.readAllStandardError.return_value = b''

500 mock_process.readAllStandardOutput.return_value = b''

501 mock_qprocess.return_value = mock_process

502

503 input_file = os.path.join(temp_workspace, "sub-07", "anat", "brain.nii")

504 os.makedirs(os.path.dirname(input_file), exist_ok=True)

505 with open(input_file, 'w') as f:

506 f.write("data")

507

508 thread = SkullStripThread([input_file], temp_workspace, {}, True, "hd-bet")

509

510 def mock_move(src, dst):

511 os.makedirs(os.path.dirname(dst), exist_ok=True)

512 with open(dst, 'w') as f:

513 f.write("stripped")

514

515 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

516 thread.run()

517

518 json_file = os.path.join(

519 temp_workspace, 'derivatives', 'skullstrips', 'sub-07', 'anat',

520 'brain_hd-bet_brain.json'

521 )

522

523 assert os.path.exists(json_file)

524

525 with open(json_file, 'r') as f:

526 metadata = json.load(f)

527

528 assert metadata["SkullStrippingMethod"] == "HD-BET"

529

530 @patch('main.threads.skull_strip_thread.setup_fsl_env')

531 @patch('main.threads.skull_strip_thread.QProcess')

532 def test_output_filename_format_bet(self, mock_qprocess, mock_fsl_env, temp_workspace):

533 """Test output filename format for BET"""

534 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

535

536 mock_process = Mock()

537 mock_process.waitForFinished.return_value = True

538 mock_process.exitCode.return_value = 0

539 mock_process.readAllStandardError.return_value = b''

540 mock_process.readAllStandardOutput.return_value = b''

541 mock_qprocess.return_value = mock_process

542

543 input_file = os.path.join(temp_workspace, "sub-08", "anat", "T1w.nii.gz")

544 os.makedirs(os.path.dirname(input_file), exist_ok=True)

545 with open(input_file, 'w') as f:

546 f.write("data")

547

548 # f_val = 0.7 -> f07

549 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.7}, False, "fsl-bet")

550

551 def mock_move(src, dst):

552 os.makedirs(os.path.dirname(dst), exist_ok=True)

553 with open(dst, 'w') as f:

554 f.write("stripped")

555

556 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

557 thread.run()

558

559 # Verify filename contains f07

560 output_file = os.path.join(

561 temp_workspace, 'derivatives', 'skullstrips', 'sub-08', 'anat',

562 'T1w_f07_brain.nii.gz'

563 )

564

565 assert os.path.exists(output_file)

566

567 @patch('main.threads.skull_strip_thread.get_bin_path')

568 @patch('main.threads.skull_strip_thread.QProcess')

569 def test_json_metadata_creation_synthstrip(self, mock_qprocess, mock_get_bin, temp_workspace):

570 """Test JSON metadata creation for SynthStrip"""

571 mock_get_bin.return_value = "/usr/bin/mri_synthstrip"

572

573 mock_process = Mock()

574 mock_process.waitForFinished.return_value = True

575 mock_process.exitCode.return_value = 0

576 mock_process.readAllStandardError.return_value = b''

577 mock_process.readAllStandardOutput.return_value = b''

578 mock_qprocess.return_value = mock_process

579

580 input_file = os.path.join(temp_workspace, "sub-07", "anat", "brain.nii")

581 os.makedirs(os.path.dirname(input_file), exist_ok=True)

582 with open(input_file, 'w') as f:

583 f.write("data")

584

585 thread = SkullStripThread([input_file], temp_workspace, {}, True, "synthstrip")

586

587 def mock_move(src, dst):

588 os.makedirs(os.path.dirname(dst), exist_ok=True)

589 with open(dst, 'w') as f:

590 f.write("stripped")

591

592 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

593 thread.run()

594

595 json_file = os.path.join(

596 temp_workspace, 'derivatives', 'skullstrips', 'sub-07', 'anat',

597 'brain_synthstrip_brain.json'

598 )

599

600 assert os.path.exists(json_file)

601

602 with open(json_file, 'r') as f:

603 metadata = json.load(f)

604

605 assert metadata["SkullStrippingMethod"] == "SynthStrip"

606

607

608class TestBatchProcessing:

609 """Tests for batch processing of multiple files"""

610

611 @patch('main.threads.skull_strip_thread.setup_fsl_env')

612 @patch('main.threads.skull_strip_thread.QProcess')

613 def test_multiple_files_processing(self, mock_qprocess, mock_fsl_env, temp_workspace):

614 """Test processing of multiple files"""

615 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

616

617 mock_process = Mock()

618 mock_process.waitForFinished.return_value = True

619 mock_process.exitCode.return_value = 0

620 mock_process.readAllStandardError.return_value = b''

621 mock_process.readAllStandardOutput.return_value = b''

622 mock_qprocess.return_value = mock_process

623

624 # Create multiple files

625 files = []

626 for i in range(1, 4):

627 file_path = os.path.join(temp_workspace, f"sub-0{i}", "anat", f"T1w_{i}.nii")

628 os.makedirs(os.path.dirname(file_path), exist_ok=True)

629 with open(file_path, 'w') as f:

630 f.write("data")

631 files.append(file_path)

632

633 thread = SkullStripThread(files, temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

634

635 file_started_count = [0]

636 file_completed_count = [0]

637

638 thread.file_started.connect(lambda f: file_started_count.__setitem__(0, file_started_count[0] + 1))

639 thread.file_completed.connect(lambda f, s, m: file_completed_count.__setitem__(0, file_completed_count[0] + 1))

640

641 def mock_move(src, dst):

642 os.makedirs(os.path.dirname(dst), exist_ok=True)

643 with open(dst, 'w') as f:

644 f.write("stripped")

645

646 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

647 thread.run()

648

649 # Verify signals emitted for each file

650 assert file_started_count[0] == 3

651 assert file_completed_count[0] == 3

652 assert thread.success_count == 3

653

654 @patch('main.threads.skull_strip_thread.setup_fsl_env')

655 @patch('main.threads.skull_strip_thread.QProcess')

656 def test_progress_tracking_multiple_files(self, mock_qprocess, mock_fsl_env, temp_workspace):

657 """Test progress tracking with multiple files"""

658 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

659

660 mock_process = Mock()

661 mock_process.waitForFinished.return_value = True

662 mock_process.exitCode.return_value = 0

663 mock_process.readAllStandardError.return_value = b''

664 mock_process.readAllStandardOutput.return_value = b''

665 mock_qprocess.return_value = mock_process

666

667 files = []

668 for i in range(5):

669 file_path = os.path.join(temp_workspace, f"sub-{i:02d}", "anat", "brain.nii")

670 os.makedirs(os.path.dirname(file_path), exist_ok=True)

671 with open(file_path, 'w') as f:

672 f.write("data")

673 files.append(file_path)

674

675 thread = SkullStripThread(files, temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

676

677 progress_values = []

678 thread.progress_value_updated.connect(lambda val: progress_values.append(val))

679

680 def mock_move(src, dst):

681 os.makedirs(os.path.dirname(dst), exist_ok=True)

682 with open(dst, 'w') as f:

683 f.write("stripped")

684

685 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

686 thread.run()

687

688 # Verify incremental progress

689 assert len(progress_values) == 5

690 # Progress should increase

691 for i in range(1, len(progress_values)):

692 assert progress_values[i] >= progress_values[i - 1]

693

694 @patch('main.threads.skull_strip_thread.setup_fsl_env')

695 @patch('main.threads.skull_strip_thread.QProcess')

696 def test_all_completed_signal(self, mock_qprocess, mock_fsl_env, temp_workspace):

697 """Test all_completed signal at the end"""

698 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

699

700 mock_process = Mock()

701 mock_process.waitForFinished.return_value = True

702 mock_process.exitCode.return_value = 0

703 mock_process.readAllStandardError.return_value = b''

704 mock_process.readAllStandardOutput.return_value = b''

705 mock_qprocess.return_value = mock_process

706

707 files = []

708 for i in range(3):

709 file_path = os.path.join(temp_workspace, f"sub-{i:02d}", "anat", "scan.nii")

710 os.makedirs(os.path.dirname(file_path), exist_ok=True)

711 with open(file_path, 'w') as f:

712 f.write("data")

713 files.append(file_path)

714

715 thread = SkullStripThread(files, temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

716

717 completed_results = []

718 thread.all_completed.connect(lambda count, failed: completed_results.append((count, failed)))

719

720 def mock_move(src, dst):

721 os.makedirs(os.path.dirname(dst), exist_ok=True)

722 with open(dst, 'w') as f:

723 f.write("stripped")

724

725 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

726 thread.run()

727

728 assert len(completed_results) == 1

729 success_count, failed_files = completed_results[0]

730 assert success_count == 3

731 assert len(failed_files) == 0

732

733

734class TestErrorHandling:

735 """Tests for error handling"""

736

737 @patch('main.threads.skull_strip_thread.setup_fsl_env')

738 @patch('main.threads.skull_strip_thread.QProcess')

739 def test_process_failure_nonzero_exit(self, mock_qprocess, mock_fsl_env, temp_workspace):

740 """Test process error handling with non-zero exit code"""

741 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

742

743 mock_process = Mock()

744 mock_process.waitForFinished.return_value = True

745 mock_process.exitCode.return_value = 1 # Error exit code

746 mock_process.readAllStandardError.return_value = b'BET error: invalid parameters'

747 mock_process.readAllStandardOutput.return_value = b''

748 mock_qprocess.return_value = mock_process

749

750 input_file = os.path.join(temp_workspace, "sub-01", "anat", "broken.nii")

751 os.makedirs(os.path.dirname(input_file), exist_ok=True)

752 with open(input_file, 'w') as f:

753 f.write("data")

754

755 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

756

757 completed_files = []

758 thread.file_completed.connect(lambda f, s, m: completed_files.append((f, s, m)))

759

760 with patch('shutil.rmtree'):

761 thread.run()

762

763 # Verify that the file is marked as failed

764 assert len(completed_files) == 1

765 filename, success, message = completed_files[0]

766 assert success is False

767 assert "error" in message.lower() or "invalid" in message.lower()

768 assert len(thread.failed_files) == 1

769

770 @patch('main.threads.skull_strip_thread.setup_fsl_env')

771 @patch('main.threads.skull_strip_thread.QProcess')

772 def test_missing_subject_id(self, mock_qprocess, mock_fsl_env, temp_workspace):

773 """Test handling of files without BIDS subject ID"""

774 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

775

776 # File without BIDS structure (no sub-XX)

777 input_file = os.path.join(temp_workspace, "random_folder", "brain.nii")

778 os.makedirs(os.path.dirname(input_file), exist_ok=True)

779 with open(input_file, 'w') as f:

780 f.write("data")

781

782 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

783

784 completed_files = []

785 thread.file_completed.connect(lambda f, s, m: completed_files.append((f, s, m)))

786

787 thread.run()

788

789 # Should fail due to missing subject ID

790 assert len(completed_files) == 1

791 filename, success, message = completed_files[0]

792 assert success is False

793 assert "Impossibile estrarre" in message or "Cannot extract" in message

794 assert len(thread.failed_files) == 1

795

796 @patch('main.threads.skull_strip_thread.setup_fsl_env')

797 @patch('main.threads.skull_strip_thread.QProcess')

798 def test_exception_during_processing(self, mock_qprocess, mock_fsl_env, temp_workspace):

799 """Test generic exception handling during processing"""

800 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

801

802 mock_process = Mock()

803 mock_process.start.side_effect = RuntimeError("Process start failed")

804 mock_qprocess.return_value = mock_process

805

806 input_file = os.path.join(temp_workspace, "sub-02", "anat", "test.nii")

807 os.makedirs(os.path.dirname(input_file), exist_ok=True)

808 with open(input_file, 'w') as f:

809 f.write("data")

810

811 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

812

813 completed_files = []

814 thread.file_completed.connect(lambda f, s, m: completed_files.append((f, s, m)))

815

816 with patch('shutil.rmtree'):

817 thread.run()

818

819 # Should catch the exception

820 assert len(completed_files) == 1

821 filename, success, message = completed_files[0]

822 assert success is False

823 assert "failed" in message.lower() or "error" in message.lower()

824

825 @patch('main.threads.skull_strip_thread.setup_fsl_env')

826 @patch('main.threads.skull_strip_thread.QProcess')

827 def test_partial_failure_multiple_files(self, mock_qprocess, mock_fsl_env, temp_workspace):

828 """Test with some files failing and others succeeding"""

829 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

830

831 # Mock process alternating success/failure

832 call_count = [0]

833

834 def get_mock_process():

835 mock_process = Mock()

836 mock_process.waitForFinished.return_value = True

837 # File 0 and 2 succeed, file 1 fails

838 if call_count[0] == 1:

839 mock_process.exitCode.return_value = 1

840 mock_process.readAllStandardError.return_value = b'Error'

841 else:

842 mock_process.exitCode.return_value = 0

843 mock_process.readAllStandardError.return_value = b''

844 mock_process.readAllStandardOutput.return_value = b''

845 call_count[0] += 1

846 return mock_process

847

848 mock_qprocess.side_effect = get_mock_process

849

850 files = []

851 for i in range(3):

852 file_path = os.path.join(temp_workspace, f"sub-{i:02d}", "anat", "brain.nii")

853 os.makedirs(os.path.dirname(file_path), exist_ok=True)

854 with open(file_path, 'w') as f:

855 f.write("data")

856 files.append(file_path)

857

858 thread = SkullStripThread(files, temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

859

860 def mock_move(src, dst):

861 os.makedirs(os.path.dirname(dst), exist_ok=True)

862 with open(dst, 'w') as f:

863 f.write("stripped")

864

865 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

866 thread.run()

867

868 # Should have 2 successes and 1 failure

869 assert thread.success_count == 2

870 assert len(thread.failed_files) == 1

871

872

873class TestCancellationDuringProcessing:

874 """Tests for cancellation during processing"""

875

876 @patch('main.threads.skull_strip_thread.setup_fsl_env')

877 @patch('main.threads.skull_strip_thread.QProcess')

878 def test_cancel_during_single_file(self, mock_qprocess, mock_fsl_env, temp_workspace):

879 """Test cancellation during single file processing"""

880 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

881

882 mock_process = Mock()

883

884 # Simulate long process that gets cancelled

885 def wait_with_cancel(*args):

886 if not thread.is_cancelled:

887 thread.cancel()

888 return True

889

890 mock_process.waitForFinished.side_effect = wait_with_cancel

891 mock_process.exitCode.return_value = 0

892 mock_process.readAllStandardError.return_value = b''

893 mock_process.readAllStandardOutput.return_value = b''

894 mock_qprocess.return_value = mock_process

895

896 input_file = os.path.join(temp_workspace, "sub-01", "anat", "brain.nii")

897 os.makedirs(os.path.dirname(input_file), exist_ok=True)

898 with open(input_file, 'w') as f:

899 f.write("data")

900

901 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

902

903 with patch('shutil.rmtree'):

904 thread.run()

905

906 # Process should be killed

907 mock_process.kill.assert_called()

908 assert thread.is_cancelled is True

909

910 @patch('main.threads.skull_strip_thread.setup_fsl_env')

911 @patch('main.threads.skull_strip_thread.QProcess')

912 def test_cancel_stops_batch_processing(self, mock_qprocess, mock_fsl_env, temp_workspace):

913 """Test that cancellation stops batch processing"""

914 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

915

916 mock_process = Mock()

917 mock_process.waitForFinished.return_value = True

918 mock_process.exitCode.return_value = 0

919 mock_process.readAllStandardError.return_value = b''

920 mock_process.readAllStandardOutput.return_value = b''

921 mock_qprocess.return_value = mock_process

922

923 files = []

924 for i in range(5):

925 file_path = os.path.join(temp_workspace, f"sub-{i:02d}", "anat", "brain.nii")

926 os.makedirs(os.path.dirname(file_path), exist_ok=True)

927 with open(file_path, 'w') as f:

928 f.write("data")

929 files.append(file_path)

930

931 thread = SkullStripThread(files, temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

932

933 file_count = [0]

934

935 def on_file_started(f):

936 file_count[0] += 1

937 if file_count[0] == 2: # Cancel after 2nd file

938 thread.cancel()

939

940 thread.file_started.connect(on_file_started)

941

942 def mock_move(src, dst):

943 os.makedirs(os.path.dirname(dst), exist_ok=True)

944 with open(dst, 'w') as f:

945 f.write("stripped")

946

947 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

948 thread.run()

949

950 # Should only process 2 files

951 assert file_count[0] == 2

952 assert thread.success_count < 5

953

954

955class TestSubjectIDExtraction:

956 """Tests for subject ID extraction"""

957

958 @patch('main.threads.skull_strip_thread.setup_fsl_env')

959 @patch('main.threads.skull_strip_thread.QProcess')

960 def test_extract_subject_id_standard_bids(self, mock_qprocess, mock_fsl_env, temp_workspace):

961 """Test extraction of subject ID from standard BIDS path"""

962 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

963

964 mock_process = Mock()

965 mock_process.waitForFinished.return_value = True

966 mock_process.exitCode.return_value = 0

967 mock_process.readAllStandardError.return_value = b''

968 mock_process.readAllStandardOutput.return_value = b''

969 mock_qprocess.return_value = mock_process

970

971 # BIDS path: workspace/sub-XX/anat/file.nii

972 input_file = os.path.join(temp_workspace, "sub-123", "anat", "T1w.nii.gz")

973 os.makedirs(os.path.dirname(input_file), exist_ok=True)

974 with open(input_file, 'w') as f:

975 f.write("data")

976

977 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

978

979 def mock_move(src, dst):

980 os.makedirs(os.path.dirname(dst), exist_ok=True)

981 with open(dst, 'w') as f:

982 f.write("stripped")

983 # Verify that the output path contains sub-123

984 assert "sub-123" in dst

985

986 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

987 thread.run()

988

989 assert thread.success_count == 1

990

991 @patch('main.threads.skull_strip_thread.setup_fsl_env')

992 @patch('main.threads.skull_strip_thread.QProcess')

993 def test_extract_subject_id_nested_path(self, mock_qprocess, mock_fsl_env, temp_workspace):

994 """Test extraction of subject ID from a nested path"""

995 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

996

997 mock_process = Mock()

998 mock_process.waitForFinished.return_value = True

999 mock_process.exitCode.return_value = 0

1000 mock_process.readAllStandardError.return_value = b''

1001 mock_process.readAllStandardOutput.return_value = b''

1002 mock_qprocess.return_value = mock_process

1003

1004 # Path with extra subdirectories

1005 input_file = os.path.join(

1006 temp_workspace, "derivatives", "imported", "sub-456", "anat", "scan.nii"

1007 )

1008 os.makedirs(os.path.dirname(input_file), exist_ok=True)

1009 with open(input_file, 'w') as f:

1010 f.write("data")

1011

1012 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

1013

1014 def mock_move(src, dst):

1015 os.makedirs(os.path.dirname(dst), exist_ok=True)

1016 with open(dst, 'w') as f:

1017 f.write("stripped")

1018 assert "sub-456" in dst

1019

1020 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

1021 thread.run()

1022

1023 assert thread.success_count == 1

1024

1025

1026class TestTempDirectoryHandling:

1027 """Tests for temporary directory management"""

1028

1029 @patch('main.threads.skull_strip_thread.setup_fsl_env')

1030 @patch('main.threads.skull_strip_thread.QProcess')

1031 @patch('tempfile.mkdtemp')

1032 def test_temp_directory_created(self, mock_mkdtemp, mock_qprocess, mock_fsl_env, temp_workspace):

1033 """Test that the temporary directory is created"""

1034 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

1035 temp_dir_path = os.path.join(temp_workspace, "temp_skullstrip")

1036 mock_mkdtemp.return_value = temp_dir_path

1037 os.makedirs(temp_dir_path, exist_ok=True)

1038

1039 mock_process = Mock()

1040 mock_process.waitForFinished.return_value = True

1041 mock_process.exitCode.return_value = 0

1042 mock_process.readAllStandardError.return_value = b''

1043 mock_process.readAllStandardOutput.return_value = b''

1044 mock_qprocess.return_value = mock_process

1045

1046 input_file = os.path.join(temp_workspace, "sub-01", "anat", "brain.nii")

1047 os.makedirs(os.path.dirname(input_file), exist_ok=True)

1048 with open(input_file, 'w') as f:

1049 f.write("data")

1050

1051 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

1052

1053 with patch('shutil.move'), patch('shutil.rmtree') as mock_rmtree:

1054 thread.run()

1055

1056 # Verify that the temp directory was created and then removed

1057 mock_mkdtemp.assert_called()

1058 mock_rmtree.assert_called()

1059

1060 @patch('main.threads.skull_strip_thread.setup_fsl_env')

1061 @patch('main.threads.skull_strip_thread.QProcess')

1062 def test_temp_directory_cleanup_on_error(self, mock_qprocess, mock_fsl_env, temp_workspace):

1063 """Test cleanup of temp directory in case of error"""

1064 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

1065

1066 mock_process = Mock()

1067 mock_process.waitForFinished.return_value = True

1068 mock_process.exitCode.return_value = 1 # Error

1069 mock_process.readAllStandardError.return_value = b'Error'

1070 mock_process.readAllStandardOutput.return_value = b''

1071 mock_qprocess.return_value = mock_process

1072

1073 input_file = os.path.join(temp_workspace, "sub-02", "anat", "brain.nii")

1074 os.makedirs(os.path.dirname(input_file), exist_ok=True)

1075 with open(input_file, 'w') as f:

1076 f.write("data")

1077

1078 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

1079

1080 with patch('shutil.rmtree') as mock_rmtree:

1081 thread.run()

1082

1083 # Should clean up the temp directory even after an error

1084 assert mock_rmtree.called

1085

1086 @patch('main.threads.skull_strip_thread.setup_fsl_env')

1087 @patch('main.threads.skull_strip_thread.QProcess')

1088 def test_temp_directory_cleanup_on_cancel(self, mock_qprocess, mock_fsl_env, temp_workspace):

1089 """Test cleanup of temp directory upon cancellation"""

1090 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

1091

1092 mock_process = Mock()

1093

1094 def wait_and_cancel(*args):

1095 thread.cancel()

1096 return True

1097

1098 mock_process.waitForFinished.side_effect = wait_and_cancel

1099 mock_qprocess.return_value = mock_process

1100

1101 input_file = os.path.join(temp_workspace, "sub-03", "anat", "scan.nii")

1102 os.makedirs(os.path.dirname(input_file), exist_ok=True)

1103 with open(input_file, 'w') as f:

1104 f.write("data")

1105

1106 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

1107

1108 with patch('shutil.rmtree') as mock_rmtree:

1109 thread.run()

1110

1111 # Temp directory should be cleaned up

1112 assert mock_rmtree.called

1113

1114class TestEdgeCases:

1115 """Tests for edge cases"""

1116

1117 @patch('main.threads.skull_strip_thread.setup_fsl_env')

1118 @patch('main.threads.skull_strip_thread.QProcess')

1119 def test_empty_file_list(self, mock_qprocess, mock_fsl_env, temp_workspace):

1120 """Test with empty file list"""

1121 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

1122

1123 thread = SkullStripThread([], temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

1124

1125 # Should not raise exceptions

1126 # Note: might cause division by zero in progress_per_file

1127 try:

1128 thread.run()

1129 except ZeroDivisionError:

1130 pytest.fail("Division by zero with empty list")

1131

1132 @patch('main.threads.skull_strip_thread.setup_fsl_env')

1133 @patch('main.threads.skull_strip_thread.QProcess')

1134 def test_file_with_spaces_in_name(self, mock_qprocess, mock_fsl_env, temp_workspace):

1135 """Test file with spaces in name"""

1136 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

1137

1138 mock_process = Mock()

1139 mock_process.waitForFinished.return_value = True

1140 mock_process.exitCode.return_value = 0

1141 mock_process.readAllStandardError.return_value = b''

1142 mock_process.readAllStandardOutput.return_value = b''

1143 mock_qprocess.return_value = mock_process

1144

1145 input_file = os.path.join(temp_workspace, "sub-01", "anat", "T1 weighted image.nii.gz")

1146 os.makedirs(os.path.dirname(input_file), exist_ok=True)

1147 with open(input_file, 'w') as f:

1148 f.write("data")

1149

1150 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

1151

1152 def mock_move(src, dst):

1153 os.makedirs(os.path.dirname(dst), exist_ok=True)

1154 with open(dst, 'w') as f:

1155 f.write("stripped")

1156

1157 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

1158 thread.run()

1159

1160 assert thread.success_count == 1

1161

1162 @patch('main.threads.skull_strip_thread.setup_fsl_env')

1163 @patch('main.threads.skull_strip_thread.QProcess')

1164 def test_nifti_without_gz_extension(self, mock_qprocess, mock_fsl_env, temp_workspace):

1165 """Test uncompressed NIfTI file (.nii)"""

1166 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

1167

1168 mock_process = Mock()

1169 mock_process.waitForFinished.return_value = True

1170 mock_process.exitCode.return_value = 0

1171 mock_process.readAllStandardError.return_value = b''

1172 mock_process.readAllStandardOutput.return_value = b''

1173 mock_qprocess.return_value = mock_process

1174

1175 input_file = os.path.join(temp_workspace, "sub-04", "anat", "brain.nii")

1176 os.makedirs(os.path.dirname(input_file), exist_ok=True)

1177 with open(input_file, 'w') as f:

1178 f.write("data")

1179

1180 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

1181

1182 def mock_move(src, dst):

1183 os.makedirs(os.path.dirname(dst), exist_ok=True)

1184 with open(dst, 'w') as f:

1185 f.write("stripped")

1186 # Output should still be .nii.gz

1187 assert dst.endswith(".nii.gz")

1188

1189 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

1190 thread.run()

1191

1192 assert thread.success_count == 1

1193

1194 @patch('main.threads.skull_strip_thread.setup_fsl_env')

1195 @patch('main.threads.skull_strip_thread.QProcess')

1196 def test_extreme_f_values(self, mock_qprocess, mock_fsl_env, temp_workspace):

1197 """Test with extreme f_val values"""

1198 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

1199

1200 mock_process = Mock()

1201 mock_process.waitForFinished.return_value = True

1202 mock_process.exitCode.return_value = 0

1203 mock_process.readAllStandardError.return_value = b''

1204 mock_process.readAllStandardOutput.return_value = b''

1205 mock_qprocess.return_value = mock_process

1206

1207 extreme_values = [0.0, 0.1, 0.9, 1.0]

1208

1209 for f_val in extreme_values:

1210 input_file = os.path.join(

1211 temp_workspace, f"sub-f{int(f_val * 10)}", "anat", "brain.nii"

1212 )

1213 os.makedirs(os.path.dirname(input_file), exist_ok=True)

1214 with open(input_file, 'w') as f:

1215 f.write("data")

1216

1217 thread = SkullStripThread([input_file], temp_workspace, {'f_val': f_val}, False, "fsl-bet")

1218

1219 def mock_move(src, dst):

1220 os.makedirs(os.path.dirname(dst), exist_ok=True)

1221 with open(dst, 'w') as f:

1222 f.write("stripped")

1223

1224 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

1225 thread.run()

1226

1227 assert thread.success_count == 1

1228

1229

1230class TestSignalEmissions:

1231 """Tests for signal emissions"""

1232

1233 @patch('main.threads.skull_strip_thread.setup_fsl_env')

1234 @patch('main.threads.skull_strip_thread.QProcess')

1235 def test_progress_updated_signal(self, mock_qprocess, mock_fsl_env, temp_workspace):

1236 """Test progress_updated signal emission"""

1237 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

1238

1239 mock_process = Mock()

1240 mock_process.waitForFinished.return_value = True

1241 mock_process.exitCode.return_value = 0

1242 mock_process.readAllStandardError.return_value = b''

1243 mock_process.readAllStandardOutput.return_value = b''

1244 mock_qprocess.return_value = mock_process

1245

1246 input_file = os.path.join(temp_workspace, "sub-01", "anat", "T1w.nii")

1247 os.makedirs(os.path.dirname(input_file), exist_ok=True)

1248 with open(input_file, 'w') as f:

1249 f.write("data")

1250

1251 thread = SkullStripThread([input_file], temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

1252

1253 progress_messages = []

1254 thread.progress_updated.connect(lambda msg: progress_messages.append(msg))

1255

1256 def mock_move(src, dst):

1257 os.makedirs(os.path.dirname(dst), exist_ok=True)

1258 with open(dst, 'w') as f:

1259 f.write("stripped")

1260

1261 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

1262 thread.run()

1263

1264 assert len(progress_messages) > 0

1265 # Message should contain info about processed file

1266 assert any("T1w.nii" in msg or "Processing" in msg for msg in progress_messages)

1267

1268 @patch('main.threads.skull_strip_thread.setup_fsl_env')

1269 @patch('main.threads.skull_strip_thread.QProcess')

1270 def test_file_started_signal_order(self, mock_qprocess, mock_fsl_env, temp_workspace):

1271 """Test file_started signal emission order"""

1272 mock_fsl_env.return_value = ("/usr/local/fsl", "NIFTI_GZ")

1273

1274 mock_process = Mock()

1275 mock_process.waitForFinished.return_value = True

1276 mock_process.exitCode.return_value = 0

1277 mock_process.readAllStandardError.return_value = b''

1278 mock_process.readAllStandardOutput.return_value = b''

1279 mock_qprocess.return_value = mock_process

1280

1281 files = []

1282 filenames = ["file1.nii", "file2.nii", "file3.nii"]

1283 for i, fname in enumerate(filenames):

1284 file_path = os.path.join(temp_workspace, f"sub-{i:02d}", "anat", fname)

1285 os.makedirs(os.path.dirname(file_path), exist_ok=True)

1286 with open(file_path, 'w') as f:

1287 f.write("data")

1288 files.append(file_path)

1289

1290 thread = SkullStripThread(files, temp_workspace, {'f_val': 0.5}, False, "fsl-bet")

1291

1292 started_files = []

1293 thread.file_started.connect(lambda f: started_files.append(f))

1294

1295 def mock_move(src, dst):

1296 os.makedirs(os.path.dirname(dst), exist_ok=True)

1297 with open(dst, 'w') as f:

1298 f.write("stripped")

1299

1300 with patch('shutil.move', side_effect=mock_move), patch('shutil.rmtree'):

1301 thread.run()

1302

1303 # Verify order

1304 assert len(started_files) == 3

1305 assert started_files == filenames

Coverage for tests/threads/test_skull_strip_thread.py: 99%

852 statements