Coverage for tests/threads/test_utils

1"""

2test_copy_delete_thread.py - Test Suite for CopyDeleteThread

4This suite tests all functionalities of the copy/delete thread:

5- Copying single files

6- Copying recursive directories

7- Deleting single files

8- Deleting directories

9- Combined operations

10- Error handling and edge cases

11"""

13import os

14import shutil

15import tempfile

16from unittest.mock import Mock, patch, call

17import pytest

19from main.threads.utils_threads import CopyDeleteThread

22class TestCopyDeleteThreadInitialization:

23 """Tests for CopyDeleteThread initialization"""

25 def test_init_copy_file(self, temp_workspace):

26 """Test initialization for file copy"""

27 src = os.path.join(temp_workspace, "source.txt")

28 dst = os.path.join(temp_workspace, "dest.txt")

30 thread = CopyDeleteThread(

31 src=src,

32 dst=dst,

33 is_folder=False,

34 copy=True,

35 delete=False

36 )

38 assert thread.src == src

39 assert thread.dst == dst

40 assert thread.is_folder is False

41 assert thread.copy is True

42 assert thread.delete is False

44 def test_init_delete_folder(self, temp_workspace):

45 """Test initialization for folder deletion"""

46 src = os.path.join(temp_workspace, "folder_to_delete")

48 thread = CopyDeleteThread(

49 src=src,

50 dst=None,

51 is_folder=True,

52 copy=False,

53 delete=True

54 )

56 assert thread.src == src

57 assert thread.dst is None

58 assert thread.is_folder is True

59 assert thread.copy is False

60 assert thread.delete is True

62 def test_init_copy_and_delete(self, temp_workspace):

63 """Test initialization for combined operation"""

64 src = os.path.join(temp_workspace, "source")

65 dst = os.path.join(temp_workspace, "dest")

67 thread = CopyDeleteThread(

68 src=src,

69 dst=dst,

70 is_folder=True,

71 copy=True,

72 delete=True

73 )

75 assert thread.copy is True

76 assert thread.delete is True

78 def test_signals_exist(self, temp_workspace):

79 """Verify that signals are defined correctly"""

80 thread = CopyDeleteThread(src="dummy", dst=None)

82 assert hasattr(thread, 'finished')

83 assert hasattr(thread, 'error')

86class TestFileCopy:

87 """Tests for single file copy"""

89 def test_copy_file_success(self, temp_workspace):

90 """Test successful file copy"""

91 src = os.path.join(temp_workspace, "source.txt")

92 dst = os.path.join(temp_workspace, "destination.txt")

94 with open(src, 'w') as f:

95 f.write("test content")

97 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

99 finished_msgs = []

100 thread.finished.connect(lambda msg: finished_msgs.append(msg))

101

102 thread.run()

103

104 # Verify file was copied

105 assert os.path.exists(dst)

106 with open(dst, 'r') as f:

107 assert f.read() == "test content"

108

109 # Verify signal

110 assert len(finished_msgs) == 1

111 assert "copied" or "copiato" in finished_msgs[0].lower()

112

113 def test_copy_file_overwrites_existing(self, temp_workspace):

114 """Test that copy overwrites existing file"""

115 src = os.path.join(temp_workspace, "source.txt")

116 dst = os.path.join(temp_workspace, "existing.txt")

117

118 with open(src, 'w') as f:

119 f.write("new content")

120

121 # Create existing destination

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

123 f.write("old content")

124

125 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

126 thread.run()

127

128 # Verify overwrite

129 with open(dst, 'r') as f:

130 assert f.read() == "new content"

131

132 def test_copy_file_to_subfolder(self, temp_workspace):

133 """Test copying file into subfolder"""

134 src = os.path.join(temp_workspace, "file.txt")

135 subfolder = os.path.join(temp_workspace, "subfolder")

136 dst = os.path.join(subfolder, "file.txt")

137

138 with open(src, 'w') as f:

139 f.write("data")

140

141 # Create subfolder

142 os.makedirs(subfolder)

143

144 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

145 thread.run()

146

147 assert os.path.exists(dst)

148

149 def test_copy_binary_file(self, temp_workspace):

150 """Test copying binary file"""

151 src = os.path.join(temp_workspace, "binary.bin")

152 dst = os.path.join(temp_workspace, "binary_copy.bin")

153

154 # Create binary file

155 binary_data = bytes([0, 1, 2, 255, 128, 64])

156 with open(src, 'wb') as f:

157 f.write(binary_data)

158

159 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

160 thread.run()

161

162 # Verify binary content

163 with open(dst, 'rb') as f:

164 assert f.read() == binary_data

165

166 def test_copy_large_file(self, temp_workspace):

167 """Test copying large file"""

168 src = os.path.join(temp_workspace, "large.dat")

169 dst = os.path.join(temp_workspace, "large_copy.dat")

170

171 # Create "large" file (1 MB)

172 large_content = "x" * (1024 * 1024)

173 with open(src, 'w') as f:

174 f.write(large_content)

175

176 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

177 thread.run()

178

179 assert os.path.exists(dst)

180 assert os.path.getsize(dst) == os.path.getsize(src)

181

182

183class TestFolderCopy:

184 """Tests for directory copy"""

185

186 def test_copy_folder_success(self, temp_workspace):

187 """Test successful folder copy"""

188 src = os.path.join(temp_workspace, "source_folder")

189 dst = os.path.join(temp_workspace, "dest_folder")

190

191 # Create source folder with files

192 os.makedirs(src)

193 with open(os.path.join(src, "file1.txt"), 'w') as f:

194 f.write("content 1")

195 with open(os.path.join(src, "file2.txt"), 'w') as f:

196 f.write("content 2")

197

198 thread = CopyDeleteThread(src=src, dst=dst, is_folder=True, copy=True)

199

200 finished_msgs = []

201 thread.finished.connect(lambda msg: finished_msgs.append(msg))

202

203 thread.run()

204

205 # Verify folder was copied

206 assert os.path.exists(dst)

207 assert os.path.exists(os.path.join(dst, "file1.txt"))

208 assert os.path.exists(os.path.join(dst, "file2.txt"))

209

210 # Verify contents

211 with open(os.path.join(dst, "file1.txt"), 'r') as f:

212 assert f.read() == "content 1"

213

214 assert len(finished_msgs) == 1

215

216 def test_copy_nested_folder(self, temp_workspace):

217 """Test copying folder with subfolders"""

218 src = os.path.join(temp_workspace, "nested_src")

219 dst = os.path.join(temp_workspace, "nested_dst")

220

221 # Create nested structure

222 os.makedirs(os.path.join(src, "sub1", "sub2"))

223 with open(os.path.join(src, "root.txt"), 'w') as f:

224 f.write("root")

225 with open(os.path.join(src, "sub1", "file1.txt"), 'w') as f:

226 f.write("sub1")

227 with open(os.path.join(src, "sub1", "sub2", "file2.txt"), 'w') as f:

228 f.write("sub2")

229

230 thread = CopyDeleteThread(src=src, dst=dst, is_folder=True, copy=True)

231 thread.run()

232

233 # Verify the entire structure

234 assert os.path.exists(os.path.join(dst, "root.txt"))

235 assert os.path.exists(os.path.join(dst, "sub1", "file1.txt"))

236 assert os.path.exists(os.path.join(dst, "sub1", "sub2", "file2.txt"))

237

238 def test_copy_folder_with_dirs_exist_ok(self, temp_workspace):

239 """Test copying into existing folder (dirs_exist_ok=True)"""

240 src = os.path.join(temp_workspace, "source")

241 dst = os.path.join(temp_workspace, "existing_dest")

242

243 os.makedirs(src)

244 with open(os.path.join(src, "new_file.txt"), 'w') as f:

245 f.write("new")

246

247 # Create existing destination

248 os.makedirs(dst)

249 with open(os.path.join(dst, "old_file.txt"), 'w') as f:

250 f.write("old")

251

252 thread = CopyDeleteThread(src=src, dst=dst, is_folder=True, copy=True)

253 thread.run()

254

255 # Both files should exist

256 assert os.path.exists(os.path.join(dst, "new_file.txt"))

257 assert os.path.exists(os.path.join(dst, "old_file.txt"))

258

259 def test_copy_empty_folder(self, temp_workspace):

260 """Test copying empty folder"""

261 src = os.path.join(temp_workspace, "empty_src")

262 dst = os.path.join(temp_workspace, "empty_dst")

263

264 os.makedirs(src)

265

266 thread = CopyDeleteThread(src=src, dst=dst, is_folder=True, copy=True)

267 thread.run()

268

269 assert os.path.exists(dst)

270 assert os.path.isdir(dst)

271 assert len(os.listdir(dst)) == 0

272

273

274class TestFileDelete:

275 """Tests for file deletion"""

276

277 def test_delete_file_success(self, temp_workspace):

278 """Test successful file deletion"""

279 file_path = os.path.join(temp_workspace, "to_delete.txt")

280

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

282 f.write("delete me")

283

284 assert os.path.exists(file_path)

285

286 thread = CopyDeleteThread(src=file_path, is_folder=False, delete=True)

287

288 finished_msgs = []

289 thread.finished.connect(lambda msg: finished_msgs.append(msg))

290

291 thread.run()

292

293 # Verify file was deleted

294 assert not os.path.exists(file_path)

295 assert len(finished_msgs) == 1

296 assert "deleted" in finished_msgs[0].lower() or "eliminato" in finished_msgs[0].lower()

297

298 def test_delete_readonly_file(self, temp_workspace):

299 """Test deleting read-only file"""

300 file_path = os.path.join(temp_workspace, "readonly.txt")

301

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

303 f.write("readonly")

304

305 # Make read-only

306 os.chmod(file_path, 0o444)

307

308 thread = CopyDeleteThread(src=file_path, is_folder=False, delete=True)

309

310 error_msgs = []

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

312

313 thread.run()

314

315 # Might fail on Unix/Linux systems

316 # But should still emit error signal if it fails

317 if os.path.exists(file_path):

318 assert len(error_msgs) > 0

319 # Cleanup

320 os.chmod(file_path, 0o644)

321 os.remove(file_path)

322

323

324class TestFolderDelete:

325 """Tests for directory deletion"""

326

327 def test_delete_folder_success(self, temp_workspace):

328 """Test successful folder deletion"""

329 folder_path = os.path.join(temp_workspace, "folder_to_delete")

330

331 # Create folder with content

332 os.makedirs(folder_path)

333 with open(os.path.join(folder_path, "file.txt"), 'w') as f:

334 f.write("content")

335

336 assert os.path.exists(folder_path)

337

338 thread = CopyDeleteThread(src=folder_path, is_folder=True, delete=True)

339

340 finished_msgs = []

341 thread.finished.connect(lambda msg: finished_msgs.append(msg))

342

343 thread.run()

344

345 # Verify folder was deleted

346 assert not os.path.exists(folder_path)

347 assert len(finished_msgs) == 1

348

349 def test_delete_nested_folder(self, temp_workspace):

350 """Test deleting folder with subfolders"""

351 folder_path = os.path.join(temp_workspace, "nested_delete")

352

353 # Create complex structure

354 os.makedirs(os.path.join(folder_path, "sub1", "sub2"))

355 for i in range(5):

356 with open(os.path.join(folder_path, f"file{i}.txt"), 'w') as f:

357 f.write(f"content {i}")

358

359 thread = CopyDeleteThread(src=folder_path, is_folder=True, delete=True)

360 thread.run()

361

362 # The entire structure should be deleted

363 assert not os.path.exists(folder_path)

364

365 def test_delete_empty_folder(self, temp_workspace):

366 """Test deleting empty folder"""

367 folder_path = os.path.join(temp_workspace, "empty_folder")

368 os.makedirs(folder_path)

369

370 thread = CopyDeleteThread(src=folder_path, is_folder=True, delete=True)

371 thread.run()

372

373 assert not os.path.exists(folder_path)

374

375

376class TestCombinedOperations:

377 """Tests for combined operations (copy + delete)"""

378

379 def test_copy_then_delete_file(self, temp_workspace):

380 """Test copy then delete file"""

381 src = os.path.join(temp_workspace, "original.txt")

382 dst = os.path.join(temp_workspace, "copy.txt")

383

384 with open(src, 'w') as f:

385 f.write("move me")

386

387 # Combined operation = "move"

388 thread = CopyDeleteThread(

389 src=src,

390 dst=dst,

391 is_folder=False,

392 copy=True,

393 delete=True

394 )

395

396 finished_count = [0]

397 thread.finished.connect(lambda msg: finished_count.__setitem__(0, finished_count[0] + 1))

398

399 thread.run()

400

401 # Copied file should exist

402 assert os.path.exists(dst)

403 with open(dst, 'r') as f:

404 assert f.read() == "move me"

405

406 # Original file should be deleted

407 assert not os.path.exists(src)

408

409 # Two finished signals (one for copy, one for delete)

410 assert finished_count[0] == 2

411

412 def test_copy_then_delete_folder(self, temp_workspace):

413 """Test copy then delete folder (equivalent to move)"""

414 src = os.path.join(temp_workspace, "src_folder")

415 dst = os.path.join(temp_workspace, "dst_folder")

416

417 os.makedirs(src)

418 with open(os.path.join(src, "data.txt"), 'w') as f:

419 f.write("folder data")

420

421 thread = CopyDeleteThread(

422 src=src,

423 dst=dst,

424 is_folder=True,

425 copy=True,

426 delete=True

427 )

428

429 thread.run()

430

431 # Destination should exist

432 assert os.path.exists(dst)

433 assert os.path.exists(os.path.join(dst, "data.txt"))

434

435 # Source should be deleted

436 assert not os.path.exists(src)

437

438

439class TestErrorHandling:

440 """Tests for error handling"""

441

442 def test_copy_missing_source(self, temp_workspace):

443 """Test copy with missing source file"""

444 src = os.path.join(temp_workspace, "nonexistent.txt")

445 dst = os.path.join(temp_workspace, "dest.txt")

446

447 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

448

449 error_msgs = []

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

451

452 thread.run()

453

454 # Should emit error

455 assert len(error_msgs) == 1

456 assert "error" in error_msgs[0].lower() or src in error_msgs[0]

457

458 def test_copy_missing_src_parameter(self, temp_workspace):

459 """Test copy with missing src parameter"""

460 thread = CopyDeleteThread(src=None, dst="dest", copy=True)

461

462 error_msgs = []

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

464

465 thread.run()

466

467 assert len(error_msgs) == 1

468 assert "missing" in error_msgs[0].lower() or "manca" in error_msgs[0].lower()

469

470 def test_copy_missing_dst_parameter(self, temp_workspace):

471 """Test copy with missing dst parameter"""

472 thread = CopyDeleteThread(src="src", dst=None, copy=True)

473

474 error_msgs = []

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

476

477 thread.run()

478

479 assert len(error_msgs) == 1

480 assert "missing" in error_msgs[0].lower() or "manca" in error_msgs[0].lower()

481

482 def test_delete_missing_source(self, temp_workspace):

483 """Test deleting non-existent file"""

484 src = os.path.join(temp_workspace, "does_not_exist.txt")

485

486 thread = CopyDeleteThread(src=src, is_folder=False, delete=True)

487

488 error_msgs = []

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

490

491 thread.run()

492

493 # Should emit error

494 assert len(error_msgs) == 1

495

496 def test_delete_missing_src_parameter(self, temp_workspace):

497 """Test deletion with missing src parameter"""

498 thread = CopyDeleteThread(src=None, delete=True)

499

500 error_msgs = []

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

502

503 thread.run()

504

505 assert len(error_msgs) == 1

506 assert "missing" in error_msgs[0].lower() or "manca" in error_msgs[0].lower()

507

508 def test_copy_to_invalid_destination(self, temp_workspace):

509 """Test copy to invalid destination"""

510 src = os.path.join(temp_workspace, "source.txt")

511 # Destination in non-existent directory

512 dst = os.path.join(temp_workspace, "nonexistent_dir", "subdir", "dest.txt")

513

514 with open(src, 'w') as f:

515 f.write("data")

516

517 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

518

519 error_msgs = []

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

521

522 thread.run()

523

524 # Should fail

525 assert len(error_msgs) == 1

526

527 @patch('shutil.copy')

528 def test_copy_permission_error(self, mock_copy, temp_workspace):

529 """Test handling permission error during copy"""

530 mock_copy.side_effect = PermissionError("Permission denied")

531

532 src = os.path.join(temp_workspace, "source.txt")

533 dst = os.path.join(temp_workspace, "dest.txt")

534

535 with open(src, 'w') as f:

536 f.write("data")

537

538 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

539

540 error_msgs = []

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

542

543 thread.run()

544

545 assert len(error_msgs) == 1

546 assert "permission" in error_msgs[0].lower() or "error" in error_msgs[0].lower()

547

548

549class TestEdgeCases:

550 """Tests for edge cases"""

551

552 def test_copy_file_to_itself(self, temp_workspace):

553 """Test copying file onto itself"""

554 file_path = os.path.join(temp_workspace, "same.txt")

555

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

557 f.write("content")

558

559 thread = CopyDeleteThread(

560 src=file_path,

561 dst=file_path,

562 is_folder=False,

563 copy=True

564 )

565

566 error_msgs = []

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

568

569 thread.run()

570

571 # shutil.copy should handle this case

572 # It might emit an error or succeed depending on the implementation

573 # We just verify it doesn't raise unhandled exceptions

574

575 def test_copy_symlink(self, temp_workspace):

576 """Test copying symlink"""

577 if os.name == 'nt':

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

579

580 src = os.path.join(temp_workspace, "original.txt")

581 link = os.path.join(temp_workspace, "link.txt")

582 dst = os.path.join(temp_workspace, "copy.txt")

583

584 with open(src, 'w') as f:

585 f.write("original data")

586

587 os.symlink(src, link)

588

589 thread = CopyDeleteThread(src=link, dst=dst, is_folder=False, copy=True)

590 thread.run()

591

592 # Should copy the content, not the link

593 assert os.path.exists(dst)

594 with open(dst, 'r') as f:

595 assert f.read() == "original data"

596

597 def test_delete_symlink(self, temp_workspace):

598 """Test deleting symlink"""

599 if os.name == 'nt':

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

601

602 target = os.path.join(temp_workspace, "target.txt")

603 link = os.path.join(temp_workspace, "link.txt")

604

605 with open(target, 'w') as f:

606 f.write("target")

607

608 os.symlink(target, link)

609

610 thread = CopyDeleteThread(src=link, is_folder=False, delete=True)

611 thread.run()

612

613 # Link should be deleted, target should still exist

614 assert not os.path.exists(link)

615 assert os.path.exists(target)

616

617 def test_copy_file_with_special_characters(self, temp_workspace):

618 """Test copying file with special characters in name"""

619 src = os.path.join(temp_workspace, "file (copy) #1.txt")

620 dst = os.path.join(temp_workspace, "destination [final].txt")

621

622 with open(src, 'w') as f:

623 f.write("special chars")

624

625 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

626 thread.run()

627

628 assert os.path.exists(dst)

629 with open(dst, 'r') as f:

630 assert f.read() == "special chars"

631

632 def test_copy_unicode_filename(self, temp_workspace):

633 """Test copying file with Unicode name"""

634 src = os.path.join(temp_workspace, "文件.txt")

635 dst = os.path.join(temp_workspace, "файл.txt")

636

637 try:

638 with open(src, 'w', encoding='utf-8') as f:

639 f.write("unicode content")

640 except (OSError, UnicodeError):

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

642

643 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

644 thread.run()

645

646 if os.path.exists(dst):

647 with open(dst, 'r', encoding='utf-8') as f:

648 assert f.read() == "unicode content"

649

650 def test_copy_hidden_file(self, temp_workspace):

651 """Test copying hidden file"""

652 src = os.path.join(temp_workspace, ".hidden_file")

653 dst = os.path.join(temp_workspace, ".hidden_copy")

654

655 with open(src, 'w') as f:

656 f.write("hidden content")

657

658 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

659 thread.run()

660

661 assert os.path.exists(dst)

662

663 def test_no_operation_specified(self, temp_workspace):

664 """Test with no operations (neither copy nor delete)"""

665 src = os.path.join(temp_workspace, "file.txt")

666 with open(src, 'w') as f:

667 f.write("data")

668

669 thread = CopyDeleteThread(src=src, copy=False, delete=False)

670

671 finished_count = [0]

672 error_count = [0]

673 thread.finished.connect(lambda msg: finished_count.__setitem__(0, finished_count[0] + 1))

674 thread.error.connect(lambda msg: error_count.__setitem__(0, error_count[0] + 1))

675

676 thread.run()

677

678 # Should do nothing, but also shouldn't crash

679 assert finished_count[0] == 0

680 assert error_count[0] == 0

681

682

683class TestSignalEmissions:

684 """Tests for signal emissions"""

685

686 def test_finished_signal_contains_paths(self, temp_workspace):

687 """Test that finished signal contains paths"""

688 src = os.path.join(temp_workspace, "source.txt")

689 dst = os.path.join(temp_workspace, "dest.txt")

690

691 with open(src, 'w') as f:

692 f.write("data")

693

694 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

695

696 finished_msgs = []

697 thread.finished.connect(lambda msg: finished_msgs.append(msg))

698

699 thread.run()

700

701 assert len(finished_msgs) == 1

702 msg = finished_msgs[0]

703

704 # Message should contain src and dst

705 assert src in msg or os.path.basename(src) in msg

706 assert dst in msg or os.path.basename(dst) in msg

707

708 def test_error_signal_contains_details(self, temp_workspace):

709 """Test that error signal contains details"""

710 src = os.path.join(temp_workspace, "nonexistent.txt")

711 dst = os.path.join(temp_workspace, "dest.txt")

712

713 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

714

715 error_msgs = []

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

717

718 thread.run()

719

720 assert len(error_msgs) == 1

721 msg = error_msgs[0]

722

723 # Error message should contain src

724 assert src in msg

725

726 def test_multiple_finished_signals_on_combined_operation(self, temp_workspace):

727 """Test multiple finished signal emissions on combined operation"""

728 src = os.path.join(temp_workspace, "file.txt")

729 dst = os.path.join(temp_workspace, "copy.txt")

730

731 with open(src, 'w') as f:

732 f.write("data")

733

734 thread = CopyDeleteThread(

735 src=src,

736 dst=dst,

737 is_folder=False,

738 copy=True,

739 delete=True

740 )

741

742 finished_msgs = []

743 thread.finished.connect(lambda msg: finished_msgs.append(msg))

744

745 thread.run()

746

747 assert len(finished_msgs) == 2

748 assert any("copied" in msg.lower() or "copiato" in msg.lower() for msg in finished_msgs)

749 assert any("deleted" in msg.lower() or "eliminato" in msg.lower() for msg in finished_msgs)

750

751

752class TestConcurrency:

753 """Tests for concurrency and thread safety"""

754

755 def test_multiple_copy_threads_concurrent(self, temp_workspace):

756 """Test concurrent execution of multiple copy threads"""

757 threads = []

758 files_to_create = 5

759

760 for i in range(files_to_create):

761 src = os.path.join(temp_workspace, f"source_{i}.txt")

762 dst = os.path.join(temp_workspace, f"dest_{i}.txt")

763

764 with open(src, 'w') as f:

765 f.write(f"content {i}")

766

767 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

768 threads.append(thread)

769

770 # Start all threads

771 for thread in threads:

772 thread.start()

773

774 # Wait for completion

775 for thread in threads:

776 thread.wait(5000) # 5 second timeout

777

778 # Verify all files were copied

779 for i in range(files_to_create):

780 dst = os.path.join(temp_workspace, f"dest_{i}.txt")

781 assert os.path.exists(dst)

782

783 def test_multiple_delete_threads_concurrent(self, temp_workspace):

784 """Test concurrent deletion of multiple files"""

785 threads = []

786 files_to_delete = 5

787

788 # Create files

789 for i in range(files_to_delete):

790 file_path = os.path.join(temp_workspace, f"delete_{i}.txt")

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

792 f.write(f"data {i}")

793

794 thread = CopyDeleteThread(src=file_path, is_folder=False, delete=True)

795 threads.append((thread, file_path))

796

797 # Start all

798 for thread, _ in threads:

799 thread.start()

800

801 # Wait

802 for thread, _ in threads:

803 thread.wait(5000)

804

805 # Verify all were deleted

806 for _, file_path in threads:

807 assert not os.path.exists(file_path)

808

809

810class TestPerformance:

811 """Tests for performance with large files/folders"""

812

813 def test_copy_many_files_in_folder(self, temp_workspace):

814 """Test copying folder with many files"""

815 src = os.path.join(temp_workspace, "many_files_src")

816 dst = os.path.join(temp_workspace, "many_files_dst")

817

818 os.makedirs(src)

819

820 # Create 100 files

821 num_files = 100

822 for i in range(num_files):

823 with open(os.path.join(src, f"file_{i:03d}.txt"), 'w') as f:

824 f.write(f"content {i}")

825

826 thread = CopyDeleteThread(src=src, dst=dst, is_folder=True, copy=True)

827 thread.run()

828

829 # Verify all were copied

830 assert len(os.listdir(dst)) == num_files

831

832 def test_delete_folder_with_many_files(self, temp_workspace):

833 """Test deleting folder with many files"""

834 folder = os.path.join(temp_workspace, "many_files_delete")

835 os.makedirs(folder)

836

837 # Create many files

838 for i in range(50):

839 with open(os.path.join(folder, f"file_{i}.txt"), 'w') as f:

840 f.write("data")

841

842 thread = CopyDeleteThread(src=folder, is_folder=True, delete=True)

843 thread.run()

844

845 assert not os.path.exists(folder)

846

847 def test_copy_deep_nested_structure(self, temp_workspace):

848 """Test copying deeply nested structure"""

849 src = os.path.join(temp_workspace, "deep_src")

850 dst = os.path.join(temp_workspace, "deep_dst")

851

852 # Create deep structure

853 current = src

854 for i in range(10):

855 current = os.path.join(current, f"level_{i}")

856 os.makedirs(current)

857 with open(os.path.join(current, f"file_{i}.txt"), 'w') as f:

858 f.write(f"level {i}")

859

860 thread = CopyDeleteThread(src=src, dst=dst, is_folder=True, copy=True)

861 thread.run()

862

863 # Verify copied structure

864 assert os.path.exists(dst)

865

866 # Verify depth

867 current_check = dst

868 for i in range(10):

869 current_check = os.path.join(current_check, f"level_{i}")

870 assert os.path.exists(current_check)

871 assert os.path.exists(os.path.join(current_check, f"file_{i}.txt"))

872

873

874class TestRealWorldScenarios:

875 """Tests for realistic scenarios"""

876

877 def test_backup_workspace_folder(self, temp_workspace):

878 """Test backing up a workspace folder"""

879 # Simulate workspace with BIDS structure

880 workspace = os.path.join(temp_workspace, "workspace")

881 backup = os.path.join(temp_workspace, "backup")

882

883 # Create structure

884 for i in range(3):

885 subject_dir = os.path.join(workspace, f"sub-{i:02d}", "anat")

886 os.makedirs(subject_dir)

887 with open(os.path.join(subject_dir, "T1w.nii"), 'w') as f:

888 f.write(f"subject {i} brain data")

889

890 thread = CopyDeleteThread(src=workspace, dst=backup, is_folder=True, copy=True)

891 thread.run()

892

893 # Verify complete backup

894 assert os.path.exists(backup)

895 for i in range(3):

896 assert os.path.exists(

897 os.path.join(backup, f"sub-{i:02d}", "anat", "T1w.nii")

898 )

899

900 def test_cleanup_temp_derivatives(self, temp_workspace):

901 """Test cleaning up temporary derivatives"""

902 derivatives = os.path.join(temp_workspace, "derivatives", "temp_processing")

903 os.makedirs(derivatives)

904

905 # Create temporary files

906 for i in range(5):

907 with open(os.path.join(derivatives, f"temp_{i}.nii"), 'w') as f:

908 f.write("temp data")

909

910 thread = CopyDeleteThread(src=derivatives, is_folder=True, delete=True)

911 thread.run()

912

913 assert not os.path.exists(derivatives)

914

915 def test_move_processed_results(self, temp_workspace):

916 """Test moving processed results"""

917 processing = os.path.join(temp_workspace, "processing")

918 results = os.path.join(temp_workspace, "results")

919

920 os.makedirs(processing)

921 with open(os.path.join(processing, "output.nii"), 'w') as f:

922 f.write("processed brain")

923 with open(os.path.join(processing, "output.json"), 'w') as f:

924 f.write('{"processed": true}')

925

926 # Move = copy + delete

927 thread = CopyDeleteThread(

928 src=processing,

929 dst=results,

930 is_folder=True,

931 copy=True,

932 delete=True

933 )

934 thread.run()

935

936 # Results should exist

937 assert os.path.exists(results)

938 assert os.path.exists(os.path.join(results, "output.nii"))

939 assert os.path.exists(os.path.join(results, "output.json"))

940

941 # Processing folder should be deleted

942 assert not os.path.exists(processing)

943

944

945class TestPathNormalization:

946 """Tests for path normalization"""

947

948 def test_copy_with_trailing_slash(self, temp_workspace):

949 """Test copy with trailing slash in paths"""

950 src = os.path.join(temp_workspace, "source") + os.sep

951 dst = os.path.join(temp_workspace, "dest") + os.sep

952

953 os.makedirs(src.rstrip(os.sep))

954 with open(os.path.join(src, "file.txt"), 'w') as f:

955 f.write("data")

956

957 thread = CopyDeleteThread(src=src, dst=dst, is_folder=True, copy=True)

958 thread.run()

959

960 assert os.path.exists(dst.rstrip(os.sep))

961

962 def test_copy_with_relative_path(self, temp_workspace):

963 """Test copy with relative path"""

964 # Change directory temporarily

965 original_cwd = os.getcwd()

966 try:

967 os.chdir(temp_workspace)

968

969 src = "relative_source.txt"

970 dst = "relative_dest.txt"

971

972 with open(src, 'w') as f:

973 f.write("relative data")

974

975 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

976 thread.run()

977

978 assert os.path.exists(dst)

979 finally:

980 os.chdir(original_cwd)

981

982 def test_copy_with_absolute_path(self, temp_workspace):

983 """Test copy with absolute path"""

984 src = os.path.abspath(os.path.join(temp_workspace, "abs_source.txt"))

985 dst = os.path.abspath(os.path.join(temp_workspace, "abs_dest.txt"))

986

987 with open(src, 'w') as f:

988 f.write("absolute path data")

989

990 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

991 thread.run()

992

993 assert os.path.exists(dst)

994

995

996class TestErrorMessages:

997 """Tests for error messages"""

998

999 def test_error_message_format(self, temp_workspace):

1000 """Test error message format"""

1001 src = os.path.join(temp_workspace, "missing.txt")

1002 dst = os.path.join(temp_workspace, "dest.txt")

1003

1004 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

1005

1006 error_msgs = []

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

1008

1009 thread.run()

1010

1011 assert len(error_msgs) == 1

1012 msg = error_msgs[0]

1013

1014 # Message should contain src, dst, and error description

1015 assert "src:" in msg.lower() or src in msg

1016 assert "dst:" in msg.lower() or dst in msg

1017

1018 def test_translatable_messages(self, temp_workspace):

1019 """Test that messages are translatable"""

1020 src = os.path.join(temp_workspace, "source.txt")

1021 dst = os.path.join(temp_workspace, "dest.txt")

1022

1023 with open(src, 'w') as f:

1024 f.write("data")

1025

1026 thread = CopyDeleteThread(src=src, dst=dst, is_folder=False, copy=True)

1027

1028 finished_msgs = []

1029 thread.finished.connect(lambda msg: finished_msgs.append(msg))

1030

1031 thread.run()

1032

1033 # Messages should come from QCoreApplication.translate

1034 # So they should be readable strings

1035 assert len(finished_msgs) == 1

1036 assert len(finished_msgs[0]) > 0

1037

1038

1039# Parametrized tests

1040@pytest.mark.parametrize("is_folder", [True, False])

1041def test_copy_parametrized_folder_flag(is_folder, temp_workspace):

1042 """Parametrized test for is_folder flag"""

1043 if is_folder:

1044 src = os.path.join(temp_workspace, "folder_src")

1045 dst = os.path.join(temp_workspace, "folder_dst")

1046 os.makedirs(src)

1047 with open(os.path.join(src, "file.txt"), 'w') as f:

1048 f.write("data")

1049 else:

1050 src = os.path.join(temp_workspace, "file_src.txt")

1051 dst = os.path.join(temp_workspace, "file_dst.txt")

1052 with open(src, 'w') as f:

1053 f.write("data")

1054

1055 thread = CopyDeleteThread(src=src, dst=dst, is_folder=is_folder, copy=True)

1056 thread.run()

1057

1058 assert os.path.exists(dst)

1059

1060

1061@pytest.mark.parametrize("operation,flag_name", [

1062 ("copy", "copy"),

1063 ("delete", "delete"),

1064])

1065def test_operation_flags_parametrized(operation, flag_name, temp_workspace):

1066 """Parametrized test for operation flags"""

1067 src = os.path.join(temp_workspace, f"{operation}_test.txt")

1068 with open(src, 'w') as f:

1069 f.write("data")

1070

1071 kwargs = {

1072 'src': src,

1073 'is_folder': False,

1074 flag_name: True

1075 }

1076

1077 if operation == "copy":

1078 kwargs['dst'] = os.path.join(temp_workspace, "dest.txt")

1079

1080 thread = CopyDeleteThread(**kwargs)

1081

1082 assert getattr(thread, flag_name) is True

1083

1084

1085if __name__ == "__main__":

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

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

589 statements