Coverage for src / ezcompiler / adapters / disk_uploader.py: 22.67%

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.file_utils import FileUtils 

31from ..utils.uploader_utils import UploaderUtils 

32from .base_uploader import BaseUploader 

33 

34# /////////////////////////////////////////////////////////////// 

35# CLASSES 

36# /////////////////////////////////////////////////////////////// 

37 

38 

39class DiskUploader(BaseUploader): 

40 """ 

41 Uploader for local disk operations. 

42 

43 Handles copying files and directories to local disk locations with 

44 configurable behavior for permissions, overwrites, and backups. 

45 

46 Configuration keys: 

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

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

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

50 

51 Example: 

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

53 >>> uploader = DiskUploader(config) 

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

55 """ 

56 

57 # //////////////////////////////////////////////// 

58 # INITIALIZATION 

59 # //////////////////////////////////////////////// 

60 

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

62 """ 

63 Initialize the disk uploader. 

64 

65 Args: 

66 config: Optional configuration dictionary with keys: 

67 - preserve_permissions (bool): Preserve file permissions 

68 - overwrite (bool): Allow overwriting existing files 

69 - create_backup (bool): Create backup before overwrite 

70 """ 

71 default_config = UploaderUtils.get_default_disk_config() 

72 

73 if config: 

74 default_config.update(config) 

75 

76 super().__init__(default_config) 

77 

78 # //////////////////////////////////////////////// 

79 # PUBLIC METHODS 

80 # //////////////////////////////////////////////// 

81 

82 def get_uploader_name(self) -> str: 

83 """ 

84 Get the name of this uploader. 

85 

86 Returns: 

87 str: Name of the uploader 

88 """ 

89 return "Disk Uploader" 

90 

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

92 """ 

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

94 

95 Args: 

96 source_path: Path to the source file or directory 

97 destination: Destination path on local disk 

98 

99 Raises: 

100 UploadError: If upload fails 

101 

102 Note: 

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

104 """ 

105 try: 

106 self._validate_source_path(source_path) 

107 dest_path = Path(destination) 

108 

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

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

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

112 if source_path.is_file(): 

113 FileUtils.create_directory_if_not_exists(dest_path) 

114 dest_path = dest_path / source_path.name 

115 else: 

116 FileUtils.create_directory_if_not_exists(dest_path) 

117 

118 # Handle existing destination 

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

120 if self._config["create_backup"]: 

121 self._create_backup(dest_path) 

122 else: 

123 raise UploadError( 

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

125 ) 

126 

127 # Perform upload 

128 if source_path.is_file(): 

129 self._upload_file(source_path, dest_path) 

130 else: 

131 self._upload_directory(source_path, dest_path) 

132 

133 except UploadError: 

134 raise 

135 except Exception as e: 

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

137 

138 # //////////////////////////////////////////////// 

139 # PRIVATE METHODS 

140 # //////////////////////////////////////////////// 

141 

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

143 """ 

144 Upload a single file. 

145 

146 Args: 

147 source_path: Source file path 

148 dest_path: Destination file path 

149 

150 Note: 

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

152 """ 

153 shutil.copy2(source_path, dest_path) 

154 

155 # Preserve permissions if configured 

156 if self._config["preserve_permissions"]: 

157 with contextlib.suppress(OSError, AttributeError): 

158 shutil.copystat(source_path, dest_path) 

159 

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

161 """ 

162 Upload a directory recursively. 

163 

164 Args: 

165 source_path: Source directory path 

166 dest_path: Destination directory path 

167 

168 Note: 

169 Removes destination if it exists and overwrite is enabled. 

170 """ 

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

172 shutil.rmtree(dest_path) 

173 

174 shutil.copytree( 

175 source_path, 

176 dest_path, 

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

178 copy_function=( 

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

180 ), 

181 ) 

182 

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

184 """ 

185 Create a backup of an existing file. 

186 

187 Args: 

188 file_path: Path to file to backup 

189 

190 Note: 

191 Uses UploaderUtils to generate unique backup path. 

192 """ 

193 backup_path = UploaderUtils.generate_backup_path(file_path) 

194 shutil.copy2(file_path, backup_path) 

195 

196 # //////////////////////////////////////////////// 

197 # VALIDATION METHODS 

198 # //////////////////////////////////////////////// 

199 

200 def _validate_config(self) -> None: 

201 """ 

202 Validate disk uploader configuration. 

203 

204 Raises: 

205 UploadError: If configuration is invalid 

206 

207 Note: 

208 Ensures all required boolean flags are present and valid. 

209 """ 

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

211 

212 for key in required_keys: 

213 if key not in self._config: 

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

215 

216 if not isinstance(self._config[key], bool): 

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