summaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2024-05-04 00:24:37 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2024-05-04 00:24:37 +0000
commit1022b2cebe73db426241c2f420d4ee9f6f3c1bed (patch)
treea5c38ccfaa66e8a52767dec01d3598b67a7422a8 /test
parentInitial commit. (diff)
downloadpython-ansible-compat-1022b2cebe73db426241c2f420d4ee9f6f3c1bed.tar.xz
python-ansible-compat-1022b2cebe73db426241c2f420d4ee9f6f3c1bed.zip
Adding upstream version 4.1.11.upstream/4.1.11
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'test')
-rw-r--r--test/__init__.py1
-rw-r--r--test/assets/galaxy_paths/.bar/galaxy.yml0
-rw-r--r--test/assets/galaxy_paths/foo/galaxy.yml0
-rw-r--r--test/assets/requirements-invalid-collection.yml3
-rw-r--r--test/assets/requirements-invalid-role.yml3
-rw-r--r--test/assets/validate0_data.json1
-rw-r--r--test/assets/validate0_expected.json22
-rw-r--r--test/assets/validate0_schema.json9
-rw-r--r--test/collections/acme.broken/galaxy.yml1
-rw-r--r--test/collections/acme.goodies/galaxy.yml34
-rw-r--r--test/collections/acme.goodies/molecule/default/converge.yml7
-rw-r--r--test/collections/acme.goodies/molecule/default/molecule.yml11
-rw-r--r--test/collections/acme.goodies/roles/baz/molecule/deep_scenario/converge.yml7
-rw-r--r--test/collections/acme.goodies/roles/baz/molecule/deep_scenario/molecule.yml11
-rw-r--r--test/collections/acme.goodies/roles/baz/tasks/main.yml3
-rw-r--r--test/collections/acme.goodies/tests/requirements.yml3
-rw-r--r--test/collections/acme.minimal/galaxy.yml30
-rw-r--r--test/conftest.py127
-rw-r--r--test/roles/acme.missing_deps/meta/main.yml8
-rw-r--r--test/roles/acme.missing_deps/requirements.yml2
-rw-r--r--test/roles/acme.sample2/meta/main.yml16
-rw-r--r--test/roles/ansible-role-sample/meta/main.yml16
-rw-r--r--test/roles/sample3/meta/main.yml16
-rw-r--r--test/roles/sample4/meta/main.yml16
-rw-r--r--test/test_api.py5
-rw-r--r--test/test_config.py86
-rw-r--r--test/test_configuration_example.py12
-rw-r--r--test/test_loaders.py9
-rw-r--r--test/test_prerun.py11
-rw-r--r--test/test_runtime.py893
-rw-r--r--test/test_runtime_example.py24
-rw-r--r--test/test_runtime_scan_path.py102
-rw-r--r--test/test_schema.py73
33 files changed, 1562 insertions, 0 deletions
diff --git a/test/__init__.py b/test/__init__.py
new file mode 100644
index 0000000..689eb7b
--- /dev/null
+++ b/test/__init__.py
@@ -0,0 +1 @@
+"""Tests for ansible_compat package."""
diff --git a/test/assets/galaxy_paths/.bar/galaxy.yml b/test/assets/galaxy_paths/.bar/galaxy.yml
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/test/assets/galaxy_paths/.bar/galaxy.yml
diff --git a/test/assets/galaxy_paths/foo/galaxy.yml b/test/assets/galaxy_paths/foo/galaxy.yml
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/test/assets/galaxy_paths/foo/galaxy.yml
diff --git a/test/assets/requirements-invalid-collection.yml b/test/assets/requirements-invalid-collection.yml
new file mode 100644
index 0000000..6ace6cf
--- /dev/null
+++ b/test/assets/requirements-invalid-collection.yml
@@ -0,0 +1,3 @@
+# "ansible-galaxy collection install" is expected to fail this invalid file
+collections:
+ - foo: bar
diff --git a/test/assets/requirements-invalid-role.yml b/test/assets/requirements-invalid-role.yml
new file mode 100644
index 0000000..e02c64e
--- /dev/null
+++ b/test/assets/requirements-invalid-role.yml
@@ -0,0 +1,3 @@
+# file expected to make "ansible-galaxy role install" to fail
+roles:
+ - this_role_does_not_exist
diff --git a/test/assets/validate0_data.json b/test/assets/validate0_data.json
new file mode 100644
index 0000000..e9f6f2e
--- /dev/null
+++ b/test/assets/validate0_data.json
@@ -0,0 +1 @@
+{ "environment": { "a": false, "b": true, "c": "foo" } }
diff --git a/test/assets/validate0_expected.json b/test/assets/validate0_expected.json
new file mode 100644
index 0000000..ea36da9
--- /dev/null
+++ b/test/assets/validate0_expected.json
@@ -0,0 +1,22 @@
+[
+ {
+ "message": "False is not of type 'string'",
+ "data_path": "environment.a",
+ "json_path": "$.environment.a",
+ "schema_path": "properties.environment.additionalProperties.type",
+ "relative_schema": { "type": "string" },
+ "expected": "string",
+ "validator": "type",
+ "found": "False"
+ },
+ {
+ "message": "True is not of type 'string'",
+ "data_path": "environment.b",
+ "json_path": "$.environment.b",
+ "schema_path": "properties.environment.additionalProperties.type",
+ "relative_schema": { "type": "string" },
+ "expected": "string",
+ "validator": "type",
+ "found": "True"
+ }
+]
diff --git a/test/assets/validate0_schema.json b/test/assets/validate0_schema.json
new file mode 100644
index 0000000..e642fb0
--- /dev/null
+++ b/test/assets/validate0_schema.json
@@ -0,0 +1,9 @@
+{
+ "$schema": "http://json-schema.org/draft-07/schema#",
+ "properties": {
+ "environment": {
+ "type": "object",
+ "additionalProperties": { "type": "string" }
+ }
+ }
+}
diff --git a/test/collections/acme.broken/galaxy.yml b/test/collections/acme.broken/galaxy.yml
new file mode 100644
index 0000000..599fd5b
--- /dev/null
+++ b/test/collections/acme.broken/galaxy.yml
@@ -0,0 +1 @@
+foo: that is not a valid collection!
diff --git a/test/collections/acme.goodies/galaxy.yml b/test/collections/acme.goodies/galaxy.yml
new file mode 100644
index 0000000..9682115
--- /dev/null
+++ b/test/collections/acme.goodies/galaxy.yml
@@ -0,0 +1,34 @@
+name: goodies
+namespace: acme
+version: 1.0.0
+readme: README.md
+authors:
+ - Red Hat
+description: Sample collection to use with molecule
+dependencies:
+ community.molecule: ">=0.1.0" # used to also test '=>' condition
+ ansible.utils: "*" # used to also test '*'
+ git+https://github.com/ansible-collections/community.crypto.git: main # tests ability to install from git
+build_ignore:
+ - "*.egg-info"
+ - .DS_Store
+ - .eggs
+ - .gitignore
+ - .mypy_cache
+ - .pytest_cache
+ - .stestr
+ - .stestr.conf
+ - .tox
+ - .vscode
+ - MANIFEST.in
+ - build
+ - dist
+ - doc
+ - report.html
+ - setup.cfg
+ - setup.py
+ - "tests/unit/*.*"
+ - README.rst
+ - tox.ini
+
+license_file: LICENSE
diff --git a/test/collections/acme.goodies/molecule/default/converge.yml b/test/collections/acme.goodies/molecule/default/converge.yml
new file mode 100644
index 0000000..b85e064
--- /dev/null
+++ b/test/collections/acme.goodies/molecule/default/converge.yml
@@ -0,0 +1,7 @@
+---
+- name: Converge
+ hosts: localhost
+ tasks:
+ - name: "Include sample role from current collection"
+ include_role:
+ name: acme.goodies.baz
diff --git a/test/collections/acme.goodies/molecule/default/molecule.yml b/test/collections/acme.goodies/molecule/default/molecule.yml
new file mode 100644
index 0000000..74c8557
--- /dev/null
+++ b/test/collections/acme.goodies/molecule/default/molecule.yml
@@ -0,0 +1,11 @@
+---
+dependency:
+ name: galaxy
+driver:
+ name: delegated
+platforms:
+ - name: instance
+provisioner:
+ name: ansible
+verifier:
+ name: ansible
diff --git a/test/collections/acme.goodies/roles/baz/molecule/deep_scenario/converge.yml b/test/collections/acme.goodies/roles/baz/molecule/deep_scenario/converge.yml
new file mode 100644
index 0000000..c18086f
--- /dev/null
+++ b/test/collections/acme.goodies/roles/baz/molecule/deep_scenario/converge.yml
@@ -0,0 +1,7 @@
+---
+- name: Converge
+ hosts: localhost
+ tasks:
+ - name: "Sample testing task part of deep_scenario"
+ include_role:
+ name: acme.goodies.baz
diff --git a/test/collections/acme.goodies/roles/baz/molecule/deep_scenario/molecule.yml b/test/collections/acme.goodies/roles/baz/molecule/deep_scenario/molecule.yml
new file mode 100644
index 0000000..74c8557
--- /dev/null
+++ b/test/collections/acme.goodies/roles/baz/molecule/deep_scenario/molecule.yml
@@ -0,0 +1,11 @@
+---
+dependency:
+ name: galaxy
+driver:
+ name: delegated
+platforms:
+ - name: instance
+provisioner:
+ name: ansible
+verifier:
+ name: ansible
diff --git a/test/collections/acme.goodies/roles/baz/tasks/main.yml b/test/collections/acme.goodies/roles/baz/tasks/main.yml
new file mode 100644
index 0000000..f5fc693
--- /dev/null
+++ b/test/collections/acme.goodies/roles/baz/tasks/main.yml
@@ -0,0 +1,3 @@
+- name: "some task inside foo.bar collection"
+ debug:
+ msg: "hello world!"
diff --git a/test/collections/acme.goodies/tests/requirements.yml b/test/collections/acme.goodies/tests/requirements.yml
new file mode 100644
index 0000000..b004fa9
--- /dev/null
+++ b/test/collections/acme.goodies/tests/requirements.yml
@@ -0,0 +1,3 @@
+collections:
+ - name: ansible.posix
+ version: ">=1.0"
diff --git a/test/collections/acme.minimal/galaxy.yml b/test/collections/acme.minimal/galaxy.yml
new file mode 100644
index 0000000..a15e418
--- /dev/null
+++ b/test/collections/acme.minimal/galaxy.yml
@@ -0,0 +1,30 @@
+name: minimal
+namespace: acme
+version: 1.0.0
+readme: README.md
+authors:
+ - Red Hat
+description: Sample collection to use with molecule
+build_ignore:
+ - "*.egg-info"
+ - .DS_Store
+ - .eggs
+ - .gitignore
+ - .mypy_cache
+ - .pytest_cache
+ - .stestr
+ - .stestr.conf
+ - .tox
+ - .vscode
+ - MANIFEST.in
+ - build
+ - dist
+ - doc
+ - report.html
+ - setup.cfg
+ - setup.py
+ - "tests/unit/*.*"
+ - README.rst
+ - tox.ini
+
+license_file: LICENSE
diff --git a/test/conftest.py b/test/conftest.py
new file mode 100644
index 0000000..a1e4893
--- /dev/null
+++ b/test/conftest.py
@@ -0,0 +1,127 @@
+"""Pytest fixtures."""
+import importlib.metadata
+import json
+import pathlib
+import subprocess
+import sys
+from collections.abc import Generator
+from pathlib import Path
+from typing import Callable
+
+import pytest
+
+from ansible_compat.runtime import Runtime
+
+
+@pytest.fixture()
+# pylint: disable=unused-argument
+def runtime(scope: str = "session") -> Generator[Runtime, None, None]: # noqa: ARG001
+ """Isolated runtime fixture."""
+ instance = Runtime(isolated=True)
+ yield instance
+ instance.clean()
+
+
+@pytest.fixture()
+# pylint: disable=unused-argument
+def runtime_tmp(
+ tmp_path: pathlib.Path,
+ scope: str = "session", # noqa: ARG001
+) -> Generator[Runtime, None, None]:
+ """Isolated runtime fixture using a temp directory."""
+ instance = Runtime(project_dir=tmp_path, isolated=True)
+ yield instance
+ instance.clean()
+
+
+def query_pkg_version(pkg: str) -> str:
+ """Get the version of a current installed package.
+
+ :param pkg: Package name
+ :return: Package version
+ """
+ return importlib.metadata.version(pkg)
+
+
+@pytest.fixture()
+def pkg_version() -> Callable[[str], str]:
+ """Get the version of a current installed package.
+
+ :return: Callable function to get package version
+ """
+ return query_pkg_version
+
+
+class VirtualEnvironment:
+ """Virtualenv wrapper."""
+
+ def __init__(self, path: Path) -> None:
+ """Initialize.
+
+ :param path: Path to virtualenv
+ """
+ self.project = path
+ self.venv_path = self.project / "venv"
+ self.venv_bin_path = self.venv_path / "bin"
+ self.venv_python_path = self.venv_bin_path / "python"
+
+ def create(self) -> None:
+ """Create virtualenv."""
+ cmd = [str(sys.executable), "-m", "venv", str(self.venv_path)]
+ subprocess.check_call(args=cmd)
+ # Install this package into the virtual environment
+ self.install(f"{__file__}/../..")
+
+ def install(self, *packages: str) -> None:
+ """Install packages in virtualenv.
+
+ :param packages: Packages to install
+ """
+ cmd = [str(self.venv_python_path), "-m", "pip", "install", *packages]
+ subprocess.check_call(args=cmd)
+
+ def python_script_run(self, script: str) -> subprocess.CompletedProcess[str]:
+ """Run command in project dir using venv.
+
+ :param args: Command to run
+ """
+ proc = subprocess.run(
+ args=[self.venv_python_path, "-c", script],
+ capture_output=True,
+ cwd=self.project,
+ check=False,
+ text=True,
+ )
+ return proc
+
+ def site_package_dirs(self) -> list[Path]:
+ """Get site packages.
+
+ :return: List of site packages dirs
+ """
+ script = "import json, site; print(json.dumps(site.getsitepackages()))"
+ proc = subprocess.run(
+ args=[self.venv_python_path, "-c", script],
+ capture_output=True,
+ check=False,
+ text=True,
+ )
+ dirs = json.loads(proc.stdout)
+ if not isinstance(dirs, list):
+ msg = "Expected list of site packages"
+ raise TypeError(msg)
+ sanitized = list({Path(d).resolve() for d in dirs})
+ return sanitized
+
+
+@pytest.fixture(scope="module")
+def venv_module(tmp_path_factory: pytest.TempPathFactory) -> VirtualEnvironment:
+ """Create a virtualenv in a temporary directory.
+
+ :param tmp_path: pytest fixture for temp path
+ :return: VirtualEnvironment instance
+ """
+ test_project = tmp_path_factory.mktemp(basename="test_project-", numbered=True)
+ _venv = VirtualEnvironment(test_project)
+ _venv.create()
+ return _venv
diff --git a/test/roles/acme.missing_deps/meta/main.yml b/test/roles/acme.missing_deps/meta/main.yml
new file mode 100644
index 0000000..69b0417
--- /dev/null
+++ b/test/roles/acme.missing_deps/meta/main.yml
@@ -0,0 +1,8 @@
+---
+galaxy_info:
+ name: missing_deps
+ namespace: acme
+ description: foo
+ license: GPL
+ min_ansible_version: "2.10"
+ platforms: []
diff --git a/test/roles/acme.missing_deps/requirements.yml b/test/roles/acme.missing_deps/requirements.yml
new file mode 100644
index 0000000..53c5937
--- /dev/null
+++ b/test/roles/acme.missing_deps/requirements.yml
@@ -0,0 +1,2 @@
+collections:
+ - foo.bar # collection that does not exist, so we can test offline mode
diff --git a/test/roles/acme.sample2/meta/main.yml b/test/roles/acme.sample2/meta/main.yml
new file mode 100644
index 0000000..b682a84
--- /dev/null
+++ b/test/roles/acme.sample2/meta/main.yml
@@ -0,0 +1,16 @@
+---
+dependencies: []
+
+galaxy_info:
+ # role_name is missing in order to test deduction from folder name
+ author: acme
+ description: ACME sample role
+ company: "ACME LTD"
+ license: MIT
+ min_ansible_version: "2.9"
+ platforms:
+ - name: Debian
+ versions:
+ - any
+ galaxy_tags:
+ - samples
diff --git a/test/roles/ansible-role-sample/meta/main.yml b/test/roles/ansible-role-sample/meta/main.yml
new file mode 100644
index 0000000..bfddeb7
--- /dev/null
+++ b/test/roles/ansible-role-sample/meta/main.yml
@@ -0,0 +1,16 @@
+---
+dependencies: []
+
+galaxy_info:
+ role_name: sample
+ author: acme
+ description: ACME sample role
+ company: "ACME LTD"
+ license: MIT
+ min_ansible_version: "2.9"
+ platforms:
+ - name: Debian
+ versions:
+ - any
+ galaxy_tags:
+ - samples
diff --git a/test/roles/sample3/meta/main.yml b/test/roles/sample3/meta/main.yml
new file mode 100644
index 0000000..f479788
--- /dev/null
+++ b/test/roles/sample3/meta/main.yml
@@ -0,0 +1,16 @@
+---
+dependencies: []
+
+galaxy_info:
+ # role_name is missing in order to test deduction from folder name
+ author: acme
+ description: ACME samble role
+ company: "ACME LTD"
+ license: MIT
+ min_ansible_version: "2.9"
+ platforms:
+ - name: Debian
+ versions:
+ - any
+ galaxy_tags:
+ - samples
diff --git a/test/roles/sample4/meta/main.yml b/test/roles/sample4/meta/main.yml
new file mode 100644
index 0000000..f479788
--- /dev/null
+++ b/test/roles/sample4/meta/main.yml
@@ -0,0 +1,16 @@
+---
+dependencies: []
+
+galaxy_info:
+ # role_name is missing in order to test deduction from folder name
+ author: acme
+ description: ACME samble role
+ company: "ACME LTD"
+ license: MIT
+ min_ansible_version: "2.9"
+ platforms:
+ - name: Debian
+ versions:
+ - any
+ galaxy_tags:
+ - samples
diff --git a/test/test_api.py b/test/test_api.py
new file mode 100644
index 0000000..80b38ba
--- /dev/null
+++ b/test/test_api.py
@@ -0,0 +1,5 @@
+"""Tests for ansible_compat package."""
+
+
+def test_placeholder() -> None:
+ """Placeholder test."""
diff --git a/test/test_config.py b/test/test_config.py
new file mode 100644
index 0000000..4f854ae
--- /dev/null
+++ b/test/test_config.py
@@ -0,0 +1,86 @@
+"""Tests for ansible_compat.config submodule."""
+import copy
+import subprocess
+
+import pytest
+from _pytest.monkeypatch import MonkeyPatch
+from packaging.version import Version
+
+from ansible_compat.config import AnsibleConfig, ansible_version, parse_ansible_version
+from ansible_compat.errors import InvalidPrerequisiteError, MissingAnsibleError
+
+
+def test_config() -> None:
+ """Checks that config vars are loaded with their expected type."""
+ config = AnsibleConfig()
+ assert isinstance(config.ACTION_WARNINGS, bool)
+ assert isinstance(config.CACHE_PLUGIN_PREFIX, str)
+ assert isinstance(config.CONNECTION_FACTS_MODULES, dict)
+ assert config.ANSIBLE_COW_PATH is None
+ assert isinstance(config.NETWORK_GROUP_MODULES, list)
+ assert isinstance(config.DEFAULT_GATHER_TIMEOUT, (int, type(None)))
+
+ # check lowercase and older name aliasing
+ assert isinstance(config.collections_paths, list)
+ assert isinstance(config.collections_path, list)
+ assert config.collections_paths == config.collections_path
+
+ # check if we can access the special data member
+ assert config.data["ACTION_WARNINGS"] == config.ACTION_WARNINGS
+
+ with pytest.raises(AttributeError):
+ _ = config.THIS_DOES_NOT_EXIST
+
+
+def test_config_with_dump() -> None:
+ """Tests that config can parse given dumps."""
+ config = AnsibleConfig(config_dump="ACTION_WARNINGS(default) = True")
+ assert config.ACTION_WARNINGS is True
+
+
+def test_config_copy() -> None:
+ """Checks ability to use copy/deepcopy."""
+ config = AnsibleConfig()
+ new_config = copy.copy(config)
+ assert isinstance(new_config, AnsibleConfig)
+ assert new_config is not config
+ # deepcopy testing
+ new_config = copy.deepcopy(config)
+ assert isinstance(new_config, AnsibleConfig)
+ assert new_config is not config
+
+
+def test_parse_ansible_version_fail() -> None:
+ """Checks that parse_ansible_version raises an error on invalid input."""
+ with pytest.raises(
+ InvalidPrerequisiteError,
+ match="Unable to parse ansible cli version",
+ ):
+ parse_ansible_version("foo")
+
+
+def test_ansible_version_missing(monkeypatch: MonkeyPatch) -> None:
+ """Validate ansible_version behavior when ansible is missing."""
+ monkeypatch.setattr(
+ "subprocess.run",
+ lambda *args, **kwargs: subprocess.CompletedProcess( # noqa: ARG005
+ args=[],
+ returncode=1,
+ ),
+ )
+ with pytest.raises(
+ MissingAnsibleError,
+ match="Unable to find a working copy of ansible executable.",
+ ):
+ # bypassing lru cache
+ ansible_version.__wrapped__()
+
+
+def test_ansible_version() -> None:
+ """Validate ansible_version behavior."""
+ assert ansible_version() >= Version("1.0")
+
+
+def test_ansible_version_arg() -> None:
+ """Validate ansible_version behavior."""
+ assert ansible_version("2.0") >= Version("1.0")
diff --git a/test/test_configuration_example.py b/test/test_configuration_example.py
new file mode 100644
index 0000000..3a2c9b7
--- /dev/null
+++ b/test/test_configuration_example.py
@@ -0,0 +1,12 @@
+"""Sample usage of AnsibleConfig."""
+from ansible_compat.config import AnsibleConfig
+
+
+def test_example_config() -> None:
+ """Test basic functionality of AnsibleConfig."""
+ cfg = AnsibleConfig()
+ assert isinstance(cfg.ACTION_WARNINGS, bool)
+ # you can also use lowercase:
+ assert isinstance(cfg.action_warnings, bool)
+ # you can also use it as dictionary
+ assert cfg["action_warnings"] == cfg.action_warnings
diff --git a/test/test_loaders.py b/test/test_loaders.py
new file mode 100644
index 0000000..7a91a4c
--- /dev/null
+++ b/test/test_loaders.py
@@ -0,0 +1,9 @@
+"""Test for ansible_compat.loaders module."""
+from pathlib import Path
+
+from ansible_compat.loaders import colpath_from_path
+
+
+def test_colpath_from_path() -> None:
+ """Test colpath_from_path non existing path."""
+ assert colpath_from_path(Path("/foo/bar/")) is None
diff --git a/test/test_prerun.py b/test/test_prerun.py
new file mode 100644
index 0000000..1549756
--- /dev/null
+++ b/test/test_prerun.py
@@ -0,0 +1,11 @@
+"""Tests for ansible_compat.prerun module."""
+from pathlib import Path
+
+from ansible_compat.prerun import get_cache_dir
+
+
+def test_get_cache_dir_relative() -> None:
+ """Test behaviors of get_cache_dir."""
+ relative_path = Path()
+ abs_path = relative_path.resolve()
+ assert get_cache_dir(relative_path) == get_cache_dir(abs_path)
diff --git a/test/test_runtime.py b/test/test_runtime.py
new file mode 100644
index 0000000..2af343d
--- /dev/null
+++ b/test/test_runtime.py
@@ -0,0 +1,893 @@
+"""Tests for Runtime class."""
+# pylint: disable=protected-access
+from __future__ import annotations
+
+import logging
+import os
+import pathlib
+import subprocess
+from contextlib import contextmanager
+from pathlib import Path
+from shutil import rmtree
+from typing import TYPE_CHECKING, Any
+
+import pytest
+from packaging.version import Version
+
+from ansible_compat.config import ansible_version
+from ansible_compat.constants import INVALID_PREREQUISITES_RC
+from ansible_compat.errors import (
+ AnsibleCommandError,
+ AnsibleCompatError,
+ InvalidPrerequisiteError,
+)
+from ansible_compat.runtime import (
+ CompletedProcess,
+ Runtime,
+ _get_galaxy_role_name,
+ is_url,
+ search_galaxy_paths,
+)
+
+if TYPE_CHECKING:
+ from collections.abc import Iterator
+
+ from _pytest.monkeypatch import MonkeyPatch
+ from pytest_mock import MockerFixture
+
+
+def test_runtime_version(runtime: Runtime) -> None:
+ """Tests version property."""
+ version = runtime.version
+ assert isinstance(version, Version)
+ # tests that caching property value worked (coverage)
+ assert version == runtime.version
+
+
+@pytest.mark.parametrize(
+ "require_module",
+ (True, False),
+ ids=("module-required", "module-unrequired"),
+)
+def test_runtime_version_outdated(require_module: bool) -> None:
+ """Checks that instantiation raises if version is outdated."""
+ with pytest.raises(RuntimeError, match="Found incompatible version of ansible"):
+ Runtime(min_required_version="9999.9.9", require_module=require_module)
+
+
+def test_runtime_missing_ansible_module(monkeypatch: MonkeyPatch) -> None:
+ """Checks that we produce a RuntimeError when ansible module is missing."""
+
+ class RaiseException:
+ """Class to raise an exception."""
+
+ def __init__(
+ self,
+ *args: Any, # noqa: ARG002,ANN401
+ **kwargs: Any, # noqa: ARG002,ANN401
+ ) -> None:
+ raise ModuleNotFoundError
+
+ monkeypatch.setattr("importlib.import_module", RaiseException)
+
+ with pytest.raises(RuntimeError, match="Unable to find Ansible python module."):
+ Runtime(require_module=True)
+
+
+def test_runtime_mismatch_ansible_module(monkeypatch: MonkeyPatch) -> None:
+ """Test that missing module is detected."""
+ monkeypatch.setattr("ansible.release.__version__", "0.0.0", raising=False)
+ with pytest.raises(RuntimeError, match="versions do not match"):
+ Runtime(require_module=True)
+
+
+def test_runtime_require_module() -> None:
+ """Check that require_module successful pass."""
+ Runtime(require_module=True)
+ # Now we try to set the collection path, something to check if that is
+ # causing an exception, as 2.15 introduced new init code.
+ from ansible.utils.collection_loader import ( # pylint: disable=import-outside-toplevel
+ AnsibleCollectionConfig,
+ )
+
+ AnsibleCollectionConfig.playbook_paths = "."
+ # Calling it again in order to see that it does not produce UserWarning: AnsibleCollectionFinder has already been configured
+ # which is done by Ansible core 2.15+. We added special code inside Runtime
+ # that should avoid initializing twice and raise that warning.
+ Runtime(require_module=True)
+
+
+def test_runtime_version_fail_module(mocker: MockerFixture) -> None:
+ """Tests for failure to detect Ansible version."""
+ patched = mocker.patch(
+ "ansible_compat.runtime.parse_ansible_version",
+ autospec=True,
+ )
+ patched.side_effect = InvalidPrerequisiteError(
+ "Unable to parse ansible cli version",
+ )
+ runtime = Runtime()
+ with pytest.raises(
+ InvalidPrerequisiteError,
+ match="Unable to parse ansible cli version",
+ ):
+ _ = runtime.version # pylint: disable=pointless-statement
+
+
+def test_runtime_version_fail_cli(mocker: MockerFixture) -> None:
+ """Tests for failure to detect Ansible version."""
+ mocker.patch(
+ "ansible_compat.runtime.Runtime.run",
+ return_value=CompletedProcess(
+ ["x"],
+ returncode=123,
+ stdout="oops",
+ stderr="some error",
+ ),
+ autospec=True,
+ )
+ runtime = Runtime()
+ with pytest.raises(
+ RuntimeError,
+ match="Unable to find a working copy of ansible executable.",
+ ):
+ _ = runtime.version # pylint: disable=pointless-statement
+
+
+def test_runtime_prepare_ansible_paths_validation() -> None:
+ """Check that we validate collection_path."""
+ runtime = Runtime()
+ runtime.config.collections_paths = "invalid-value" # type: ignore[assignment]
+ with pytest.raises(RuntimeError, match="Unexpected ansible configuration"):
+ runtime._prepare_ansible_paths()
+
+
+@pytest.mark.parametrize(
+ ("folder", "role_name", "isolated"),
+ (
+ ("ansible-role-sample", "acme.sample", True),
+ ("acme.sample2", "acme.sample2", True),
+ ("sample3", "acme.sample3", True),
+ ("sample4", "acme.sample4", False),
+ ),
+ ids=("1", "2", "3", "4"),
+)
+def test_runtime_install_role(
+ caplog: pytest.LogCaptureFixture,
+ folder: str,
+ role_name: str,
+ isolated: bool,
+) -> None:
+ """Checks that we can install roles."""
+ caplog.set_level(logging.INFO)
+ project_dir = Path(__file__).parent / "roles" / folder
+ runtime = Runtime(isolated=isolated, project_dir=project_dir)
+ runtime.prepare_environment(install_local=True)
+ # check that role appears as installed now
+ result = runtime.run(["ansible-galaxy", "list"])
+ assert result.returncode == 0, result
+ assert role_name in result.stdout
+ if isolated:
+ assert pathlib.Path(f"{runtime.cache_dir}/roles/{role_name}").is_symlink()
+ else:
+ assert pathlib.Path(
+ f"{Path(runtime.config.default_roles_path[0]).expanduser()}/{role_name}",
+ ).is_symlink()
+ runtime.clean()
+ # also test that clean does not break when cache_dir is missing
+ tmp_dir = runtime.cache_dir
+ runtime.cache_dir = None
+ runtime.clean()
+ runtime.cache_dir = tmp_dir
+
+
+def test_prepare_environment_with_collections(tmp_path: pathlib.Path) -> None:
+ """Check that collections are correctly installed."""
+ runtime = Runtime(isolated=True, project_dir=tmp_path)
+ runtime.prepare_environment(required_collections={"community.molecule": "0.1.0"})
+
+
+def test_runtime_install_requirements_missing_file() -> None:
+ """Check that missing requirements file is ignored."""
+ # Do not rely on this behavior, it may be removed in the future
+ runtime = Runtime()
+ runtime.install_requirements(Path("/that/does/not/exist"))
+
+
+@pytest.mark.parametrize(
+ ("file", "exc", "msg"),
+ (
+ (
+ Path("/dev/null"),
+ InvalidPrerequisiteError,
+ "file is not a valid Ansible requirements file",
+ ),
+ (
+ Path(__file__).parent / "assets" / "requirements-invalid-collection.yml",
+ AnsibleCommandError,
+ "Got 1 exit code while running: ansible-galaxy",
+ ),
+ (
+ Path(__file__).parent / "assets" / "requirements-invalid-role.yml",
+ AnsibleCommandError,
+ "Got 1 exit code while running: ansible-galaxy",
+ ),
+ ),
+ ids=("empty", "invalid-collection", "invalid-role"),
+)
+def test_runtime_install_requirements_invalid_file(
+ file: Path,
+ exc: type[Any],
+ msg: str,
+) -> None:
+ """Check that invalid requirements file is raising."""
+ runtime = Runtime()
+ with pytest.raises(
+ exc,
+ match=msg,
+ ):
+ runtime.install_requirements(file)
+
+
+@contextmanager
+def cwd(path: Path) -> Iterator[None]:
+ """Context manager for temporary changing current working directory."""
+ old_pwd = Path.cwd()
+ os.chdir(path)
+ try:
+ yield
+ finally:
+ os.chdir(old_pwd)
+
+
+def test_prerun_reqs_v1(caplog: pytest.LogCaptureFixture) -> None:
+ """Checks that the linter can auto-install requirements v1 when found."""
+ runtime = Runtime(verbosity=1)
+ path = Path(__file__).parent.parent / "examples" / "reqs_v1"
+ with cwd(path):
+ runtime.prepare_environment()
+ assert any(
+ msg.startswith("Running ansible-galaxy role install") for msg in caplog.messages
+ )
+ assert all(
+ "Running ansible-galaxy collection install" not in msg
+ for msg in caplog.messages
+ )
+
+
+def test_prerun_reqs_v2(caplog: pytest.LogCaptureFixture) -> None:
+ """Checks that the linter can auto-install requirements v2 when found."""
+ runtime = Runtime(verbosity=1)
+ path = (Path(__file__).parent.parent / "examples" / "reqs_v2").resolve()
+ with cwd(path):
+ runtime.prepare_environment()
+ assert any(
+ msg.startswith("Running ansible-galaxy role install")
+ for msg in caplog.messages
+ )
+ assert any(
+ msg.startswith("Running ansible-galaxy collection install")
+ for msg in caplog.messages
+ )
+
+
+def test_prerun_reqs_broken(runtime: Runtime) -> None:
+ """Checks that the we report invalid requirements.yml file."""
+ path = (Path(__file__).parent.parent / "examples" / "reqs_broken").resolve()
+ with cwd(path), pytest.raises(InvalidPrerequisiteError):
+ runtime.prepare_environment()
+
+
+def test__update_env_no_old_value_no_default_no_value(monkeypatch: MonkeyPatch) -> None:
+ """Make sure empty value does not touch environment."""
+ monkeypatch.delenv("DUMMY_VAR", raising=False)
+
+ runtime = Runtime()
+ runtime._update_env("DUMMY_VAR", [])
+
+ assert "DUMMY_VAR" not in runtime.environ
+
+
+def test__update_env_no_old_value_no_value(monkeypatch: MonkeyPatch) -> None:
+ """Make sure empty value does not touch environment."""
+ monkeypatch.delenv("DUMMY_VAR", raising=False)
+
+ runtime = Runtime()
+ runtime._update_env("DUMMY_VAR", [], "a:b")
+
+ assert "DUMMY_VAR" not in runtime.environ
+
+
+def test__update_env_no_default_no_value(monkeypatch: MonkeyPatch) -> None:
+ """Make sure empty value does not touch environment."""
+ monkeypatch.setenv("DUMMY_VAR", "a:b")
+
+ runtime = Runtime()
+ runtime._update_env("DUMMY_VAR", [])
+
+ assert runtime.environ["DUMMY_VAR"] == "a:b"
+
+
+@pytest.mark.parametrize(
+ ("value", "result"),
+ (
+ (["a"], "a"),
+ (["a", "b"], "a:b"),
+ (["a", "b", "c"], "a:b:c"),
+ ),
+)
+def test__update_env_no_old_value_no_default(
+ monkeypatch: MonkeyPatch,
+ value: list[str],
+ result: str,
+) -> None:
+ """Values are concatenated using : as the separator."""
+ monkeypatch.delenv("DUMMY_VAR", raising=False)
+
+ runtime = Runtime()
+ runtime._update_env("DUMMY_VAR", value)
+
+ assert runtime.environ["DUMMY_VAR"] == result
+
+
+@pytest.mark.parametrize(
+ ("default", "value", "result"),
+ (
+ ("a:b", ["c"], "c:a:b"),
+ ("a:b", ["c:d"], "c:d:a:b"),
+ ),
+)
+def test__update_env_no_old_value(
+ monkeypatch: MonkeyPatch,
+ default: str,
+ value: list[str],
+ result: str,
+) -> None:
+ """Values are appended to default value."""
+ monkeypatch.delenv("DUMMY_VAR", raising=False)
+
+ runtime = Runtime()
+ runtime._update_env("DUMMY_VAR", value, default)
+
+ assert runtime.environ["DUMMY_VAR"] == result
+
+
+@pytest.mark.parametrize(
+ ("old_value", "value", "result"),
+ (
+ ("a:b", ["c"], "c:a:b"),
+ ("a:b", ["c:d"], "c:d:a:b"),
+ ),
+)
+def test__update_env_no_default(
+ monkeypatch: MonkeyPatch,
+ old_value: str,
+ value: list[str],
+ result: str,
+) -> None:
+ """Values are appended to preexisting value."""
+ monkeypatch.setenv("DUMMY_VAR", old_value)
+
+ runtime = Runtime()
+ runtime._update_env("DUMMY_VAR", value)
+
+ assert runtime.environ["DUMMY_VAR"] == result
+
+
+@pytest.mark.parametrize(
+ ("old_value", "default", "value", "result"),
+ (
+ ("", "", ["e"], "e"),
+ ("a", "", ["e"], "e:a"),
+ ("", "c", ["e"], "e"),
+ ("a", "c", ["e:f"], "e:f:a"),
+ ),
+)
+def test__update_env(
+ monkeypatch: MonkeyPatch,
+ old_value: str,
+ default: str, # pylint: disable=unused-argument # noqa: ARG001
+ value: list[str],
+ result: str,
+) -> None:
+ """Defaults are ignored when preexisting value is present."""
+ monkeypatch.setenv("DUMMY_VAR", old_value)
+
+ runtime = Runtime()
+ runtime._update_env("DUMMY_VAR", value)
+
+ assert runtime.environ["DUMMY_VAR"] == result
+
+
+def test_require_collection_wrong_version(runtime: Runtime) -> None:
+ """Tests behaviour of require_collection."""
+ subprocess.check_output(
+ [ # noqa: S603
+ "ansible-galaxy",
+ "collection",
+ "install",
+ "examples/reqs_v2/community-molecule-0.1.0.tar.gz",
+ "-p",
+ "~/.ansible/collections",
+ ],
+ )
+ with pytest.raises(InvalidPrerequisiteError) as pytest_wrapped_e:
+ runtime.require_collection("community.molecule", "9999.9.9")
+ assert pytest_wrapped_e.type == InvalidPrerequisiteError
+ assert pytest_wrapped_e.value.code == INVALID_PREREQUISITES_RC
+
+
+def test_require_collection_invalid_name(runtime: Runtime) -> None:
+ """Check that require_collection raise with invalid collection name."""
+ with pytest.raises(
+ InvalidPrerequisiteError,
+ match="Invalid collection name supplied:",
+ ):
+ runtime.require_collection("that-is-invalid")
+
+
+def test_require_collection_invalid_collections_path(runtime: Runtime) -> None:
+ """Check that require_collection raise with invalid collections path."""
+ runtime.config.collections_paths = "/that/is/invalid" # type: ignore[assignment]
+ with pytest.raises(
+ InvalidPrerequisiteError,
+ match="Unable to determine ansible collection paths",
+ ):
+ runtime.require_collection("community.molecule")
+
+
+def test_require_collection_preexisting_broken(tmp_path: pathlib.Path) -> None:
+ """Check that require_collection raise with broken pre-existing collection."""
+ runtime = Runtime(isolated=True, project_dir=tmp_path)
+ dest_path: str = runtime.config.collections_paths[0]
+ dest = pathlib.Path(dest_path) / "ansible_collections" / "foo" / "bar"
+ dest.mkdir(parents=True, exist_ok=True)
+ with pytest.raises(InvalidPrerequisiteError, match="missing MANIFEST.json"):
+ runtime.require_collection("foo.bar")
+
+
+def test_require_collection(runtime_tmp: Runtime) -> None:
+ """Check that require collection successful install case."""
+ runtime_tmp.require_collection("community.molecule", "0.1.0")
+
+
+@pytest.mark.parametrize(
+ ("name", "version", "install"),
+ (
+ ("fake_namespace.fake_name", None, True),
+ ("fake_namespace.fake_name", "9999.9.9", True),
+ ("fake_namespace.fake_name", None, False),
+ ),
+ ids=("a", "b", "c"),
+)
+def test_require_collection_missing(
+ name: str,
+ version: str,
+ install: bool,
+ runtime: Runtime,
+) -> None:
+ """Tests behaviour of require_collection, missing case."""
+ with pytest.raises(AnsibleCompatError) as pytest_wrapped_e:
+ runtime.require_collection(name=name, version=version, install=install)
+ assert pytest_wrapped_e.type == InvalidPrerequisiteError
+ assert pytest_wrapped_e.value.code == INVALID_PREREQUISITES_RC
+
+
+def test_install_collection(runtime: Runtime) -> None:
+ """Check that valid collection installs do not fail."""
+ runtime.install_collection("examples/reqs_v2/community-molecule-0.1.0.tar.gz")
+
+
+def test_install_collection_git(runtime: Runtime) -> None:
+ """Check that valid collection installs do not fail."""
+ runtime.install_collection(
+ "git+https://github.com/ansible-collections/ansible.posix,main",
+ )
+
+
+def test_install_collection_dest(runtime: Runtime, tmp_path: pathlib.Path) -> None:
+ """Check that valid collection to custom destination passes."""
+ # Since Ansible 2.15.3 there is no guarantee that this will install the collection at requested path
+ # as it might decide to not install anything if requirement is already present at another location.
+ runtime.install_collection(
+ "examples/reqs_v2/community-molecule-0.1.0.tar.gz",
+ destination=tmp_path,
+ )
+ runtime.load_collections()
+ for collection in runtime.collections:
+ if collection == "community.molecule":
+ return
+ msg = "Failed to find collection as installed."
+ raise AssertionError(msg)
+
+
+def test_install_collection_fail(runtime: Runtime) -> None:
+ """Check that invalid collection install fails."""
+ with pytest.raises(AnsibleCompatError) as pytest_wrapped_e:
+ runtime.install_collection("community.molecule:>=9999.0")
+ assert pytest_wrapped_e.type == InvalidPrerequisiteError
+ assert pytest_wrapped_e.value.code == INVALID_PREREQUISITES_RC
+
+
+def test_install_galaxy_role(runtime_tmp: Runtime) -> None:
+ """Check install role with empty galaxy file."""
+ pathlib.Path(f"{runtime_tmp.project_dir}/galaxy.yml").touch()
+ pathlib.Path(f"{runtime_tmp.project_dir}/meta").mkdir()
+ pathlib.Path(f"{runtime_tmp.project_dir}/meta/main.yml").touch()
+ # this should only raise a warning
+ runtime_tmp._install_galaxy_role(runtime_tmp.project_dir, role_name_check=1)
+ # this should test the bypass role name check path
+ runtime_tmp._install_galaxy_role(runtime_tmp.project_dir, role_name_check=2)
+ # this should raise an error
+ with pytest.raises(
+ InvalidPrerequisiteError,
+ match="does not follow current galaxy requirements",
+ ):
+ runtime_tmp._install_galaxy_role(runtime_tmp.project_dir, role_name_check=0)
+
+
+def test_install_galaxy_role_unlink(
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """Test ability to unlink incorrect symlinked roles."""
+ runtime_tmp = Runtime(verbosity=1)
+ runtime_tmp.prepare_environment()
+ pathlib.Path(f"{runtime_tmp.cache_dir}/roles").mkdir(parents=True, exist_ok=True)
+ pathlib.Path(f"{runtime_tmp.cache_dir}/roles/acme.get_rich").symlink_to("/dev/null")
+ pathlib.Path(f"{runtime_tmp.project_dir}/meta").mkdir()
+ pathlib.Path(f"{runtime_tmp.project_dir}/meta/main.yml").write_text(
+ """galaxy_info:
+ role_name: get_rich
+ namespace: acme
+""",
+ encoding="utf-8",
+ )
+ runtime_tmp._install_galaxy_role(runtime_tmp.project_dir)
+ assert "symlink to current repository" in caplog.text
+
+
+def test_install_galaxy_role_bad_namespace(runtime_tmp: Runtime) -> None:
+ """Check install role with bad namespace in galaxy info."""
+ pathlib.Path(f"{runtime_tmp.project_dir}/meta").mkdir()
+ pathlib.Path(f"{runtime_tmp.project_dir}/meta/main.yml").write_text(
+ """galaxy_info:
+ role_name: foo
+ author: bar
+ namespace: ["xxx"]
+""",
+ )
+ # this should raise an error regardless the role_name_check value
+ with pytest.raises(AnsibleCompatError, match="Role namespace must be string, not"):
+ runtime_tmp._install_galaxy_role(runtime_tmp.project_dir, role_name_check=1)
+
+
+@pytest.mark.parametrize(
+ "galaxy_info",
+ (
+ """galaxy_info:
+ role_name: foo-bar
+ namespace: acme
+""",
+ """galaxy_info:
+ role_name: foo-bar
+""",
+ ),
+ ids=("bad-name", "bad-name-without-namespace"),
+)
+def test_install_galaxy_role_name_role_name_check_equals_to_1(
+ runtime_tmp: Runtime,
+ galaxy_info: str,
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """Check install role with bad role name in galaxy info."""
+ caplog.set_level(logging.WARN)
+ pathlib.Path(f"{runtime_tmp.project_dir}/meta").mkdir()
+ pathlib.Path(f"{runtime_tmp.project_dir}/meta/main.yml").write_text(
+ galaxy_info,
+ encoding="utf-8",
+ )
+
+ runtime_tmp._install_galaxy_role(runtime_tmp.project_dir, role_name_check=1)
+ assert "Computed fully qualified role name of " in caplog.text
+
+
+def test_install_galaxy_role_no_checks(runtime_tmp: Runtime) -> None:
+ """Check install role with bad namespace in galaxy info."""
+ runtime_tmp.prepare_environment()
+ pathlib.Path(f"{runtime_tmp.project_dir}/meta").mkdir()
+ pathlib.Path(f"{runtime_tmp.project_dir}/meta/main.yml").write_text(
+ """galaxy_info:
+ role_name: foo
+ author: bar
+ namespace: acme
+""",
+ )
+ runtime_tmp._install_galaxy_role(runtime_tmp.project_dir, role_name_check=2)
+ result = runtime_tmp.run(["ansible-galaxy", "list"])
+ assert "- acme.foo," in result.stdout
+ assert result.returncode == 0, result
+
+
+def test_upgrade_collection(runtime_tmp: Runtime) -> None:
+ """Check that collection upgrade is possible."""
+ # ensure that we inject our tmp folders in ansible paths
+ runtime_tmp.prepare_environment()
+
+ # we install specific oudated version of a collection
+ runtime_tmp.install_collection("examples/reqs_v2/community-molecule-0.1.0.tar.gz")
+ with pytest.raises(
+ InvalidPrerequisiteError,
+ match="Found community.molecule collection 0.1.0 but 9.9.9 or newer is required.",
+ ):
+ # we check that when install=False, we raise error
+ runtime_tmp.require_collection("community.molecule", "9.9.9", install=False)
+ # this should not fail, as we have this version
+ runtime_tmp.require_collection("community.molecule", "0.1.0")
+
+
+def test_require_collection_no_cache_dir() -> None:
+ """Check require_collection without a cache directory."""
+ runtime = Runtime()
+ assert not runtime.cache_dir
+ runtime.require_collection("community.molecule", "0.1.0", install=True)
+
+
+def test_runtime_env_ansible_library(monkeypatch: MonkeyPatch) -> None:
+ """Verify that custom path specified using ANSIBLE_LIBRARY is not lost."""
+ path_name = "foo"
+ monkeypatch.setenv("ANSIBLE_LIBRARY", path_name)
+
+ path_name = os.path.realpath(path_name)
+ runtime = Runtime()
+ runtime.prepare_environment()
+ assert path_name in runtime.config.default_module_path
+
+
+@pytest.mark.parametrize(
+ ("lower", "upper", "expected"),
+ (
+ ("1.0", "9999.0", True),
+ (None, "9999.0", True),
+ ("1.0", None, True),
+ ("9999.0", None, False),
+ (None, "1.0", False),
+ ),
+ ids=("1", "2", "3", "4", "5"),
+)
+def test_runtime_version_in_range(
+ lower: str | None,
+ upper: str | None,
+ expected: bool,
+) -> None:
+ """Validate functioning of version_in_range."""
+ runtime = Runtime()
+ assert runtime.version_in_range(lower=lower, upper=upper) is expected
+
+
+@pytest.mark.parametrize(
+ ("path", "scenario", "expected_collections"),
+ (
+ pytest.param(
+ "test/collections/acme.goodies",
+ "default",
+ [
+ "ansible.posix", # from tests/requirements.yml
+ "ansible.utils", # from galaxy.yml
+ "community.molecule", # from galaxy.yml
+ "community.crypto", # from galaxy.yml as a git dependency
+ ],
+ id="normal",
+ ),
+ pytest.param(
+ "test/collections/acme.goodies/roles/baz",
+ "deep_scenario",
+ ["community.molecule"],
+ id="deep",
+ ),
+ ),
+)
+def test_install_collection_from_disk(
+ path: str,
+ scenario: str,
+ expected_collections: list[str],
+) -> None:
+ """Tests ability to install a local collection."""
+ # ensure we do not have acme.goodies installed in user directory as it may
+ # produce false positives
+ rmtree(
+ pathlib.Path(
+ "~/.ansible/collections/ansible_collections/acme/goodies",
+ ).expanduser(),
+ ignore_errors=True,
+ )
+ with cwd(Path(path)):
+ runtime = Runtime(isolated=True)
+ # this should call install_collection_from_disk(".")
+ runtime.prepare_environment(install_local=True)
+ # that molecule converge playbook can be used without molecule and
+ # should validate that the installed collection is available.
+ result = runtime.run(["ansible-playbook", f"molecule/{scenario}/converge.yml"])
+ assert result.returncode == 0, result.stdout
+ runtime.load_collections()
+ for collection_name in expected_collections:
+ assert (
+ collection_name in runtime.collections
+ ), f"{collection_name} not found in {runtime.collections.keys()}"
+ runtime.clean()
+
+
+def test_install_collection_from_disk_fail() -> None:
+ """Tests that we fail to install a broken collection."""
+ with cwd(Path("test/collections/acme.broken")):
+ runtime = Runtime(isolated=True)
+ with pytest.raises(RuntimeError) as exc_info:
+ runtime.prepare_environment(install_local=True)
+ # based on version of Ansible used, we might get a different error,
+ # but both errors should be considered acceptable
+ assert exc_info.type in (
+ RuntimeError,
+ AnsibleCompatError,
+ AnsibleCommandError,
+ InvalidPrerequisiteError,
+ )
+ assert exc_info.match(
+ "(is missing the following mandatory|Got 1 exit code while running: ansible-galaxy collection build)",
+ )
+
+
+def test_prepare_environment_offline_role() -> None:
+ """Ensure that we can make use of offline roles."""
+ with cwd(Path("test/roles/acme.missing_deps")):
+ runtime = Runtime(isolated=True)
+ runtime.prepare_environment(install_local=True, offline=True)
+
+
+def test_runtime_run(runtime: Runtime) -> None:
+ """Check if tee and non tee mode return same kind of results."""
+ result1 = runtime.run(["seq", "10"])
+ result2 = runtime.run(["seq", "10"], tee=True)
+ assert result1.returncode == result2.returncode
+ assert result1.stderr == result2.stderr
+ assert result1.stdout == result2.stdout
+
+
+def test_runtime_exec_cwd(runtime: Runtime) -> None:
+ """Check if passing cwd works as expected."""
+ path = Path("/")
+ result1 = runtime.run(["pwd"], cwd=path)
+ result2 = runtime.run(["pwd"])
+ assert result1.stdout.rstrip() == str(path)
+ assert result1.stdout != result2.stdout
+
+
+def test_runtime_exec_env(runtime: Runtime) -> None:
+ """Check if passing env works."""
+ result = runtime.run(["printenv", "FOO"])
+ assert not result.stdout
+
+ result = runtime.run(["printenv", "FOO"], env={"FOO": "bar"})
+ assert result.stdout.rstrip() == "bar"
+
+ runtime.environ["FOO"] = "bar"
+ result = runtime.run(["printenv", "FOO"])
+ assert result.stdout.rstrip() == "bar"
+
+
+def test_runtime_plugins(runtime: Runtime) -> None:
+ """Tests ability to access detected plugins."""
+ assert len(runtime.plugins.cliconf) == 0
+ # ansible.netcommon.restconf might be in httpapi
+ assert isinstance(runtime.plugins.httpapi, dict)
+ # "ansible.netcommon.default" might be in runtime.plugins.netconf
+ assert isinstance(runtime.plugins.netconf, dict)
+ assert isinstance(runtime.plugins.role, dict)
+ assert "become" in runtime.plugins.keyword
+
+ if ansible_version() < Version("2.14.0"):
+ assert "sudo" in runtime.plugins.become
+ assert "memory" in runtime.plugins.cache
+ assert "default" in runtime.plugins.callback
+ assert "local" in runtime.plugins.connection
+ assert "ini" in runtime.plugins.inventory
+ assert "env" in runtime.plugins.lookup
+ assert "sh" in runtime.plugins.shell
+ assert "host_group_vars" in runtime.plugins.vars
+ assert "file" in runtime.plugins.module
+ assert "free" in runtime.plugins.strategy
+ # ansible-doc below 2.14 does not support listing 'test' and 'filter' types:
+ with pytest.raises(RuntimeError):
+ assert "is_abs" in runtime.plugins.test
+ with pytest.raises(RuntimeError):
+ assert "bool" in runtime.plugins.filter
+ else:
+ assert "ansible.builtin.sudo" in runtime.plugins.become
+ assert "ansible.builtin.memory" in runtime.plugins.cache
+ assert "ansible.builtin.default" in runtime.plugins.callback
+ assert "ansible.builtin.local" in runtime.plugins.connection
+ assert "ansible.builtin.ini" in runtime.plugins.inventory
+ assert "ansible.builtin.env" in runtime.plugins.lookup
+ assert "ansible.builtin.sh" in runtime.plugins.shell
+ assert "ansible.builtin.host_group_vars" in runtime.plugins.vars
+ assert "ansible.builtin.file" in runtime.plugins.module
+ assert "ansible.builtin.free" in runtime.plugins.strategy
+ assert "ansible.builtin.is_abs" in runtime.plugins.test
+ assert "ansible.builtin.bool" in runtime.plugins.filter
+
+
+@pytest.mark.parametrize(
+ ("path", "result"),
+ (
+ pytest.param(
+ "test/assets/galaxy_paths",
+ ["test/assets/galaxy_paths/foo/galaxy.yml"],
+ id="1",
+ ),
+ pytest.param(
+ "test/collections",
+ [], # should find nothing because these folders are not valid namespaces
+ id="2",
+ ),
+ pytest.param(
+ "test/assets/galaxy_paths/foo",
+ ["test/assets/galaxy_paths/foo/galaxy.yml"],
+ id="3",
+ ),
+ ),
+)
+def test_galaxy_path(path: str, result: list[str]) -> None:
+ """Check behavior of galaxy path search."""
+ assert search_galaxy_paths(Path(path)) == result
+
+
+@pytest.mark.parametrize(
+ ("name", "result"),
+ (
+ pytest.param(
+ "foo",
+ False,
+ id="0",
+ ),
+ pytest.param(
+ "git+git",
+ True,
+ id="1",
+ ),
+ pytest.param(
+ "git@acme.com",
+ True,
+ id="2",
+ ),
+ ),
+)
+def test_is_url(name: str, result: bool) -> None:
+ """Checks functionality of is_url."""
+ assert is_url(name) == result
+
+
+def test_prepare_environment_repair_broken_symlink(
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """Ensure we can deal with broken symlinks in collections."""
+ caplog.set_level(logging.INFO)
+ project_dir = Path(__file__).parent / "collections" / "acme.minimal"
+ runtime = Runtime(isolated=True, project_dir=project_dir)
+ assert runtime.cache_dir
+ acme = runtime.cache_dir / "collections" / "ansible_collections" / "acme"
+ acme.mkdir(parents=True, exist_ok=True)
+ goodies = acme / "minimal"
+ rmtree(goodies, ignore_errors=True)
+ goodies.unlink(missing_ok=True)
+ goodies.symlink_to("/invalid/destination")
+ runtime.prepare_environment(install_local=True)
+ assert any(
+ msg.startswith("Collection is symlinked, but not pointing to")
+ for msg in caplog.messages
+ )
+
+
+def test_get_galaxy_role_name_invalid() -> None:
+ """Verifies that function returns empty string on invalid input."""
+ galaxy_infos = {
+ "role_name": False, # <-- invalid data, should be string
+ }
+ assert _get_galaxy_role_name(galaxy_infos) == ""
diff --git a/test/test_runtime_example.py b/test/test_runtime_example.py
new file mode 100644
index 0000000..e500e59
--- /dev/null
+++ b/test/test_runtime_example.py
@@ -0,0 +1,24 @@
+"""Sample use of Runtime class."""
+from ansible_compat.runtime import Runtime
+
+
+def test_runtime_example() -> None:
+ """Test basic functionality of Runtime class."""
+ # instantiate the runtime using isolated mode, so installing new
+ # roles/collections do not pollute the default setup.
+ runtime = Runtime(isolated=True, max_retries=3)
+
+ # Print Ansible core version
+ _ = runtime.version # 2.9.10 (Version object)
+ # Get configuration info from runtime
+ _ = runtime.config.collections_path
+
+ # Detect if current project is a collection and install its requirements
+ runtime.prepare_environment(install_local=True) # will retry 3 times if needed
+
+ # Install a new collection (will retry 3 times if needed)
+ runtime.install_collection("examples/reqs_v2/community-molecule-0.1.0.tar.gz")
+
+ # Execute a command
+ result = runtime.run(["ansible-doc", "--list"])
+ assert result.returncode == 0
diff --git a/test/test_runtime_scan_path.py b/test/test_runtime_scan_path.py
new file mode 100644
index 0000000..be44f1c
--- /dev/null
+++ b/test/test_runtime_scan_path.py
@@ -0,0 +1,102 @@
+"""Test the scan path functionality of the runtime."""
+
+import json
+import textwrap
+from dataclasses import dataclass, fields
+from pathlib import Path
+
+import pytest
+from _pytest.monkeypatch import MonkeyPatch
+
+from ansible_compat.runtime import Runtime
+
+from .conftest import VirtualEnvironment
+
+V2_COLLECTION_TARBALL = Path("examples/reqs_v2/community-molecule-0.1.0.tar.gz")
+V2_COLLECTION_NAMESPACE = "community"
+V2_COLLECTION_NAME = "molecule"
+V2_COLLECTION_VERSION = "0.1.0"
+V2_COLLECTION_FULL_NAME = f"{V2_COLLECTION_NAMESPACE}.{V2_COLLECTION_NAME}"
+
+
+@dataclass
+class ScanSysPath:
+ """Parameters for scan tests."""
+
+ scan: bool
+ raises_not_found: bool
+
+ def __str__(self) -> str:
+ """Return a string representation of the object."""
+ parts = [
+ f"{field.name}{str(getattr(self, field.name))[0]}" for field in fields(self)
+ ]
+ return "-".join(parts)
+
+
+@pytest.mark.parametrize(
+ ("param"),
+ (
+ ScanSysPath(scan=False, raises_not_found=True),
+ ScanSysPath(scan=True, raises_not_found=False),
+ ),
+ ids=str,
+)
+def test_scan_sys_path(
+ venv_module: VirtualEnvironment,
+ monkeypatch: MonkeyPatch,
+ runtime_tmp: Runtime,
+ tmp_path: Path,
+ param: ScanSysPath,
+) -> None:
+ """Confirm sys path is scanned for collections.
+
+ :param venv_module: Fixture for a virtual environment
+ :param monkeypatch: Fixture for monkeypatching
+ :param runtime_tmp: Fixture for a Runtime object
+ :param tmp_dir: Fixture for a temporary directory
+ :param param: The parameters for the test
+ """
+ first_site_package_dir = venv_module.site_package_dirs()[0]
+
+ installed_to = (
+ first_site_package_dir
+ / "ansible_collections"
+ / V2_COLLECTION_NAMESPACE
+ / V2_COLLECTION_NAME
+ )
+ if not installed_to.exists():
+ # Install the collection into the venv site packages directory, force
+ # as of yet this test is not isolated from the rest of the system
+ runtime_tmp.install_collection(
+ collection=V2_COLLECTION_TARBALL,
+ destination=first_site_package_dir,
+ force=True,
+ )
+ # Confirm the collection is installed
+ assert installed_to.exists()
+ # Set the sys scan path environment variable
+ monkeypatch.setenv("ANSIBLE_COLLECTIONS_SCAN_SYS_PATH", str(param.scan))
+ # Set the ansible collections paths to avoid bleed from other tests
+ monkeypatch.setenv("ANSIBLE_COLLECTIONS_PATH", str(tmp_path))
+
+ script = textwrap.dedent(
+ f"""
+ import json;
+ from ansible_compat.runtime import Runtime;
+ r = Runtime();
+ fv, cp = r.require_collection(name="{V2_COLLECTION_FULL_NAME}", version="{V2_COLLECTION_VERSION}", install=False);
+ print(json.dumps({{"found_version": str(fv), "collection_path": str(cp)}}));
+ """,
+ )
+
+ proc = venv_module.python_script_run(script)
+ if param.raises_not_found:
+ assert proc.returncode != 0, (proc.stdout, proc.stderr)
+ assert "InvalidPrerequisiteError" in proc.stderr
+ assert "'community.molecule' not found" in proc.stderr
+ else:
+ assert proc.returncode == 0, (proc.stdout, proc.stderr)
+ result = json.loads(proc.stdout)
+ assert result["found_version"] == V2_COLLECTION_VERSION
+ assert result["collection_path"] == str(installed_to)
diff --git a/test/test_schema.py b/test/test_schema.py
new file mode 100644
index 0000000..b253cb5
--- /dev/null
+++ b/test/test_schema.py
@@ -0,0 +1,73 @@
+"""Tests for schema utilities."""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+import pytest
+
+from ansible_compat.schema import JsonSchemaError, json_path, validate
+
+if TYPE_CHECKING:
+ from ansible_compat.types import JSON
+
+expected_results = [
+ JsonSchemaError(
+ message="False is not of type 'string'",
+ data_path="environment.a",
+ json_path="$.environment.a",
+ schema_path="properties.environment.additionalProperties.type",
+ relative_schema='{"type": "string"}',
+ expected="string",
+ validator="type",
+ found="False",
+ ),
+ JsonSchemaError(
+ message="True is not of type 'string'",
+ data_path="environment.b",
+ json_path="$.environment.b",
+ schema_path="properties.environment.additionalProperties.type",
+ relative_schema='{"type": "string"}',
+ expected="string",
+ validator="type",
+ found="True",
+ ),
+]
+
+
+def json_from_asset(file_name: str) -> JSON:
+ """Load a json file from disk."""
+ file = Path(__file__).parent / file_name
+ with file.open(encoding="utf-8") as f:
+ return json.load(f) # type: ignore[no-any-return]
+
+
+def jsonify(data: Any) -> JSON: # noqa: ANN401
+ """Convert object in JSON data structure."""
+ return json.loads(json.dumps(data, default=vars, sort_keys=True)) # type: ignore[no-any-return]
+
+
+@pytest.mark.parametrize("index", range(1))
+def test_schema(index: int) -> None:
+ """Test the schema validator."""
+ schema = json_from_asset(f"assets/validate{index}_schema.json")
+ data = json_from_asset(f"assets/validate{index}_data.json")
+ expected = json_from_asset(f"assets/validate{index}_expected.json")
+
+ # ensure we produce consistent results between runs
+ for _ in range(1, 100):
+ found_errors = validate(schema=schema, data=data)
+ # ensure returned results are already sorted, as we assume our class
+ # knows how to sort itself
+ assert sorted(found_errors) == found_errors, "multiple errors not sorted"
+
+ found_errors_json = jsonify(found_errors)
+ assert (
+ found_errors_json == expected
+ ), f"inconsistent returns: {found_errors_json}"
+
+
+def test_json_path() -> None:
+ """Test json_path function."""
+ assert json_path(["a", 1, "b"]) == "$.a[1].b"