Adding upstream version 6.17.2.upstream/6.17.2

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

2 files changed, 222 insertions, 0 deletions

diff --git a/src/ansiblelint/testing/__init__.py b/src/ansiblelint/testing/__init__.py
new file mode 100644
index 0000000..e7f6c1b
--- /dev/null
+++ b/src/ansiblelint/testing/__init__.py
@@ -0,0 +1,159 @@
+"""Test utils for ansible-lint."""
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+from ansiblelint.app import get_app
+
+if TYPE_CHECKING:
+    # https://github.com/PyCQA/pylint/issues/3240
+    # pylint: disable=unsubscriptable-object
+    CompletedProcess = subprocess.CompletedProcess[Any]
+    from ansiblelint.errors import MatchError
+    from ansiblelint.rules import RulesCollection
+else:
+    CompletedProcess = subprocess.CompletedProcess
+
+# pylint: disable=wrong-import-position
+from ansiblelint.runner import Runner
+
+
+class RunFromText:
+    """Use Runner on temp files created from testing text snippets."""
+
+    app = None
+
+    def __init__(self, collection: RulesCollection) -> None:
+        """Initialize a RunFromText instance with rules collection."""
+        # Emulate command line execution initialization as without it Ansible module
+        # would be loaded with incomplete module/role/collection list.
+        if not self.app:  # pragma: no cover
+            self.app = get_app(offline=True)
+
+        self.collection = collection
+
+    def _call_runner(self, path: Path) -> list[MatchError]:
+        runner = Runner(path, rules=self.collection)
+        return runner.run()
+
+    def run(self, filename: Path) -> list[MatchError]:
+        """Lints received filename."""
+        return self._call_runner(filename)
+
+    def run_playbook(
+        self,
+        playbook_text: str,
+        prefix: str = "playbook",
+    ) -> list[MatchError]:
+        """Lints received text as a playbook."""
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".yml", prefix=prefix) as fh:
+            fh.write(playbook_text)
+            fh.flush()
+            results = self._call_runner(Path(fh.name))
+        return results
+
+    def run_role_tasks_main(
+        self,
+        tasks_main_text: str,
+        tmp_path: Path,
+    ) -> list[MatchError]:
+        """Lints received text as tasks."""
+        role_path = tmp_path
+        tasks_path = role_path / "tasks"
+        tasks_path.mkdir(parents=True, exist_ok=True)
+        with (tasks_path / "main.yml").open("w", encoding="utf-8") as fh:
+            fh.write(tasks_main_text)
+            fh.flush()
+        results = self._call_runner(role_path)
+        shutil.rmtree(role_path)
+        return results
+
+    def run_role_meta_main(
+        self,
+        meta_main_text: str,
+        temp_path: Path,
+    ) -> list[MatchError]:
+        """Lints received text as meta."""
+        role_path = temp_path
+        meta_path = role_path / "meta"
+        meta_path.mkdir(parents=True, exist_ok=True)
+        with (meta_path / "main.yml").open("w", encoding="utf-8") as fh:
+            fh.write(meta_main_text)
+            fh.flush()
+        results = self._call_runner(role_path)
+        shutil.rmtree(role_path)
+        return results
+
+    def run_role_defaults_main(
+        self,
+        defaults_main_text: str,
+        tmp_path: Path,
+    ) -> list[MatchError]:
+        """Lints received text as vars file in defaults."""
+        role_path = tmp_path
+        defaults_path = role_path / "defaults"
+        defaults_path.mkdir(parents=True, exist_ok=True)
+        with (defaults_path / "main.yml").open("w", encoding="utf-8") as fh:
+            fh.write(defaults_main_text)
+            fh.flush()
+        results = self._call_runner(role_path)
+        shutil.rmtree(role_path)
+        return results
+
+
+def run_ansible_lint(
+    *argv: str | Path,
+    cwd: Path | None = None,
+    executable: str | None = None,
+    env: dict[str, str] | None = None,
+    offline: bool = True,
+) -> CompletedProcess:
+    """Run ansible-lint on a given path and returns its output."""
+    args = [str(item) for item in argv]
+    if offline:  # pragma: no cover
+        args.insert(0, "--offline")
+
+    if not executable:
+        executable = sys.executable
+        args = [sys.executable, "-m", "ansiblelint", *args]
+    else:
+        args = [executable, *args]
+
+    # It is not safe to pass entire env for testing as other tests would
+    # pollute the env, causing weird behaviors, so we pass only a safe list of
+    # vars.
+    safe_list = [
+        "COVERAGE_FILE",
+        "COVERAGE_PROCESS_START",
+        "HOME",
+        "LANG",
+        "LC_ALL",
+        "LC_CTYPE",
+        "NO_COLOR",
+        "PATH",
+        "PYTHONIOENCODING",
+        "PYTHONPATH",
+        "TERM",
+        "VIRTUAL_ENV",
+    ]
+
+    _env = {} if env is None else env
+    for v in safe_list:
+        if v in os.environ and v not in _env:
+            _env[v] = os.environ[v]
+
+    return subprocess.run(
+        args,
+        capture_output=True,
+        shell=False,  # needed when command is a list
+        check=False,
+        cwd=cwd,
+        env=_env,
+        text=True,
+    )
diff --git a/src/ansiblelint/testing/fixtures.py b/src/ansiblelint/testing/fixtures.py
new file mode 100644
index 0000000..814a076
--- /dev/null
+++ b/src/ansiblelint/testing/fixtures.py
@@ -0,0 +1,63 @@
+"""PyTest Fixtures.
+
+They should not be imported, instead add code below to your root conftest.py
+file:
+
+pytest_plugins = ['ansiblelint.testing']
+"""
+from __future__ import annotations
+
+import copy
+from typing import TYPE_CHECKING
+
+import pytest
+
+from ansiblelint.config import Options, options
+from ansiblelint.constants import DEFAULT_RULESDIR
+from ansiblelint.rules import RulesCollection
+from ansiblelint.testing import RunFromText
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+    from _pytest.fixtures import SubRequest
+
+
+# The sessions scope does not apply to xdist, so we will still have one
+# session for each worker, but at least it will a limited number.
+@pytest.fixture(name="default_rules_collection", scope="session")
+def fixture_default_rules_collection() -> RulesCollection:
+    """Return default rule collection."""
+    assert DEFAULT_RULESDIR.is_dir()
+    # For testing we want to manually enable opt-in rules
+    test_options = copy.deepcopy(options)
+    test_options.enable_list = ["no-same-owner"]
+    # That is instantiated very often and do want to avoid ansible-galaxy
+    # install errors due to concurrency.
+    test_options.offline = True
+    return RulesCollection(rulesdirs=[DEFAULT_RULESDIR], options=test_options)
+
+
+@pytest.fixture()
+def default_text_runner(default_rules_collection: RulesCollection) -> RunFromText:
+    """Return RunFromText instance for the default set of collections."""
+    return RunFromText(default_rules_collection)
+
+
+@pytest.fixture()
+def rule_runner(request: SubRequest, config_options: Options) -> RunFromText:
+    """Return runner for a specific rule class."""
+    rule_class = request.param
+    config_options.enable_list.append(rule_class().id)
+    collection = RulesCollection(options=config_options)
+    collection.register(rule_class())
+    return RunFromText(collection)
+
+
+@pytest.fixture(name="config_options")
+def fixture_config_options() -> Iterator[Options]:
+    """Return configuration options that will be restored after testrun."""
+    global options  # pylint: disable=global-statement,invalid-name # noqa: PLW0603
+    original_options = copy.deepcopy(options)
+    yield options
+    options = original_options