Coverage for tests/threads/test_import

1"""

2test_import_thread.py - Test Suite for ImportThread

4This suite tests all the main functionalities of ImportThread:

5- Importing already organized BIDS folders

6- DICOM → NIfTI conversion

7- Handling single vs. multiple patients

8- Structure detection heuristics

9- Import cancellation

10- Error handling

11"""

13import os

14from types import SimpleNamespace

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

16import pytest

17from PyQt6.QtCore import QCoreApplication

19from main.threads.import_thread import ImportThread

22class TestImportThreadInitialization:

23 """Tests for ImportThread initialization"""

25 def test_init_single_path(self, mock_context, temp_workspace):

26 """Test initialization with a single path"""

27 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

29 assert thread.context == mock_context

30 assert thread.folders_path == [temp_workspace]

31 assert thread.workspace_path == temp_workspace

32 assert thread.current_progress == 0

33 assert thread._is_canceled is False

34 assert thread.process is None

36 def test_init_multiple_paths(self, mock_context, temp_workspace):

37 """Test initialization with multiple paths"""

38 paths = [temp_workspace, os.path.join(temp_workspace, "sub-01")]

39 thread = ImportThread(mock_context, paths, temp_workspace)

41 assert len(thread.folders_path) == 2

42 assert thread.folders_path == paths

44 def test_signals_exist(self, mock_context, temp_workspace):

45 """Verify that signals are defined correctly"""

46 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

48 assert hasattr(thread, 'finished')

49 assert hasattr(thread, 'error')

50 assert hasattr(thread, 'progress')

53class TestFileDetection:

54 """Tests for file detection methods"""

56 def test_is_nifti_file(self, mock_context, temp_workspace):

57 """Test NIfTI file detection"""

58 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

60 assert thread._is_nifti_file("brain.nii") is True

61 assert thread._is_nifti_file("brain.nii.gz") is True

62 assert thread._is_nifti_file("brain.json") is False

63 assert thread._is_nifti_file("brain.dcm") is False

65 def test_is_dicom_file_valid(self, mock_context, temp_workspace):

66 """Test valid DICOM detection"""

67 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

69 # Create a mock DICOM file

70 dicom_path = os.path.join(temp_workspace, "test.dcm")

71 with open(dicom_path, "wb") as f:

72 f.write(b'\x00' * 128) # 128 bytes of padding

73 f.write(b'DICM') # DICOM magic bytes

74 f.write(b'\x00' * 100)

76 assert thread._is_dicom_file(dicom_path) is True

78 def test_is_dicom_file_invalid(self, mock_context, temp_workspace):

79 """Test non-DICOM file detection"""

80 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

82 # File without DICOM magic bytes

83 non_dicom_path = os.path.join(temp_workspace, "notdicom.txt")

84 with open(non_dicom_path, "w") as f:

85 f.write("This is not a DICOM file")

87 assert thread._is_dicom_file(non_dicom_path) is False

89 def test_is_dicom_file_exception(self, mock_context, temp_workspace):

90 """Test exception handling during DICOM read"""

91 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

93 # Non-existent file

94 assert thread._is_dicom_file("/nonexistent/file.dcm") is False

97class TestBIDSDetection:

98 """Tests for BIDS structure detection"""

100 def test_is_bids_folder_valid(self, mock_context, temp_workspace):

101 """Test valid BIDS folder detection"""

102 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

103

104 # Create BIDS structure

105 bids_dir = os.path.join(temp_workspace, "bids_test")

106 # **EDIT**: The folder name passed to the method must start with "sub-"

107 # according to the provided code logic.

108 sub_folder = os.path.join(bids_dir, "sub-01")

109 anat_dir = os.path.join(sub_folder, "anat")

110 os.makedirs(anat_dir, exist_ok=True)

111

112 # Add NIfTI file

113 nifti_path = os.path.join(anat_dir, "sub-01_T1w.nii.gz")

114 with open(nifti_path, "w") as f:

115 f.write("nifti data")

116

117 # **FIX**: The assertion must test 'sub_folder', not 'bids_dir'

118 assert thread._is_bids_folder(sub_folder) is True

119

120 def test_is_bids_folder_invalid(self, mock_context, temp_workspace):

121 """Test detection of non-BIDS folder (wrong name)"""

122 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

123 invalid_dir = os.path.join(temp_workspace, "not_bids")

124 os.makedirs(invalid_dir, exist_ok=True)

125 assert thread._is_bids_folder(invalid_dir) is False

126

127 def test_is_bids_folder_no_sub_prefix(self, mock_context, temp_workspace):

128 """Test folder without 'sub-' prefix"""

129 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

130

131 # Folder with similar structure but without the correct prefix

132 fake_bids = os.path.join(temp_workspace, "fake_bids")

133 anat_dir = os.path.join(fake_bids, "patient01", "anat")

134 os.makedirs(anat_dir, exist_ok=True)

135

136 with open(os.path.join(anat_dir, "scan.nii"), "w") as f:

137 f.write("data")

138

139 assert thread._is_bids_folder(fake_bids) is False

140

141

142class TestSubjectIDGeneration:

143 """Tests for subject ID generation"""

144

145 def test_get_next_sub_id_empty_workspace(self, mock_context, temp_workspace):

146 """Test ID generation on an empty workspace"""

147 empty_ws = os.path.join(temp_workspace, "empty")

148 os.makedirs(empty_ws)

149

150 thread = ImportThread(mock_context, [temp_workspace], empty_ws)

151 sub_id = thread._get_next_sub_id()

152

153 assert sub_id == "sub-01"

154

155 def test_get_next_sub_id_existing_subjects(self, mock_context, temp_workspace):

156 """Test ID generation with existing subjects"""

157 # Add exist_ok=True

158 os.makedirs(os.path.join(temp_workspace, "sub-01"), exist_ok=True)

159 os.makedirs(os.path.join(temp_workspace, "sub-02"), exist_ok=True)

160

161 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

162 sub_id = thread._get_next_sub_id()

163

164 assert sub_id == "sub-03"

165

166 def test_get_next_sub_id_non_sequential(self, mock_context, temp_workspace):

167 """Test with non-sequential IDs"""

168 # Add exist_ok=True

169 os.makedirs(os.path.join(temp_workspace, "sub-01"), exist_ok=True)

170 os.makedirs(os.path.join(temp_workspace, "sub-05")) # Skip

171

172 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

173 sub_id = thread._get_next_sub_id()

174

175 # Should be sub-06 (max + 1)

176 assert sub_id == "sub-06"

177

178 def test_get_next_sub_id_invalid_names(self, mock_context, temp_workspace):

179 """Test with invalid folder names"""

180 # Add exist_ok=True

181 os.makedirs(os.path.join(temp_workspace, "sub-01"), exist_ok=True)

182 os.makedirs(os.path.join(temp_workspace, "sub-02"), exist_ok=True)

183 os.makedirs(os.path.join(temp_workspace, "sub-invalid"))

184 os.makedirs(os.path.join(temp_workspace, "not-a-sub"))

185

186 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

187 sub_id = thread._get_next_sub_id()

188

189 # Should ignore invalid folders and use sub-01, sub-02

190 assert sub_id == "sub-03"

191

192

193class TestPatientDetectionHeuristics:

194 """Tests for single/multiple patient detection heuristics"""

195

196 def test_subfolders_look_like_different_patients(self, mock_context, temp_workspace):

197 """Test multi-patient folder detection"""

198 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

199

200 # Create folders that look like different patients

201 patient_folders = [

202 os.path.join(temp_workspace, "sub-01"),

203 os.path.join(temp_workspace, "sub-02"),

204 os.path.join(temp_workspace, "patient_03")

205 ]

206

207 for folder in patient_folders:

208 os.makedirs(folder, exist_ok=True)

209

210 assert thread._subfolders_look_like_different_patients(patient_folders) is True

211

212 def test_subfolders_single_patient(self, mock_context, temp_workspace):

213 """Test folders belonging to a single patient"""

214 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

215

216 # Folders without multi-patient pattern

217 series_folders = [

218 os.path.join(temp_workspace, "series_001"),

219 os.path.join(temp_workspace, "series_002"),

220 os.path.join(temp_workspace, "t1_weighted")

221 ]

222

223 for folder in series_folders:

224 os.makedirs(folder, exist_ok=True)

225

226 assert thread._subfolders_look_like_different_patients(series_folders) is False

227

228 @patch('main.threads.import_thread.pydicom.dcmread')

229 def test_are_dicom_series_of_same_patient_true(self, mock_dcmread, mock_context, temp_workspace):

230 """Test DICOM files from the same patient"""

231 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

232

233 # Setup mock DICOM with the same PatientID

234 mock_dcm = Mock()

235 mock_dcm.PatientID = "PATIENT001"

236 mock_dcm.PatientName = "Doe^John"

237 mock_dcmread.return_value = mock_dcm

238

239 # Create folders with DICOM

240 series_folders = []

241 for i in range(3):

242 folder = os.path.join(temp_workspace, f"series_{i}")

243 os.makedirs(folder, exist_ok=True)

244

245 # Create mock DICOM file

246 dicom_file = os.path.join(folder, f"image_{i}.dcm")

247 with open(dicom_file, "wb") as f:

248 f.write(b'\x00' * 128)

249 f.write(b'DICM')

250

251 series_folders.append(folder)

252

253 result = thread._are_dicom_series_of_same_patient(series_folders)

254 assert result is True

255

256 @patch('main.threads.import_thread.pydicom.dcmread')

257 def test_are_dicom_series_of_same_patient_false(self, mock_dcmread, mock_context, temp_workspace):

258 """Test DICOM files from different patients"""

259 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

260

261 # All files are recognized as DICOM

262 thread._is_dicom_file = Mock(return_value=True)

263

264 # Each call to dcmread returns a different patient

265 mock_dcmread.side_effect = [

266 SimpleNamespace(PatientID="PATIENT01"),

267 SimpleNamespace(PatientID="PATIENT02"),

268 SimpleNamespace(PatientID="PATIENT03"),

269 ]

270

271 # Create fake DICOM folders and files

272 series_folders = []

273 for i in range(3):

274 folder = os.path.join(temp_workspace, f"patient_{i}")

275 os.makedirs(folder, exist_ok=True)

276 dicom_file = os.path.join(folder, "scan.dcm")

277 with open(dicom_file, "wb") as f:

278 f.write(b'\x00' * 128)

279 f.write(b'DICM')

280 series_folders.append(folder)

281

282 result = thread._are_dicom_series_of_same_patient(series_folders)

283

284 assert result is False

285

286 def test_are_dicom_series_no_dicom_files(self, mock_context, temp_workspace):

287 """Test with folders containing no DICOM files"""

288 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

289

290 # Empty folders or with other files

291 folders = []

292 for i in range(3):

293 folder = os.path.join(temp_workspace, f"empty_{i}")

294 os.makedirs(folder, exist_ok=True)

295 folders.append(folder)

296

297 result = thread._are_dicom_series_of_same_patient(folders)

298 assert result is False

299

300

301class TestCancellation:

302 """Tests for import cancellation"""

303

304 def test_cancel_flag_set(self, mock_context, temp_workspace):

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

306 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

307

308 assert thread._is_canceled is False

309 thread.cancel()

310 assert thread._is_canceled is True

311

312 def test_cancel_terminates_process(self, mock_context, temp_workspace):

313 """Test termination of external process during cancellation"""

314 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

315

316 # Mock an external running process

317 mock_process = Mock()

318 thread.process = mock_process

319

320 thread.cancel()

321

322 mock_process.terminate.assert_called_once()

323 mock_process.wait.assert_called_once()

324 mock_process.kill.assert_called_once()

325

326 def test_cancel_no_process(self, mock_context, temp_workspace):

327 """Test cancellation with no active process"""

328 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

329

330 # Should not raise exceptions

331 thread.cancel()

332 assert thread._is_canceled is True

333

334

335class TestBIDSImport:

336 """Tests for BIDS folder import"""

337

338 @patch.object(ImportThread, '_is_bids_folder')

339 @patch.object(ImportThread, '_get_next_sub_id')

340 def test_import_bids_folder_directly(self, mock_get_id, mock_is_bids,

341 mock_context, temp_workspace):

342 """Test direct copy of BIDS folder"""

343 # Setup

344 mock_is_bids.return_value = True

345 mock_get_id.return_value = "sub-003"

346

347 # Create source BIDS folder

348 src_bids = os.path.join(temp_workspace, "source_bids")

349 anat_dir = os.path.join(src_bids, "sub-01", "anat")

350 os.makedirs(anat_dir)

351

352 nifti_file = os.path.join(anat_dir, "T1w.nii.gz")

353 with open(nifti_file, "w") as f:

354 f.write("test nifti data")

355

356 # Create destination workspace

357 dest_ws = os.path.join(temp_workspace, "dest_workspace")

358 os.makedirs(dest_ws)

359

360 thread = ImportThread(mock_context, [src_bids], dest_ws)

361

362 # Connect signal to verify emission

363 finished_called = [False]

364

365 def on_finished():

366 finished_called[0] = True

367

368 thread.finished.connect(on_finished)

369

370 thread.run()

371

372 assert finished_called[0] is True

373 dest_folder = os.path.join(dest_ws, "sub-003")

374 assert os.path.exists(dest_folder)

375

376

377class TestDICOMConversion:

378 """Tests for DICOM → NIfTI conversion"""

379

380 @patch('main.threads.import_thread.subprocess.Popen')

381 @patch('main.threads.import_thread.get_bin_path')

382 def test_convert_dicom_folder_success(self, mock_get_bin, mock_Popen,

383 mock_context, temp_workspace):

384 """Test successful DICOM conversion"""

385 mock_get_bin.return_value = "/fake/path/dcm2niix"

386

387 # Configure the Popen mock to simulate a successful process

388 mock_process = Mock()

389 mock_process.communicate.return_value = (b'Conversion complete', b'') # stdout, stderr (bytes)

390 mock_process.returncode = 0

391 mock_Popen.return_value = mock_process # Popen() returns our mock

392

393 src_folder = os.path.join(temp_workspace, "dicom_source")

394 dest_folder = os.path.join(temp_workspace, "nifti_dest")

395 os.makedirs(src_folder)

396 # os.makedirs(dest_folder) # The function itself creates this folder

397

398 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

399

400 # Run the function

401 thread._convert_dicom_folder_to_nifti(src_folder, dest_folder)

402

403 # Verify that Popen was called correctly

404 mock_Popen.assert_called_once()

405 mock_process.communicate.assert_called_once()

406 # Verify that no error was raised (the test would fail earlier if it were)

407

408 @patch('main.threads.import_thread.get_bin_path')

409 def test_convert_dicom_folder_missing_tool(self, mock_get_bin,

410 mock_context, temp_workspace):

411 """Test when dcm2niix is not available"""

412 # Simulate FileNotFoundError when searching for the binary

413 mock_get_bin.side_effect = FileNotFoundError("dcm2niix not found")

414

415 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

416

417 # **FIX**: The code now catches the exception and raises a RuntimeError.

418 # The old test expected nothing to happen.

419 with pytest.raises(RuntimeError, match="dcm2niix not found"):

420 thread._convert_dicom_folder_to_nifti(temp_workspace, temp_workspace)

421

422 @patch('main.threads.import_thread.subprocess.Popen')

423 @patch('main.threads.import_thread.get_bin_path')

424 def test_convert_dicom_folder_process_error(self, mock_get_bin, mock_Popen,

425 mock_context, temp_workspace):

426 """Test error during dcm2niix execution (returncode != 0)"""

427 mock_get_bin.return_value = "/fake/path/dcm2niix"

428

429 # Simulate a process that fails

430 mock_process = Mock()

431 mock_process.communicate.return_value = (b'', b'Errore fatale') # Message on stderr

432 mock_process.returncode = 1 # Error exit code

433 mock_Popen.return_value = mock_process

434

435 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

436

437 # **FIX**: The code detects the returncode != 0 and raises a RuntimeError.

438 with pytest.raises(RuntimeError, match="dcm2niix failed: Errore fatale"):

439 thread._convert_dicom_folder_to_nifti(temp_workspace, temp_workspace)

440

441

442class TestBIDSStructureConversion:

443 """Tests for BIDS structure conversion"""

444

445 @patch.object(ImportThread, '_get_next_sub_id', return_value="sub-01")

446 def test_convert_to_bids_structure(self, mock_get_id, mock_context, temp_workspace):

447 """Test folder conversion to BIDS structure"""

448 src_folder = os.path.join(temp_workspace, "source")

449 os.makedirs(src_folder)

450

451 # Create fictitious NIfTI and JSON with minimal metadata

452 nifti_file = os.path.join(src_folder, "brain.nii.gz")

453 json_file = os.path.join(src_folder, "brain.json")

454

455 with open(nifti_file, "w") as f: f.write("nifti data")

456 # The code requires 'Modality' metadata

457 with open(json_file, "w") as f: f.write('{"Modality": "MR", "ProtocolName": "T1w"}')

458

459 dest_ws = os.path.join(temp_workspace, "workspace")

460 os.makedirs(dest_ws)

461

462 thread = ImportThread(mock_context, [temp_workspace], dest_ws)

463 thread._convert_to_bids_structure(src_folder)

464

465 # Verify created structure (with correct ID)

466 expected_sub = os.path.join(dest_ws, "sub-01", "anat") # <- CORRECT

467 assert os.path.exists(expected_sub)

468 expected_file = os.path.join(expected_sub, "sub-01_run-1_T1w.nii.gz")

469 assert os.path.exists(expected_file)

470

471 @patch.object(ImportThread, '_get_next_sub_id', return_value="sub-01")

472 def test_convert_to_bids_structure_nested(self, mock_get_id, mock_context, temp_workspace):

473 """Test conversion with files in subfolders"""

474 src_folder = os.path.join(temp_workspace, "nested_source")

475 sub_dir = os.path.join(src_folder, "subdir")

476 os.makedirs(sub_dir)

477

478 # File in subfolder

479 nifti_file = os.path.join(sub_dir, "scan.nii.gz") # Added .gz for consistency

480 json_file = os.path.join(sub_dir, "scan.json")

481 with open(nifti_file, "w") as f: f.write("scan data")

482 with open(json_file, "w") as f: f.write('{"Modality": "PT", "Radiopharmaceutical": "FDG"}')

483

484 dest_ws = os.path.join(temp_workspace, "ws")

485 os.makedirs(dest_ws)

486

487 thread = ImportThread(mock_context, [temp_workspace], dest_ws)

488 thread._convert_to_bids_structure(src_folder)

489

490 # File should be copied to pet/ (according to PT logic)

491 expected_pet = os.path.join(dest_ws, "sub-01", "ses-01", "pet") # <- CORRECT

492 assert os.path.exists(expected_pet)

493 # The filename is changed

494 expected_file_path = os.path.join(expected_pet, "sub-01_task-unknown_run-1_pet.nii.gz")

495 assert os.path.exists(expected_file_path)

496

497

498class TestSinglePatientProcessing:

499 """Tests for single-patient folder processing"""

500

501 @patch.object(ImportThread, '_convert_dicom_folder_to_nifti')

502 @patch.object(ImportThread, '_convert_to_bids_structure')

503 def test_process_single_patient_folder_with_nifti(self, mock_convert_bids,

504 mock_convert_dicom,

505 mock_context, temp_workspace):

506 """Test processing folder with existing NIfTI files"""

507 # Patient folder with NIfTI

508 patient_folder = os.path.join(temp_workspace, "patient_data")

509 os.makedirs(patient_folder)

510

511 nifti_file = os.path.join(patient_folder, "t1.nii.gz")

512 with open(nifti_file, "w") as f:

513 f.write("nifti")

514

515 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

516 thread._process_single_patient_folder(patient_folder)

517

518 # Should convert to BIDS but not call dcm2niix

519 mock_convert_bids.assert_called_once()

520 mock_convert_dicom.assert_not_called()

521

522 @patch.object(ImportThread, '_convert_dicom_folder_to_nifti')

523 @patch.object(ImportThread, '_convert_to_bids_structure')

524 def test_process_single_patient_folder_with_dicom(self, mock_convert_bids,

525 mock_convert_dicom,

526 mock_context, temp_workspace):

527 """Test processing folder with DICOM files"""

528 patient_folder = os.path.join(temp_workspace, "patient_dicom")

529 os.makedirs(patient_folder)

530

531 # Create DICOM file

532 dicom_file = os.path.join(patient_folder, "image.dcm")

533 with open(dicom_file, "wb") as f:

534 f.write(b'\x00' * 128)

535 f.write(b'DICM')

536

537 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

538 thread._process_single_patient_folder(patient_folder)

539

540 # Both conversions should be called

541 mock_convert_dicom.assert_called_once()

542 mock_convert_bids.assert_called_once()

543

544 def test_process_single_patient_folder_cancelled(self, mock_context, temp_workspace):

545 """Test cancellation during processing"""

546 patient_folder = os.path.join(temp_workspace, "patient")

547 os.makedirs(patient_folder)

548

549 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

550 thread.cancel() # Cancel before starting

551

552 # Should not raise exceptions

553 thread._process_single_patient_folder(patient_folder)

554

555

556class TestErrorHandling:

557 """Tests for error handling"""

558

559 def test_run_invalid_folder_path(self, mock_context, temp_workspace):

560 """Test with invalid folder path"""

561 invalid_path = os.path.join(temp_workspace, "nonexistent")

562 thread = ImportThread(mock_context, [invalid_path], temp_workspace)

563

564 error_called = [False]

565 error_msg = [None]

566

567 def on_error(msg):

568 error_called[0] = True

569 error_msg[0] = msg

570

571 thread.error.connect(on_error)

572 thread.run()

573

574 assert error_called[0] is True

575 assert "Invalid" in error_msg[0] or "non valido" in error_msg[0]

576

577 def test_run_empty_folders_list(self, mock_context, temp_workspace):

578 """Test with empty folder list"""

579 thread = ImportThread(mock_context, [], temp_workspace)

580

581 error_called = [False]

582

583 def on_error(msg):

584 error_called[0] = True

585

586 thread.error.connect(on_error)

587 thread.run()

588

589 assert error_called[0] is True

590

591 def test_run_cancelled_no_error_emit(self, mock_context, temp_workspace):

592 """Test that cancellation does not emit errors"""

593 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

594

595 error_called = [False]

596

597 def on_error(msg):

598 error_called[0] = True

599

600 thread.error.connect(on_error)

601 thread.cancel()

602 thread.run()

603

604 # Should not emit an error when canceled

605 assert error_called[0] is False

606

607

608class TestProgressEmission:

609 """Tests for progress emission"""

610

611 @patch('main.threads.import_thread.subprocess.Popen')

612 def test_progress_increases_monotonically(self, mock_Popen, mock_context, temp_workspace):

613 """Test that progress increases monotonically"""

614

615 # Simulate Popen to avoid FileNotFoundError

616 mock_process = Mock()

617 mock_process.communicate.return_value = (b'Success', b'')

618 mock_process.returncode = 0

619 mock_Popen.return_value = mock_process

620

621 # Create a folder that is NOT BIDS (so it follows "Case C")

622 nifti_folder = os.path.join(temp_workspace, "nifti_src")

623 os.makedirs(nifti_folder)

624 with open(os.path.join(nifti_folder, "scan.nii.gz"), "w") as f:

625 f.write("x")

626 with open(os.path.join(nifti_folder, "scan.json"), "w") as f:

627 f.write('{"Modality": "MR", "ProtocolName": "T1w"}')

628

629 dest = os.path.join(temp_workspace, "out")

630 os.makedirs(dest)

631

632 # Use the "nifti_folder" (non-BIDS) as input

633 thread = ImportThread(mock_context, [nifti_folder], dest)

634

635 progress_values = []

636 thread.progress.connect(lambda v: progress_values.append(v))

637

638 # Run the thread

639 thread.run()

640

641 # **FIX**: More robust assertions. The strange value (1251907152)

642 # was likely a symptom of the test being interrupted abruptly.

643 assert len(progress_values) > 0

644 assert progress_values[-1] == 100 # Must end at 100

645

646 # Verify that values never decrease

647 last_val = -1

648 for val in progress_values:

649 assert val >= last_val

650 last_val = val

651

652

653class TestMultipleFoldersImport:

654 """Tests for multiple folder import"""

655

656 @patch.object(ImportThread, '_handle_import')

657 def test_run_multiple_folders(self, mock_handle, mock_context, temp_workspace):

658 """Test handling of multiple folders"""

659 folders = [

660 os.path.join(temp_workspace, "folder1"),

661 os.path.join(temp_workspace, "folder2"),

662 os.path.join(temp_workspace, "folder3")

663 ]

664

665 for folder in folders:

666 os.makedirs(folder, exist_ok=True)

667

668 thread = ImportThread(mock_context, folders, temp_workspace)

669

670 finished_called = [False]

671 thread.finished.connect(lambda: finished_called.__setitem__(0, True))

672

673 thread.run()

674

675 # Should call _handle_import for each folder

676 assert mock_handle.call_count == 3

677 assert finished_called[0] is True

678

679 def test_run_multiple_folders_progress(self, mock_context, temp_workspace):

680 """Test progress with multiple folders"""

681 folders = []

682 for i in range(3):

683 folder = os.path.join(temp_workspace, f"bids{i}")

684 anat = os.path.join(folder, "sub-01", "anat")

685 os.makedirs(anat)

686 with open(os.path.join(anat, "scan.nii"), "w") as f:

687 f.write("data")

688 folders.append(folder)

689

690 dest = os.path.join(temp_workspace, "output")

691 os.makedirs(dest)

692

693 thread = ImportThread(mock_context, folders, dest)

694

695 progress_values = []

696 thread.progress.connect(lambda v: progress_values.append(v))

697 thread.run()

698

699 # Progress should reach 100

700 assert 100 in progress_values

701 # Should have intermediate values

702 assert len(progress_values) > 2

703

704

705class TestHandleImport:

706 """Tests for the _handle_import method which manages single folders"""

707

708 @patch.object(ImportThread, '_is_bids_folder')

709 @patch.object(ImportThread, '_get_next_sub_id')

710 def test_handle_import_bids_folder(self, mock_get_id, mock_is_bids,

711 mock_context, temp_workspace):

712 """Test _handle_import with BIDS folder"""

713 mock_is_bids.return_value = True

714 mock_get_id.return_value = "sub-005"

715

716 bids_src = os.path.join(temp_workspace, "bids_source")

717 anat_dir = os.path.join(bids_src, "sub-01", "anat")

718 os.makedirs(anat_dir)

719 with open(os.path.join(anat_dir, "T1w.nii"), "w") as f:

720 f.write("brain")

721

722 dest_ws = os.path.join(temp_workspace, "workspace")

723 os.makedirs(dest_ws)

724

725 thread = ImportThread(mock_context, [temp_workspace], dest_ws)

726 thread._handle_import(bids_src)

727

728 # Verify direct copy

729 dest_folder = os.path.join(dest_ws, "sub-005")

730 assert os.path.exists(dest_folder)

731

732 @patch.object(ImportThread, '_process_single_patient_folder')

733 def test_handle_import_direct_medical_files(self, mock_process,

734 mock_context, temp_workspace):

735 """Test _handle_import with direct medical files"""

736 folder = os.path.join(temp_workspace, "patient_folder")

737 os.makedirs(folder)

738

739 # Direct NIfTI file

740 with open(os.path.join(folder, "brain.nii.gz"), "w") as f:

741 f.write("nifti")

742

743 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

744 thread._handle_import(folder)

745

746 # Should process as a single patient

747 mock_process.assert_called_once_with(folder)

748

749 @patch.object(ImportThread, '_are_dicom_series_of_same_patient')

750 @patch.object(ImportThread, '_process_single_patient_folder')

751 def test_handle_import_dicom_series_same_patient(self, mock_process, mock_same_patient,

752 mock_context, temp_workspace):

753 """Test _handle_import with DICOM series from the same patient"""

754 mock_same_patient.return_value = True

755

756 root_folder = os.path.join(temp_workspace, "dicom_root")

757 os.makedirs(root_folder)

758

759 # Create DICOM subfolders

760 for i in range(3):

761 series_dir = os.path.join(root_folder, f"series_{i}")

762 os.makedirs(series_dir)

763 dicom_file = os.path.join(series_dir, f"img_{i}.dcm")

764 with open(dicom_file, "wb") as f:

765 f.write(b'\x00' * 128 + b'DICM')

766

767 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

768 thread._handle_import(root_folder)

769

770 # Should process as a single patient

771 mock_process.assert_called_once_with(root_folder)

772

773 @patch.object(ImportThread, '_subfolders_look_like_different_patients')

774 def test_handle_import_multiple_patients(self, mock_different, mock_context, temp_workspace):

775 mock_different.return_value = True

776 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

777

778 root = os.path.join(temp_workspace, "multi_patient")

779 os.makedirs(root)

780 patient_folders = [os.path.join(root, f"sub-{i:02d}") for i in range(3)]

781 for p in patient_folders:

782 os.makedirs(p)

783

784 # Internal patch, not a decorator

785 with patch.object(thread, "_handle_import", wraps=thread._handle_import) as mock_handle_recursive:

786 thread._handle_import(root)

787

788 # Now internal calls are counted

789 assert mock_handle_recursive.call_count >= 3

790

791 def test_handle_import_empty_folder(self, mock_context, temp_workspace):

792 """Test _handle_import with an empty folder"""

793 empty_folder = os.path.join(temp_workspace, "empty")

794 os.makedirs(empty_folder)

795

796 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

797

798 # Should not raise exceptions

799 thread._handle_import(empty_folder)

800

801 def test_handle_import_not_directory(self, mock_context, temp_workspace):

802 """Test _handle_import with a file instead of a directory"""

803 file_path = os.path.join(temp_workspace, "file.txt")

804 with open(file_path, "w") as f:

805 f.write("not a directory")

806

807 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

808

809 # Should handle gracefully and do nothing

810 thread._handle_import(file_path)

811

812

813class TestComplexScenarios:

814 """Tests for complex scenarios and edge cases"""

815

816 def test_nested_bids_structure(self, mock_context, temp_workspace):

817 """Test with complex nested BIDS structure"""

818 bids_root = os.path.join(temp_workspace, "complex_bids")

819

820 # Multiple subjects

821 for sub_id in ["sub-01", "sub-02"]:

822 mod_dir = os.path.join(bids_root, sub_id, "anat") # Simplified for the test

823 os.makedirs(mod_dir)

824 nifti = os.path.join(mod_dir, f"{sub_id}_anat.nii.gz")

825 with open(nifti, "w") as f: f.write("nifti data")

826

827 thread = ImportThread(mock_context, [bids_root], temp_workspace)

828

829 # **FIX**: Test one of the created 'sub-' folders, not the root

830 assert thread._is_bids_folder(os.path.join(bids_root, "sub-01")) is True

831 assert thread._is_bids_folder(os.path.join(bids_root, "sub-02")) is True

832 assert thread._is_bids_folder(bids_root) is False # The root is not a BIDS folder

833

834 def test_mixed_content_folder(self, mock_context, temp_workspace):

835 """Test with folder containing a mix of different files"""

836 mixed_folder = os.path.join(temp_workspace, "mixed")

837 os.makedirs(mixed_folder)

838

839 with open(os.path.join(mixed_folder, "brain.nii"), "w") as f:

840 f.write("nifti")

841

842 dicom_file = os.path.join(mixed_folder, "scan.dcm")

843 with open(dicom_file, "wb") as f:

844 f.write(b'\x00' * 128 + b'DICM')

845

846 with open(os.path.join(mixed_folder, "report.pdf"), "w") as f:

847 f.write("pdf")

848 with open(os.path.join(mixed_folder, "metadata.json"), "w") as f:

849 f.write('{}')

850

851 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

852

853 # Should detect direct medical files

854 has_medical = any(

855 thread._is_nifti_file(f) or thread._is_dicom_file(os.path.join(mixed_folder, f))

856 for f in os.listdir(mixed_folder)

857 )

858 assert has_medical is True

859

860 @patch.object(ImportThread, '_convert_dicom_folder_to_nifti')

861 @patch.object(ImportThread, '_convert_dicom_folder_to_nifti')

862 def test_large_dataset_simulation(self, mock_convert, mock_context, temp_workspace):

863 """Test large dataset simulation (NIfTI)"""

864 root = os.path.join(temp_workspace, "large_dataset")

865 os.makedirs(root)

866

867 # Simulate many NIfTI files

868 for i in range(20):

869 nifti_file = os.path.join(root, f"scan_{i:03d}.nii.gz")

870 json_file = os.path.join(root, f"scan_{i:03d}.json")

871

872 with open(nifti_file, "w") as f:

873 f.write(f"scan {i}")

874

875 # --- FIX HERE ---

876 # Add minimal metadata for BIDS conversion

877 with open(json_file, "w") as f:

878 f.write('{"Modality": "MR", "ProtocolName": "T1w"}')

879 # ---------------------

880

881 dest_ws = os.path.join(temp_workspace, "dest")

882 os.makedirs(dest_ws)

883

884 thread = ImportThread(mock_context, [root], dest_ws)

885

886 progress_updates = []

887 thread.progress.connect(lambda v: progress_updates.append(v))

888

889 thread.run()

890

891 # Should complete successfully

892 assert 100 in progress_updates

893 # Now this assertion will work

894 assert os.path.exists(os.path.join(dest_ws, "sub-01", "anat"))

895

896 def test_special_characters_in_filenames(self, mock_context, temp_workspace):

897 """Test with special characters in filenames"""

898 folder = os.path.join(temp_workspace, "special_chars")

899 os.makedirs(folder)

900

901 # File with special characters

902 special_names = [

903 "brain (copy).nii",

904 "scan_01-02-2024.nii.gz",

905 "patient#123.json",

906 "data [final].nii"

907 ]

908

909 for name in special_names:

910 safe_name = name.replace("[", "_").replace("]", "_")

911 file_path = os.path.join(folder, safe_name)

912 with open(file_path, "w") as f:

913 f.write("data")

914

915 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

916

917 # Should handle without errors

918 nifti_files = [f for f in os.listdir(folder) if thread._is_nifti_file(f)]

919 assert len(nifti_files) > 0

920

921 @patch('main.threads.import_thread.pydicom.dcmread')

922 def test_dicom_missing_patient_info(self, mock_dcmread, mock_context, temp_workspace):

923 """Test DICOM without patient information"""

924 # Mock DICOM without PatientID or PatientName

925 mock_dcm = Mock()

926 mock_dcm.PatientID = None

927 mock_dcm.PatientName = None

928 mock_dcmread.return_value = mock_dcm

929

930 folder = os.path.join(temp_workspace, "dicom_no_info")

931 os.makedirs(folder)

932

933 dicom_file = os.path.join(folder, "anon.dcm")

934 with open(dicom_file, "wb") as f:

935 f.write(b'\x00' * 128 + b'DICM')

936

937 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

938

939 # Should handle gracefully

940 result = thread._are_dicom_series_of_same_patient([folder])

941 # Without PatientID, it considers them the same patient

942 assert result is True

943

944

945class TestConcurrencyAndThreadSafety:

946 """Tests for thread safety and concurrency"""

947

948 def test_cancel_during_file_copy(self, mock_context, temp_workspace):

949 """Test cancellation during file copy"""

950 # Create many files to copy

951 src = os.path.join(temp_workspace, "source_large")

952 os.makedirs(src)

953

954 for i in range(100):

955 with open(os.path.join(src, f"file_{i}.nii"), "w") as f:

956 f.write("x" * 1000)

957

958 dest = os.path.join(temp_workspace, "dest")

959 os.makedirs(dest)

960

961 thread = ImportThread(mock_context, [src], dest)

962

963 # Cancel after a short delay

964 import threading

965 def cancel_after_delay():

966 import time

967 time.sleep(0.01)

968 thread.cancel()

969

970 cancel_thread = threading.Thread(target=cancel_after_delay)

971 cancel_thread.start()

972

973 thread.run()

974 cancel_thread.join()

975

976 # Should be canceled

977 assert thread._is_canceled is True

978

979 def test_multiple_progress_emissions(self, mock_context, temp_workspace):

980 """Test multiple progress emissions"""

981 bids = os.path.join(temp_workspace, "bids_multi")

982 anat = os.path.join(bids, "sub-01", "anat")

983 os.makedirs(anat)

984 with open(os.path.join(anat, "T1.nii"), "w") as f:

985 f.write("x")

986

987 dest = os.path.join(temp_workspace, "out")

988 os.makedirs(dest)

989

990 thread = ImportThread(mock_context, [bids], dest)

991

992 progress_count = [0]

993

994 def count_progress(val):

995 progress_count[0] += 1

996

997 thread.progress.connect(count_progress)

998 thread.run()

999

1000 # Should have emitted progress multiple times

1001 assert progress_count[0] >= 3

1002

1003

1004class TestEdgeCases:

1005 """Tests for edge cases"""

1006

1007 def test_symlink_handling(self, mock_context, temp_workspace):

1008 """Test symlink handling"""

1009 if os.name == 'nt': # Skip on Windows

1010 pytest.skip("Symlinks not reliable on Windows")

1011

1012 real_folder = os.path.join(temp_workspace, "real")

1013 os.makedirs(real_folder)

1014 with open(os.path.join(real_folder, "data.nii"), "w") as f:

1015 f.write("nifti")

1016

1017 symlink = os.path.join(temp_workspace, "symlink")

1018 os.symlink(real_folder, symlink)

1019

1020 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

1021

1022 # Should follow symlink

1023 assert os.path.isdir(symlink)

1024

1025 def test_readonly_source_files(self, mock_context, temp_workspace):

1026 """Test with read-only source files"""

1027 src = os.path.join(temp_workspace, "readonly_src")

1028 os.makedirs(src)

1029

1030 readonly_file = os.path.join(src, "readonly.nii")

1031 with open(readonly_file, "w") as f:

1032 f.write("nifti")

1033

1034 # Make read-only

1035 os.chmod(readonly_file, 0o444)

1036

1037 dest = os.path.join(temp_workspace, "dest")

1038 os.makedirs(dest)

1039

1040 thread = ImportThread(mock_context, [src], dest)

1041

1042 # Should copy anyway

1043 try:

1044 thread.run()

1045 finally:

1046 # Cleanup: remove read-only to allow cleanup

1047 os.chmod(readonly_file, 0o644)

1048

1049 def test_very_long_path(self, mock_context, temp_workspace):

1050 """Test with very long path"""

1051 # Create a very long path

1052 long_path = temp_workspace

1053 for i in range(10):

1054 long_path = os.path.join(long_path, f"nested_folder_{i}")

1055

1056 try:

1057 os.makedirs(long_path)

1058 with open(os.path.join(long_path, "file.nii"), "w") as f:

1059 f.write("data")

1060 except OSError:

1061 pytest.skip("Path too long for filesystem")

1062

1063 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

1064 # Should handle without errors

1065 assert os.path.exists(long_path)

1066

1067 def test_unicode_filenames(self, mock_context, temp_workspace):

1068 """Test with Unicode filenames"""

1069 folder = os.path.join(temp_workspace, "unicode")

1070 os.makedirs(folder)

1071

1072 unicode_names = [

1073 "cervello_日本語.nii",

1074 "мозг_данные.nii.gz",

1075 "cerebro_中文.json"

1076 ]

1077

1078 for name in unicode_names:

1079 file_path = os.path.join(folder, name)

1080 try:

1081 with open(file_path, "w", encoding="utf-8") as f:

1082 f.write("data")

1083 except (OSError, UnicodeError):

1084 pytest.skip("Filesystem doesn't support Unicode")

1085

1086 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

1087

1088 # Verify NIfTI detection with Unicode

1089 nifti_count = sum(1 for f in os.listdir(folder) if thread._is_nifti_file(f))

1090 assert nifti_count >= 2

1091

1092

1093class TestIntegrationScenarios:

1094 """Tests for end-to-end integration"""

1095

1096 @patch('main.threads.import_thread.subprocess.Popen')

1097 @patch('main.threads.import_thread.get_bin_path')

1098 def test_full_dicom_to_bids_workflow(self, mock_get_bin, mock_Popen,

1099 mock_context, temp_workspace):

1100 """Test complete DICOM → BIDS workflow"""

1101 mock_get_bin.return_value = "/fake/path/dcm2niix"

1102

1103 # Simulate dcm2niix output

1104 def mock_popen_side_effect(*args, **kwargs):

1105 command = args[0]

1106 output_dir = command[command.index("-o") + 1]

1107

1108 # Create fictitious NIfTI and JSON files in the temporary output

1109 with open(os.path.join(output_dir, "converted.nii.gz"), "w") as f:

1110 f.write("converted nifti")

1111 # Add metadata for BIDS conversion

1112 with open(os.path.join(output_dir, "converted.json"), "w") as f:

1113 f.write('{"Modality": "MR", "ProtocolName": "T1w"}')

1114

1115 mock_process = Mock()

1116 mock_process.communicate.return_value = (b'Success', b'')

1117 mock_process.returncode = 0

1118 return mock_process

1119

1120 mock_Popen.side_effect = mock_popen_side_effect

1121

1122 # Create fictitious DICOM folder

1123 dicom_folder = os.path.join(temp_workspace, "dicom_data")

1124 os.makedirs(dicom_folder)

1125 # Use pydicom to create a "minimally valid" DICOM file

1126 try:

1127 import pydicom

1128 from pydicom.dataset import Dataset, FileMetaDataset

1129 from pydicom.uid import ImplicitVRLittleEndian

1130

1131 file_meta = FileMetaDataset()

1132 file_meta.MediaStorageSOPClassUID = '1.2.840.10008.5.1.4.1.1.2'

1133 file_meta.MediaStorageSOPInstanceUID = "1.2.3.4"

1134 file_meta.ImplementationClassUID = "1.2.3.4"

1135 file_meta.TransferSyntaxUID = ImplicitVRLittleEndian

1136

1137 ds = Dataset()

1138 ds.PatientID = "12345"

1139 ds.file_meta = file_meta

1140 ds.is_little_endian = True

1141 ds.is_implicit_VR = True

1142

1143 ds.save_as(os.path.join(dicom_folder, "image_001.dcm"), write_like_original=False)

1144

1145 except ImportError:

1146 # Fallback if pydicom is not installed in the test environment

1147 with open(os.path.join(dicom_folder, "image_001.dcm"), "wb") as f:

1148 f.write(b'\x00' * 128 + b'DICM' + b'\x00' * 100)

1149

1150 dest_ws = os.path.join(temp_workspace, "workspace")

1151 os.makedirs(dest_ws)

1152

1153 thread = ImportThread(mock_context, [dicom_folder], dest_ws)

1154

1155 finished = [False]

1156 thread.finished.connect(lambda: finished.__setitem__(0, True))

1157

1158 thread.run()

1159

1160 # **FIX**: The assertion now passes because the Popen mock

1161 # prevents the FileNotFoundError and allows the thread to emit 'finished'.

1162 assert finished[0] is True

1163

1164 # Add an assertion to verify the BIDS structure was created

1165 expected_anat = os.path.join(dest_ws, "sub-01", "anat")

1166 assert os.path.exists(expected_anat)

1167 expected_file = os.path.join(expected_anat, "sub-01_run-1_T1w.nii.gz")

1168 assert os.path.exists(expected_file)

1169

1170 def test_mixed_patient_dataset_import(self, mock_context, temp_workspace):

1171 """Test importing dataset with multiple patients and mixed formats"""

1172 root = os.path.join(temp_workspace, "mixed_dataset")

1173 os.makedirs(root)

1174

1175 # Patient 1: NIfTI

1176 p1 = os.path.join(root, "sub-01")

1177 os.makedirs(p1)

1178 with open(os.path.join(p1, "T1w.nii.gz"), "w") as f:

1179 f.write("patient 1 nifti")

1180

1181 # Patient 2: Already organized BIDS

1182 p2_anat = os.path.join(root, "sub-02", "anat")

1183 os.makedirs(p2_anat)

1184 with open(os.path.join(p2_anat, "sub-02_T1w.nii"), "w") as f:

1185 f.write("patient 2")

1186

1187 # Patient 3: DICOM

1188 p3 = os.path.join(root, "patient_003")

1189 os.makedirs(p3)

1190 with open(os.path.join(p3, "scan.dcm"), "wb") as f:

1191 f.write(b'\x00' * 128 + b'DICM')

1192

1193 dest = os.path.join(temp_workspace, "output")

1194 os.makedirs(dest)

1195

1196 thread = ImportThread(mock_context, [root], dest)

1197 thread.run()

1198

1199 # Should process as multiple patients

1200 # (thanks to the folder name heuristic)

1201 assert thread._subfolders_look_like_different_patients([p1, p2_anat, p3])

1202

1203

1204# Parametrized tests for reuse

1205@pytest.mark.parametrize("extension", [".nii", ".nii.gz"])

1206def test_nifti_extensions(extension, mock_context, temp_workspace):

1207 """Parametrized test for different NIfTI extensions"""

1208 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

1209 filename = f"brain{extension}"

1210 assert thread._is_nifti_file(filename) is True

1211

1212

1213@pytest.mark.parametrize("invalid_ext", [".dcm", ".txt", ".json", ".pdf", ""])

1214def test_non_nifti_extensions(invalid_ext, mock_context, temp_workspace):

1215 """Parametrized test for non-NIfTI extensions"""

1216 thread = ImportThread(mock_context, [temp_workspace], temp_workspace)

1217 filename = f"file{invalid_ext}"

1218 assert thread._is_nifti_file(filename) is False

1219

1220

1221if __name__ == "__main__":

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

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

683 statements