diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-13 12:06:49 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-13 12:06:49 +0000 |
commit | 2fe34b6444502079dc0b84365ce82dbc92de308e (patch) | |
tree | 8fedcab52bbbc3db6c5aa909a88a7a7b81685018 /src/ansiblelint/testing | |
parent | Initial commit. (diff) | |
download | ansible-lint-2fe34b6444502079dc0b84365ce82dbc92de308e.tar.xz ansible-lint-2fe34b6444502079dc0b84365ce82dbc92de308e.zip |
Adding upstream version 6.17.2.upstream/6.17.2
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'src/ansiblelint/testing')
-rw-r--r-- | src/ansiblelint/testing/__init__.py | 159 | ||||
-rw-r--r-- | src/ansiblelint/testing/fixtures.py | 63 |
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 |