Coverage for plugins/module_utils/common.py: 48%

1# -*- coding: utf-8 -*- 

2 

3# Copyright (c) 2024 Simon Dodsley, <simon@everpuredata.com> 

4# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) 

5 

6from __future__ import absolute_import, division, print_function 

7 

8__metaclass__ = type 

9 

10""" 

11This module adds shared functions for the FlashArray modules 

12""" 

13 

14import platform 

15import os 

16import re 

17 

18from ansible.module_utils.common.process import get_bin_path 

19from ansible.module_utils.facts.utils import get_file_content 

20 

21from ansible_collections.everpure.flashblade.plugins.module_utils.version import ( 

22 LooseVersion, 

23) 

24 

25 

26def _findstr(text, match): 

27 """Find first line in text containing match string. 

28 

29 Args: 

30 text: Multi-line text to search 

31 match: String to search for 

32 

33 Returns: 

34 First line containing match, or None if not found 

35 """ 

36 for line in text.splitlines(): 

37 if match in line: 

38 return line 

39 return None 

40 

41 

42def remove_duplicates(items): 

43 """Remove duplicates from a list while preserving order. 

44 

45 Args: 

46 items: List that may contain duplicates 

47 

48 Returns: 

49 List with duplicates removed, order preserved 

50 """ 

51 return list(dict.fromkeys(items)) 

52 

53 

54def get_error_message(response, default="Unknown error"): 

55 """Safely extract error message from API response. 

56 

57 Safely retrieves the first error message from a FlashBlade API response 

58 object, avoiding IndexError if the errors list is empty or missing. 

59 

60 Args: 

61 response: API response object with potential errors attribute 

62 default: Default message to return if no error found (default: 'Unknown error') 

63 

64 Returns: 

65 str: Error message from response.errors[0].message, or default if unavailable 

66 

67 Examples: 

68 >>> # Safe - handles empty errors list 

69 >>> res = blade.get_file_systems(names=['nonexistent']) 

70 >>> error_msg = get_error_message(res) 

71 >>> module.fail_json(msg=f"Failed to get filesystem: {error_msg}") 

72 >>> 

73 >>> # With custom default message 

74 >>> error_msg = get_error_message(res, "Operation failed") 

75 

76 Note: 

77 This function prevents IndexError crashes that occur when accessing 

78 res.errors[0].message directly without checking if errors exist. 

79 """ 

80 if hasattr(response, "errors") and response.errors: 

81 errors = list(response.errors) 

82 if errors: 

83 return errors[0].message 

84 return default 

85 

86 

87def get_filesystem(module, blade): 

88 """Get filesystem from FlashBlade. 

89 

90 Retrieves a filesystem by name, with optional context support for 

91 multi-tenancy (API version 2.17+). 

92 

93 Args: 

94 module: Ansible module object with params['name'] and optional params['context'] 

95 blade: FlashBlade client object 

96 

97 Returns: 

98 Filesystem object if found, None otherwise 

99 

100 Example: 

101 >>> fs = get_filesystem(module, blade) 

102 >>> if fs: 

103 >>> print(fs.name) 

104 """ 

105 CONTEXT_API_VERSION = "2.17" 

106 

107 api_version = list(blade.get_versions().items) 

108 if CONTEXT_API_VERSION in api_version and module.params.get("context"): 

109 res = blade.get_file_systems( 

110 names=[module.params["name"]], context_names=[module.params["context"]] 

111 ) 

112 else: 

113 res = blade.get_file_systems(names=[module.params["name"]]) 

114 

115 if res.status_code == 200: 

116 items = list(res.items) 

117 if items: 

118 return items[0] 

119 return None 

120 

121 

122def get_rest_api_version(blade): 

123 """Return the highest REST API version the FlashBlade supports as a string. 

124 

125 Mirrors FlashArray's ``array.get_rest_version()``, which pypureclient does 

126 not expose for FlashBlade. ``blade.get_versions().items`` is the full list 

127 of supported versions; this helper picks the highest using LooseVersion 

128 ordering so the result can be fed straight into LooseVersion comparisons. 

129 """ 

130 

131 versions = list(blade.get_versions().items) 

132 return str(max(versions, key=LooseVersion)) 

133 

134 

135def human_to_bytes(size): 

136 """Convert human-readable byte string to bytes. 

137 

138 Converts strings with size suffixes (K, M, G, T, P, B) to integer bytes. 

139 

140 Args: 

141 size: Human-readable size string (e.g., '2G', '30M', '512K') 

142 

143 Returns: 

144 int: Number of bytes, or 0 if format is invalid 

145 

146 Examples: 

147 >>> human_to_bytes('2G') 

148 2147483648 

149 >>> human_to_bytes('512M') 

150 536870912 

151 >>> human_to_bytes('1K') 

152 1024 

153 """ 

154 bytes = size[:-1] 

155 unit = size[-1].upper() 

156 if bytes.isdigit(): 

157 bytes = int(bytes) 

158 if unit == "P": 

159 bytes *= 1125899906842624 

160 elif unit == "T": 

161 bytes *= 1099511627776 

162 elif unit == "G": 

163 bytes *= 1073741824 

164 elif unit == "M": 

165 bytes *= 1048576 

166 elif unit == "K": 

167 bytes *= 1024 

168 elif unit == "B": 

169 bytes *= 1 

170 else: 

171 bytes = 0 

172 else: 

173 bytes = 0 

174 return bytes 

175 

176 

177def human_to_real(iops): 

178 """Convert human-readable IOPS string to real number. 

179 

180 Converts strings with K (thousands) or M (millions) suffixes to integers. 

181 

182 Args: 

183 iops: Human-readable IOPS string (e.g., '2K', '30M', '5000') 

184 

185 Returns: 

186 int: Real number of IOPS, or 0 if format is invalid 

187 

188 Examples: 

189 >>> human_to_real('2K') 

190 2000 

191 >>> human_to_real('30M') 

192 30000000 

193 >>> human_to_real('5000') 

194 5000 

195 """ 

196 digit = iops[:-1] 

197 unit = iops[-1].upper() 

198 if unit.isdigit(): 

199 digit = iops 

200 elif digit.isdigit(): 

201 digit = int(digit) 

202 if unit == "M": 

203 digit *= 1000000 

204 elif unit == "K": 

205 digit *= 1000 

206 else: 

207 digit = 0 

208 else: 

209 digit = 0 

210 return digit 

211 

212 

213def get_local_tz(module, timezone="UTC"): 

214 """Get local timezone of the server running the module. 

215 

216 Attempts to detect the system timezone using platform-specific methods. 

217 Falls back to UTC if detection fails or platform is unsupported. 

218 

219 Supported platforms: 

220 - Linux (via timedatectl or /etc/timezone) 

221 - SunOS (via /etc/default/init) 

222 - Darwin/macOS (via systemsetup) 

223 - BSD variants (via /etc/timezone) 

224 - AIX 6.1+ (via /etc/environment) 

225 

226 Args: 

227 module: Ansible module object for running commands and warnings 

228 timezone: Default timezone if detection fails (default: 'UTC') 

229 

230 Returns: 

231 str: Timezone string (e.g., 'America/New_York', 'UTC') 

232 

233 Note: 

234 Windows is not supported and will always return UTC. 

235 Linux has been tested; other operating systems should work but may 

236 fall back to UTC on failure. 

237 """ 

238 if platform.system() == "Linux": 

239 timedatectl = get_bin_path("timedatectl") 

240 if timedatectl is not None: 

241 rcode, stdout, stderr = module.run_command(timedatectl) 

242 if rcode == 0 and stdout: 

243 line = _findstr(stdout, "Time zone") 

244 full_tz = line.split(":", 1)[1].rstrip() 

245 timezone = full_tz.split()[0] 

246 return timezone 

247 else: 

248 module.warn("Incorrect timedatectl output. Timezone will be set to UTC") 

249 else: 

250 if os.path.exists("/etc/timezone"): 

251 timezone = get_file_content("/etc/timezone") 

252 else: 

253 module.warn("Could not find /etc/timezone. Assuming UTC") 

254 

255 elif platform.system() == "SunOS": 

256 if os.path.exists("/etc/default/init"): 

257 for line in get_file_content("/etc/default/init", "").splitlines(): 

258 if line.startswith("TZ="): 

259 timezone = line.split("=", 1)[1] 

260 return timezone 

261 else: 

262 module.warn("Could not find /etc/default/init. Assuming UTC") 

263 

264 elif re.match("^Darwin", platform.platform()): 

265 systemsetup = get_bin_path("systemsetup") 

266 if systemsetup is not None: 

267 rcode, stdout, stderr = module.execute(systemsetup, "-gettimezone") 

268 if rcode == 0 and stdout: 

269 timezone = stdout.split(":", 1)[1].lstrip() 

270 else: 

271 module.warn("Could not run systemsetup. Assuming UTC") 

272 else: 

273 module.warn("Could not find systemsetup. Assuming UTC") 

274 

275 elif re.match("^(Free|Net|Open)BSD", platform.platform()): 

276 if os.path.exists("/etc/timezone"): 

277 timezone = get_file_content("/etc/timezone") 

278 else: 

279 module.warn("Could not find /etc/timezone. Assuming UTC") 

280 

281 elif platform.system() == "AIX": 

282 aix_oslevel = int(platform.version() + platform.release()) 

283 if aix_oslevel >= 61: 

284 if os.path.exists("/etc/environment"): 

285 for line in get_file_content("/etc/environment", "").splitlines(): 

286 if line.startswith("TZ="): 

287 timezone = line.split("=", 1)[1] 

288 return timezone 

289 else: 

290 module.warn("Could not find /etc/environment. Assuming UTC") 

291 else: 

292 module.warn( 

293 "Cannot determine timezone when AIX os level < 61. Assuming UTC" 

294 ) 

295 else: 

296 module.warn("Could not find /etc/timezone. Assuming UTC") 

297 return timezone