diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-28 16:04:21 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-28 16:04:21 +0000 |
commit | 8a754e0858d922e955e71b253c139e071ecec432 (patch) | |
tree | 527d16e74bfd1840c85efd675fdecad056c54107 /lib/ansible/plugins/test/core.py | |
parent | Initial commit. (diff) | |
download | ansible-core-upstream/2.14.3.tar.xz ansible-core-upstream/2.14.3.zip |
Adding upstream version 2.14.3.upstream/2.14.3upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to '')
-rw-r--r-- | lib/ansible/plugins/test/core.py | 287 |
1 files changed, 287 insertions, 0 deletions
diff --git a/lib/ansible/plugins/test/core.py b/lib/ansible/plugins/test/core.py new file mode 100644 index 0000000..d9e7e8b --- /dev/null +++ b/lib/ansible/plugins/test/core.py @@ -0,0 +1,287 @@ +# (c) 2012, Jeroen Hoekx <jeroen@hoekx.be> +# +# This file is part of Ansible +# +# Ansible is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Ansible is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Ansible. If not, see <http://www.gnu.org/licenses/>. + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import re +import operator as py_operator + +from collections.abc import MutableMapping, MutableSequence + +from ansible.module_utils.compat.version import LooseVersion, StrictVersion + +from ansible import errors +from ansible.module_utils._text import to_native, to_text +from ansible.module_utils.parsing.convert_bool import boolean +from ansible.utils.display import Display +from ansible.utils.version import SemanticVersion + +try: + from packaging.version import Version as PEP440Version + HAS_PACKAGING = True +except ImportError: + HAS_PACKAGING = False + +display = Display() + + +def failed(result): + ''' Test if task result yields failed ''' + if not isinstance(result, MutableMapping): + raise errors.AnsibleFilterError("The 'failed' test expects a dictionary") + return result.get('failed', False) + + +def success(result): + ''' Test if task result yields success ''' + return not failed(result) + + +def unreachable(result): + ''' Test if task result yields unreachable ''' + if not isinstance(result, MutableMapping): + raise errors.AnsibleFilterError("The 'unreachable' test expects a dictionary") + return result.get('unreachable', False) + + +def reachable(result): + ''' Test if task result yields reachable ''' + return not unreachable(result) + + +def changed(result): + ''' Test if task result yields changed ''' + if not isinstance(result, MutableMapping): + raise errors.AnsibleFilterError("The 'changed' test expects a dictionary") + if 'changed' not in result: + changed = False + if ( + 'results' in result and # some modules return a 'results' key + isinstance(result['results'], MutableSequence) and + isinstance(result['results'][0], MutableMapping) + ): + for res in result['results']: + if res.get('changed', False): + changed = True + break + else: + changed = result.get('changed', False) + return changed + + +def skipped(result): + ''' Test if task result yields skipped ''' + if not isinstance(result, MutableMapping): + raise errors.AnsibleFilterError("The 'skipped' test expects a dictionary") + return result.get('skipped', False) + + +def started(result): + ''' Test if async task has started ''' + if not isinstance(result, MutableMapping): + raise errors.AnsibleFilterError("The 'started' test expects a dictionary") + if 'started' in result: + # For async tasks, return status + # NOTE: The value of started is 0 or 1, not False or True :-/ + return result.get('started', 0) == 1 + else: + # For non-async tasks, warn user, but return as if started + display.warning("The 'started' test expects an async task, but a non-async task was tested") + return True + + +def finished(result): + ''' Test if async task has finished ''' + if not isinstance(result, MutableMapping): + raise errors.AnsibleFilterError("The 'finished' test expects a dictionary") + if 'finished' in result: + # For async tasks, return status + # NOTE: The value of finished is 0 or 1, not False or True :-/ + return result.get('finished', 0) == 1 + else: + # For non-async tasks, warn user, but return as if finished + display.warning("The 'finished' test expects an async task, but a non-async task was tested") + return True + + +def regex(value='', pattern='', ignorecase=False, multiline=False, match_type='search'): + ''' Expose `re` as a boolean filter using the `search` method by default. + This is likely only useful for `search` and `match` which already + have their own filters. + ''' + # In addition to ensuring the correct type, to_text here will ensure + # _fail_with_undefined_error happens if the value is Undefined + value = to_text(value, errors='surrogate_or_strict') + flags = 0 + if ignorecase: + flags |= re.I + if multiline: + flags |= re.M + _re = re.compile(pattern, flags=flags) + return bool(getattr(_re, match_type, 'search')(value)) + + +def vault_encrypted(value): + """Evaulate whether a variable is a single vault encrypted value + + .. versionadded:: 2.10 + """ + return getattr(value, '__ENCRYPTED__', False) and value.is_encrypted() + + +def match(value, pattern='', ignorecase=False, multiline=False): + ''' Perform a `re.match` returning a boolean ''' + return regex(value, pattern, ignorecase, multiline, 'match') + + +def search(value, pattern='', ignorecase=False, multiline=False): + ''' Perform a `re.search` returning a boolean ''' + return regex(value, pattern, ignorecase, multiline, 'search') + + +def version_compare(value, version, operator='eq', strict=None, version_type=None): + ''' Perform a version comparison on a value ''' + op_map = { + '==': 'eq', '=': 'eq', 'eq': 'eq', + '<': 'lt', 'lt': 'lt', + '<=': 'le', 'le': 'le', + '>': 'gt', 'gt': 'gt', + '>=': 'ge', 'ge': 'ge', + '!=': 'ne', '<>': 'ne', 'ne': 'ne' + } + + type_map = { + 'loose': LooseVersion, + 'strict': StrictVersion, + 'semver': SemanticVersion, + 'semantic': SemanticVersion, + 'pep440': PEP440Version, + } + + if strict is not None and version_type is not None: + raise errors.AnsibleFilterError("Cannot specify both 'strict' and 'version_type'") + + if not value: + raise errors.AnsibleFilterError("Input version value cannot be empty") + + if not version: + raise errors.AnsibleFilterError("Version parameter to compare against cannot be empty") + + if version_type == 'pep440' and not HAS_PACKAGING: + raise errors.AnsibleFilterError("The pep440 version_type requires the Python 'packaging' library") + + Version = LooseVersion + if strict: + Version = StrictVersion + elif version_type: + try: + Version = type_map[version_type] + except KeyError: + raise errors.AnsibleFilterError( + "Invalid version type (%s). Must be one of %s" % (version_type, ', '.join(map(repr, type_map))) + ) + + if operator in op_map: + operator = op_map[operator] + else: + raise errors.AnsibleFilterError( + 'Invalid operator type (%s). Must be one of %s' % (operator, ', '.join(map(repr, op_map))) + ) + + try: + method = getattr(py_operator, operator) + return method(Version(to_text(value)), Version(to_text(version))) + except Exception as e: + raise errors.AnsibleFilterError('Version comparison failed: %s' % to_native(e)) + + +def truthy(value, convert_bool=False): + """Evaluate as value for truthiness using python ``bool`` + + Optionally, attempt to do a conversion to bool from boolean like values + such as ``"false"``, ``"true"``, ``"yes"``, ``"no"``, ``"on"``, ``"off"``, etc. + + .. versionadded:: 2.10 + """ + if convert_bool: + try: + value = boolean(value) + except TypeError: + pass + + return bool(value) + + +def falsy(value, convert_bool=False): + """Evaluate as value for falsiness using python ``bool`` + + Optionally, attempt to do a conversion to bool from boolean like values + such as ``"false"``, ``"true"``, ``"yes"``, ``"no"``, ``"on"``, ``"off"``, etc. + + .. versionadded:: 2.10 + """ + return not truthy(value, convert_bool=convert_bool) + + +class TestModule(object): + ''' Ansible core jinja2 tests ''' + + def tests(self): + return { + # failure testing + 'failed': failed, + 'failure': failed, + 'succeeded': success, + 'success': success, + 'successful': success, + 'reachable': reachable, + 'unreachable': unreachable, + + # changed testing + 'changed': changed, + 'change': changed, + + # skip testing + 'skipped': skipped, + 'skip': skipped, + + # async testing + 'finished': finished, + 'started': started, + + # regex + 'match': match, + 'search': search, + 'regex': regex, + + # version comparison + 'version_compare': version_compare, + 'version': version_compare, + + # lists + 'any': any, + 'all': all, + + # truthiness + 'truthy': truthy, + 'falsy': falsy, + + # vault + 'vault_encrypted': vault_encrypted, + } |