Adding upstream version 4.1.11.upstream/4.1.11

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 893 insertions, 0 deletions

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