Coverage for src/ezcompiler/utils/_uploader_utils.py: 96.67%

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

2# UPLOADER_UTILS - Uploader-specific utility functions 

3# Project: ezcompiler 

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

5 

6""" 

7Uploader utilities - Uploader-specific utility functions for EzCompiler. 

8 

9This module provides specialized utility functions for uploader operations, 

10including source path validation, configuration validation, and backup 

11operations. Uses thematic utils (FileUtils, ValidationUtils) internally. 

12 

13Utils layer can only use DEBUG and ERROR log levels. 

14""" 

15 

16from __future__ import annotations 

17 

18# /////////////////////////////////////////////////////////////// 

19# IMPORTS 

20# /////////////////////////////////////////////////////////////// 

21# Standard library imports 

22from collections.abc import Iterable 

23from pathlib import Path 

24from typing import Any 

25 

26# Local imports 

27from ..shared.exceptions.utils import ( 

28 ServerConfigError, 

29 SourcePathError, 

30 UploaderTypeError, 

31) 

32 

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

34# CLASSES 

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

36 

37 

38class UploaderUtils: 

39 """ 

40 Utility class for uploader-specific operations. 

41 

42 Provides static methods for uploader-related tasks such as source path 

43 validation, configuration validation, and backup operations. 

44 Uses thematic utils (FileUtils) internally. 

45 

46 Example: 

47 >>> UploaderUtils.validate_source_path(Path("file.zip")) 

48 >>> UploaderUtils.validate_upload_type("disk") 

49 >>> backup_path = UploaderUtils.generate_backup_path(Path("file.zip")) 

50 """ 

51 

52 # //////////////////////////////////////////////// 

53 # VALIDATION METHODS 

54 # //////////////////////////////////////////////// 

55 

56 @staticmethod 

57 def validate_source_path(source_path: Path) -> None: 

58 """ 

59 Validate that the source path exists and is accessible. 

60 

61 Args: 

62 source_path: Path to validate 

63 

64 Raises: 

65 SourcePathError: If source path is invalid or inaccessible 

66 

67 Note: 

68 Uses FileUtils for file/directory existence checks. 

69 

70 Example: 

71 >>> UploaderUtils.validate_source_path(Path("file.zip")) 

72 """ 

73 if not source_path.exists(): 

74 raise SourcePathError(f"Source path does not exist: {source_path}") 

75 

76 if not source_path.is_file() and not source_path.is_dir(): 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 raise SourcePathError( 

78 f"Source path is neither file nor directory: {source_path}" 

79 ) 

80 

81 @staticmethod 

82 def validate_upload_type(upload_type: str) -> None: 

83 """ 

84 Validate upload type string. 

85 

86 Args: 

87 upload_type: Upload type to validate 

88 

89 Raises: 

90 UploaderTypeError: If upload type is not supported 

91 

92 Example: 

93 >>> UploaderUtils.validate_upload_type("disk") 

94 >>> UploaderUtils.validate_upload_type("invalid") 

95 UploaderTypeError: Unsupported upload type: invalid 

96 """ 

97 valid_types = ["disk", "server", "r2"] 

98 if upload_type.lower() not in valid_types: 

99 raise UploaderTypeError(f"Unsupported upload type: {upload_type}") 

100 

101 @staticmethod 

102 def validate_server_url(server_url: str) -> None: 

103 """ 

104 Validate server URL format. 

105 

106 Args: 

107 server_url: Server URL to validate 

108 

109 Raises: 

110 ServerConfigError: If server URL is invalid 

111 

112 Example: 

113 >>> UploaderUtils.validate_server_url("https://example.com") 

114 >>> UploaderUtils.validate_server_url("invalid") 

115 ServerConfigError: server_url must start with http:// or https:// 

116 """ 

117 if not server_url: 

118 raise ServerConfigError("server_url is required") 

119 

120 if not server_url.startswith(("http://", "https://")): 

121 raise ServerConfigError("server_url must start with http:// or https://") 

122 

123 # //////////////////////////////////////////////// 

124 # BACKUP METHODS 

125 # //////////////////////////////////////////////// 

126 

127 @staticmethod 

128 def generate_backup_path(file_path: Path) -> Path: 

129 """ 

130 Generate a unique backup path for a file. 

131 

132 Args: 

133 file_path: Path to file that needs backup 

134 

135 Returns: 

136 Path: Unique backup path (with .backup suffix and counter if needed) 

137 

138 Example: 

139 >>> backup = UploaderUtils.generate_backup_path(Path("file.zip")) 

140 >>> print(backup) 

141 file.zip.backup 

142 """ 

143 backup_path = file_path.with_suffix(f"{file_path.suffix}.backup") 

144 counter = 1 

145 

146 # Find unique backup name 

147 while backup_path.exists(): 

148 backup_path = file_path.with_suffix(f"{file_path.suffix}.backup.{counter}") 

149 counter += 1 

150 

151 return backup_path 

152 

153 # //////////////////////////////////////////////// 

154 # CONFIGURATION HELPERS 

155 # //////////////////////////////////////////////// 

156 

157 @staticmethod 

158 def reject_unknown_config_keys( 

159 config: dict[str, Any] | None, 

160 allowed: Iterable[str], 

161 *, 

162 uploader: str, 

163 ) -> None: 

164 """ 

165 Refuse an upload configuration carrying an unrecognized key. 

166 

167 A typo in ``[tool.ezcompiler.upload]`` would otherwise be swallowed and 

168 leave the intended option at its default — nothing would distinguish 

169 "not configured" from "misconfigured" (a misspelled ``cert`` silently 

170 disables mTLS). 

171 

172 Args: 

173 config: Caller-supplied configuration, if any. 

174 allowed: Recognized keys for this uploader. 

175 uploader: Uploader name, for the error message. 

176 

177 Raises: 

178 ServerConfigError: If ``config`` holds a key outside ``allowed``. 

179 """ 

180 if not config: 

181 return 

182 unknown = sorted(set(config) - set(allowed)) 

183 if unknown: 

184 raise ServerConfigError( 

185 f"Unknown configuration key for the {uploader} uploader: " 

186 f"{', '.join(unknown)}. Accepted keys: " 

187 f"{', '.join(sorted(allowed))}." 

188 ) 

189 

190 @staticmethod 

191 def get_default_disk_config() -> dict[str, Any]: 

192 """ 

193 Get default configuration for disk uploader. 

194 

195 Returns: 

196 dict[str, Any]: Default disk uploader configuration 

197 

198 Example: 

199 >>> config = UploaderUtils.get_default_disk_config() 

200 >>> print(config) 

201 {'preserve_permissions': True, 'overwrite': True, 'create_backup': False} 

202 """ 

203 return { 

204 "preserve_permissions": True, 

205 "overwrite": True, 

206 "create_backup": False, 

207 } 

208 

209 @staticmethod 

210 def get_default_server_config() -> dict[str, Any]: 

211 """ 

212 Get default configuration for server uploader. 

213 

214 Returns: 

215 dict[str, Any]: Default server uploader configuration 

216 

217 Example: 

218 >>> config = UploaderUtils.get_default_server_config() 

219 >>> print(config) 

220 {'server_url': '', 'username': '', 'password': '', ...} 

221 """ 

222 return { 

223 "server_url": "", 

224 "username": "", 

225 "password": "", # nosec B105 - empty placeholder, not a hardcoded secret 

226 "api_key": "", 

227 "timeout": 30, 

228 "verify_ssl": True, 

229 "chunk_size": 8192, 

230 "retry_attempts": 3, 

231 "proxies": {}, 

232 "extra_headers": {}, 

233 "cert": None, 

234 }