Coverage for tests/threads/test_nifti_utils

1"""

2test_save_load_threads.py - Test Suite for SaveNiftiThread and ImageLoadThread

4This suite tests all functionalities of the saving and loading threads:

5- SaveNiftiThread: NIfTI saving + JSON with BIDS metadata

6- ImageLoadThread: loading and normalization of 3D/4D NIfTI

7- Error handling and progress tracking

8- Intensity normalization with percentiles

9"""

11import json

12import os

13import tempfile

14import shutil

15from unittest.mock import Mock, patch, MagicMock

16import pytest

17import numpy as np

18import nibabel as nib

20from main.threads.nifti_utils_threads import SaveNiftiThread, ImageLoadThread

23class TestSaveNiftiThreadInitialization:

24 """Tests for SaveNiftiThread initialization"""

26 def test_init_with_all_parameters(self, temp_workspace):

27 """Test initialization with all parameters"""

28 data = np.random.rand(10, 10, 10)

29 affine = np.eye(4)

30 path = os.path.join(temp_workspace, "output.nii.gz")

31 json_path = os.path.join(temp_workspace, "output.json")

32 relative_path = "sub-01/anat/T1w.nii.gz"

33 radius = 5.0

34 difference = 0.3

35 source_dict = {"radius": radius, "difference": difference}

37 thread = SaveNiftiThread(

38 data=data,

39 affine=affine,

40 path=path,

41 json_path=json_path,

42 relative_path=relative_path,

43 source_dict=source_dict

44 )

46 assert thread.data is data

47 assert np.array_equal(thread.affine, affine)

48 assert thread.path == path

49 assert thread.json_path == json_path

50 assert thread.relative_path == relative_path

51 assert thread.source_dict["radius"] == radius

52 assert thread.source_dict["difference"] == difference

54 def test_signals_exist(self, temp_workspace):

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

56 data = np.zeros((5, 5, 5))

57 affine = np.eye(4)

58 source_dict = {"radius": 1.0, "difference": 0.1}

59 thread = SaveNiftiThread(

60 data, affine, "path.nii", "path.json", "rel/path.nii", source_dict

61 )

63 assert hasattr(thread, 'success')

64 assert hasattr(thread, 'error')

67class TestSaveNiftiThreadExecution:

68 """Tests for SaveNiftiThread execution"""

70 def test_successful_save(self, temp_workspace):

71 """Test successful saving of NIfTI and JSON"""

72 # Test data

73 data = np.random.randint(0, 255, (10, 10, 10), dtype=np.uint8)

74 affine = np.eye(4)

75 affine[0, 0] = 2.0 # Voxel size

77 nifti_path = os.path.join(temp_workspace, "mask.nii.gz")

78 json_path = os.path.join(temp_workspace, "mask.json")

79 relative_path = "sub-01/anat/T1w.nii.gz"

81 radius = 3.5

82 difference = 0.25

83 source_dict = {"radius": radius, "difference": difference}

85 thread = SaveNiftiThread(

86 data=data,

87 affine=affine,

88 path=nifti_path,

89 json_path=json_path,

90 relative_path=relative_path,

91 source_dict=source_dict

92 )

94 # Connect signal

95 success_paths = []

97 def on_success(path, json_p):

98 success_paths.append((path, json_p))

100 thread.success.connect(on_success)

101

102 # Execute

103 thread.run()

104

105 # Verify signal emitted

106 assert len(success_paths) == 1

107 assert success_paths[0][0] == nifti_path

108 assert success_paths[0][1] == json_path

109

110 # Verify NIfTI file created

111 assert os.path.exists(nifti_path)

112 loaded_img = nib.load(nifti_path)

113 loaded_data = loaded_img.get_fdata()

114 assert loaded_data.shape == data.shape

115 np.testing.assert_array_almost_equal(loaded_data, data, decimal=0)

116

117 # Verify affine

118 np.testing.assert_array_almost_equal(loaded_img.affine, affine)

119

120 def test_json_metadata_structure(self, temp_workspace):

121 """Test structure and content of the JSON file"""

122 data = np.ones((5, 5, 5), dtype=np.uint8)

123 affine = np.eye(4)

124

125 nifti_path = os.path.join(temp_workspace, "roi.nii.gz")

126 json_path = os.path.join(temp_workspace, "roi.json")

127 relative_path = "sub-02/anat/brain.nii.gz"

128 radius = 4.2

129 difference = 0.15

130 source_dict = {"radius": radius, "difference": difference}

131

132 thread = SaveNiftiThread(

133 data, affine, nifti_path, json_path, relative_path, source_dict

134 )

135 thread.run()

136

137 # Verify JSON created

138 assert os.path.exists(json_path)

139

140 with open(json_path, 'r') as f:

141 metadata = json.load(f)

142

143 # Verify structure

144 assert "Type" in metadata

145 assert metadata["Type"] == "ROI"

146

147 assert "Sources" in metadata

148 assert isinstance(metadata["Sources"], list)

149 assert len(metadata["Sources"]) == 1

150 assert metadata["Sources"][0] == f"bids:{relative_path}"

151

152 assert "Description" in metadata

153 assert "roi mask" in metadata["Description"].lower()

154

155 assert "Origin" in metadata

156 params = metadata["Origin"]

157 assert params["radius"] == radius

158 assert params["difference"] == difference

159

160 def test_json_indentation(self, temp_workspace):

161 """Test that the JSON is formatted with indentation"""

162 data = np.zeros((3, 3, 3), dtype=np.uint8)

163 affine = np.eye(4)

164

165 nifti_path = os.path.join(temp_workspace, "test.nii")

166 json_path = os.path.join(temp_workspace, "test.json")

167

168 radius = 1.0

169 difference = 0.1

170 source_dict = {"radius": radius, "difference": difference}

171

172 thread = SaveNiftiThread(

173 data, affine, nifti_path, json_path, "path.nii", source_dict

174 )

175 thread.run()

176

177 with open(json_path, 'r') as f:

178 content = f.read()

179

180 # Verify indentation (formatted JSON)

181 assert " " in content # 4 spaces indentation

182 assert "\n" in content # Newlines between fields

183

184 def test_data_type_conversion_to_uint8(self, temp_workspace):

185 """Test automatic conversion of data to uint8"""

186 # Float data

187 data = np.random.rand(5, 5, 5) * 255

188 affine = np.eye(4)

189

190 nifti_path = os.path.join(temp_workspace, "converted.nii.gz")

191 json_path = os.path.join(temp_workspace, "converted.json")

192

193 source_dict = {"radius": 1.0, "difference": 0.1}

194

195 thread = SaveNiftiThread(

196 data, affine, nifti_path, json_path, "rel.nii", source_dict

197 )

198 thread.run()

199

200 # Verify saved data type

201 loaded_img = nib.load(nifti_path)

202 loaded_data = loaded_img.get_fdata()

203

204 # Should be converted to uint8

205 assert loaded_data.dtype in [np.uint8, np.float32, np.float64]

206 assert loaded_data.min() >= 0

207 assert loaded_data.max() <= 255

208

209 def test_error_handling_invalid_path(self, temp_workspace):

210 """Test error handling with invalid path"""

211 data = np.zeros((3, 3, 3))

212 affine = np.eye(4)

213

214 # Invalid path (nonexistent directory)

215 invalid_path = os.path.join(temp_workspace, "nonexistent", "dir", "file.nii")

216 json_path = os.path.join(temp_workspace, "nonexistent", "dir", "file.json")

217

218 source_dict = {"radius": 1.0, "difference": 0.1}

219

220 thread = SaveNiftiThread(

221 data, affine, invalid_path, json_path, "rel.nii", source_dict

222 )

223

224 error_msgs = []

225 thread.error.connect(lambda msg: error_msgs.append(msg))

226

227 thread.run()

228

229 # Should emit an error

230 assert len(error_msgs) == 1

231 assert len(error_msgs[0]) > 0

232

233 @patch('nibabel.save')

234 def test_error_during_nifti_save(self, mock_save, temp_workspace):

235 """Test error handling during NIfTI saving"""

236 mock_save.side_effect = IOError("Disk full")

237

238 data = np.zeros((3, 3, 3))

239 affine = np.eye(4)

240

241 nifti_path = os.path.join(temp_workspace, "fail.nii")

242 json_path = os.path.join(temp_workspace, "fail.json")

243

244 source_dict = {"radius": 1.0, "difference": 0.1}

245

246 thread = SaveNiftiThread(

247 data, affine, nifti_path, json_path, "rel.nii", source_dict

248 )

249

250 error_msgs = []

251 thread.error.connect(lambda msg: error_msgs.append(msg))

252

253 thread.run()

254

255 assert len(error_msgs) == 1

256 assert "Disk full" in error_msgs[0]

257

258 def test_different_affine_matrices(self, temp_workspace):

259 """Test with different affine matrices"""

260 data = np.ones((4, 4, 4), dtype=np.uint8)

261

262 # Affine with different voxel sizes

263 affine = np.diag([2.5, 3.0, 3.5, 1.0])

264 affine[:3, 3] = [10, 20, 30] # Offset

265

266 nifti_path = os.path.join(temp_workspace, "custom_affine.nii")

267 json_path = os.path.join(temp_workspace, "custom_affine.json")

268

269 source_dict = {"radius": 2.0, "difference": 0.2}

270

271 thread = SaveNiftiThread(

272 data, affine, nifti_path, json_path, "path.nii", source_dict

273 )

274 thread.run()

275

276 # Verify affine saved correctly

277 loaded_img = nib.load(nifti_path)

278 np.testing.assert_array_almost_equal(loaded_img.affine, affine)

279

280

281class TestImageLoadThreadInitialization:

282 """Tests for ImageLoadThread initialization"""

283

284 def test_init_regular_image(self, temp_workspace):

285 """Test initialization for regular image"""

286 file_path = os.path.join(temp_workspace, "test.nii")

287 thread = ImageLoadThread(file_path, is_overlay=False)

288

289 assert thread.file_path == file_path

290 assert thread.is_overlay is False

291

292 def test_init_overlay_image(self, temp_workspace):

293 """Test initialization for overlay image"""

294 file_path = os.path.join(temp_workspace, "overlay.nii.gz")

295 thread = ImageLoadThread(file_path, is_overlay=True)

296

297 assert thread.file_path == file_path

298 assert thread.is_overlay is True

299

300 def test_signals_exist(self, temp_workspace):

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

302 thread = ImageLoadThread("dummy.nii", False)

303

304 assert hasattr(thread, 'finished')

305 assert hasattr(thread, 'error')

306 assert hasattr(thread, 'progress')

307

308

309class TestImageLoadThread3D:

310 """Tests for loading 3D NIfTI images"""

311

312 def test_load_3d_nifti_success(self, temp_workspace):

313 """Test successful loading of 3D NIfTI"""

314 # Create 3D test image

315 data_3d = np.random.rand(20, 20, 20).astype(np.float32) * 100

316 affine = np.eye(4)

317 img = nib.Nifti1Image(data_3d, affine)

318

319 nifti_path = os.path.join(temp_workspace, "test_3d.nii.gz")

320 nib.save(img, nifti_path)

321

322 thread = ImageLoadThread(nifti_path, is_overlay=False)

323

324 # Connect signal

325 results = []

326

327 def on_finished(img_data, dims, aff, is_4d, is_overlay):

328 results.append({

329 'img_data': img_data,

330 'dims': dims,

331 'affine': aff,

332 'is_4d': is_4d,

333 'is_overlay': is_overlay

334 })

335

336 thread.finished.connect(on_finished)

337

338 # Execute

339 thread.run()

340

341 # Verify results

342 assert len(results) == 1

343 result = results[0]

344

345 assert result['dims'] == (20, 20, 20)

346 assert result['is_4d'] is False

347 assert result['is_overlay'] is False

348 assert result['img_data'].shape == (20, 20, 20)

349 assert result['img_data'].dtype == np.float32

350

351 # Verify normalization (values between 0 and 1)

352 assert result['img_data'].min() >= 0

353 assert result['img_data'].max() <= 1

354

355 def test_load_3d_progress_emissions(self, temp_workspace):

356 """Test progress emissions during 3D loading"""

357 # Create simple image

358 data = np.ones((10, 10, 10))

359 img = nib.Nifti1Image(data, np.eye(4))

360

361 nifti_path = os.path.join(temp_workspace, "progress_test.nii")

362 nib.save(img, nifti_path)

363

364 thread = ImageLoadThread(nifti_path, False)

365

366 progress_values = []

367 thread.progress.connect(lambda val: progress_values.append(val))

368

369 thread.run()

370

371 # Verify progress emitted

372 assert len(progress_values) > 0

373 assert 10 in progress_values

374 assert 100 in progress_values

375

376 # Verify monotonic increase

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

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

379

380

381class TestImageLoadThread4D:

382 """Tests for loading 4D NIfTI images"""

383

384 def test_load_4d_nifti_success(self, temp_workspace):

385 """Test successful loading of 4D NIfTI"""

386 # Create 4D image (e.g., fMRI time series)

387 data_4d = np.random.rand(15, 15, 15, 10).astype(np.float32) * 50

388 affine = np.eye(4)

389 img = nib.Nifti1Image(data_4d, affine)

390

391 nifti_path = os.path.join(temp_workspace, "test_4d.nii.gz")

392 nib.save(img, nifti_path)

393

394 thread = ImageLoadThread(nifti_path, is_overlay=True)

395

396 results = []

397 thread.finished.connect(

398 lambda img_data, dims, aff, is_4d, is_overlay:

399 results.append((img_data, dims))

400 )

401

402 thread.run()

403

404 assert len(results) == 1

405 img_data, dims = results[0]

406 assert dims == (15, 15, 15, 10)

407 assert img_data.shape == (15, 15, 15, 10)

408

409 def test_anisotropic_voxels(self, temp_workspace):

410 """Test with anisotropic voxels"""

411 data = np.random.rand(10, 20, 30).astype(np.float32)

412

413 # Affine with different voxel sizes for each axis

414 affine = np.diag([0.5, 1.0, 2.0, 1.0])

415

416 img = nib.Nifti1Image(data, affine)

417 nifti_path = os.path.join(temp_workspace, "anisotropic.nii")

418 nib.save(img, nifti_path)

419

420 thread = ImageLoadThread(nifti_path, False)

421

422 results = []

423 thread.finished.connect(

424 lambda img_data, dims, aff, is_4d, is_overlay:

425 results.append((img_data, aff))

426 )

427

428 thread.run()

429

430 assert len(results) == 1

431 img_data, loaded_affine = results[0]

432

433 # Dimensions preserved (or reordered after canonicalization)

434 assert img_data.ndim == 3

435 assert isinstance(loaded_affine, np.ndarray)

436

437 def test_different_data_types(self, temp_workspace):

438 """Test with different input dtypes"""

439 dtypes_to_test = [np.uint8, np.int16, np.float32, np.float64]

440

441 for dtype in dtypes_to_test:

442 data = (np.random.rand(8, 8, 8) * 100).astype(dtype)

443 img = nib.Nifti1Image(data, np.eye(4))

444

445 nifti_path = os.path.join(temp_workspace, f"dtype_{dtype.__name__}.nii")

446 nib.save(img, nifti_path)

447

448 thread = ImageLoadThread(nifti_path, False)

449

450 results = []

451 thread.finished.connect(

452 lambda img_data, dims, aff, is_4d, is_overlay:

453 results.append(img_data)

454 )

455

456 thread.run()

457

458 # Should always normalize to float32

459 assert len(results) == 1

460 assert results[0].dtype == np.float32

461 assert results[0].min() >= 0

462 assert results[0].max() <= 1

463

464 def test_negative_values_handling(self, temp_workspace):

465 """Test handling of negative values"""

466 # Data with negative values (e.g., parametric maps)

467 data = np.random.randn(15, 15, 15).astype(np.float32)

468

469 img = nib.Nifti1Image(data, np.eye(4))

470 nifti_path = os.path.join(temp_workspace, "negative_values.nii")

471 nib.save(img, nifti_path)

472

473 thread = ImageLoadThread(nifti_path, False)

474

475 results = []

476 thread.finished.connect(

477 lambda img_data, dims, aff, is_4d, is_overlay:

478 results.append(img_data)

479 )

480

481 thread.run()

482

483 normalized = results[0]

484

485 # Even with negative values, normalization must be in [0, 1]

486 assert normalized.min() >= 0

487 assert normalized.max() <= 1

488

489

490class TestSaveLoadThreadsIntegration:

491 """Integration tests between SaveNiftiThread and ImageLoadThread"""

492

493 def test_save_and_reload_cycle(self, temp_workspace):

494 """Full cycle test: save and reload"""

495 # Original data

496 original_data = np.random.randint(0, 255, (12, 12, 12), dtype=np.uint8)

497 affine = np.eye(4)

498 affine[0, 0] = 2.0

499

500 # Save

501 nifti_path = os.path.join(temp_workspace, "cycle_test.nii.gz")

502 json_path = os.path.join(temp_workspace, "cycle_test.json")

503

504 radius = 2.5

505 difference = 0.2

506 source_dict = {"radius": radius, "difference": difference}

507

508 save_thread = SaveNiftiThread(

509 data=original_data,

510 affine=affine,

511 path=nifti_path,

512 json_path=json_path,

513 relative_path="sub-01/anat/T1w.nii.gz",

514 source_dict=source_dict

515 )

516

517 save_success = []

518 save_thread.success.connect(

519 lambda p, j: save_success.append(True)

520 )

521

522 save_thread.run()

523 assert len(save_success) == 1

524

525 # Reload

526 load_thread = ImageLoadThread(nifti_path, is_overlay=False)

527

528 loaded_results = []

529 load_thread.finished.connect(

530 lambda img_data, dims, aff, is_4d, is_overlay:

531 loaded_results.append((img_data, dims, aff))

532 )

533

534 load_thread.run()

535

536 assert len(loaded_results) == 1

537 loaded_data, loaded_dims, loaded_affine = loaded_results[0]

538

539 # Verify dimensions

540 assert loaded_dims == (12, 12, 12)

541

542 # Verify affine (may slightly differ after canonicalization)

543 # but voxel size should be similar

544 assert loaded_affine.shape == (4, 4)

545

546 # Verify JSON

547 assert os.path.exists(json_path)

548 with open(json_path, 'r') as f:

549 metadata = json.load(f)

550 assert metadata["Type"] == "ROI"

551 assert metadata["Origin"]["radius"] == 2.5

552

553 def test_save_load_multiple_files(self, temp_workspace):

554 """Test saving and loading multiple files"""

555 num_files = 5

556 saved_paths = []

557

558 for i in range(num_files):

559 data = np.random.randint(0, 100, (8, 8, 8), dtype=np.uint8)

560 nifti_path = os.path.join(temp_workspace, f"multi_{i}.nii")

561 json_path = os.path.join(temp_workspace, f"multi_{i}.json")

562

563 source_dict = {"radius": float(i), "difference": float(i) * 0.1}

564

565 save_thread = SaveNiftiThread(

566 data, np.eye(4), nifti_path, json_path,

567 f"path_{i}.nii", source_dict

568 )

569 save_thread.run()

570 saved_paths.append(nifti_path)

571

572 # Reload all

573 for path in saved_paths:

574 load_thread = ImageLoadThread(path, False)

575

576 results = []

577 load_thread.finished.connect(

578 lambda img_data, dims, aff, is_4d, is_overlay:

579 results.append(True)

580 )

581

582 load_thread.run()

583 assert len(results) == 1

584

585 def test_save_4d_reload_as_4d(self, temp_workspace):

586 """Test saving and reloading 4D data"""

587 # Create 4D data

588 data_4d = np.random.randint(0, 255, (10, 10, 10, 5), dtype=np.uint8)

589 affine = np.eye(4)

590

591 nifti_path = os.path.join(temp_workspace, "test_4d_cycle.nii.gz")

592 json_path = os.path.join(temp_workspace, "test_4d_cycle.json")

593

594 source_dict = {"radius": 3.0, "difference": 0.15}

595

596 # SaveNiftiThread should also handle 4D

597 save_thread = SaveNiftiThread(

598 data_4d, affine, nifti_path, json_path,

599 "rel_4d.nii", source_dict

600 )

601

602 save_success = []

603 save_thread.success.connect(lambda p, j: save_success.append(True))

604 save_thread.run()

605

606 assert len(save_success) == 1

607

608 # Reload as 4D

609 load_thread = ImageLoadThread(nifti_path, False)

610

611 results = []

612 load_thread.finished.connect(

613 lambda img_data, dims, aff, is_4d, is_overlay:

614 results.append((dims, is_4d))

615 )

616

617 load_thread.run()

618

619 dims, is_4d = results[0]

620 assert len(dims) == 4

621 assert is_4d is True

622

623

624class TestConcurrencyAndThreadSafety:

625 """Tests for concurrency and thread safety"""

626

627 def test_multiple_save_threads_concurrent(self, temp_workspace):

628 """Test concurrent execution of multiple SaveNiftiThread"""

629 threads = []

630 success_count = [0]

631

632 def on_success(p, j):

633 success_count[0] += 1

634

635 for i in range(3):

636 data = np.ones((5, 5, 5), dtype=np.uint8) * i

637 nifti_path = os.path.join(temp_workspace, f"concurrent_save_{i}.nii")

638 json_path = os.path.join(temp_workspace, f"concurrent_save_{i}.json")

639

640 source_dict = {"radius": 1.0, "difference": 0.1}

641

642 thread = SaveNiftiThread(

643 data, np.eye(4), nifti_path, json_path,

644 f"path_{i}.nii", source_dict

645 )

646 thread.success.connect(on_success)

647 threads.append(thread)

648

649 # Start all threads

650 for thread in threads:

651 thread.run()

652

653 # Wait for completion

654 for thread in threads:

655 thread.wait(5000) # 5-second timeout

656

657 assert success_count[0] == 3

658

659 def test_multiple_load_threads_concurrent(self, temp_workspace):

660 """Test concurrent loading of multiple files"""

661 # Create test files

662 file_paths = []

663 for i in range(3):

664 data = np.random.rand(6, 6, 6).astype(np.float32)

665 img = nib.Nifti1Image(data, np.eye(4))

666 path = os.path.join(temp_workspace, f"concurrent_load_{i}.nii")

667 nib.save(img, path)

668 file_paths.append(path)

669

670 threads = []

671 finished_count = [0]

672

673 def on_finished(img_data, dims, aff, is_4d, is_overlay):

674 finished_count[0] += 1

675

676 for path in file_paths:

677 thread = ImageLoadThread(path, False)

678 thread.finished.connect(on_finished)

679 threads.append(thread)

680

681 # Start all

682 for thread in threads:

683 thread.run()

684

685 # Wait

686 for thread in threads:

687 thread.wait(5000)

688

689 assert finished_count[0] == 3

690

691

692class TestSpecialCases:

693 """Tests for special cases and boundary conditions"""

694

695 def test_save_zero_volume(self, temp_workspace):

696 """Test saving a volume with all zeros"""

697 data = np.zeros((10, 10, 10), dtype=np.uint8)

698 nifti_path = os.path.join(temp_workspace, "all_zeros.nii")

699 json_path = os.path.join(temp_workspace, "all_zeros.json")

700

701 source_dict = {"radius": 1.0, "difference": 0.1}

702

703 thread = SaveNiftiThread(

704 data, np.eye(4), nifti_path, json_path,

705 "zeros.nii", source_dict

706 )

707

708 success = []

709 thread.success.connect(lambda p, j: success.append(True))

710 thread.run()

711

712 assert len(success) == 1

713 assert os.path.exists(nifti_path)

714

715 def test_load_zero_volume(self, temp_workspace):

716 """Test loading a volume with all zeros"""

717 data = np.zeros((8, 8, 8))

718 img = nib.Nifti1Image(data, np.eye(4))

719

720 nifti_path = os.path.join(temp_workspace, "zeros_load.nii")

721 nib.save(img, nifti_path)

722

723 thread = ImageLoadThread(nifti_path, False)

724

725 results = []

726 thread.finished.connect(

727 lambda img_data, dims, aff, is_4d, is_overlay:

728 results.append(img_data)

729 )

730

731 thread.run()

732

733 # Should handle it correctly

734 assert len(results) == 1

735 normalized = results[0]

736 assert not np.any(np.isnan(normalized))

737

738 def test_save_binary_mask(self, temp_workspace):

739 """Test saving a binary mask"""

740 # Binary mask (0 and 1)

741 mask = np.random.randint(0, 2, (15, 15, 15), dtype=np.uint8)

742

743 nifti_path = os.path.join(temp_workspace, "binary_mask.nii.gz")

744 json_path = os.path.join(temp_workspace, "binary_mask.json")

745

746 source_dict = {"radius": 0.0, "difference": 0.0}

747

748 thread = SaveNiftiThread(

749 mask, np.eye(4), nifti_path, json_path,

750 "mask.nii", source_dict

751 )

752

753 success = []

754 thread.success.connect(lambda p, j: success.append(True))

755 thread.run()

756

757 assert len(success) == 1

758

759 # Verify integrity

760 loaded = nib.load(nifti_path)

761 loaded_data = loaded.get_fdata()

762

763 # Should preserve binary values

764 unique_vals = np.unique(loaded_data)

765 assert len(unique_vals) <= 2

766

767 def test_load_compressed_vs_uncompressed(self, temp_workspace):

768 """Test loading compressed vs uncompressed files"""

769 data = np.random.rand(10, 10, 10).astype(np.float32)

770 img = nib.Nifti1Image(data, np.eye(4))

771

772 # Save compressed version

773 compressed_path = os.path.join(temp_workspace, "compressed.nii.gz")

774 nib.save(img, compressed_path)

775

776 # Save uncompressed version

777 uncompressed_path = os.path.join(temp_workspace, "uncompressed.nii")

778 nib.save(img, uncompressed_path)

779

780 # Load both

781 results_compressed = []

782 thread_compressed = ImageLoadThread(compressed_path, False)

783 thread_compressed.finished.connect(

784 lambda img_data, dims, aff, is_4d, is_overlay:

785 results_compressed.append(img_data)

786 )

787 thread_compressed.run()

788

789 results_uncompressed = []

790 thread_uncompressed = ImageLoadThread(uncompressed_path, False)

791 thread_uncompressed.finished.connect(

792 lambda img_data, dims, aff, is_4d, is_overlay:

793 results_uncompressed.append(img_data)

794 )

795 thread_uncompressed.run()

796

797 # They should produce equivalent results

798 assert len(results_compressed) == 1

799 assert len(results_uncompressed) == 1

800

801 np.testing.assert_array_almost_equal(

802 results_compressed[0],

803 results_uncompressed[0],

804 decimal=5

805 )

806

807 def test_save_with_special_characters_in_path(self, temp_workspace):

808 """Test saving with special characters in the path"""

809 data = np.ones((5, 5, 5), dtype=np.uint8)

810

811 # Filename with spaces and special characters

812 nifti_path = os.path.join(temp_workspace, "test file (copy) #1.nii")

813 json_path = os.path.join(temp_workspace, "test file (copy) #1.json")

814

815 source_dict = {"radius": 1.0, "difference": 0.1}

816

817 thread = SaveNiftiThread(

818 data, np.eye(4), nifti_path, json_path,

819 "relative/path.nii", source_dict

820 )

821

822 success = []

823 thread.success.connect(lambda p, j: success.append(True))

824 thread.run()

825

826 assert len(success) == 1

827 assert os.path.exists(nifti_path)

828

829

830class TestParameterValidation:

831 """Tests for parameter validation"""

832

833 def test_save_with_extreme_radius_values(self, temp_workspace):

834 """Test with extreme values for radius"""

835 data = np.ones((5, 5, 5), dtype=np.uint8)

836

837 extreme_radii = [0.0, 100.0, 1e6, -5.0]

838

839 for radius in extreme_radii:

840 nifti_path = os.path.join(temp_workspace, f"radius_{radius}.nii")

841 json_path = os.path.join(temp_workspace, f"radius_{radius}.json")

842

843 difference = 0.1

844 source_dict = {"radius": radius, "difference": difference}

845

846 thread = SaveNiftiThread(

847 data, np.eye(4), nifti_path, json_path,

848 "path.nii", source_dict

849 )

850

851 success = []

852 thread.success.connect(lambda p, j: success.append(True))

853 thread.run()

854

855 # Should still save even with extreme values

856 assert len(success) == 1

857

858 # Verify JSON contains the value

859 with open(json_path, 'r') as f:

860 metadata = json.load(f)

861 assert metadata["Origin"]["radius"] == radius

862

863 def test_save_with_extreme_difference_values(self, temp_workspace):

864 """Test with extreme values for difference"""

865 data = np.ones((5, 5, 5), dtype=np.uint8)

866

867 extreme_diffs = [0.0, 1.0, 10.0, -1.0]

868

869 for diff in extreme_diffs:

870 nifti_path = os.path.join(temp_workspace, f"diff_{diff}.nii")

871 json_path = os.path.join(temp_workspace, f"diff_{diff}.json")

872

873 source_dict = {"radius": 1.0, "difference": diff}

874

875 thread = SaveNiftiThread(

876 data, np.eye(4), nifti_path, json_path,

877 "path.nii", source_dict

878 )

879

880 success = []

881 thread.success.connect(lambda p, j: success.append(True))

882 thread.run()

883

884 assert len(success) == 1

885

886 with open(json_path, 'r') as f:

887 metadata = json.load(f)

888 assert metadata["Origin"]["difference"] == diff

889

890

891class TestPerformanceAndScalability:

892 """Tests for performance and scalability"""

893

894 def test_large_4d_dataset(self, temp_workspace):

895 """Test with a large 4D dataset"""

896 # Simulated 4D dataset (not too large for CI)

897 data_4d = np.random.rand(40, 40, 40, 20).astype(np.float32)

898 img = nib.Nifti1Image(data_4d, np.eye(4))

899

900 nifti_path = os.path.join(temp_workspace, "large_4d.nii.gz")

901 nib.save(img, nifti_path)

902

903 thread = ImageLoadThread(nifti_path, False)

904

905 results = []

906 progress_updates = []

907

908 thread.finished.connect(

909 lambda img_data, dims, aff, is_4d, is_overlay:

910 results.append(dims)

911 )

912 thread.progress.connect(lambda val: progress_updates.append(val))

913

914 thread.run()

915

916 assert len(results) == 1

917 assert results[0] == (40, 40, 40, 20)

918

919 # Progress should be emitted

920 assert len(progress_updates) > 0

921 assert 100 in progress_updates

922

923 def test_progress_tracking_accuracy(self, temp_workspace):

924 """Test accuracy of progress tracking"""

925 data = np.random.rand(25, 25, 25).astype(np.float32)

926 img = nib.Nifti1Image(data, np.eye(4))

927

928 nifti_path = os.path.join(temp_workspace, "progress_track.nii")

929 nib.save(img, nifti_path)

930

931 thread = ImageLoadThread(nifti_path, False)

932

933 progress_values = []

934 thread.progress.connect(lambda val: progress_values.append(val))

935

936 thread.run()

937

938 # Verify progress sequence

939 expected_milestones = [10, 30, 50, 70, 80, 100]

940 for milestone in expected_milestones:

941 assert milestone in progress_values

942

943 # Verify increasing order

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

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

946

947

948class TestNiftiThreadParameterized:

949 """Parameterized tests and normalization tests for NIfTI threads"""

950

951 @pytest.mark.parametrize("shape", [

952 (5, 5, 5),

953 (10, 20, 15),

954 (3, 3, 3),

955 (50, 50, 50)

956 ])

957 def test_save_various_shapes(self, shape, temp_workspace):

958 """Parameterized test for different volume shapes"""

959 data = np.random.randint(0, 255, shape, dtype=np.uint8)

960 nifti_path = os.path.join(temp_workspace, f"shape_{'x'.join(map(str, shape))}.nii")

961 json_path = nifti_path.replace('.nii', '.json')

962

963 source_dict = {"radius": 1.0, "difference": 0.1}

964

965 thread = SaveNiftiThread(

966 data, np.eye(4), nifti_path, json_path,

967 "path.nii", source_dict

968 )

969

970 success = []

971 thread.success.connect(lambda p, j: success.append(True))

972 thread.run()

973

974 assert len(success) == 1

975

976 # Verify loading

977 loaded = nib.load(nifti_path)

978 assert loaded.shape == shape

979

980 @pytest.mark.parametrize("is_overlay", [True, False])

981 def test_load_overlay_flag(self, is_overlay, temp_workspace):

982 """Parameterized test for the overlay flag"""

983 data = np.random.rand(8, 8, 8).astype(np.float32)

984 img = nib.Nifti1Image(data, np.eye(4))

985

986 nifti_path = os.path.join(temp_workspace, f"overlay_{is_overlay}.nii")

987 nib.save(img, nifti_path)

988

989 thread = ImageLoadThread(nifti_path, is_overlay=is_overlay)

990

991 results = []

992 thread.finished.connect(

993 lambda img_data, dims, aff, is_4d, is_ov:

994 results.append(is_ov)

995 )

996

997 thread.run()

998

999 assert len(results) == 1

1000 assert results[0] == is_overlay

1001

1002 def test_load_4d_normalization_per_volume(self, temp_workspace):

1003 """Test independent normalization for each 4D volume"""

1004 # Create 4D with different intensities per volume (add variability to avoid constant volumes)

1005 volume1 = np.ones((10, 10, 10)) * 100

1006 volume1[0, 0, 0] = 101 # Add variability

1007 volume2 = np.ones((10, 10, 10)) * 200

1008 volume2[0, 0, 0] = 201

1009 volume3 = np.ones((10, 10, 10)) * 50

1010 volume3[0, 0, 0] = 51

1011

1012 data_4d = np.stack([volume1, volume2, volume3], axis=-1)

1013 img = nib.Nifti1Image(data_4d, np.eye(4))

1014

1015 nifti_path = os.path.join(temp_workspace, "4d_norm.nii")

1016 nib.save(img, nifti_path)

1017

1018 thread = ImageLoadThread(nifti_path, False)

1019

1020 results = []

1021 thread.finished.connect(

1022 lambda img_data, dims, aff, is_4d, is_overlay:

1023 results.append(img_data)

1024 )

1025

1026 thread.run()

1027

1028 normalized_data = results[0]

1029

1030 # Each volume should be normalized independently

1031 for i in range(3):

1032 vol = normalized_data[..., i]

1033 assert vol.min() >= 0

1034 assert vol.max() <= 1

1035 assert vol.max() - vol.min() > 0

1036

1037

1038class TestImageLoadThreadNormalization:

1039 """Tests for percentile-based normalization"""

1040

1041 def test_normalize_data_matplotlib_style(self, temp_workspace):

1042 """Test normalization using percentiles"""

1043 # Create data with outliers

1044 data = np.random.randn(30, 30, 30).astype(np.float32) * 100

1045 data[0, 0, 0] = 10000 # High outlier

1046 data[1, 1, 1] = -5000 # Low outlier

1047

1048 img = nib.Nifti1Image(data, np.eye(4))

1049 nifti_path = os.path.join(temp_workspace, "outliers.nii")

1050 nib.save(img, nifti_path)

1051

1052 thread = ImageLoadThread(nifti_path, False)

1053

1054 results = []

1055 thread.finished.connect(

1056 lambda img_data, dims, aff, is_4d, is_overlay:

1057 results.append(img_data)

1058 )

1059

1060 thread.run()

1061

1062 normalized = results[0]

1063

1064 # Verify that outliers do not dominate normalization

1065 # Most values should be reasonably distributed

1066 assert normalized.min() >= 0

1067 assert normalized.max() <= 1

1068

1069 # Intermediate percentiles should be well distributed

1070 p25 = np.percentile(normalized, 25)

1071 p75 = np.percentile(normalized, 75)

1072 assert 0.1 < p25 < 0.9

1073 assert 0.1 < p75 < 0.9

1074

1075 def test_normalize_uniform_data(self, temp_workspace):

1076 """Test normalization with uniform data"""

1077 # All voxels have the same value

1078 data = np.ones((10, 10, 10)) * 42.0

1079 img = nib.Nifti1Image(data, np.eye(4))

1080

1081 nifti_path = os.path.join(temp_workspace, "uniform.nii")

1082 nib.save(img, nifti_path)

1083

1084 thread = ImageLoadThread(nifti_path, False)

1085

1086 results = []

1087 thread.finished.connect(

1088 lambda img_data, dims, aff, is_4d, is_overlay:

1089 results.append(img_data)

1090 )

1091

1092 thread.run()

1093

1094 normalized = results[0]

1095

1096 # With uniform data, normalization should handle it gracefully

1097 assert not np.any(np.isnan(normalized))

1098 assert not np.any(np.isinf(normalized))

1099 assert normalized.min() >= 0

1100 assert normalized.max() <= 1

1101

1102 def test_normalize_with_nan_values(self, temp_workspace):

1103 """Test normalization with NaN values"""

1104 data = np.random.rand(10, 10, 10).astype(np.float32)

1105 data[5, 5, 5] = np.nan

1106 data[3, 3, 3] = np.inf

1107

1108 img = nib.Nifti1Image(data, np.eye(4))

1109 nifti_path = os.path.join(temp_workspace, "with_nan.nii")

1110 nib.save(img, nifti_path)

1111

1112 thread = ImageLoadThread(nifti_path, False)

1113

1114 results = []

1115 thread.finished.connect(

1116 lambda img_data, dims, aff, is_4d, is_overlay:

1117 results.append(img_data)

1118 )

1119

1120 thread.run()

1121

1122 normalized = results[0]

1123

1124 # Should handle NaN/Inf without propagating them

1125 finite_count = np.isfinite(normalized).sum()

1126 assert finite_count > 0 # At least some finite values

1127

1128 def test_normalize_empty_volume(self):

1129 """Test normalization of an empty volume"""

1130 thread = ImageLoadThread("dummy.nii", False)

1131

1132 # Direct test of the method

1133 empty_data = np.array([])

1134 result = thread.normalize_data_matplotlib_style(empty_data)

1135

1136 assert result.size == 0

1137

1138 def test_normalize_volume_all_invalid(self):

1139 """Test normalization with all invalid values"""

1140 thread = ImageLoadThread("dummy.nii", False)

1141

1142 # Volume with only NaN values

1143 nan_volume = np.full((5, 5, 5), np.nan)

1144 result = thread.normalize_data_matplotlib_style(nan_volume)

1145

1146 # Should return zeros

1147 assert result.shape == nan_volume.shape

1148 assert np.all(result == 0)

1149

1150

1151class TestImageLoadThreadCanonicalOrientation:

1152 """Tests for conversion to canonical RAS+ orientation"""

1153

1154 def test_canonical_orientation_applied(self, temp_workspace):

1155 """Test that the image is converted to RAS+ orientation"""

1156 # Create an image with non-canonical orientation

1157 data = np.random.rand(10, 12, 8).astype(np.float32)

1158

1159 # Affine with a different orientation (e.g., LAS)

1160 affine = np.array([

1161 [-2, 0, 0, 10],

1162 [0, 2, 0, 20],

1163 [0, 0, 2, 30],

1164 [0, 0, 0, 1]

1165 ])

1166

1167 img = nib.Nifti1Image(data, affine)

1168 nifti_path = os.path.join(temp_workspace, "non_canonical.nii")

1169 nib.save(img, nifti_path)

1170

1171 thread = ImageLoadThread(nifti_path, False)

1172

1173 results = []

1174 thread.finished.connect(

1175 lambda img_data, dims, aff, is_4d, is_overlay:

1176 results.append((img_data, aff))

1177 )

1178

1179 thread.run()

1180

1181 img_data, loaded_affine = results[0]

1182

1183 # The image should now be in canonical orientation

1184 assert img_data.shape[0] > 0

1185 assert img_data.shape[1] > 0

1186 assert img_data.shape[2] > 0

1187

1188 # Affine should be updated

1189 assert isinstance(loaded_affine, np.ndarray)

1190 assert loaded_affine.shape == (4, 4)

1191

1192

1193class TestImageLoadThreadErrorHandling:

1194 """Tests for error handling"""

1195

1196 def test_error_file_not_found(self, temp_workspace):

1197 """Test error when file does not exist"""

1198 nonexistent_path = os.path.join(temp_workspace, "does_not_exist.nii")

1199 thread = ImageLoadThread(nonexistent_path, False)

1200

1201 error_msgs = []

1202 thread.error.connect(lambda msg: error_msgs.append(msg))

1203

1204 thread.run()

1205

1206 assert len(error_msgs) == 1

1207 assert len(error_msgs[0]) > 0

1208

1209 def test_error_invalid_nifti_file(self, temp_workspace):

1210 """Test error with a non-NIfTI file"""

1211 # Create a text file instead of a NIfTI

1212 invalid_path = os.path.join(temp_workspace, "invalid.nii")

1213 with open(invalid_path, 'w') as f:

1214 f.write("This is not a NIfTI file")

1215

1216 thread = ImageLoadThread(invalid_path, False)

1217

1218 error_msgs = []

1219 thread.error.connect(lambda msg: error_msgs.append(msg))

1220

1221 thread.run()

1222

1223 assert len(error_msgs) == 1

1224

1225 def test_error_corrupted_nifti(self, temp_workspace):

1226 """Test error with a corrupted NIfTI file"""

1227 corrupted_path = os.path.join(temp_workspace, "corrupted.nii.gz")

1228

1229 # Create a partially valid but corrupted file

1230 with open(corrupted_path, 'wb') as f:

1231 f.write(b'corrupted data that looks like gzip')

1232

1233 thread = ImageLoadThread(corrupted_path, False)

1234

1235 error_msgs = []

1236 thread.error.connect(lambda msg: error_msgs.append(msg))

1237

1238 thread.run()

1239

1240 assert len(error_msgs) == 1

1241

1242 @patch('nibabel.load')

1243 def test_error_during_load(self, mock_load, temp_workspace):

1244 """Test generic load error handling"""

1245 mock_load.side_effect = RuntimeError("Memory error")

1246

1247 nifti_path = os.path.join(temp_workspace, "test.nii")

1248 thread = ImageLoadThread(nifti_path, False)

1249

1250 error_msgs = []

1251 thread.error.connect(lambda msg: error_msgs.append(msg))

1252

1253 thread.run()

1254

1255 assert len(error_msgs) == 1

1256 assert "Memory error" in error_msgs[0]

1257

1258

1259class TestImageLoadThreadNifti2Support:

1260 """Tests for NIfTI-2 support"""

1261

1262 def test_nifti2_image_loads(self, temp_workspace):

1263 """Test loading a NIfTI-2 image"""

1264 data = np.random.rand(8, 8, 8).astype(np.float32)

1265 affine = np.eye(4)

1266 img = nib.Nifti2Image(data, affine)

1267

1268 nifti_path = os.path.join(temp_workspace, "test_nifti2.nii")

1269 nib.save(img, nifti_path)

1270

1271 thread = ImageLoadThread(nifti_path, False)

1272

1273 results = []

1274 thread.finished.connect(

1275 lambda img_data, dims, aff, is_4d, is_overlay:

1276 results.append(img_data)

1277 )

1278

1279 thread.run()

1280

1281 assert len(results) == 1

1282 assert results[0].shape == (8, 8, 8)

1283

1284

1285class TestImageLoadThreadMemoryMapping:

1286 """Tests for memory mapping"""

1287

1288 @patch('nibabel.load')

1289 def test_memory_mapping_used(self, mock_load, temp_workspace):

1290 """Test that memory mapping is used"""

1291 # Mock to check mmap argument

1292 mock_img = Mock(spec=nib.Nifti1Image)

1293 mock_img.header.get_data_shape.return_value = (10, 10, 10)

1294 mock_img.affine = np.eye(4)

1295 mock_img.dataobj = np.zeros((10, 10, 10))

1296

1297 mock_canonical = Mock()

1298 mock_canonical.header.get_data_shape.return_value = (10, 10, 10)

1299 mock_canonical.affine = np.eye(4)

1300 mock_canonical.dataobj = np.zeros((10, 10, 10))

1301

1302 with patch('nibabel.as_closest_canonical', return_value=mock_canonical):

1303 mock_load.return_value = mock_img

1304

1305 thread = ImageLoadThread("dummy.nii", False)

1306 thread.run()

1307

1308 # Verify mmap was used

1309 mock_load.assert_called_once()

1310 call_args = mock_load.call_args

1311 assert call_args[1].get('mmap') == 'c'

1312

1313

1314class TestEdgeCasesAndIntegration:

1315 """Tests for edge cases and integration scenarios"""

1316

1317 def test_very_small_image(self, temp_workspace):

1318 """Test with a very small image (1×1×1)"""

1319 data = np.array([[[42.0]]])

1320 img = nib.Nifti1Image(data, np.eye(4))

1321

1322 nifti_path = os.path.join(temp_workspace, "tiny.nii")

1323 nib.save(img, nifti_path)

1324

1325 thread = ImageLoadThread(nifti_path, False)

1326

1327 results = []

1328 thread.finished.connect(

1329 lambda img_data, dims, aff, is_4d, is_overlay:

1330 results.append(img_data)

1331 )

1332

1333 thread.run()

1334

1335 assert len(results) == 1

1336 assert results[0].shape == (1, 1, 1)

1337 assert np.isclose(results[0][0, 0, 0], 0.0) or np.isclose(results[0][0, 0, 0], 1.0)

1338

1339 def test_large_dimensions(self, temp_workspace):

1340 """Test with a large image"""

1341 # Simulate a large but manageable volume

1342 data = np.zeros((100, 100, 100), dtype=np.float32)

1343 data[50, 50, 50] = 100 # add high-intensity point

1344

1345 img = nib.Nifti1Image(data, np.eye(4))

1346 nifti_path = os.path.join(temp_workspace, "large.nii.gz")

1347 nib.save(img, nifti_path)

1348

1349 thread = ImageLoadThread(nifti_path, False)

1350

1351 results = []

1352 thread.finished.connect(

1353 lambda img_data, dims, aff, is_4d, is_overlay:

1354 results.append((img_data, dims, is_4d))

1355 )

1356

1357 thread.run()

1358

1359 assert len(results) == 1

1360 img_data, dims, is_4d = results[0]

1361

1362 assert dims == (100, 100, 100)

1363 assert not is_4d

1364 assert img_data.shape == (100, 100, 100)

1365 assert np.all((img_data >= 0) & (img_data <= 1)), "Normalized values must be between 0 and 1"

1366 assert np.isclose(img_data[50, 50, 50], 1.0, atol=1e-3)

1367

1368 def test_invalid_file_emits_error(self, temp_workspace):

1369 """Test that a non-NIfTI file emits an error signal"""

1370 invalid_path = os.path.join(temp_workspace, "not_nifti.txt")

1371 with open(invalid_path, "w") as f:

1372 f.write("this is not an image")

1373

1374 thread = ImageLoadThread(invalid_path, False)

1375

1376 errors = []

1377 thread.error.connect(lambda msg: errors.append(msg))

1378

1379 thread.run()

1380

1381 assert len(errors) == 1

1382 assert any(keyword in errors[0] for keyword in [

1383 "Not a valid NIfTI",

1384 "No such file",

1385 "Cannot work out file type"

1386 ])

1387

1388 def test_save_nifti_thread_success(self, temp_workspace):

1389 """Integration test: NIfTI + JSON save"""

1390 data = np.ones((5, 5, 5), dtype=np.uint8)

1391 affine = np.eye(4)

1392 nifti_path = os.path.join(temp_workspace, "roi.nii.gz")

1393 json_path = os.path.join(temp_workspace, "roi.json")

1394

1395 radius = 2.0

1396 difference = 0.5

1397 source_dict = {"radius": radius, "difference": difference}

1398

1399 thread = SaveNiftiThread(

1400 data=data,

1401 affine=affine,

1402 path=nifti_path,

1403 json_path=json_path,

1404 relative_path="sub-01/anat/T1w.nii.gz",

1405 source_dict=source_dict

1406 )

1407

1408 results = []

1409 thread.success.connect(lambda npath, jpath: results.append((npath, jpath)))

1410

1411 thread.run()

1412

1413 # Check that files were created

1414 assert os.path.exists(nifti_path)

1415 assert os.path.exists(json_path)

1416

1417 assert len(results) == 1

1418 npath, jpath = results[0]

1419 assert npath.endswith(".nii.gz")

1420 assert jpath.endswith(".json")

1421

1422 # Check JSON contains correct fields

1423 import json

1424 with open(json_path) as f:

1425 meta = json.load(f)

1426

1427 assert "Origin" in meta

1428 assert meta["Origin"]["radius"] == 2.0

1429

1430 def test_save_nifti_thread_error(self, tmp_path):

1431 """Error test: unwritable path"""

1432 bad_path = "/invalid/path/roi.nii.gz"

1433 json_path = os.path.join(tmp_path, "roi.json")

1434

1435 radius = 1.0

1436 difference = 0.2

1437 source_dict = {"radius": radius, "difference": difference}

1438

1439 thread = SaveNiftiThread(

1440 data=np.ones((2, 2, 2)),

1441 affine=np.eye(4),

1442 path=bad_path,

1443 json_path=json_path,

1444 relative_path="sub-01/anat/T1w.nii.gz",

1445 source_dict=source_dict

1446 )

1447

1448 errors = []

1449 thread.error.connect(lambda msg: errors.append(msg))

1450

1451 thread.run()

1452

1453 assert len(errors) == 1

1454 assert isinstance(errors[0], str)

1455

1456

1457if __name__ == "__main__":

1458 pytest.main([__file__, "-v"])

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

757 statements