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

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 UploaderUtils.reject_unknown_config_keys( 

72 config, 

73 # destination_path is injected by UploaderService.upload(). 

74 (*default_config, "destination_path"), 

75 uploader="disk", 

76 ) 

77 

78 if config: 

79 default_config.update(config) 

80 

81 super().__init__(default_config) 

82 

83 # //////////////////////////////////////////////// 

84 # PUBLIC METHODS 

85 # //////////////////////////////////////////////// 

86 

87 def get_uploader_name(self) -> str: 

88 """ 

89 Get the name of this uploader. 

90 

91 Returns: 

92 str: Name of the uploader 

93 """ 

94 return "Disk Uploader" 

95 

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

97 """ 

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

99 

100 Args: 

101 source_path: Path to the source file or directory 

102 destination: Destination path on local disk 

103 

104 Raises: 

105 UploadError: If upload fails 

106 

107 Note: 

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

109 """ 

110 try: 

111 self._validate_source_path(source_path) 

112 dest_path = Path(destination) 

113 

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

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

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

117 if source_path.is_file(): 

118 FileUtils.create_directory_if_not_exists(dest_path) 

119 dest_path = dest_path / source_path.name 

120 else: 

121 FileUtils.create_directory_if_not_exists(dest_path) 

122 

123 # Handle existing destination 

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

125 if self._config["create_backup"]: 

126 self._create_backup(dest_path) 

127 else: 

128 raise UploadError( 

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

130 ) 

131 

132 # Perform upload 

133 if source_path.is_file(): 

134 self._upload_file(source_path, dest_path) 

135 else: 

136 self._upload_directory(source_path, dest_path) 

137 

138 except UploadError: 

139 raise 

140 except Exception as e: 

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

142 

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

144 """ 

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

146 

147 Args: 

148 remote_source: Filesystem path of the remote tree. 

149 local_dir: Local directory to populate. 

150 

151 Raises: 

152 UploadError: If the copy fails. 

153 

154 Note: 

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

156 """ 

157 try: 

158 source = Path(remote_source) 

159 if not source.exists(): 

160 return 

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

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

163 except Exception as e: 

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

165 

166 # //////////////////////////////////////////////// 

167 # PRIVATE METHODS 

168 # //////////////////////////////////////////////// 

169 

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

171 """ 

172 Upload a single file. 

173 

174 Args: 

175 source_path: Source file path 

176 dest_path: Destination file path 

177 

178 Note: 

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

180 """ 

181 shutil.copy2(source_path, dest_path) 

182 

183 # Preserve permissions if configured 

184 if self._config["preserve_permissions"]: 

185 with contextlib.suppress(OSError, AttributeError): 

186 shutil.copystat(source_path, dest_path) 

187 

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

189 """ 

190 Upload a directory recursively. 

191 

192 Args: 

193 source_path: Source directory path 

194 dest_path: Destination directory path 

195 

196 Note: 

197 Removes destination if it exists and overwrite is enabled. 

198 """ 

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

200 shutil.rmtree(dest_path) 

201 

202 shutil.copytree( 

203 source_path, 

204 dest_path, 

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

206 copy_function=( 

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

208 ), 

209 ) 

210 

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

212 """ 

213 Create a backup of an existing file. 

214 

215 Args: 

216 file_path: Path to file to backup 

217 

218 Note: 

219 Uses UploaderUtils to generate unique backup path. 

220 """ 

221 backup_path = UploaderUtils.generate_backup_path(file_path) 

222 shutil.copy2(file_path, backup_path) 

223 

224 # //////////////////////////////////////////////// 

225 # VALIDATION METHODS 

226 # //////////////////////////////////////////////// 

227 

228 def _validate_config(self) -> None: 

229 """ 

230 Validate disk uploader configuration. 

231 

232 Raises: 

233 UploadError: If configuration is invalid 

234 

235 Note: 

236 Ensures all required boolean flags are present and valid. 

237 """ 

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

239 

240 for key in required_keys: 

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

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

243 

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

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