Coverage for src/ezcompiler/adapters/_disk_uploader.py: 43.53%

1# /////////////////////////////////////////////////////////////// 

2# DISK_UPLOADER - Local disk uploader implementation 

3# Project: ezcompiler 

4# /////////////////////////////////////////////////////////////// 

5 

6""" 

7Disk uploader - Local disk upload handler for EzCompiler. 

8 

9This module provides functionality for uploading files and directories to 

10local disk locations, with support for backup creation, permission preservation, 

11and overwrite control. 

12 

13Note: Protocols layer should not perform logging directly. Logging is handled 

14by the service layer that orchestrates upload operations. 

15""" 

16 

17from __future__ import annotations 

18 

19# /////////////////////////////////////////////////////////////// 

20# IMPORTS 

21# /////////////////////////////////////////////////////////////// 

22# Standard library imports 

23import contextlib 

24import shutil 

25from pathlib import Path 

26from typing import Any 

27 

28# Local imports 

29from ..shared.exceptions import UploadError 

30from ..utils import FileUtils, UploaderUtils 

31from .base_uploader import BaseUploader 

32 

33# /////////////////////////////////////////////////////////////// 

34# CLASSES 

35# /////////////////////////////////////////////////////////////// 

36 

37 

38class DiskUploader(BaseUploader): 

39 """ 

40 Uploader for local disk operations. 

41 

42 Handles copying files and directories to local disk locations with 

43 configurable behavior for permissions, overwrites, and backups. 

44 

45 Configuration keys: 

46 preserve_permissions (bool): Preserve file permissions (default: True) 

47 overwrite (bool): Allow overwriting existing files (default: True) 

48 create_backup (bool): Create backup before overwrite (default: False) 

49 

50 Example: 

51 >>> config = {"preserve_permissions": True, "overwrite": True} 

52 >>> uploader = DiskUploader(config) 

53 >>> uploader.upload(Path("source.zip"), "/path/to/destination.zip") 

54 """ 

55 

56 # //////////////////////////////////////////////// 

57 # INITIALIZATION 

58 # //////////////////////////////////////////////// 

59 

60 def __init__(self, config: dict[str, Any] | None = None) -> None: 

61 """ 

62 Initialize the disk uploader. 

63 

64 Args: 

65 config: Optional configuration dictionary with keys: 

66 - preserve_permissions (bool): Preserve file permissions 

67 - overwrite (bool): Allow overwriting existing files 

68 - create_backup (bool): Create backup before overwrite 

69 """ 

70 default_config = UploaderUtils.get_default_disk_config() 

71 

72 if config: 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true

73 default_config.update(config) 

74 

75 super().__init__(default_config) 

76 

77 # //////////////////////////////////////////////// 

78 # PUBLIC METHODS 

79 # //////////////////////////////////////////////// 

80 

81 def get_uploader_name(self) -> str: 

82 """ 

83 Get the name of this uploader. 

84 

85 Returns: 

86 str: Name of the uploader 

87 """ 

88 return "Disk Uploader" 

89 

90 def upload(self, source_path: Path, destination: str) -> None: 

91 """ 

92 Upload a file or directory to a local disk location. 

93 

94 Args: 

95 source_path: Path to the source file or directory 

96 destination: Destination path on local disk 

97 

98 Raises: 

99 UploadError: If upload fails 

100 

101 Note: 

102 Creates parent directories automatically if they don't exist. 

103 """ 

104 try: 

105 self._validate_source_path(source_path) 

106 dest_path = Path(destination) 

107 

108 # For file uploads, treat destination as a directory and 

109 # preserve the source filename (e.g., "releases/Nuitka" + "build.zip" 

110 # -> "releases/Nuitka/build.zip") 

111 if source_path.is_file(): 

112 FileUtils.create_directory_if_not_exists(dest_path) 

113 dest_path = dest_path / source_path.name 

114 else: 

115 FileUtils.create_directory_if_not_exists(dest_path) 

116 

117 # Handle existing destination 

118 if not self._config["overwrite"] and dest_path.exists(): 

119 if self._config["create_backup"]: 

120 self._create_backup(dest_path) 

121 else: 

122 raise UploadError( 

123 f"Destination already exists and overwrite is disabled: {dest_path}" 

124 ) 

125 

126 # Perform upload 

127 if source_path.is_file(): 

128 self._upload_file(source_path, dest_path) 

129 else: 

130 self._upload_directory(source_path, dest_path) 

131 

132 except UploadError: 

133 raise 

134 except Exception as e: 

135 raise UploadError(f"Disk upload failed: {e}") from e 

136 

137 def download(self, remote_source: str, local_dir: Path) -> None: 

138 """ 

139 Copy a remote disk tree into ``local_dir`` (recursive). 

140 

141 Args: 

142 remote_source: Filesystem path of the remote tree. 

143 local_dir: Local directory to populate. 

144 

145 Raises: 

146 UploadError: If the copy fails. 

147 

148 Note: 

149 Missing remote source is a no-op (first run: empty channel). 

150 """ 

151 try: 

152 source = Path(remote_source) 

153 if not source.exists(): 

154 return 

155 local_dir.mkdir(parents=True, exist_ok=True) 

156 shutil.copytree(source, local_dir, dirs_exist_ok=True) 

157 except Exception as e: 

158 raise UploadError(f"Disk download failed: {e}") from e 

159 

160 # //////////////////////////////////////////////// 

161 # PRIVATE METHODS 

162 # //////////////////////////////////////////////// 

163 

164 def _upload_file(self, source_path: Path, dest_path: Path) -> None: 

165 """ 

166 Upload a single file. 

167 

168 Args: 

169 source_path: Source file path 

170 dest_path: Destination file path 

171 

172 Note: 

173 Uses copy2 to preserve metadata, then optionally preserves permissions. 

174 """ 

175 shutil.copy2(source_path, dest_path) 

176 

177 # Preserve permissions if configured 

178 if self._config["preserve_permissions"]: 

179 with contextlib.suppress(OSError, AttributeError): 

180 shutil.copystat(source_path, dest_path) 

181 

182 def _upload_directory(self, source_path: Path, dest_path: Path) -> None: 

183 """ 

184 Upload a directory recursively. 

185 

186 Args: 

187 source_path: Source directory path 

188 dest_path: Destination directory path 

189 

190 Note: 

191 Removes destination if it exists and overwrite is enabled. 

192 """ 

193 if dest_path.exists() and self._config["overwrite"]: 

194 shutil.rmtree(dest_path) 

195 

196 shutil.copytree( 

197 source_path, 

198 dest_path, 

199 dirs_exist_ok=self._config["overwrite"], 

200 copy_function=( 

201 shutil.copy2 if self._config["preserve_permissions"] else shutil.copy 

202 ), 

203 ) 

204 

205 def _create_backup(self, file_path: Path) -> None: 

206 """ 

207 Create a backup of an existing file. 

208 

209 Args: 

210 file_path: Path to file to backup 

211 

212 Note: 

213 Uses UploaderUtils to generate unique backup path. 

214 """ 

215 backup_path = UploaderUtils.generate_backup_path(file_path) 

216 shutil.copy2(file_path, backup_path) 

217 

218 # //////////////////////////////////////////////// 

219 # VALIDATION METHODS 

220 # //////////////////////////////////////////////// 

221 

222 def _validate_config(self) -> None: 

223 """ 

224 Validate disk uploader configuration. 

225 

226 Raises: 

227 UploadError: If configuration is invalid 

228 

229 Note: 

230 Ensures all required boolean flags are present and valid. 

231 """ 

232 required_keys = ["preserve_permissions", "overwrite", "create_backup"] 

233 

234 for key in required_keys: 

235 if key not in self._config: 235 ↛ 236line 235 didn't jump to line 236 because the condition on line 235 was never true

236 raise UploadError(f"Missing required configuration key: {key}") 

237 

238 if not isinstance(self._config[key], bool): 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true

239 raise UploadError(f"Configuration key '{key}' must be a boolean")