diff options
Diffstat (limited to 'test/lib/ansible_test/_internal/commands/sanity/yamllint.py')
-rw-r--r-- | test/lib/ansible_test/_internal/commands/sanity/yamllint.py | 125 |
1 files changed, 125 insertions, 0 deletions
diff --git a/test/lib/ansible_test/_internal/commands/sanity/yamllint.py b/test/lib/ansible_test/_internal/commands/sanity/yamllint.py new file mode 100644 index 0000000..a0d859f --- /dev/null +++ b/test/lib/ansible_test/_internal/commands/sanity/yamllint.py @@ -0,0 +1,125 @@ +"""Sanity test using yamllint.""" +from __future__ import annotations + +import json +import os +import typing as t + +from . import ( + SanitySingleVersion, + SanityMessage, + SanityFailure, + SanitySuccess, + SanityTargets, + SANITY_ROOT, +) + +from ...test import ( + TestResult, +) + +from ...target import ( + TestTarget, +) + +from ...util import ( + SubprocessError, + display, + is_subdir, +) + +from ...util_common import ( + run_command, +) + +from ...config import ( + SanityConfig, +) + +from ...data import ( + data_context, +) + +from ...host_configs import ( + PythonConfig, +) + + +class YamllintTest(SanitySingleVersion): + """Sanity test using yamllint.""" + @property + def error_code(self) -> t.Optional[str]: + """Error code for ansible-test matching the format used by the underlying test program, or None if the program does not use error codes.""" + return 'ansible-test' + + @property + def require_libyaml(self) -> bool: + """True if the test requires PyYAML to have libyaml support.""" + return True + + def filter_targets(self, targets: list[TestTarget]) -> list[TestTarget]: + """Return the given list of test targets, filtered to include only those relevant for the test.""" + yaml_targets = [target for target in targets if os.path.splitext(target.path)[1] in ('.yml', '.yaml')] + + for plugin_type, plugin_path in sorted(data_context().content.plugin_paths.items()): + if plugin_type == 'module_utils': + continue + + yaml_targets.extend([target for target in targets if + os.path.splitext(target.path)[1] == '.py' and + os.path.basename(target.path) != '__init__.py' and + is_subdir(target.path, plugin_path)]) + + return yaml_targets + + def test(self, args: SanityConfig, targets: SanityTargets, python: PythonConfig) -> TestResult: + settings = self.load_processor(args) + + paths = [target.path for target in targets.include] + + results = self.test_paths(args, paths, python) + results = settings.process_errors(results, paths) + + if results: + return SanityFailure(self.name, messages=results) + + return SanitySuccess(self.name) + + @staticmethod + def test_paths(args: SanityConfig, paths: list[str], python: PythonConfig) -> list[SanityMessage]: + """Test the specified paths using the given Python and return the results.""" + cmd = [ + python.path, + os.path.join(SANITY_ROOT, 'yamllint', 'yamllinter.py'), + ] + + data = '\n'.join(paths) + + display.info(data, verbosity=4) + + try: + stdout, stderr = run_command(args, cmd, data=data, capture=True) + status = 0 + except SubprocessError as ex: + stdout = ex.stdout + stderr = ex.stderr + status = ex.status + + if stderr: + raise SubprocessError(cmd=cmd, status=status, stderr=stderr, stdout=stdout) + + if args.explain: + return [] + + results = json.loads(stdout)['messages'] + + results = [SanityMessage( + code=r['code'], + message=r['message'], + path=r['path'], + line=int(r['line']), + column=int(r['column']), + level=r['level'], + ) for r in results] + + return results |