Krikri::PythonModuleRunner

BASIC_PY_SHIM = "import json\nimport os\nimport subprocess\nimport sys\nimport tempfile\n\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 self._apply_argument_spec(raw_args)\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 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 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 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 if isinstance(args, str):\n argv = args.split()\n else:\n argv = [str(a) for a in args]\n env = os.environ.copy()\n if environ_update:\n env.update({k: str(v) for k, v in environ_update.items()})\n proc = subprocess.Popen(argv, cwd=cwd, stdout=subprocess.PIPE,\n stderr=subprocess.PIPE, env=env)\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 real 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 result = dict(kwargs)\n result.setdefault('changed', False)\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 result = dict(kwargs)\n result['failed'] = True\n result['msg'] = msg\n if self._warnings:\n result['warnings'] = self._warnings\n sys.stdout.write(json.dumps(result) + '\\n')\n sys.exit(1)"

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)"

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)"

write_module_utils_bundle(work_dir : String) : Nil

build_kv_argv(params : Hash(String, String)) : Array(String)

build_module_args(params : Hash(String, String), check_mode : Bool) : String

find_source(module_name : String, role_path : String | Nil, playbook_dir : String | Nil) : String | Nil

missing_interpreter_line?(source : String, new_style : Bool) : Bool

new_style?(source : String) : Bool

parse_module_output(stdout : String) : JSON::Any | Nil

short_name(module_name : String) : String