module

Krikri::PythonModuleRunner

Constants

ANSIBLE_RELEASE_PY_SHIM = "__version__ = '#{ANSIBLE_TARGET_VERSION}'\n__author__ = 'Ansible, Inc.'\n__codename__ = 'What Is and What Should Never Be'"

ansible-core's ansible/module_utils/ansible_release.py is the running release's own metadata, and modules read __version__ out of it (a module built on amazon.aws' module_utils, which import it, failed with "No module named 'ansible.module_utils.ansible_release'" under the shim bundle - issue #843 - while ansible-playbook, whose AnsiballZ payload bundles the file, ran it). The version is the ansible-core release this engine reproduces (ANSIBLE_TARGET_VERSION, the same value the ansible_version magic variable reports) rather than a second literal, so it tracks a compatibility-target bump.

BASIC_PY_SHIM = "import json\nimport os\nimport platform\nimport shlex\nimport subprocess\nimport sys\nimport tempfile\n\n\n# Real basic.py ships this import-gate helper beside AnsibleModule,\n# and role-private modules import both names in one line\n# (linux-system-roles.selinux's library/selinux_modules_facts.py on\n# a target without python3-libsemanage, issue #557): without it\n# here the import itself raised ImportError, the module printed no\n# result JSON and the task died as MODULE FAILURE where real's\n# bundled basic.py lets the module fail through fail_json. Verbatim\n# from ansible-core 2.19's basic.py so the composed text - library,\n# hostname, interpreter path, fixed advice - matches the [ERROR]\n# block real prints for the same failure.\ndef missing_required_lib(library, reason=None, url=None):\n hostname = platform.node()\n msg = \"Failed to import the required Python library (%s) on %s's Python %s.\" % (library, hostname, sys.executable)\n if reason:\n msg += \" This is required %s.\" % reason\n if url:\n msg += \" See %s for more info.\" % url\n msg += (\" Please read the module documentation and install it in the appropriate location.\"\n \" If the required library is installed, but Ansible is using the wrong Python interpreter,\"\n \" please consult the documentation on ansible_python_interpreter\")\n return msg\n\n\n# Real basic.py re-exports env_fallback from common.parameters beside\n# AnsibleModule, and modules import both names in one line to declare\n# argument-spec fallbacks (fallback=(env_fallback, ['MY_REGION'])) -\n# custom modules that declare env-var fallbacks the way the Ansible\n# docs teach fail at this very import otherwise (issue #752).\nclass AnsibleFallbackNotFound(Exception):\n pass\n\ndef env_fallback(*args, **kwargs):\n \"\"\"Load value from environment variable\"\"\"\n\n for arg in args:\n if arg in os.environ:\n return os.environ[arg]\n raise AnsibleFallbackNotFound\n\nclass AnsibleModule(object):\n def __init__(self, argument_spec=None, bypass_checks=False, no_log=False,\n supports_check_mode=False, **kwargs):\n self.argument_spec = argument_spec or {}\n self.supports_check_mode = supports_check_mode\n self._warnings = []\n self.tmpdir = tempfile.gettempdir()\n self.check_mode = False\n self.params = {}\n self._name = os.path.splitext(os.path.basename(sys.argv[0]))[0]\n raw_args = self._read_args()\n self.check_mode = bool(\n raw_args.pop('_ansible_check_mode', False)\n or os.environ.get('ANSIBLE_CHECK_MODE') == '1')\n # Real 2.21's action layer passes the INJECT_INVOCATION config\n # resolution with every module run and _return_formatted\n # injects or pops `invocation` off it (basic.py:\n # \"_ansible_inject_invocation\"); the shim mirrors that\n # instead of the pre-2.21 always-inject.\n self._inject_invocation = bool(raw_args.pop('_ansible_inject_invocation', False))\n # Real basic.py's PASS_VARS always exposes `_diff` (from\n # `_ansible_diff`, default False) on the instance, and\n # amazon.aws' AnsibleAWSModule copies `self._module._diff`\n # right after constructing its AnsibleModule - without the\n # attribute a collection-backed custom module died with\n # \"'AnsibleModule' object has no attribute '_diff'\" before\n # it ever looked at its params (issue #955).\n self._diff = bool(raw_args.pop('_ansible_diff', False))\n self._apply_argument_spec(raw_args)\n # Real basic.py closes __init__ with this same gate\n # (basic.py: \"if self.check_mode and not\n # self.supports_check_mode: self.exit_json(skipped=True, ...)\"),\n # placed AFTER argument validation: a module that never\n # declared supports_check_mode must never reach its own body\n # under --check - a check-mode run must not make real\n # changes, and this shim previously let the body run\n # (issue #693: the module's side effects happened during a\n # dry run).\n if self.check_mode and not self.supports_check_mode:\n self.exit_json(skipped=True, msg='remote module (%s) does not support check mode' % self._name)\n\n def _read_args(self):\n env_args = os.environ.get('ANSIBLE_MODULE_ARGS')\n if env_args:\n try:\n return json.loads(env_args)\n except ValueError:\n self.fail_json(msg='ANSIBLE_MODULE_ARGS env var is not valid JSON')\n raw = ''\n try:\n if not sys.stdin.isatty():\n raw = sys.stdin.read()\n except Exception:\n raw = ''\n raw = raw.strip()\n if not raw:\n return {}\n try:\n parsed = json.loads(raw)\n except ValueError:\n self.fail_json(msg='Failed to decode JSON module parameters.')\n if isinstance(parsed, dict) and 'ANSIBLE_MODULE_ARGS' in parsed:\n parsed = parsed['ANSIBLE_MODULE_ARGS']\n if not isinstance(parsed, dict):\n self.fail_json(msg='Module parameters must be a JSON object.')\n return parsed\n\n def _cast(self, name, value, spec):\n kind = spec.get('type', 'str')\n if kind == 'bool':\n if isinstance(value, bool):\n return value\n text = str(value).strip().lower()\n if text in ('yes', 'on', '1', 'true'):\n return True\n if text in ('no', 'off', '0', 'false', ''):\n return False\n self.fail_json(msg=\"argument '%s' is not a valid boolean\" % name)\n if kind == 'int':\n try:\n return int(value)\n except (TypeError, ValueError):\n self.fail_json(msg=\"argument '%s' is not a valid integer\" % name)\n if kind == 'float':\n try:\n return float(value)\n except (TypeError, ValueError):\n self.fail_json(msg=\"argument '%s' is not a valid float\" % name)\n if kind in ('dict', 'json'):\n if isinstance(value, dict):\n return value\n try:\n parsed = json.loads(value)\n except (TypeError, ValueError):\n self.fail_json(msg=\"argument '%s' is not valid JSON\" % name)\n if not isinstance(parsed, dict):\n self.fail_json(msg=\"argument '%s' is not a dict\" % name)\n return parsed\n if kind == 'list':\n if isinstance(value, list):\n return value\n try:\n parsed = json.loads(value)\n except (TypeError, ValueError):\n parsed = str(value).split(',')\n return parsed\n if kind == 'path':\n return os.path.expanduser(os.path.expandvars(str(value)))\n # Real check_type_str (common/validation.py): a str passes\n # through unchanged, None becomes '' (\"approximate pre-2.19\n # templating None->empty str equivalency\"), and anything else\n # - a real int included, which is exactly what a wide-int\n # param arrives as after the args' bigint marker unquoting -\n # is to_native()'d into its str(). Returning the value\n # unchanged instead left a str-typed param holding the int\n # the caller sent, diverging from real's coerced echo\n # (issue #915: this shim now serves targets that used to\n # reach their own real basic.py, where the coercion ran).\n # 'raw' deliberately falls through to the unchanged value -\n # real's raw type never converts.\n if kind == 'str':\n if isinstance(value, str):\n return value\n if value is None:\n return ''\n if isinstance(value, bytes):\n try:\n return value.decode('utf-8')\n except UnicodeDecodeError:\n return value.decode('utf-8', 'surrogateescape')\n return str(value)\n return value\n\n def _apply_argument_spec(self, raw_args):\n for key, spec in self.argument_spec.items():\n for alias in spec.get('aliases', []) or []:\n if alias in raw_args and key not in raw_args:\n raw_args[key] = raw_args[alias]\n # Real validates args through ModuleArgumentSpecValidator,\n # which runs set_fallbacks (common/parameters.py) BEFORE\n # _set_defaults: a param the caller did not pass gets its\n # fallback strategy's value (env_fallback reads the named\n # env vars), a not-found fallback leaves the param to the\n # default/None path, and an explicitly passed value never\n # consults the fallback at all - so basic.py's own\n # unsafe_writes takes ANSIBLE_UNSAFE_WRITES over its\n # default=False only when the env var is set (issue #752).\n for key, spec in self.argument_spec.items():\n if key in raw_args or 'fallback' not in spec:\n continue\n fallback = spec['fallback']\n if not fallback:\n continue\n fallback_args = []\n fallback_kwargs = {}\n for item in fallback[1:]:\n if isinstance(item, dict):\n fallback_kwargs = item\n else:\n fallback_args = item\n try:\n fallback_value = fallback[0](*fallback_args, **fallback_kwargs)\n except AnsibleFallbackNotFound:\n continue\n raw_args[key] = fallback_value\n missing = []\n for key, spec in self.argument_spec.items():\n if key in raw_args:\n value = self._cast(key, raw_args[key], spec)\n if spec.get('type') == 'list' and spec.get('elements'):\n element_spec = {'type': spec['elements']}\n value = [self._cast(key, element, element_spec)\n for element in value]\n self.params[key] = value\n elif 'default' in spec:\n self.params[key] = spec['default']\n elif spec.get('required'):\n missing.append(key)\n else:\n self.params[key] = None\n if missing:\n self.fail_json(msg='missing required arguments: %s'\n % ', '.join(sorted(missing)))\n for key, value in raw_args.items():\n if key.startswith('_ansible_') or key in self.params:\n continue\n self.params[key] = value\n\n def _pull_global_warnings(self):\n try:\n from ansible.module_utils.common.warnings import get_warning_messages\n for text in get_warning_messages():\n if str(text) not in self._warnings:\n self._warnings.append(str(text))\n except Exception:\n pass\n\n def warn(self, message):\n self._warnings.append(str(message))\n\n def deprecate(self, message, **kwargs):\n self._warnings.append('DEPRECATED: %s' % message)\n\n # Real basic.py logs to the systemd journal (when the target has\n # python-systemd) or syslog with the ident\n # 'ansible-<module_name>' at LOG_INFO; where neither is reachable\n # (containers, sandboxed exec contexts) python's syslog module\n # itself silently no-ops. Real basic.py only raises when *msg*\n # isn't a string; the actual syslog write never fails the module\n # - so here a swallowed exception is the documented worst case,\n # never an AttributeError like before (sr_fingerprint via\n # linux-system-roles.firewall/.kdump).\n def log(self, msg, log_args=None):\n if isinstance(msg, bytes):\n msg = msg.decode('utf-8', 'replace')\n try:\n import syslog\n syslog.openlog('ansible-%s' % self._name, 0, syslog.LOG_USER)\n syslog.syslog(syslog.LOG_INFO, str(msg))\n except Exception:\n pass\n\n def run_command(self, args, check_rc=False, cwd=None,\n environ_update=None, **kwargs):\n # A string command is shlex.split (not str.split) in real\n # basic.py, so quoting survives; the split tokens are what\n # _clean_args re-quotes into the cmd: key of a failure.\n if isinstance(args, str):\n argv = shlex.split(args)\n else:\n argv = [str(a) for a in args]\n clean_args = ' '.join(shlex.quote(a) for a in argv)\n env = os.environ.copy()\n if environ_update:\n env.update({k: str(v) for k, v in environ_update.items()})\n # Real basic.py runs with ignore_invalid_cwd=True: a cwd that\n # is not an existing directory is silently dropped (effectively\n # cwd=None) instead of killing Popen - the manageiq role calls\n # run_command(['rails', ...], cwd='/var/www/miq/vmdb') before\n # that checkout exists (issue #384).\n if cwd and os.path.isdir(cwd):\n cwd = os.path.abspath(cwd)\n else:\n cwd = None\n # Real Popen gets a bytes argv (to_bytes per element), so its\n # OSError text names the missing binary as b'rails' - the\n # exact spelling the composed error text must carry.\n b_args = [str(a).encode('utf-8', 'surrogateescape') for a in argv]\n try:\n proc = subprocess.Popen(b_args, cwd=cwd, stdout=subprocess.PIPE,\n stderr=subprocess.PIPE, env=env)\n except OSError as ex:\n self.fail_json(rc=ex.errno, stdout='', stderr='',\n msg='Error executing command.', cmd=clean_args,\n exception=ex)\n except Exception as ex:\n self.fail_json(rc=257, stdout='', stderr='',\n msg='Error executing command.', cmd=clean_args,\n exception=ex)\n out, err = proc.communicate()\n rc = proc.returncode\n if check_rc and rc != 0:\n self.fail_json(msg='Command failed with rc %d: %s'\n % (rc, err.decode('utf-8', 'replace')))\n return (rc, out.decode('utf-8', 'replace'),\n err.decode('utf-8', 'replace'))\n\n # Mirrors real basic.py's AnsibleModule.get_bin_path\n # (delegate to module_utils.common.process.get_bin_path):\n # absolute paths pass through, then opt_dirs, then PATH. Not\n # found + required fails via fail_json like real basic.py;\n # not required raises ValueError for the caller to catch\n # (systemd_units via linux-system-roles.systemd calls it\n # with neither, and ansible-playbook still succeeds\n # there because systemctl is found).\n def get_bin_path(self, arg, required=False, opt_dirs=None):\n paths = []\n if os.path.isabs(arg):\n paths.append(arg)\n for d in (opt_dirs or []):\n paths.append(d)\n paths.extend(os.environ.get('PATH', os.defpath).split(os.pathsep))\n for d in paths:\n candidate = os.path.join(d, arg)\n if os.path.isfile(candidate) and os.access(candidate, os.X_OK):\n return candidate\n msg = ('Failed to find required executable \"%s\" in paths: %s'\n % (arg, ':'.join(paths)))\n if required:\n self.fail_json(msg=msg)\n raise ValueError(msg)\n\n def exit_json(self, **kwargs):\n self._pull_global_warnings()\n result = dict(kwargs)\n result.setdefault('changed', False)\n # Real _return_formatted (2.21) injects `invocation` only when\n # the action layer passed `_ansible_inject_invocation`, and\n # pops it otherwise - before the derived *_lines and warnings\n # keys are added, so `invocation` sits before those in the\n # dumped dict when present.\n if self._inject_invocation:\n if 'invocation' not in result:\n result['invocation'] = {'module_args': self.params}\n else:\n result.pop('invocation', None)\n # Real _return_formatted adds the <key>_lines companions for\n # stdout/stderr whenever those keys exist (empty list for an\n # empty string) - a result carrying stdout without them\n # diverges from real's dumped result shape.\n if 'stdout' in result:\n result['stdout_lines'] = str(result['stdout']).splitlines()\n if 'stderr' in result:\n result['stderr_lines'] = str(result['stderr']).splitlines()\n if self._warnings:\n result['warnings'] = self._warnings\n sys.stdout.write(json.dumps(result) + '\\n')\n sys.exit(0)\n\n def fail_json(self, msg='Module failed', **kwargs):\n self._pull_global_warnings()\n # Real fail_json accepts an exception kwarg (run_command's\n # Popen OSError, issue #384): ansible-core 2.19 composes the\n # task error header as \"<msg>: <exception str>\" from the\n # ErrorSummary chain while the dumped result keeps the bare\n # msg and drops the exception key entirely. The exception str\n # rides the internal _ansible_error_exception key instead -\n # the py_module runner forwards _ansible_* keys and the\n # controller composes from it without ever dumping it.\n exception = kwargs.pop('exception', None)\n result = dict(kwargs)\n result['failed'] = True\n result['msg'] = msg\n # Same inject-or-pop gate as exit_json above (real\n # _return_formatted runs for both exits).\n if self._inject_invocation:\n if 'invocation' not in result:\n result['invocation'] = {'module_args': self.params}\n else:\n result.pop('invocation', None)\n # Only a real exception object builds an ErrorSummary chain in\n # real fail_json; a plain-string traceback is just hidden\n # (issue #557) and must not reach the error header.\n if isinstance(exception, BaseException):\n result['_ansible_error_exception'] = str(exception)\n if 'stdout' in result:\n result['stdout_lines'] = str(result['stdout']).splitlines()\n if 'stderr' in result:\n result['stderr_lines'] = str(result['stderr']).splitlines()\n if self._warnings:\n result['warnings'] = self._warnings\n sys.stdout.write(json.dumps(result) + '\\n')\n sys.exit(1)"

The ansible/module_utils bundle a new-style module's from ansible.module_utils.basic import AnsibleModule import needs. Ansible never relies on ansible-core being installed on the target - the AnsiballZ wrapper bundles module_utils INTO the module payload it ships - so every new-style role-private module runs on any target with a python3. This engine runs the raw module script instead, so on a target with no ansible-core installed the import died with ModuleNotFoundError and the module printed no result JSON ("MODULE FAILURE") - hard-FAILING the task where Ansible ran it successfully (found via newrelic.newrelic-infra's own "Setup agent config *NIX" task: the role ships its own library/merge_yaml.py, which took the py_module path and failed on every fresh target while ansible-playbook succeeded). The shim covers what corpus role-private modules actually use - params parsing/validation against argument_spec (with type coercion, defaults, aliases, required), check_mode, exit_json/fail_json, warn/run_command, log, get_bin_path - plus the ansible/module_utils/_text helpers modules import directly; not the whole real basic.py surface; anything beyond that fails exactly as before this shim existed.

COLLECTION_MODULE_UTILS_IMPORT = /\b(?:from|import)\s+ansible_collections\.[^.\s]+\.[^.\s]+\.plugins\.module_utils\b/

Ansible's own new-style detection (module_common.py _find_module_utils): a module importing ansible.module_utils - or one carrying the legacy INCLUDE marker, which the modularizer rewrites into from ansible.module_utils.basic import * - gets its args as a JSON dict (via the ANSIBLE_MODULE_ARGS env var its basic.py reads when no argv is given); everything else is old-style key=value argv.

NEW_STYLE_PYTHON_MODULE_RE in the same file ALSO counts a collection module_utils import (from|import ansible_collections.<ns>.<coll>.plugins.module_utils...) as new-style - such a module (a custom library/ module built on amazon.aws' AnsibleAWSModule, which wraps AnsibleModule) gets its args on stdin JSON like any module_utils importer, because ansible bundles the collection's module_utils into the AnsiballZ payload. Missing this form classified the module old-style, and the shim AnsibleModule (argv is never read) then saw no args at all: "missing required arguments: tags" for a dict arg the task passed (issue #955).

CONVERT_BOOL_PY_SHIM = "import collections.abc\n\nfrom ansible.module_utils.common.text.converters import to_text\n\n\nBOOLEANS_TRUE = frozenset(('y', 'yes', 'on', '1', 'true', 't', 1, 1.0, True))\nBOOLEANS_FALSE = frozenset(('n', 'no', 'off', '0', 'false', 'f', 0, 0.0, False))\nBOOLEANS = BOOLEANS_TRUE.union(BOOLEANS_FALSE)\n\n\ndef boolean(value, strict=True):\n if isinstance(value, bool):\n return value\n\n normalized_value = value\n\n if isinstance(value, (str, bytes)):\n normalized_value = to_text(value, errors='surrogate_or_strict').lower().strip()\n\n if not isinstance(value, collections.abc.Hashable):\n normalized_value = None\n\n if normalized_value in BOOLEANS_TRUE:\n return True\n elif normalized_value in BOOLEANS_FALSE or not strict:\n return False\n\n raise TypeError(\"The value '%s' is not a valid boolean. Valid booleans include: %s\" % (to_text(value), ', '.join(repr(i) for i in BOOLEANS)))"

ansible.module_utils.parsing.convert_bool - the BOOLEANS_TRUE/ BOOLEANS_FALSE tables and boolean() converter, which custom modules import directly when they parse a loose boolean themselves (the same tables this engine's own argspec validator emulates). Surveyed beside issue #779 as the other small pure module_utils file a role-private module most often imports that the shim bundle lacked; adopted from ansible-core's own file minus the _no_six deprecation hook (a py2-compat remnant no py3 module path exercises).

CONVERTERS_PY_SHIM = "def to_bytes(value, errors='surrogate_or_strict', encoding='utf-8'):\n if isinstance(value, bytes):\n return value\n if errors in ('surrogate_or_strict', 'surrogate_or_replace',\n 'surrogate_or_xmltext', 'surrogate_then_replace'):\n errors = 'surrogateescape'\n try:\n return str(value).encode(encoding, errors)\n except (UnicodeEncodeError, LookupError):\n return str(value).encode(encoding, 'replace')\n\ndef to_text(value, errors='surrogate_or_strict', encoding='utf-8'):\n if isinstance(value, bytes):\n if errors in ('surrogate_or_strict', 'surrogate_or_replace',\n 'surrogate_or_xmltext', 'surrogate_then_replace'):\n errors = 'surrogateescape'\n try:\n return value.decode(encoding, errors)\n except (UnicodeDecodeError, LookupError):\n return value.decode(encoding, 'replace')\n if isinstance(value, str):\n return value\n return str(value)\n\ndef to_native(value, errors='surrogate_or_strict', encoding='utf-8'):\n return to_text(value, errors, encoding)\n\ndef to_basestring(value):\n return to_text(value)"

In modern ansible-core the real text-conversion implementation moved from the Ansible module to the Ansible module - _text remains only as a deprecated re-export shim - and newer roles import the new path directly. bodsch.users' own library/multi_users.py does exactly that (from ansible.module_utils.common.text.converters import to_native, round 813275) and died with ModuleNotFoundError: No module named 'ansible.module_utils.common' on a target without ansible-core, while Ansible - which ships both paths - succeeded on the same task. So this file ships alongside _text.py, self-contained rather than importing from it, since role code may import either path (or both) and Ansible keeps both importable. Same to_bytes/to_text/to_native surface and the same surrogateescape mapping of the Ansible error-handler spellings as the _text shim above.

DICT_TRANSFORMATIONS_PY_SHIM = "import re\nfrom collections.abc import MutableMapping\nfrom copy import deepcopy\n\n\ndef camel_dict_to_snake_dict(camel_dict, reversible=False, ignore_list=()):\n def value_is_list(camel_list):\n checked_list = []\n for item in camel_list:\n if isinstance(item, dict):\n checked_list.append(camel_dict_to_snake_dict(item, reversible))\n elif isinstance(item, list):\n checked_list.append(value_is_list(item))\n else:\n checked_list.append(item)\n return checked_list\n\n snake_dict = {}\n for k, v in camel_dict.items():\n if isinstance(v, dict) and k not in ignore_list:\n snake_dict[_camel_to_snake(k, reversible=reversible)] = camel_dict_to_snake_dict(v, reversible)\n elif isinstance(v, list) and k not in ignore_list:\n snake_dict[_camel_to_snake(k, reversible=reversible)] = value_is_list(v)\n else:\n snake_dict[_camel_to_snake(k, reversible=reversible)] = v\n return snake_dict\n\n\ndef snake_dict_to_camel_dict(snake_dict, capitalize_first=False):\n def camelize(complex_type, capitalize_first=False):\n if complex_type is None:\n return\n new_type = type(complex_type)()\n if isinstance(complex_type, dict):\n for key in complex_type:\n new_type[_snake_to_camel(key, capitalize_first)] = camelize(complex_type[key], capitalize_first)\n elif isinstance(complex_type, list):\n for i in range(len(complex_type)):\n new_type.append(camelize(complex_type[i], capitalize_first))\n else:\n return complex_type\n return new_type\n\n return camelize(snake_dict, capitalize_first)\n\n\ndef _snake_to_camel(snake, capitalize_first=False):\n if capitalize_first:\n return ''.join(x.capitalize() or '_' for x in snake.split('_'))\n else:\n return snake.split('_')[0] + ''.join(x.capitalize() or '_' for x in snake.split('_')[1:])\n\n\ndef _camel_to_snake(name, reversible=False):\n def prepend_underscore_and_lower(m):\n return '_' + m.group(0).lower()\n\n if reversible:\n upper_pattern = r'[A-Z]'\n else:\n # Cope with pluralized abbreviations such as TargetGroupARNs\n # that would otherwise be rendered target_group_ar_ns\n upper_pattern = r'[A-Z]{3,}s$'\n\n s1 = re.sub(upper_pattern, prepend_underscore_and_lower, name)\n # Handle when there was nothing before the plural_pattern\n if s1.startswith(\"_\") and not name.startswith(\"_\"):\n s1 = s1[1:]\n if reversible:\n return s1\n\n first_cap_pattern = r'(.)([A-Z][a-z]+)'\n all_cap_pattern = r'([a-z0-9])([A-Z]+)'\n s2 = re.sub(first_cap_pattern, r'\\1_\\2', s1)\n return re.sub(all_cap_pattern, r'\\1_\\2', s2).lower()\n\n\ndef dict_merge(a, b):\n if not isinstance(b, dict):\n return b\n result = deepcopy(a)\n for k, v in b.items():\n if k in result and isinstance(result[k], dict):\n result[k] = dict_merge(result[k], v)\n else:\n result[k] = deepcopy(v)\n return result\n\n\ndef recursive_diff(dict1, dict2):\n if not all((isinstance(item, MutableMapping) for item in (dict1, dict2))):\n raise TypeError(\"Unable to diff 'dict1' %s and 'dict2' %s. \"\n \"Both must be a dictionary.\" % (type(dict1), type(dict2)))\n\n left = dict((k, v) for (k, v) in dict1.items() if k not in dict2)\n right = dict((k, v) for (k, v) in dict2.items() if k not in dict1)\n for k in (set(dict1.keys()) & set(dict2.keys())):\n if isinstance(dict1[k], dict) and isinstance(dict2[k], dict):\n result = recursive_diff(dict1[k], dict2[k])\n if result:\n left[k] = result[0]\n right[k] = result[1]\n elif dict1[k] != dict2[k]:\n left[k] = dict1[k]\n right[k] = dict2[k]\n if left or right:\n return left, right\n return None"

The dict-conversion helpers ansible-core ships at ansible/module_utils/common/dict_transformations - the camel/snake dict converters most amazon.aws-style modules run their facts through. A custom module importing it at top level (issue #779) died with ModuleNotFoundError: No module named 'ansible.module_utils.common.dict_transformations' on a target where the shim bundle stood in for ansible-core, while Ansible - which bundles every module_utils import into the AnsiballZ payload - ran the module fine. Adopted from ansible-core's own file (same GPL-3.0 family) so the conversion semantics - the ARNs-pluralization case in _camel_to_snake, the reversible flag, recursive_diff's TypeError shape - match exactly rather than approximately.

DIRECT_CAUSE_SEPARATOR = "The above exception was the direct cause of the following exception:"

DISTRO_PY_SHIM = "def _os_release():\n for path in (\"/etc/os-release\", \"/usr/lib/os-release\"):\n try:\n with open(path, \"r\", encoding=\"utf-8\", errors=\"replace\") as fh:\n data = {}\n for line in fh:\n line = line.strip()\n if not line or line.startswith(\"#\") or \"=\" not in line:\n continue\n key, _, value = line.partition(\"=\")\n value = value.strip()\n if len(value) >= 2 and value[0] == value[-1] and value[0] in ('\"', \"'\"):\n value = value[1:-1]\n data[key.strip()] = value\n return data\n except OSError:\n continue\n return {}\n\n\ndef _lsb_release():\n try:\n with open(\"/etc/lsb-release\", \"r\", encoding=\"utf-8\", errors=\"replace\") as fh:\n data = {}\n for line in fh:\n line = line.strip()\n if not line or line.startswith(\"#\") or \"=\" not in line:\n continue\n key, _, value = line.partition(\"=\")\n data[key.strip()] = value.strip()\n return data\n except OSError:\n return {}\n\n\ndef id():\n return _os_release().get(\"ID\", \"\").lower()\n\n\ndef name(pretty=False):\n info = _os_release()\n if pretty:\n return info.get(\"PRETTY_NAME\", \"\") or info.get(\"NAME\", \"\")\n return info.get(\"NAME\", \"\")\n\n\ndef version(best=False):\n info = _os_release()\n version_id = info.get(\"VERSION_ID\", \"\")\n if version_id:\n return version_id\n version_text = info.get(\"VERSION\", \"\")\n if version_text:\n text = version_text.split(\"(\", 1)[0].strip()\n return text.split(\" \")[0] if text else \"\"\n return \"\"\n\n\ndef codename():\n info = _os_release()\n result = info.get(\"VERSION_CODENAME\", \"\")\n if not result:\n result = _lsb_release().get(\"DISTRIB_CODENAME\", \"\")\n if not result:\n version_text = info.get(\"VERSION\", \"\")\n if \"(\" in version_text and \")\" in version_text:\n result = version_text.split(\"(\", 1)[1].split(\")\", 1)[0].strip()\n return result\n\n\ndef like():\n return _os_release().get(\"ID_LIKE\", \"\")\n\n\ndef version_parts(best=False):\n text = version(best)\n parts = (text.split(\".\") + [\"\", \"\", \"\"])[:3]\n return tuple(parts)\n\n\ndef major_version(best=False):\n return version_parts(best)[0]\n\n\ndef minor_version(best=False):\n return version_parts(best)[1]\n\n\ndef build_number(best=False):\n return version_parts(best)[2]\n\n\ndef linux_distribution(full_distribution_name=True):\n # Upstream python-distro's compat shim calls name() WITHOUT\n # pretty=True here, so the full name is os-release NAME\n # (\"Ubuntu\"), not PRETTY_NAME (\"Ubuntu 22.04.4 LTS\").\n distro_name = name() if full_distribution_name else id()\n return (distro_name, version(), codename())\n\n\ndef info(best=False):\n return {\n \"id\": id(),\n \"version\": version(best),\n \"like\": like(),\n \"codename\": codename(),\n }\n\n\ndef os_release_attr(attribute):\n return _os_release().get(attribute, \"\")\n\n\ndef lsb_release_attr(attribute):\n return _lsb_release().get(attribute, \"\")\n\n\ndef best_version():\n return version(best=True)\n\n\ndef main():\n pass"

ansible-core vendors the python-distro project inside module_utils (ansible/module_utils/distro/) and ships it in every module payload, so role-private modules detect the target distribution without python3-distro being installed on it. This engine runs the raw module script instead, so on a target without ansible-core the from ansible.module_utils import distro import died with "cannot import name 'distro' from 'ansible.module_utils'" and the module printed no result JSON - hard-FAILING the task where ansible ran it successfully (found via bodsch.php: both of its own library/php_version.py and library/php_modules.py import distro and gate their package-manager search on the returned distribution id, so the role died at its very first custom-module task). The shim covers python-distro's documented surface over the same files upstream python-distro reads - /etc/os-release, then /usr/lib/os-release, then /etc/lsb-release - including the linux_distribution(full_distribution_name=...) compatibility tuple both bodsch modules call - the shim always wins over the target's own ansible-core now (issue #915), so this surface is what every target sees.

MODULE_COMMON_MARKER = "#<<INCLUDE_ANSIBLE_MODULE_COMMON>>"

The legacy module-boilerplate marker: a module leaving #<<INCLUDE_ANSIBLE_MODULE_COMMON>> in its source expects the controller to splice the module_utils bundle in at that spot. Real ansible-core's modularizer (executor/module_common.py _find_module_utils) replaces the marker with from ansible.module_utils.basic import * and then classifies the module as NEW-style - so kunik.deploy-metadata's library/deployment_facts (which calls AnsibleModule(argument_spec = dict()) without ever importing it) ran fine under ansible-playbook while this engine - shipping the raw source - died with "name 'AnsibleModule' is not defined" (issue #543).

MODULE_IGNORE_EXTS = [".pyc", ".pyo", ".swp", ".bak", "~", ".rpm", ".md", ".txt", ".rst", ".yaml", ".yml", ".ini"] of ::String

Finds a role-private module source for module_name, or nil. Search roots mirror Ansible's two most-used locations: the current role's own library/ and the playbook-adjacent library/. First match wins (Ansible's own nearest-first order).

Takes the role's ROOT directory directly (task.role_path, always set - see role_loader.cr's task.role_path = role_dir), not role_files_dir (only set when the role actually ships a files/ subdirectory - existing_dir returns nil otherwise). A role with no files/ dir at all (linux-system-roles.storage/.logging/ .timesync, none of them ship one) could never resolve its own library/*.py modules through the old files/-derived path, so sr_fingerprint/blivet/timesync_provider fell straight back to "unavailable modules" - the exact scope cut 0.9.819 was supposed to have already closed for role-private modules. Found re-testing linux-system-roles.storage/logging/timesync. Extensions Ansible's legacy module finder never matches a module name against (ansible-core's MODULE_IGNORE_EXTS). A role shipping a DOCUMENTATION stub next to its script - linux-system-roles.timesync's library/timesync_provider.yml beside timesync_provider.sh - must resolve to the script, never the stub.

SIX_PY_SHIM = "\"\"\"Vendored `six` (its Python 3 surface, six 1.16 API).\"\"\"\n\nimport functools\nimport importlib\nimport importlib.util\nimport io\nimport operator\nimport sys\nimport types\nimport unittest\nfrom collections import abc as _collections_abc\n\n__version__ = \"1.16.0\"\n__author__ = \"Benjamin Peterson <benjamin@python.org>\"\n\nPY2 = False\nPY3 = True\nPY34 = True\nPY35 = True\nPY36 = True\nPY37 = True\nPY38 = True\nPY39 = True\n\nstring_types = (str,)\ninteger_types = (int,)\nclass_types = (type,)\ntext_type = str\nbinary_type = bytes\nMAXSIZE = sys.maxsize\nIterator = _collections_abc.Iterator\nIterable = _collections_abc.Iterable\n\nadvance_iterator = next\nnext = next\ncallable = callable\nprint_ = print\nwraps = functools.wraps\n\n\ndef add_move(move):\n \"\"\"Add an item to six.moves.\"\"\"\n setattr(moves, move.name, move)\n\n\ndef remove_move(name):\n try:\n delattr(moves, name)\n except AttributeError:\n raise AttributeError(\"no such move, %r\" % (name,))\n\n\ndef get_unbound_function(unbound):\n return unbound\n\n\ncreate_bound_method = types.MethodType\n\n\ndef create_unbound_method(func, cls):\n return func\n\n\ndef iterkeys(d, **kw):\n return iter(d.keys(**kw))\n\n\ndef itervalues(d, **kw):\n return iter(d.values(**kw))\n\n\ndef iteritems(d, **kw):\n return iter(d.items(**kw))\n\n\ndef iterlists(d, **kw):\n return iter(d.lists(**kw))\n\n\ndef keys(d, **kw):\n return d.keys(**kw)\n\n\ndef values(d, **kw):\n return d.values(**kw)\n\n\ndef items(d, **kw):\n return d.items(**kw)\n\n\ndef lists(d, **kw):\n return d.lists(**kw)\n\n\nviewkeys = keys\nviewvalues = values\nviewitems = items\nviewlists = lists\n\nget_method_function = operator.attrgetter(\"__func__\")\nget_method_self = operator.attrgetter(\"__self__\")\nget_function_closure = operator.attrgetter(\"__closure__\")\nget_function_code = operator.attrgetter(\"__code__\")\nget_function_defaults = operator.attrgetter(\"__defaults__\")\nget_function_globals = operator.attrgetter(\"__globals__\")\n\nStringIO = io.StringIO\nBytesIO = io.BytesIO\n\nassertCountEqual = unittest.TestCase.assertCountEqual\nassertRaisesRegex = unittest.TestCase.assertRaisesRegex\nassertRegex = unittest.TestCase.assertRegex\nassertNotRegex = unittest.TestCase.assertNotRegex\n\n\ndef b(s):\n return s.encode(\"latin-1\")\n\n\ndef u(s):\n return s\n\n\ndef int2byte(i):\n return bytes((i,))\n\n\ndef byte2int(bs):\n return bs[0]\n\n\ndef indexbytes(buf, i):\n return buf[i]\n\n\ndef iterbytes(buf):\n return iter(buf)\n\n\ndef with_metaclass(meta, *bases):\n class metaclass(type):\n def __new__(cls, name, this_bases, d):\n if sys.version_info[:2] >= (3, 7):\n resolved_bases = types.resolve_bases(bases)\n if resolved_bases is not bases:\n d[\"__orig_bases__\"] = bases\n else:\n resolved_bases = bases\n return meta(name, resolved_bases, d)\n\n @classmethod\n def __prepare__(cls, name, this_bases):\n return meta.__prepare__(name, bases)\n\n return type.__new__(metaclass, \"temporary_class\", (), {})\n\n\ndef add_metaclass(metaclass):\n def wrapper(cls):\n orig_vars = cls.__dict__.copy()\n slots = orig_vars.get(\"__slots__\")\n if slots is not None:\n if isinstance(slots, str):\n slots = [slots]\n for slots_var in slots:\n orig_vars.pop(slots_var)\n orig_vars.pop(\"__dict__\", None)\n orig_vars.pop(\"__weakref__\", None)\n if hasattr(cls, \"__qualname__\"):\n orig_vars[\"__qualname__\"] = cls.__qualname__\n return metaclass(cls.__name__, cls.__bases__, orig_vars)\n\n return wrapper\n\n\ndef ensure_binary(s, encoding=\"utf-8\", errors=\"strict\"):\n if isinstance(s, binary_type):\n return s\n if isinstance(s, text_type):\n return s.encode(encoding, errors)\n raise TypeError(\"not expecting type '%s'\" % type(s))\n\n\ndef ensure_str(s, encoding=\"utf-8\", errors=\"strict\"):\n if not isinstance(s, (text_type, binary_type)):\n raise TypeError(\"not expecting type '%s'\" % type(s))\n if isinstance(s, binary_type):\n s = s.decode(encoding, errors)\n return s\n\n\ndef ensure_text(s, encoding=\"utf-8\", errors=\"strict\"):\n if isinstance(s, binary_type):\n return s.decode(encoding, errors)\n elif isinstance(s, text_type):\n return s\n else:\n raise TypeError(\"not expecting type '%s'\" % type(s))\n\n\ndef python_2_unicode_compatible(klass):\n return klass\n\n\ndef reraise(tp, value, tb=None):\n try:\n if value is None:\n value = tp()\n if value.__traceback__ is not tb:\n raise value.with_traceback(tb)\n raise value\n finally:\n value = None\n tb = None\n\n\ndef raise_from(value, from_value):\n raise value from from_value\n\n\n_MOVES = {\n \"cStringIO\": (\"io\", \"StringIO\"),\n \"filter\": (\"builtins\", \"filter\"),\n \"filterfalse\": (\"itertools\", \"filterfalse\"),\n \"input\": (\"builtins\", \"input\"),\n \"intern\": (\"sys\", \"intern\"),\n \"map\": (\"builtins\", \"map\"),\n \"getcwd\": (\"os\", \"getcwd\"),\n \"getcwdb\": (\"os\", \"getcwdb\"),\n \"getoutput\": (\"subprocess\", \"getoutput\"),\n \"range\": (\"builtins\", \"range\"),\n \"reload_module\": (\"importlib\", \"reload\"),\n \"reduce\": (\"functools\", \"reduce\"),\n \"shlex_quote\": (\"shlex\", \"quote\"),\n \"StringIO\": (\"io\", \"StringIO\"),\n \"UserDict\": (\"collections\", \"UserDict\"),\n \"UserList\": (\"collections\", \"UserList\"),\n \"UserString\": (\"collections\", \"UserString\"),\n \"xrange\": (\"builtins\", \"range\"),\n \"zip\": (\"builtins\", \"zip\"),\n \"zip_longest\": (\"itertools\", \"zip_longest\"),\n \"builtins\": (\"builtins\", None),\n \"configparser\": (\"configparser\", None),\n \"collections_abc\": (\"collections.abc\", None),\n \"copyreg\": (\"copyreg\", None),\n \"dbm_gnu\": (\"dbm.gnu\", None),\n \"dbm_ndbm\": (\"dbm.ndbm\", None),\n \"_dummy_thread\": (\"_thread\", None),\n \"http_cookiejar\": (\"http.cookiejar\", None),\n \"http_cookies\": (\"http.cookies\", None),\n \"html_entities\": (\"html.entities\", None),\n \"html_parser\": (\"html.parser\", None),\n \"http_client\": (\"http.client\", None),\n \"email_mime_base\": (\"email.mime.base\", None),\n \"email_mime_image\": (\"email.mime.image\", None),\n \"email_mime_multipart\": (\"email.mime.multipart\", None),\n \"email_mime_nonmultipart\": (\"email.mime.nonmultipart\", None),\n \"email_mime_text\": (\"email.mime.text\", None),\n \"BaseHTTPServer\": (\"http.server\", None),\n \"CGIHTTPServer\": (\"http.server\", None),\n \"SimpleHTTPServer\": (\"http.server\", None),\n \"cPickle\": (\"pickle\", None),\n \"queue\": (\"queue\", None),\n \"reprlib\": (\"reprlib\", None),\n \"socketserver\": (\"socketserver\", None),\n \"_thread\": (\"_thread\", None),\n \"tkinter\": (\"tkinter\", None),\n \"tkinter_dialog\": (\"tkinter.dialog\", None),\n \"tkinter_filedialog\": (\"tkinter.filedialog\", None),\n \"tkinter_scrolledtext\": (\"tkinter.scrolledtext\", None),\n \"tkinter_simpledialog\": (\"tkinter.simpledialog\", None),\n \"tkinter_tix\": (\"tkinter.tix\", None),\n \"tkinter_ttk\": (\"tkinter.ttk\", None),\n \"tkinter_constants\": (\"tkinter.constants\", None),\n \"tkinter_dnd\": (\"tkinter.dnd\", None),\n \"tkinter_colorchooser\": (\"tkinter.colorchooser\", None),\n \"tkinter_commondialog\": (\"tkinter.commondialog\", None),\n \"tkinter_tkfiledialog\": (\"tkinter.filedialog\", None),\n \"tkinter_font\": (\"tkinter.font\", None),\n \"tkinter_messagebox\": (\"tkinter.messagebox\", None),\n \"tkinter_tksimpledialog\": (\"tkinter.simpledialog\", None),\n \"xmlrpc_client\": (\"xmlrpc.client\", None),\n \"xmlrpc_server\": (\"xmlrpc.server\", None),\n}\n\nif sys.platform == \"win32\":\n _MOVES[\"winreg\"] = (\"winreg\", None)\n\n_MOVES_PREFIX = __name__ + \".moves.\"\n\n\nclass _AliasLoader:\n def __init__(self, module):\n self._module = module\n\n def create_module(self, spec):\n return self._module\n\n def exec_module(self, module):\n pass\n\n\nclass _MovesFinder:\n @staticmethod\n def find_spec(fullname, path=None, target=None):\n if not fullname.startswith(_MOVES_PREFIX):\n return None\n entry = _MOVES.get(fullname[len(_MOVES_PREFIX):])\n if entry is None or entry[1] is not None:\n return None\n try:\n module = importlib.import_module(entry[0])\n except ImportError:\n return None\n return importlib.util.spec_from_loader(fullname, _AliasLoader(module))\n\n\nclass _MovesModule(types.ModuleType):\n def __getattr__(self, name):\n entry = _MOVES.get(name)\n if entry is None:\n raise AttributeError(\"module %r has no attribute %r\" % (self.__name__, name))\n target = importlib.import_module(entry[0])\n return target if entry[1] is None else getattr(target, entry[1])\n\n\nmoves = _MovesModule(_MOVES_PREFIX[:-1])\nmoves.__path__ = []\n\n_urllib = types.ModuleType(_MOVES_PREFIX + \"urllib\")\n_urllib.__path__ = []\nfor _suffix in (\"parse\", \"error\", \"request\", \"response\", \"robotparser\"):\n _real = importlib.import_module(\"urllib.\" + _suffix)\n setattr(_urllib, _suffix, _real)\n sys.modules[_MOVES_PREFIX + \"urllib.\" + _suffix] = _real\n sys.modules[_MOVES_PREFIX + \"urllib_\" + _suffix] = _real\nsetattr(moves, \"urllib\", _urllib)\nsys.modules[_MOVES_PREFIX + \"urllib\"] = _urllib\nsys.modules[_MOVES_PREFIX[:-1]] = moves\nsys.meta_path.append(_MovesFinder)"

ansible.module_utils.six - ansible-core's own vendored copy of the six compatibility library, bundled into every AnsiballZ module payload. Custom modules and collections import it directly (from ansible.module_utils.six import string_types, and amazon.aws' own module_utils do the same, issue #903), so it resolves under ansible-playbook on a target that has no ansible-core installed at all. This bundle is written both for an ansible-core-less target and unconditionally whenever a role ships its own module_utils tree - and in that second case it SHADOWS an installed ansible-core (the module file's own directory is sys.path[0]), which made the import fail exactly where ansible-playbook succeeded. six ships as a PACKAGE (six/__init__.py), the same shape ansible-core uses, because modules also import its six.moves submodules (from ansible.module_utils.six.moves import ..., ...six.moves.urllib.parse). Python-3 targets are all this shim has to cover - the py2 branches of six 1.16 are dead code there - but the public surface (type aliases, iterator/view helpers, moves, with_metaclass/add_metaclass, ensure_*, reraise/raise_from) is complete so a module behaves exactly as it does against the real one.

TEXT_PY_SHIM = "def to_bytes(value, errors='surrogate_or_strict', encoding='utf-8'):\n if isinstance(value, bytes):\n return value\n if errors in ('surrogate_or_strict', 'surrogate_or_replace',\n 'surrogate_or_xmltext', 'surrogate_then_replace'):\n errors = 'surrogateescape'\n try:\n return str(value).encode(encoding, errors)\n except (UnicodeEncodeError, LookupError):\n return str(value).encode(encoding, 'replace')\n\ndef to_text(value, errors='surrogate_or_strict', encoding='utf-8'):\n if isinstance(value, bytes):\n if errors in ('surrogate_or_strict', 'surrogate_or_replace',\n 'surrogate_or_xmltext', 'surrogate_then_replace'):\n errors = 'surrogateescape'\n try:\n return value.decode(encoding, errors)\n except (UnicodeDecodeError, LookupError):\n return value.decode(encoding, 'replace')\n if isinstance(value, str):\n return value\n return str(value)\n\ndef to_native(value, errors='surrogate_or_strict', encoding='utf-8'):\n return to_text(value, errors, encoding)\n\ndef to_basestring(value):\n return to_text(value)"

The ansible/module_utils/_text helpers a new-style module can import DIRECTLY alongside basic (nbde_server_tang via linux-system-roles.nbde_server does from ansible.module_utils._text import to_native) - without this file the import dies with ModuleNotFoundError at module top level, before AnsibleModule is ever constructed. Only the to_bytes/to_text/to_native surface modules actually import; the error-handler spellings real _text.py maps (surrogate_or_strict et al) become surrogateescape on py3 like the real code.

URLS_PY_SHIM = "import urllib.request as _urllib_request\nimport ssl as _ssl\n\n\ndef open_url(url, data=None, headers=None, method=None, use_proxy=True,\n force=False, timeout=10, validate_certs=True,\n url_username=None, url_password=None, http_agent=None,\n force_basic_auth=False, follow_redirects='urllib2',\n client_cert=None, client_key=None, cookies=None,\n unix_socket=None, ca_path=None, unredirected_headers=None,\n **kwargs):\n req = _urllib_request.Request(url, data=data, headers=headers or {}, method=method)\n if http_agent:\n req.add_header('User-agent', http_agent)\n context = None\n if url.lower().startswith('https'):\n context = _ssl.create_default_context(cafile=ca_path)\n if not validate_certs:\n context.check_hostname = False\n context.verify_mode = _ssl.CERT_NONE\n return _urllib_request.urlopen(req, timeout=timeout, context=context)"

ansible.module_utils.urls - role-private modules fetch URLs with open_url (kazoo-ansible.common's own library/publicip.py does from ansible.module_utils.urls import * and then open_url, issue #826: on a target without ansible-core the import died with "No module named 'ansible.module_utils.urls'", hard-failing the task where ansible - whose AnsiballZ payload bundles the real urls.py - ran the same module past the import, into its own bytes-JSON failure). Only the open_url surface such modules call is covered, over python's own urllib: the real call signature and defaults (timeout=10, validate_certs=True), a .read() that returns bytes like the real response, and 'urllib2' redirect semantics that are urllib's own. Client certificates, cookies and unix-socket transports stay uncovered - no shim-covered module uses them, and a wrong guess would mask real behavior.

WARNINGS_PY_SHIM = "_global_warnings = {}\n_global_deprecations = {}\n\n\ndef _record_warning(message):\n _global_warnings[str(message)] = None\n\n\ndef warn(warning, *, help_text=None, obj=None):\n _record_warning(warning)\n\n\ndef error_as_warning(msg, exception, *, help_text=None, obj=None):\n _record_warning(msg or str(exception))\n\n\ndef deprecate(msg, version=None, date=None, collection_name=None, *, deprecator=None, help_text=None, obj=None):\n if version:\n text = '%s (version %s)' % (msg, version)\n elif date:\n text = '%s (removed in %s)' % (msg, date)\n else:\n text = 'DEPRECATED: %s' % msg\n _global_deprecations[text] = None\n\n\ndef get_warning_messages():\n return tuple(_global_warnings)\n\n\ndef get_deprecation_messages():\n return tuple({'msg': text, 'collection_name': None, 'version': None} for text in _global_deprecations)\n\n\ndef get_warnings():\n return list(_global_warnings)\n\n\ndef get_deprecations():\n return list(_global_deprecations)"

ansible.module_utils.common.warnings - the global warning and deprecation accumulators real basic.py merges into the module result (kwargs['warnings'] = get_warnings()). Modules importing warn or deprecate directly (issue #908: a custom module importing warn, the same import amazon.aws' AnsibleAWSModule tree uses) died with ModuleNotFoundError under the shim bundle, which lacked the file, while ansible-playbook's AnsiballZ payload ran it. The shim carries the same warn/deprecate/get_warning_messages/get_deprecation_messages/ get_warnings/get_deprecations/error_as_warning surface and tracks messages as plain strings so the shim's basic.py pulls them into result['warnings'] the way real does.

Class methods

collection_import_roots(roots : Array(String) | Nil = nil) : Array(String)

Directories to put on a custom module's PYTHONPATH so an ansible_collections.* import resolves the way real ansible's module payload provides it. Real ansible never needs the collections importable on the target: its AnsiballZ wrapper bundles every ansible.module_utils.* AND ansible_collections.<ns>.<coll>.plugins.module_utils.* import INTO the zipapp it ships (ansible-core remaps collection module_utils into the payload), so a module doing from ansible_collections.amazon.aws.plugins.module_utils.retries import AWSRetry runs on a python that cannot import the collection at all. This runner executes the raw script with the target's python instead, so when that python is not the one the collections were installed into (Homebrew's isolated Cellar interpreter, issue #692; an ansible-galaxy user install under ~/.ansible/collections, which no python has on sys.path) the import died with ModuleNotFoundError where ansible-playbook ran the module fine. The roots are the same ones CollectionIndex already enumerates for module-name resolution (ANSIBLE_COLLECTIONS_PATH, the default install locations, the ansible-playbook shebang's virtualenv site-packages). A root may name either a collections ROOT (a directory CONTAINING an ansible_collections/ tree) or a path pointing directly AT an ansible_collections dir - Python needs the PARENT of ansible_collections on sys.path either way. Roots whose tree is absent are dropped (CollectionIndex keeps the two default locations even when they don't exist); roots is a test seam so a spec can pin the mapping without touching the process-wide collection env.

Constants

Class methods

Instance methods