Coverage for main/threads/utils_threads.py: 100%

1import os 

2import shutil 

3 

4from PyQt6.QtCore import QThread, pyqtSignal, QCoreApplication 

5 

6 

7class CopyDeleteThread(QThread): 

8 """ 

9 Background worker thread for performing **file or directory copy and delete operations** 

10 without blocking the PyQt GUI. 

11 

12 This class provides an asynchronous way to execute potentially long file system tasks 

13 (copying large files, removing folders, etc.) in a background thread, while emitting 

14 progress and error signals that can be handled by the main thread safely. 

15 

16 --- 

17 

18 **Signals:** 

19 

20 - `finished (str)`: Emitted when the operation completes successfully with a message. 

21 - `error (str)`: Emitted when an error occurs, providing an error message. 

22 

23 --- 

24 

25 **Parameters:** 

26 

27 - `src` (*str*): Source file or folder path. 

28 - `dst` (*str | None*): Destination path (required for copy operations). 

29 - `is_folder` (*bool*): If `True`, the operation targets a folder instead of a file. 

30 - `copy` (*bool*): If `True`, performs a copy operation. 

31 - `delete` (*bool*): If `True`, performs a delete operation. 

32 """ 

33 

34 # Signal emitted when the operation finishes successfully. 

35 finished = pyqtSignal(str) 

36 """**Signal(str):** Emitted when the operation completes successfully.  

37 The parameter is a status message describing the result or outcome. 

38 """ 

39 

40 # Signal emitted when an error occurs during execution. 

41 error = pyqtSignal(str) 

42 """**Signal(str):** Emitted when an error occurs during the operation.  

43 The parameter contains a descriptive error message. 

44 """ 

45 

46 def __init__(self, src, dst=None, is_folder=False, copy=False, delete=False): 

47 """ 

48 Initialize the worker thread with operation parameters. 

49 

50 Args: 

51 src (str): Path to the source file or directory. 

52 dst (str | None): Destination path for copy operations. 

53 is_folder (bool): Whether the source represents a folder. 

54 copy (bool): Whether to perform a copy operation. 

55 delete (bool): Whether to perform a delete operation. 

56 """ 

57 super().__init__() 

58 self.src = src 

59 self.dst = dst 

60 self.is_folder = is_folder 

61 self.copy = copy 

62 self.delete = delete 

63 

64 def run(self): 

65 """ 

66 Executes the file system operation. 

67 

68 Depending on the configuration flags (`copy`, `delete`, `is_folder`), 

69 this method performs one or both operations: 

70 - **Copy**: Copies a file or directory from `src` to `dst`. 

71 - **Delete**: Removes a file or directory at `src`. 

72 

73 Emits: 

74 - `finished (str)`: On successful completion. 

75 - `error (str)`: On any exception during execution. 

76 """ 

77 try: 

78 # --- COPY OPERATION --- 

79 if self.copy: 

80 if self.src is None or self.dst is None: 

81 raise ValueError(QCoreApplication.translate("Threads", "Missing src or dst")) 

82 

83 if self.is_folder: 

84 # Copy entire directory recursively 

85 shutil.copytree(self.src, self.dst, dirs_exist_ok=True) 

86 else: 

87 # Copy single file 

88 shutil.copy(self.src, self.dst) 

89 

90 self.finished.emit( 

91 QCoreApplication.translate( 

92 "Threads", 

93 "Successfully copied {0} to {1}" 

94 ).format(self.src, self.dst) 

95 ) 

96 

97 # --- DELETE OPERATION --- 

98 if self.delete: 

99 if self.src is None: 

100 raise ValueError(QCoreApplication.translate("Threads", "Missing src")) 

101 

102 if self.is_folder: 

103 # Remove entire folder tree 

104 shutil.rmtree(self.src) 

105 else: 

106 # Remove single file 

107 os.remove(self.src) 

108 

109 self.finished.emit( 

110 QCoreApplication.translate( 

111 "Threads", 

112 "Successfully deleted {0}" 

113 ).format(self.src) 

114 ) 

115 

116 except Exception as e: 

117 # Emit a formatted error message including source, destination, and exception details 

118 self.error.emit( 

119 QCoreApplication.translate( 

120 "Threads", 

121 "Error src:{0}, dst:{1}, {2}" 

122 ).format(self.src, self.dst, e) 

123 )