diff options
Diffstat (limited to 'src/ansible_compat/config.py')
| -rw-r--r-- | src/ansible_compat/config.py | 465 |
1 files changed, 465 insertions, 0 deletions
diff --git a/src/ansible_compat/config.py b/src/ansible_compat/config.py new file mode 100644 index 0000000..a0b41b7 --- /dev/null +++ b/src/ansible_compat/config.py @@ -0,0 +1,465 @@ +"""Store configuration options as a singleton.""" +from __future__ import annotations + +import ast +import copy +import os +import re +import subprocess +from collections import UserDict +from typing import Literal + +from packaging.version import Version + +from ansible_compat.constants import ANSIBLE_MIN_VERSION +from ansible_compat.errors import InvalidPrerequisiteError, MissingAnsibleError +from ansible_compat.ports import cache + + +# do not use lru_cache here, as environment can change between calls +def ansible_collections_path() -> str: + """Return collection path variable for current version of Ansible.""" + for env_var in [ + "ANSIBLE_COLLECTIONS_PATH", + "ANSIBLE_COLLECTIONS_PATHS", + ]: + if env_var in os.environ: + return env_var + return "ANSIBLE_COLLECTIONS_PATH" + + +def parse_ansible_version(stdout: str) -> Version: + """Parse output of 'ansible --version'.""" + # Ansible can produce extra output before displaying version in debug mode. + + # ansible-core 2.11+: 'ansible [core 2.11.3]' + match = re.search( + r"^ansible \[(?:core|base) (?P<version>[^\]]+)\]", + stdout, + re.MULTILINE, + ) + if match: + return Version(match.group("version")) + msg = f"Unable to parse ansible cli version: {stdout}\nKeep in mind that only {ANSIBLE_MIN_VERSION } or newer are supported." + raise InvalidPrerequisiteError(msg) + + +@cache +def ansible_version(version: str = "") -> Version: + """Return current Version object for Ansible. + + If version is not mentioned, it returns current version as detected. + When version argument is mentioned, it return converts the version string + to Version object in order to make it usable in comparisons. + """ + if version: + return Version(version) + + proc = subprocess.run( + ["ansible", "--version"], # noqa: S603 + text=True, + check=False, + capture_output=True, + ) + if proc.returncode != 0: + raise MissingAnsibleError(proc=proc) + + return parse_ansible_version(proc.stdout) + + +class AnsibleConfig(UserDict[str, object]): # pylint: disable=too-many-ancestors + """Interface to query Ansible configuration. + + This should allow user to access everything provided by `ansible-config dump` without having to parse the data himself. + """ + + _aliases = { + "COLLECTIONS_PATH": "COLLECTIONS_PATHS", # 2.9 -> 2.10 + } + # Expose some attributes to enable auto-complete in editors, based on + # https://docs.ansible.com/ansible/latest/reference_appendices/config.html + action_warnings: bool = True + agnostic_become_prompt: bool = True + allow_world_readable_tmpfiles: bool = False + ansible_connection_path: str | None = None + ansible_cow_acceptlist: list[str] + ansible_cow_path: str | None = None + ansible_cow_selection: str = "default" + ansible_force_color: bool = False + ansible_nocolor: bool = False + ansible_nocows: bool = False + ansible_pipelining: bool = False + any_errors_fatal: bool = False + become_allow_same_user: bool = False + become_plugin_path: list[str] = [ + "~/.ansible/plugins/become", + "/usr/share/ansible/plugins/become", + ] + cache_plugin: str = "memory" + cache_plugin_connection: str | None = None + cache_plugin_prefix: str = "ansible_facts" + cache_plugin_timeout: int = 86400 + callable_accept_list: list[str] = [] + callbacks_enabled: list[str] = [] + collections_on_ansible_version_mismatch: Literal["warning", "ignore"] = "warning" + collections_paths: list[str] = [ + "~/.ansible/collections", + "/usr/share/ansible/collections", + ] + collections_scan_sys_path: bool = True + color_changed: str = "yellow" + color_console_prompt: str = "white" + color_debug: str = "dark gray" + color_deprecate: str = "purple" + color_diff_add: str = "green" + color_diff_lines: str = "cyan" + color_diff_remove: str = "red" + color_error: str = "red" + color_highlight: str = "white" + color_ok: str = "green" + color_skip: str = "cyan" + color_unreachable: str = "bright red" + color_verbose: str = "blue" + color_warn: str = "bright purple" + command_warnings: bool = False + conditional_bare_vars: bool = False + connection_facts_modules: dict[str, str] + controller_python_warning: bool = True + coverage_remote_output: str | None + coverage_remote_paths: list[str] + default_action_plugin_path: list[str] = [ + "~/.ansible/plugins/action", + "/usr/share/ansible/plugins/action", + ] + default_allow_unsafe_lookups: bool = False + default_ask_pass: bool = False + default_ask_vault_pass: bool = False + default_become: bool = False + default_become_ask_pass: bool = False + default_become_exe: str | None = None + default_become_flags: str + default_become_method: str = "sudo" + default_become_user: str = "root" + default_cache_plugin_path: list[str] = [ + "~/.ansible/plugins/cache", + "/usr/share/ansible/plugins/cache", + ] + default_callback_plugin_path: list[str] = [ + "~/.ansible/plugins/callback", + "/usr/share/ansible/plugins/callback", + ] + default_cliconf_plugin_path: list[str] = [ + "~/.ansible/plugins/cliconf", + "/usr/share/ansible/plugins/cliconf", + ] + default_connection_plugin_path: list[str] = [ + "~/.ansible/plugins/connection", + "/usr/share/ansible/plugins/connection", + ] + default_debug: bool = False + default_executable: str = "/bin/sh" + default_fact_path: str | None = None + default_filter_plugin_path: list[str] = [ + "~/.ansible/plugins/filter", + "/usr/share/ansible/plugins/filter", + ] + default_force_handlers: bool = False + default_forks: int = 5 + default_gathering: Literal["smart", "explicit", "implicit"] = "smart" + default_gather_subset: list[str] = ["all"] + default_gather_timeout: int = 10 + default_handler_includes_static: bool = False + default_hash_behaviour: str = "replace" + default_host_list: list[str] = ["/etc/ansible/hosts"] + default_httpapi_plugin_path: list[str] = [ + "~/.ansible/plugins/httpapi", + "/usr/share/ansible/plugins/httpapi", + ] + default_internal_poll_interval: float = 0.001 + default_inventory_plugin_path: list[str] = [ + "~/.ansible/plugins/inventory", + "/usr/share/ansible/plugins/inventory", + ] + default_jinja2_extensions: list[str] = [] + default_jinja2_native: bool = False + default_keep_remote_files: bool = False + default_libvirt_lxc_noseclabel: bool = False + default_load_callback_plugins: bool = False + default_local_tmp: str = "~/.ansible/tmp" + default_log_filter: list[str] = [] + default_log_path: str | None = None + default_lookup_lugin_path: list[str] = [ + "~/.ansible/plugins/lookup", + "/usr/share/ansible/plugins/lookup", + ] + default_managed_str: str = "Ansible managed" + default_module_args: str + default_module_compression: str = "ZIP_DEFLATED" + default_module_name: str = "command" + default_module_path: list[str] = [ + "~/.ansible/plugins/modules", + "/usr/share/ansible/plugins/modules", + ] + default_module_utils_path: list[str] = [ + "~/.ansible/plugins/module_utils", + "/usr/share/ansible/plugins/module_utils", + ] + default_netconf_plugin_path: list[str] = [ + "~/.ansible/plugins/netconf", + "/usr/share/ansible/plugins/netconf", + ] + default_no_log: bool = False + default_no_target_syslog: bool = False + default_null_representation: str | None = None + default_poll_interval: int = 15 + default_private_key_file: str | None = None + default_private_role_vars: bool = False + default_remote_port: str | None = None + default_remote_user: str | None = None + # https://docs.ansible.com/ansible/latest/reference_appendices/config.html#collections-paths + default_collections_path: list[str] = [ + "~/.ansible/collections", + "/usr/share/ansible/collections", + ] + default_roles_path: list[str] = [ + "~/.ansible/roles", + "/usr/share/ansible/roles", + "/etc/ansible/roles", + ] + default_selinux_special_fs: list[str] = [ + "fuse", + "nfs", + "vboxsf", + "ramfs", + "9p", + "vfat", + ] + default_stdout_callback: str = "default" + default_strategy: str = "linear" + default_strategy_plugin_path: list[str] = [ + "~/.ansible/plugins/strategy", + "/usr/share/ansible/plugins/strategy", + ] + default_su: bool = False + default_syslog_facility: str = "LOG_USER" + default_task_includes_static: bool = False + default_terminal_plugin_path: list[str] = [ + "~/.ansible/plugins/terminal", + "/usr/share/ansible/plugins/terminal", + ] + default_test_plugin_path: list[str] = [ + "~/.ansible/plugins/test", + "/usr/share/ansible/plugins/test", + ] + default_timeout: int = 10 + default_transport: str = "smart" + default_undefined_var_behavior: bool = True + default_vars_plugin_path: list[str] = [ + "~/.ansible/plugins/vars", + "/usr/share/ansible/plugins/vars", + ] + default_vault_encrypt_identity: str | None = None + default_vault_identity: str = "default" + default_vault_identity_list: list[str] = [] + default_vault_id_match: bool = False + default_vault_password_file: str | None = None + default_verbosity: int = 0 + deprecation_warnings: bool = False + devel_warning: bool = True + diff_always: bool = False + diff_context: int = 3 + display_args_to_stdout: bool = False + display_skipped_hosts: bool = True + docsite_root_url: str = "https://docs.ansible.com/ansible/" + doc_fragment_plugin_path: list[str] = [ + "~/.ansible/plugins/doc_fragments", + "/usr/share/ansible/plugins/doc_fragments", + ] + duplicate_yaml_dict_key: Literal["warn", "error", "ignore"] = "warn" + enable_task_debugger: bool = False + error_on_missing_handler: bool = True + facts_modules: list[str] = ["smart"] + galaxy_cache_dir: str = "~/.ansible/galaxy_cache" + galaxy_display_progress: str | None = None + galaxy_ignore_certs: bool = False + galaxy_role_skeleton: str | None = None + galaxy_role_skeleton_ignore: list[str] = ["^.git$", "^.*/.git_keep$"] + galaxy_server: str = "https://galaxy.ansible.com" + galaxy_server_list: str | None = None + galaxy_token_path: str = "~/.ansible/galaxy_token" + host_key_checking: bool = True + host_pattern_mismatch: Literal["warning", "error", "ignore"] = "warning" + inject_facts_as_vars: bool = True + interpreter_python: str = "auto_legacy" + interpreter_python_distro_map: dict[str, str] + interpreter_python_fallback: list[str] + invalid_task_attribute_failed: bool = True + inventory_any_unparsed_is_failed: bool = False + inventory_cache_enabled: bool = False + inventory_cache_plugin: str | None = None + inventory_cache_plugin_connection: str | None = None + inventory_cache_plugin_prefix: str = "ansible_facts" + inventory_cache_timeout: int = 3600 + inventory_enabled: list[str] = [ + "host_list", + "script", + "auto", + "yaml", + "ini", + "toml", + ] + inventory_export: bool = False + inventory_ignore_exts: str + inventory_ignore_patterns: list[str] = [] + inventory_unparsed_is_failed: bool = False + localhost_warning: bool = True + max_file_size_for_diff: int = 104448 + module_ignore_exts: str + netconf_ssh_config: str | None = None + network_group_modules: list[str] = [ + "eos", + "nxos", + "ios", + "iosxr", + "junos", + "enos", + "ce", + "vyos", + "sros", + "dellos9", + "dellos10", + "dellos6", + "asa", + "aruba", + "aireos", + "bigip", + "ironware", + "onyx", + "netconf", + "exos", + "voss", + "slxos", + ] + old_plugin_cache_clearing: bool = False + paramiko_host_key_auto_add: bool = False + paramiko_look_for_keys: bool = True + persistent_command_timeout: int = 30 + persistent_connect_retry_timeout: int = 15 + persistent_connect_timeout: int = 30 + persistent_control_path_dir: str = "~/.ansible/pc" + playbook_dir: str | None + playbook_vars_root: Literal["top", "bottom", "all"] = "top" + plugin_filters_cfg: str | None = None + python_module_rlimit_nofile: int = 0 + retry_files_enabled: bool = False + retry_files_save_path: str | None = None + run_vars_plugins: str = "demand" + show_custom_stats: bool = False + string_conversion_action: Literal["warn", "error", "ignore"] = "warn" + string_type_filters: list[str] = [ + "string", + "to_json", + "to_nice_json", + "to_yaml", + "to_nice_yaml", + "ppretty", + "json", + ] + system_warnings: bool = True + tags_run: list[str] = [] + tags_skip: list[str] = [] + task_debugger_ignore_errors: bool = True + task_timeout: int = 0 + transform_invalid_group_chars: Literal[ + "always", + "never", + "ignore", + "silently", + ] = "never" + use_persistent_connections: bool = False + variable_plugins_enabled: list[str] = ["host_group_vars"] + variable_precedence: list[str] = [ + "all_inventory", + "groups_inventory", + "all_plugins_inventory", + "all_plugins_play", + "groups_plugins_inventory", + "groups_plugins_play", + ] + verbose_to_stderr: bool = False + win_async_startup_timeout: int = 5 + worker_shutdown_poll_count: int = 0 + worker_shutdown_poll_delay: float = 0.1 + yaml_filename_extensions: list[str] = [".yml", ".yaml", ".json"] + + def __init__( + self, + config_dump: str | None = None, + data: dict[str, object] | None = None, + ) -> None: + """Load config dictionary.""" + super().__init__() + + if data: + self.data = copy.deepcopy(data) + return + + if not config_dump: + env = os.environ.copy() + # Avoid possible ANSI garbage + env["ANSIBLE_FORCE_COLOR"] = "0" + config_dump = subprocess.check_output( + ["ansible-config", "dump"], # noqa: S603 + universal_newlines=True, + env=env, + ) + + for match in re.finditer( + r"^(?P<key>[A-Za-z0-9_]+).* = (?P<value>.*)$", + config_dump, + re.MULTILINE, + ): + key = match.groupdict()["key"] + value = match.groupdict()["value"] + try: + self[key] = ast.literal_eval(value) + except (NameError, SyntaxError, ValueError): + self[key] = value + + def __getattribute__(self, attr_name: str) -> object: + """Allow access of config options as attributes.""" + _dict = super().__dict__ # pylint: disable=no-member + if attr_name in _dict: + return _dict[attr_name] + + data = super().__getattribute__("data") + if attr_name == "data": # pragma: no cover + return data + + name = attr_name.upper() + if name in data: + return data[name] + if name in AnsibleConfig._aliases: + return data[AnsibleConfig._aliases[name]] + + return super().__getattribute__(attr_name) + + def __getitem__(self, name: str) -> object: + """Allow access to config options using indexing.""" + return super().__getitem__(name.upper()) + + def __copy__(self) -> AnsibleConfig: + """Allow users to run copy on Config.""" + return AnsibleConfig(data=self.data) + + def __deepcopy__(self, memo: object) -> AnsibleConfig: + """Allow users to run deeepcopy on Config.""" + return AnsibleConfig(data=self.data) + + +__all__ = [ + "ansible_collections_path", + "parse_ansible_version", + "ansible_version", + "AnsibleConfig", +] |