summaryrefslogtreecommitdiffstats
path: root/src/ansiblelint/testing/__init__.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/ansiblelint/testing/__init__.py')
-rw-r--r--src/ansiblelint/testing/__init__.py147
1 files changed, 147 insertions, 0 deletions
diff --git a/src/ansiblelint/testing/__init__.py b/src/ansiblelint/testing/__init__.py
new file mode 100644
index 0000000..a4ebcc8
--- /dev/null
+++ b/src/ansiblelint/testing/__init__.py
@@ -0,0 +1,147 @@
+"""Test utils for ansible-lint."""
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+from typing import TYPE_CHECKING, Any
+
+from ansiblelint.app import get_app
+from ansiblelint.rules import RulesCollection
+
+if TYPE_CHECKING:
+ # https://github.com/PyCQA/pylint/issues/3240
+ # pylint: disable=unsubscriptable-object
+ CompletedProcess = subprocess.CompletedProcess[Any]
+ from ansiblelint.errors import MatchError # noqa: E402
+else:
+ CompletedProcess = subprocess.CompletedProcess
+
+# pylint: disable=wrong-import-position
+from ansiblelint.runner import Runner # noqa: E402
+
+
+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()
+
+ self.collection = collection
+
+ def _call_runner(self, path: str) -> list[MatchError]:
+ runner = Runner(path, rules=self.collection)
+ return runner.run()
+
+ def run(self, filename: str) -> 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(fh.name)
+ return results
+
+ def run_role_tasks_main(self, tasks_main_text: str) -> list[MatchError]:
+ """Lints received text as tasks."""
+ role_path = tempfile.mkdtemp(prefix="role_")
+ tasks_path = os.path.join(role_path, "tasks")
+ os.makedirs(tasks_path)
+ with open(os.path.join(tasks_path, "main.yml"), "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) -> list[MatchError]:
+ """Lints received text as meta."""
+ role_path = tempfile.mkdtemp(prefix="role_")
+ meta_path = os.path.join(role_path, "meta")
+ os.makedirs(meta_path)
+ with open(os.path.join(meta_path, "main.yml"), "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) -> list[MatchError]:
+ """Lints received text as vars file in defaults."""
+ role_path = tempfile.mkdtemp(prefix="role_")
+ defaults_path = os.path.join(role_path, "defaults")
+ os.makedirs(defaults_path)
+ with open(os.path.join(defaults_path, "main.yml"), "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,
+ cwd: str | 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 = [*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",
+ ]
+
+ if env is None:
+ _env = {}
+ else:
+ _env = 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,
+ )