Coverage for plugins/module_utils/common.py: 48%
115 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-10-30 17:06 -0400
1# -*- coding: utf-8 -*-
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)
6from __future__ import absolute_import, division, print_function
8__metaclass__ = type
10"""
11This module adds shared functions for the FlashArray modules
12"""
14import platform
15import os
16import re
18from ansible.module_utils.common.process import get_bin_path
19from ansible.module_utils.facts.utils import get_file_content
21from ansible_collections.everpure.flashblade.plugins.module_utils.version import (
22 LooseVersion,
23)
26def _findstr(text, match):
27 """Find first line in text containing match string.
29 Args:
30 text: Multi-line text to search
31 match: String to search for
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
42def remove_duplicates(items):
43 """Remove duplicates from a list while preserving order.
45 Args:
46 items: List that may contain duplicates
48 Returns:
49 List with duplicates removed, order preserved
50 """
51 return list(dict.fromkeys(items))
54def get_error_message(response, default="Unknown error"):
55 """Safely extract error message from API response.
57 Safely retrieves the first error message from a FlashBlade API response
58 object, avoiding IndexError if the errors list is empty or missing.
60 Args:
61 response: API response object with potential errors attribute
62 default: Default message to return if no error found (default: 'Unknown error')
64 Returns:
65 str: Error message from response.errors[0].message, or default if unavailable
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")
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
87def get_filesystem(module, blade):
88 """Get filesystem from FlashBlade.
90 Retrieves a filesystem by name, with optional context support for
91 multi-tenancy (API version 2.17+).
93 Args:
94 module: Ansible module object with params['name'] and optional params['context']
95 blade: FlashBlade client object
97 Returns:
98 Filesystem object if found, None otherwise
100 Example:
101 >>> fs = get_filesystem(module, blade)
102 >>> if fs:
103 >>> print(fs.name)
104 """
105 CONTEXT_API_VERSION = "2.17"
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"]])
115 if res.status_code == 200:
116 items = list(res.items)
117 if items:
118 return items[0]
119 return None
122def get_rest_api_version(blade):
123 """Return the highest REST API version the FlashBlade supports as a string.
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 """
131 versions = list(blade.get_versions().items)
132 return str(max(versions, key=LooseVersion))
135def human_to_bytes(size):
136 """Convert human-readable byte string to bytes.
138 Converts strings with size suffixes (K, M, G, T, P, B) to integer bytes.
140 Args:
141 size: Human-readable size string (e.g., '2G', '30M', '512K')
143 Returns:
144 int: Number of bytes, or 0 if format is invalid
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
177def human_to_real(iops):
178 """Convert human-readable IOPS string to real number.
180 Converts strings with K (thousands) or M (millions) suffixes to integers.
182 Args:
183 iops: Human-readable IOPS string (e.g., '2K', '30M', '5000')
185 Returns:
186 int: Real number of IOPS, or 0 if format is invalid
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
213def get_local_tz(module, timezone="UTC"):
214 """Get local timezone of the server running the module.
216 Attempts to detect the system timezone using platform-specific methods.
217 Falls back to UTC if detection fails or platform is unsupported.
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)
226 Args:
227 module: Ansible module object for running commands and warnings
228 timezone: Default timezone if detection fails (default: 'UTC')
230 Returns:
231 str: Timezone string (e.g., 'America/New_York', 'UTC')
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")
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")
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")
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")
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