diff options
Diffstat (limited to 'src/ansible_compat/runtime.py')
| -rw-r--r-- | src/ansible_compat/runtime.py | 961 |
1 files changed, 961 insertions, 0 deletions
diff --git a/src/ansible_compat/runtime.py b/src/ansible_compat/runtime.py new file mode 100644 index 0000000..ad81132 --- /dev/null +++ b/src/ansible_compat/runtime.py @@ -0,0 +1,961 @@ +"""Ansible runtime environment manager.""" +from __future__ import annotations + +import contextlib +import importlib +import json +import logging +import os +import re +import shutil +import subprocess +import sys +import warnings +from collections import OrderedDict +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING, Any, Callable, no_type_check + +import subprocess_tee +from packaging.version import Version + +from ansible_compat.config import ( + AnsibleConfig, + ansible_collections_path, + ansible_version, + parse_ansible_version, +) +from ansible_compat.constants import ( + META_MAIN, + MSG_INVALID_FQRL, + RC_ANSIBLE_OPTIONS_ERROR, + REQUIREMENT_LOCATIONS, +) +from ansible_compat.errors import ( + AnsibleCommandError, + AnsibleCompatError, + InvalidPrerequisiteError, + MissingAnsibleError, +) +from ansible_compat.loaders import colpath_from_path, yaml_from_file +from ansible_compat.prerun import get_cache_dir + +if TYPE_CHECKING: + # https://github.com/PyCQA/pylint/issues/3240 + # pylint: disable=unsubscriptable-object + CompletedProcess = subprocess.CompletedProcess[Any] +else: + CompletedProcess = subprocess.CompletedProcess + + +_logger = logging.getLogger(__name__) +# regex to extract the first version from a collection range specifier +version_re = re.compile(":[>=<]*([^,]*)") +namespace_re = re.compile("^[a-z][a-z0-9_]+$") + + +class AnsibleWarning(Warning): + """Warnings related to Ansible runtime.""" + + +@dataclass +class Collection: + """Container for Ansible collection information.""" + + name: str + version: str + path: Path + + +class CollectionVersion(Version): + """Collection version.""" + + def __init__(self, version: str) -> None: + """Initialize collection version.""" + # As packaging Version class does not support wildcard, we convert it + # to "0", as this being the smallest version possible. + if version == "*": + version = "0" + super().__init__(version) + + +@dataclass +class Plugins: # pylint: disable=too-many-instance-attributes + """Dataclass to access installed Ansible plugins, uses ansible-doc to retrieve them.""" + + runtime: Runtime + become: dict[str, str] = field(init=False) + cache: dict[str, str] = field(init=False) + callback: dict[str, str] = field(init=False) + cliconf: dict[str, str] = field(init=False) + connection: dict[str, str] = field(init=False) + httpapi: dict[str, str] = field(init=False) + inventory: dict[str, str] = field(init=False) + lookup: dict[str, str] = field(init=False) + netconf: dict[str, str] = field(init=False) + shell: dict[str, str] = field(init=False) + vars: dict[str, str] = field(init=False) # noqa: A003 + module: dict[str, str] = field(init=False) + strategy: dict[str, str] = field(init=False) + test: dict[str, str] = field(init=False) + filter: dict[str, str] = field(init=False) # noqa: A003 + role: dict[str, str] = field(init=False) + keyword: dict[str, str] = field(init=False) + + @no_type_check + def __getattribute__(self, attr: str): # noqa: ANN204 + """Get attribute.""" + if attr in { + "become", + "cache", + "callback", + "cliconf", + "connection", + "httpapi", + "inventory", + "lookup", + "netconf", + "shell", + "vars", + "module", + "strategy", + "test", + "filter", + "role", + "keyword", + }: + try: + result = super().__getattribute__(attr) + except AttributeError as exc: + if ansible_version() < Version("2.14") and attr in {"filter", "test"}: + msg = "Ansible version below 2.14 does not support retrieving filter and test plugins." + raise RuntimeError(msg) from exc + proc = self.runtime.run( + ["ansible-doc", "--json", "-l", "-t", attr], + ) + data = json.loads(proc.stdout) + if not isinstance(data, dict): # pragma: no cover + msg = "Unexpected output from ansible-doc" + raise AnsibleCompatError(msg) from exc + result = data + else: + result = super().__getattribute__(attr) + + return result + + +# pylint: disable=too-many-instance-attributes +class Runtime: + """Ansible Runtime manager.""" + + _version: Version | None = None + collections: OrderedDict[str, Collection] = OrderedDict() + cache_dir: Path | None = None + # Used to track if we have already initialized the Ansible runtime as attempts + # to do it multiple tilmes will cause runtime warnings from within ansible-core + initialized: bool = False + plugins: Plugins + + def __init__( + self, + project_dir: Path | None = None, + *, + isolated: bool = False, + min_required_version: str | None = None, + require_module: bool = False, + max_retries: int = 0, + environ: dict[str, str] | None = None, + verbosity: int = 0, + ) -> None: + """Initialize Ansible runtime environment. + + :param project_dir: The directory containing the Ansible project. If + not mentioned it will be guessed from the current + working directory. + :param isolated: Assure that installation of collections or roles + does not affect Ansible installation, an unique cache + directory being used instead. + :param min_required_version: Minimal version of Ansible required. If + not found, a :class:`RuntimeError` + exception is raised. + :param require_module: If set, instantiation will fail if Ansible + Python module is missing or is not matching + the same version as the Ansible command line. + That is useful for consumers that expect to + also perform Python imports from Ansible. + :param max_retries: Number of times it should retry network operations. + Default is 0, no retries. + :param environ: Environment dictionary to use, if undefined + ``os.environ`` will be copied and used. + :param verbosity: Verbosity level to use. + """ + self.project_dir = project_dir or Path.cwd() + self.isolated = isolated + self.max_retries = max_retries + self.environ = environ or os.environ.copy() + self.plugins = Plugins(runtime=self) + self.verbosity = verbosity + + self.initialize_logger(level=self.verbosity) + + # Reduce noise from paramiko, unless user already defined PYTHONWARNINGS + # paramiko/transport.py:236: CryptographyDeprecationWarning: Blowfish has been deprecated + # https://github.com/paramiko/paramiko/issues/2038 + # As CryptographyDeprecationWarning is not a builtin, we cannot use + # PYTHONWARNINGS to ignore it using category but we can use message. + # https://stackoverflow.com/q/68251969/99834 + if "PYTHONWARNINGS" not in self.environ: # pragma: no cover + self.environ["PYTHONWARNINGS"] = "ignore:Blowfish has been deprecated" + + if isolated: + self.cache_dir = get_cache_dir(self.project_dir) + self.config = AnsibleConfig() + + # Add the sys.path to the collection paths if not isolated + self._add_sys_path_to_collection_paths() + + if not self.version_in_range(lower=min_required_version): + msg = f"Found incompatible version of ansible runtime {self.version}, instead of {min_required_version} or newer." + raise RuntimeError(msg) + if require_module: + self._ensure_module_available() + + # pylint: disable=import-outside-toplevel + from ansible.utils.display import Display + + # pylint: disable=unused-argument + def warning( + self: Display, # noqa: ARG001 + msg: str, + *, + formatted: bool = False, # noqa: ARG001 + ) -> None: + """Override ansible.utils.display.Display.warning to avoid printing warnings.""" + warnings.warn( + message=msg, + category=AnsibleWarning, + stacklevel=2, + source={"msg": msg}, + ) + + # Monkey patch ansible warning in order to use warnings module. + Display.warning = warning + + def initialize_logger(self, level: int = 0) -> None: + """Set up the global logging level based on the verbosity number.""" + verbosity_map = { + -2: logging.CRITICAL, + -1: logging.ERROR, + 0: logging.WARNING, + 1: logging.INFO, + 2: logging.DEBUG, + } + # Unknown logging level is treated as DEBUG + logging_level = verbosity_map.get(level, logging.DEBUG) + _logger.setLevel(logging_level) + # Use module-level _logger instance to validate it + _logger.debug("Logging initialized to level %s", logging_level) + + def _add_sys_path_to_collection_paths(self) -> None: + """Add the sys.path to the collection paths.""" + if self.config.collections_scan_sys_path: + for path in sys.path: + if ( + path not in self.config.collections_paths + and (Path(path) / "ansible_collections").is_dir() + ): + self.config.collections_paths.append( # pylint: disable=E1101 + path, + ) + + def load_collections(self) -> None: + """Load collection data.""" + self.collections = OrderedDict() + no_collections_msg = "None of the provided paths were usable" + + proc = self.run(["ansible-galaxy", "collection", "list", "--format=json"]) + if proc.returncode == RC_ANSIBLE_OPTIONS_ERROR and ( + no_collections_msg in proc.stdout or no_collections_msg in proc.stderr + ): + _logger.debug("Ansible reported no installed collections at all.") + return + if proc.returncode != 0: + _logger.error(proc) + msg = f"Unable to list collections: {proc}" + raise RuntimeError(msg) + data = json.loads(proc.stdout) + if not isinstance(data, dict): + msg = f"Unexpected collection data, {data}" + raise TypeError(msg) + for path in data: + for collection, collection_info in data[path].items(): + if not isinstance(collection, str): + msg = f"Unexpected collection data, {collection}" + raise TypeError(msg) + if not isinstance(collection_info, dict): + msg = f"Unexpected collection data, {collection_info}" + raise TypeError(msg) + + self.collections[collection] = Collection( + name=collection, + version=collection_info["version"], + path=path, + ) + + def _ensure_module_available(self) -> None: + """Assure that Ansible Python module is installed and matching CLI version.""" + ansible_release_module = None + with contextlib.suppress(ModuleNotFoundError, ImportError): + ansible_release_module = importlib.import_module("ansible.release") + + if ansible_release_module is None: + msg = "Unable to find Ansible python module." + raise RuntimeError(msg) + + ansible_module_version = Version( + ansible_release_module.__version__, + ) + if ansible_module_version != self.version: + msg = f"Ansible CLI ({self.version}) and python module ({ansible_module_version}) versions do not match. This indicates a broken execution environment." + raise RuntimeError(msg) + + # For ansible 2.15+ we need to initialize the plugin loader + # https://github.com/ansible/ansible-lint/issues/2945 + if not Runtime.initialized: + col_path = [f"{self.cache_dir}/collections"] + if self.version >= Version("2.15.0.dev0"): + # pylint: disable=import-outside-toplevel,no-name-in-module + from ansible.plugins.loader import init_plugin_loader + + init_plugin_loader(col_path) + else: + # noinspection PyProtectedMember + from ansible.utils.collection_loader._collection_finder import ( # pylint: disable=import-outside-toplevel + _AnsibleCollectionFinder, + ) + + # noinspection PyProtectedMember + # pylint: disable=protected-access + col_path += self.config.collections_paths + col_path += os.path.dirname( # noqa: PTH120 + os.environ.get(ansible_collections_path(), "."), + ).split(":") + _AnsibleCollectionFinder( # noqa: SLF001 + paths=col_path, + )._install() # pylint: disable=protected-access + Runtime.initialized = True + + def clean(self) -> None: + """Remove content of cache_dir.""" + if self.cache_dir: + shutil.rmtree(self.cache_dir, ignore_errors=True) + + def run( # ruff: disable=PLR0913 + self, + args: str | list[str], + *, + retry: bool = False, + tee: bool = False, + env: dict[str, str] | None = None, + cwd: Path | None = None, + ) -> CompletedProcess: + """Execute a command inside an Ansible environment. + + :param retry: Retry network operations on failures. + :param tee: Also pass captured stdout/stderr to system while running. + """ + if tee: + run_func: Callable[..., CompletedProcess] = subprocess_tee.run + else: + run_func = subprocess.run + env = self.environ if env is None else env.copy() + # Presence of ansible debug variable or config option will prevent us + # from parsing its JSON output due to extra debug messages on stdout. + env["ANSIBLE_DEBUG"] = "0" + + # https://github.com/ansible/ansible-lint/issues/3522 + env["ANSIBLE_VERBOSE_TO_STDERR"] = "True" + + for _ in range(self.max_retries + 1 if retry else 1): + result = run_func( + args, + universal_newlines=True, + check=False, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + cwd=str(cwd) if cwd else None, + ) + if result.returncode == 0: + break + _logger.debug("Environment: %s", env) + if retry: + _logger.warning( + "Retrying execution failure %s of: %s", + result.returncode, + " ".join(args), + ) + return result + + @property + def version(self) -> 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 self._version: + return self._version + + proc = self.run(["ansible", "--version"]) + if proc.returncode == 0: + self._version = parse_ansible_version(proc.stdout) + return self._version + + msg = "Unable to find a working copy of ansible executable." + raise MissingAnsibleError(msg, proc=proc) + + def version_in_range( + self, + lower: str | None = None, + upper: str | None = None, + ) -> bool: + """Check if Ansible version is inside a required range. + + The lower limit is inclusive and the upper one exclusive. + """ + if lower and self.version < Version(lower): + return False + if upper and self.version >= Version(upper): + return False + return True + + def install_collection( + self, + collection: str | Path, + *, + destination: Path | None = None, + force: bool = False, + ) -> None: + """Install an Ansible collection. + + Can accept arguments like: + 'foo.bar:>=1.2.3' + 'git+https://github.com/ansible-collections/ansible.posix.git,main' + """ + cmd = [ + "ansible-galaxy", + "collection", + "install", + "-vvv", # this is needed to make ansible display important info in case of failures + ] + if force: + cmd.append("--force") + + if isinstance(collection, Path): + collection = str(collection) + # As ansible-galaxy install is not able to automatically determine + # if the range requires a pre-release, we need to manually add the --pre + # flag when needed. + matches = version_re.search(collection) + + if ( + not is_url(collection) + and matches + and CollectionVersion(matches[1]).is_prerelease + ): + cmd.append("--pre") + + cpaths: list[str] = self.config.collections_paths + if destination and str(destination) not in cpaths: + # we cannot use '-p' because it breaks galaxy ability to ignore already installed collections, so + # we hack ansible_collections_path instead and inject our own path there. + # pylint: disable=no-member + cpaths.insert(0, str(destination)) + cmd.append(f"{collection}") + + _logger.info("Running from %s : %s", Path.cwd(), " ".join(cmd)) + process = self.run( + cmd, + retry=True, + env={**self.environ, ansible_collections_path(): ":".join(cpaths)}, + ) + if process.returncode != 0: + msg = f"Command returned {process.returncode} code:\n{process.stdout}\n{process.stderr}" + _logger.error(msg) + raise InvalidPrerequisiteError(msg) + + def install_collection_from_disk( + self, + path: Path, + destination: Path | None = None, + ) -> None: + """Build and install collection from a given disk path.""" + self.install_collection(path, destination=destination, force=True) + + # pylint: disable=too-many-branches + def install_requirements( # noqa: C901 + self, + requirement: Path, + *, + retry: bool = False, + offline: bool = False, + ) -> None: + """Install dependencies from a requirements.yml. + + :param requirement: path to requirements.yml file + :param retry: retry network operations on failures + :param offline: bypass installation, may fail if requirements are not met. + """ + if not Path(requirement).exists(): + return + reqs_yaml = yaml_from_file(Path(requirement)) + if not isinstance(reqs_yaml, (dict, list)): + msg = f"{requirement} file is not a valid Ansible requirements file." + raise InvalidPrerequisiteError(msg) + + if isinstance(reqs_yaml, dict): + for key in reqs_yaml: + if key not in ("roles", "collections"): + msg = f"{requirement} file is not a valid Ansible requirements file. Only 'roles' and 'collections' keys are allowed at root level. Recognized valid locations are: {', '.join(REQUIREMENT_LOCATIONS)}" + raise InvalidPrerequisiteError(msg) + + if isinstance(reqs_yaml, list) or "roles" in reqs_yaml: + cmd = [ + "ansible-galaxy", + "role", + "install", + "-r", + f"{requirement}", + ] + if self.verbosity > 0: + cmd.extend(["-" + ("v" * self.verbosity)]) + if self.cache_dir: + cmd.extend(["--roles-path", f"{self.cache_dir}/roles"]) + + if offline: + _logger.warning( + "Skipped installing old role dependencies due to running in offline mode.", + ) + else: + _logger.info("Running %s", " ".join(cmd)) + + result = self.run(cmd, retry=retry) + _logger.debug(result.stdout) + if result.returncode != 0: + _logger.error(result.stderr) + raise AnsibleCommandError(result) + + # Run galaxy collection install works on v2 requirements.yml + if "collections" in reqs_yaml and reqs_yaml["collections"] is not None: + cmd = [ + "ansible-galaxy", + "collection", + "install", + ] + if self.verbosity > 0: + cmd.extend(["-" + ("v" * self.verbosity)]) + + for collection in reqs_yaml["collections"]: + if isinstance(collection, dict) and collection.get("type", "") == "git": + _logger.info( + "Adding '--pre' to ansible-galaxy collection install because we detected one collection being sourced from git.", + ) + cmd.append("--pre") + break + if offline: + _logger.warning( + "Skipped installing collection dependencies due to running in offline mode.", + ) + else: + cmd.extend(["-r", str(requirement)]) + cpaths = self.config.collections_paths + if self.cache_dir: + # we cannot use '-p' because it breaks galaxy ability to ignore already installed collections, so + # we hack ansible_collections_path instead and inject our own path there. + dest_path = f"{self.cache_dir}/collections" + if dest_path not in cpaths: + # pylint: disable=no-member + cpaths.insert(0, dest_path) + _logger.info("Running %s", " ".join(cmd)) + result = self.run( + cmd, + retry=retry, + env={**os.environ, "ANSIBLE_COLLECTIONS_PATH": ":".join(cpaths)}, + ) + _logger.debug(result.stdout) + if result.returncode != 0: + _logger.error(result.stderr) + raise AnsibleCommandError(result) + + def prepare_environment( # noqa: C901 + self, + required_collections: dict[str, str] | None = None, + *, + retry: bool = False, + install_local: bool = False, + offline: bool = False, + role_name_check: int = 0, + ) -> None: + """Make dependencies available if needed.""" + destination: Path | None = None + if required_collections is None: + required_collections = {} + + # first one is standard for collection layout repos and the last two + # are part of Tower specification + # https://docs.ansible.com/ansible-tower/latest/html/userguide/projects.html#ansible-galaxy-support + # https://docs.ansible.com/ansible-tower/latest/html/userguide/projects.html#collections-support + for req_file in REQUIREMENT_LOCATIONS: + self.install_requirements(Path(req_file), retry=retry, offline=offline) + + self._prepare_ansible_paths() + + if not install_local: + return + + for gpath in search_galaxy_paths(self.project_dir): + # processing all found galaxy.yml files + galaxy_path = Path(gpath) + if galaxy_path.exists(): + data = yaml_from_file(galaxy_path) + if isinstance(data, dict) and "dependencies" in data: + for name, required_version in data["dependencies"].items(): + _logger.info( + "Provisioning collection %s:%s from galaxy.yml", + name, + required_version, + ) + self.install_collection( + f"{name}{',' if is_url(name) else ':'}{required_version}", + destination=destination, + ) + + if self.cache_dir: + destination = self.cache_dir / "collections" + for name, min_version in required_collections.items(): + self.install_collection( + f"{name}:>={min_version}", + destination=destination, + ) + + if (self.project_dir / "galaxy.yml").exists(): + if destination: + # while function can return None, that would not break the logic + colpath = Path( + f"{destination}/ansible_collections/{colpath_from_path(self.project_dir)}", + ) + if colpath.is_symlink(): + if os.path.realpath(colpath) == str(Path.cwd()): + _logger.warning( + "Found symlinked collection, skipping its installation.", + ) + return + _logger.warning( + "Collection is symlinked, but not pointing to %s directory, so we will remove it.", + Path.cwd(), + ) + colpath.unlink() + + # molecule scenario within a collection + self.install_collection_from_disk( + galaxy_path.parent, + destination=destination, + ) + elif ( + Path().resolve().parent.name == "roles" + and Path("../../galaxy.yml").exists() + ): + # molecule scenario located within roles/<role-name>/molecule inside + # a collection + self.install_collection_from_disk( + Path("../.."), + destination=destination, + ) + else: + # no collection, try to recognize and install a standalone role + self._install_galaxy_role( + self.project_dir, + role_name_check=role_name_check, + ignore_errors=True, + ) + # reload collections + self.load_collections() + + def require_collection( + self, + name: str, + version: str | None = None, + *, + install: bool = True, + ) -> tuple[CollectionVersion, Path]: + """Check if a minimal collection version is present or exits. + + In the future this method may attempt to install a missing or outdated + collection before failing. + + :param name: collection name + :param version: minimal version required + :param install: if True, attempt to install a missing collection + :returns: tuple of (found_version, collection_path) + """ + try: + ns, coll = name.split(".", 1) + except ValueError as exc: + msg = f"Invalid collection name supplied: {name}%s" + raise InvalidPrerequisiteError( + msg, + ) from exc + + paths: list[str] = self.config.collections_paths + if not paths or not isinstance(paths, list): + msg = f"Unable to determine ansible collection paths. ({paths})" + raise InvalidPrerequisiteError( + msg, + ) + + if self.cache_dir: + # if we have a cache dir, we want to be use that would be preferred + # destination when installing a missing collection + # https://github.com/PyCQA/pylint/issues/4667 + paths.insert(0, f"{self.cache_dir}/collections") # pylint: disable=E1101 + + for path in paths: + collpath = Path(path) / "ansible_collections" / ns / coll + if collpath.exists(): + mpath = collpath / "MANIFEST.json" + if not mpath.exists(): + msg = f"Found collection at '{collpath}' but missing MANIFEST.json, cannot get info." + _logger.fatal(msg) + raise InvalidPrerequisiteError(msg) + + with mpath.open(encoding="utf-8") as f: + manifest = json.loads(f.read()) + found_version = CollectionVersion( + manifest["collection_info"]["version"], + ) + if version and found_version < CollectionVersion(version): + if install: + self.install_collection(f"{name}:>={version}") + self.require_collection(name, version, install=False) + else: + msg = f"Found {name} collection {found_version} but {version} or newer is required." + _logger.fatal(msg) + raise InvalidPrerequisiteError(msg) + return found_version, collpath.resolve() + break + else: + if install: + self.install_collection(f"{name}:>={version}" if version else name) + return self.require_collection( + name=name, + version=version, + install=False, + ) + msg = f"Collection '{name}' not found in '{paths}'" + _logger.fatal(msg) + raise InvalidPrerequisiteError(msg) + + def _prepare_ansible_paths(self) -> None: + """Configure Ansible environment variables.""" + try: + library_paths: list[str] = self.config.default_module_path.copy() + roles_path: list[str] = self.config.default_roles_path.copy() + collections_path: list[str] = self.config.collections_paths.copy() + except AttributeError as exc: + msg = "Unexpected ansible configuration" + raise RuntimeError(msg) from exc + + alterations_list: list[tuple[list[str], str, bool]] = [ + (library_paths, "plugins/modules", True), + (roles_path, "roles", True), + ] + + alterations_list.extend( + [ + (roles_path, f"{self.cache_dir}/roles", False), + (library_paths, f"{self.cache_dir}/modules", False), + (collections_path, f"{self.cache_dir}/collections", False), + ] + if self.isolated + else [], + ) + + for path_list, path_, must_be_present in alterations_list: + path = Path(path_) + if not path.exists(): + if must_be_present: + continue + path.mkdir(parents=True, exist_ok=True) + if str(path) not in path_list: + path_list.insert(0, str(path)) + + if library_paths != self.config.DEFAULT_MODULE_PATH: + self._update_env("ANSIBLE_LIBRARY", library_paths) + if collections_path != self.config.default_collections_path: + self._update_env(ansible_collections_path(), collections_path) + if roles_path != self.config.default_roles_path: + self._update_env("ANSIBLE_ROLES_PATH", roles_path) + + def _get_roles_path(self) -> Path: + """Return roles installation path. + + If `self.isolated` is set to `True`, `self.cache_dir` would be + created, then it returns the `self.cache_dir/roles`. When `self.isolated` is + not mentioned or set to `False`, it returns the first path in + `default_roles_path`. + """ + if self.cache_dir: + path = Path(f"{self.cache_dir}/roles") + else: + path = Path(self.config.default_roles_path[0]).expanduser() + return path + + def _install_galaxy_role( + self, + project_dir: Path, + role_name_check: int = 0, + *, + ignore_errors: bool = False, + ) -> None: + """Detect standalone galaxy role and installs it. + + :param: role_name_check: logic to used to check role name + 0: exit with error if name is not compliant (default) + 1: warn if name is not compliant + 2: bypass any name checking + + :param: ignore_errors: if True, bypass installing invalid roles. + + Our implementation aims to match ansible-galaxy's behaviour for installing + roles from a tarball or scm. For example ansible-galaxy will install a role + that has both galaxy.yml and meta/main.yml present but empty. Also missing + galaxy.yml is accepted but missing meta/main.yml is not. + """ + yaml = None + galaxy_info = {} + + for meta_main in META_MAIN: + meta_filename = Path(project_dir) / meta_main + + if meta_filename.exists(): + break + else: + if ignore_errors: + return + + yaml = yaml_from_file(meta_filename) + + if yaml and "galaxy_info" in yaml: + galaxy_info = yaml["galaxy_info"] + + fqrn = _get_role_fqrn(galaxy_info, project_dir) + + if role_name_check in [0, 1]: + if not re.match(r"[a-z0-9][a-z0-9_]+\.[a-z][a-z0-9_]+$", fqrn): + msg = MSG_INVALID_FQRL.format(fqrn) + if role_name_check == 1: + _logger.warning(msg) + else: + _logger.error(msg) + raise InvalidPrerequisiteError(msg) + elif "role_name" in galaxy_info: + # when 'role-name' is in skip_list, we stick to plain role names + role_namespace = _get_galaxy_role_ns(galaxy_info) + role_name = _get_galaxy_role_name(galaxy_info) + fqrn = f"{role_namespace}{role_name}" + else: + fqrn = Path(project_dir).absolute().name + path = self._get_roles_path() + path.mkdir(parents=True, exist_ok=True) + link_path = path / fqrn + # despite documentation stating that is_file() reports true for symlinks, + # it appears that is_dir() reports true instead, so we rely on exists(). + target = Path(project_dir).absolute() + if not link_path.exists() or ( + link_path.is_symlink() and link_path.readlink() != target + ): + # must call unlink before checking exists because a broken + # link reports as not existing and we want to repair it + link_path.unlink(missing_ok=True) + # https://github.com/python/cpython/issues/73843 + link_path.symlink_to(str(target), target_is_directory=True) + _logger.info( + "Using %s symlink to current repository in order to enable Ansible to find the role using its expected full name.", + link_path, + ) + + def _update_env(self, varname: str, value: list[str], default: str = "") -> None: + """Update colon based environment variable if needed. + + New values are prepended to make sure they take precedence. + """ + if not value: + return + orig_value = self.environ.get(varname, default) + if orig_value: + value = [*value, *orig_value.split(":")] + value_str = ":".join(value) + if value_str != self.environ.get(varname, ""): + self.environ[varname] = value_str + _logger.info("Set %s=%s", varname, value_str) + + +def _get_role_fqrn(galaxy_infos: dict[str, Any], project_dir: Path) -> str: + """Compute role fqrn.""" + role_namespace = _get_galaxy_role_ns(galaxy_infos) + role_name = _get_galaxy_role_name(galaxy_infos) + + if len(role_name) == 0: + role_name = Path(project_dir).absolute().name + role_name = re.sub(r"(ansible-|ansible-role-)", "", role_name).split( + ".", + maxsplit=2, + )[-1] + + return f"{role_namespace}{role_name}" + + +def _get_galaxy_role_ns(galaxy_infos: dict[str, Any]) -> str: + """Compute role namespace from meta/main.yml, including trailing dot.""" + role_namespace = galaxy_infos.get("namespace", "") + if len(role_namespace) == 0: + role_namespace = galaxy_infos.get("author", "") + if not isinstance(role_namespace, str): + msg = f"Role namespace must be string, not {role_namespace}" + raise AnsibleCompatError(msg) + # if there's a space in the name space, it's likely author name + # and not the galaxy login, so act as if there was no namespace + if not role_namespace or re.match(r"^\w+ \w+", role_namespace): + role_namespace = "" + else: + role_namespace = f"{role_namespace}." + return role_namespace + + +def _get_galaxy_role_name(galaxy_infos: dict[str, Any]) -> str: + """Compute role name from meta/main.yml.""" + result = galaxy_infos.get("role_name", "") + if not isinstance(result, str): + return "" + return result + + +def search_galaxy_paths(search_dir: Path) -> list[str]: + """Search for galaxy paths (only one level deep).""" + galaxy_paths: list[str] = [] + for file in [".", *os.listdir(search_dir)]: + # We ignore any folders that are not valid namespaces, just like + # ansible galaxy does at this moment. + if file != "." and not namespace_re.match(file): + continue + file_path = search_dir / file / "galaxy.yml" + if file_path.is_file(): + galaxy_paths.append(str(file_path)) + return galaxy_paths + + +def is_url(name: str) -> bool: + """Return True if a dependency name looks like an URL.""" + return bool(re.match("^git[+@]", name)) |