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

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: 

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 # //////////////////////////////////////////////// 

138 # PRIVATE METHODS 

139 # //////////////////////////////////////////////// 

140 

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

142 """ 

143 Upload a single file. 

144 

145 Args: 

146 source_path: Source file path 

147 dest_path: Destination file path 

148 

149 Note: 

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

151 """ 

152 shutil.copy2(source_path, dest_path) 

153 

154 # Preserve permissions if configured 

155 if self._config["preserve_permissions"]: 

156 with contextlib.suppress(OSError, AttributeError): 

157 shutil.copystat(source_path, dest_path) 

158 

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

160 """ 

161 Upload a directory recursively. 

162 

163 Args: 

164 source_path: Source directory path 

165 dest_path: Destination directory path 

166 

167 Note: 

168 Removes destination if it exists and overwrite is enabled. 

169 """ 

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

171 shutil.rmtree(dest_path) 

172 

173 shutil.copytree( 

174 source_path, 

175 dest_path, 

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

177 copy_function=( 

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

179 ), 

180 ) 

181 

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

183 """ 

184 Create a backup of an existing file. 

185 

186 Args: 

187 file_path: Path to file to backup 

188 

189 Note: 

190 Uses UploaderUtils to generate unique backup path. 

191 """ 

192 backup_path = UploaderUtils.generate_backup_path(file_path) 

193 shutil.copy2(file_path, backup_path) 

194 

195 # //////////////////////////////////////////////// 

196 # VALIDATION METHODS 

197 # //////////////////////////////////////////////// 

198 

199 def _validate_config(self) -> None: 

200 """ 

201 Validate disk uploader configuration. 

202 

203 Raises: 

204 UploadError: If configuration is invalid 

205 

206 Note: 

207 Ensures all required boolean flags are present and valid. 

208 """ 

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

210 

211 for key in required_keys: 

212 if key not in self._config: 

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

214 

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

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