1 files changed, 270 insertions, 0 deletions
diff --git a/lib/ansiblelint/__main__.py b/lib/ansiblelint/__main__.py
new file mode 100755
index 0000000..ff8d477
--- /dev/null
+++ b/lib/ansiblelint/__main__.py
@@ -0,0 +1,270 @@
+#!/usr/bin/env python
+# Copyright (c) 2013-2014 Will Thames <will@thames.id.au>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+"""Command line implementation."""
+
+import errno
+import logging
+import os
+import pathlib
+import subprocess
+import sys
+from contextlib import contextmanager
+from typing import TYPE_CHECKING, Any, List, Set, Type, Union
+
+from rich.markdown import Markdown
+
+from ansiblelint import cli, formatters
+from ansiblelint.color import console, console_stderr
+from ansiblelint.file_utils import cwd
+from ansiblelint.generate_docs import rules_as_rich, rules_as_rst
+from ansiblelint.rules import RulesCollection
+from ansiblelint.runner import Runner
+from ansiblelint.utils import get_playbooks_and_roles, get_rules_dirs
+
+if TYPE_CHECKING:
+    from argparse import Namespace
+
+    from ansiblelint.errors import MatchError
+
+_logger = logging.getLogger(__name__)
+
+_rule_format_map = {
+    'plain': str,
+    'rich': rules_as_rich,
+    'rst': rules_as_rst
+}
+
+
+def initialize_logger(level: int = 0) -> None:
+    """Set up the global logging level based on the verbosity number."""
+    VERBOSITY_MAP = {
+        0: logging.NOTSET,
+        1: logging.INFO,
+        2: logging.DEBUG
+    }
+
+    handler = logging.StreamHandler()
+    formatter = logging.Formatter('%(levelname)-8s %(message)s')
+    handler.setFormatter(formatter)
+    logger = logging.getLogger(__package__)
+    logger.addHandler(handler)
+    # Unknown logging level is treated as DEBUG
+    logging_level = VERBOSITY_MAP.get(level, logging.DEBUG)
+    logger.setLevel(logging_level)
+    # Use module-level _logger instance to validate it
+    _logger.debug("Logging initialized to level %s", logging_level)
+
+
+def choose_formatter_factory(
+    options_list: "Namespace"
+) -> Type[formatters.BaseFormatter]:
+    """Select an output formatter based on the incoming command line arguments."""
+    r: Type[formatters.BaseFormatter] = formatters.Formatter
+    if options_list.quiet:
+        r = formatters.QuietFormatter
+    elif options_list.parseable:
+        r = formatters.ParseableFormatter
+    elif options_list.parseable_severity:
+        r = formatters.ParseableSeverityFormatter
+    return r
+
+
+def report_outcome(matches: List["MatchError"], options) -> int:
+    """Display information about how to skip found rules.
+
+    Returns exit code, 2 if errors were found, 0 when only warnings were found.
+    """
+    failure = False
+    msg = """\
+You can skip specific rules or tags by adding them to your configuration file:
+```yaml
+# .ansible-lint
+warn_list:  # or 'skip_list' to silence them completely
+"""
+    matches_unignored = [match for match in matches if not match.ignored]
+
+    matched_rules = {match.rule.id: match.rule for match in matches_unignored}
+    for id in sorted(matched_rules.keys()):
+        if {id, *matched_rules[id].tags}.isdisjoint(options.warn_list):
+            msg += f"  - '{id}'  # {matched_rules[id].shortdesc}\n"
+            failure = True
+    for match in matches:
+        if "experimental" in match.rule.tags:
+            msg += "  - experimental  # all rules tagged as experimental\n"
+            break
+    msg += "```"
+
+    if matches and not options.quiet:
+        console_stderr.print(Markdown(msg))
+
+    if failure:
+        return 2
+    else:
+        return 0
+
+
+def main() -> int:
+    """Linter CLI entry point."""
+    cwd = pathlib.Path.cwd()
+
+    options = cli.get_config(sys.argv[1:])
+
+    initialize_logger(options.verbosity)
+    _logger.debug("Options: %s", options)
+
+    formatter_factory = choose_formatter_factory(options)
+    formatter = formatter_factory(cwd, options.display_relative_path)
+
+    rulesdirs = get_rules_dirs([str(rdir) for rdir in options.rulesdir],
+                               options.use_default_rules)
+    rules = RulesCollection(rulesdirs)
+
+    if options.listrules:
+        console.print(
+            _rule_format_map[options.format](rules),
+            highlight=False)
+        return 0
+
+    if options.listtags:
+        print(rules.listtags())
+        return 0
+
+    if isinstance(options.tags, str):
+        options.tags = options.tags.split(',')
+
+    skip = set()
+    for s in options.skip_list:
+        skip.update(str(s).split(','))
+    options.skip_list = frozenset(skip)
+
+    matches = _get_matches(rules, options)
+
+    # Assure we do not print duplicates and the order is consistent
+    matches = sorted(set(matches))
+
+    mark_as_success = False
+    if matches and options.progressive:
+        _logger.info(
+            "Matches found, running again on previous revision in order to detect regressions")
+        with _previous_revision():
+            old_matches = _get_matches(rules, options)
+            # remove old matches from current list
+            matches_delta = list(set(matches) - set(old_matches))
+            if len(matches_delta) == 0:
+                _logger.warning(
+                    "Total violations not increased since previous "
+                    "commit, will mark result as success. (%s -> %s)",
+                    len(old_matches), len(matches_delta))
+                mark_as_success = True
+
+            ignored = 0
+            for match in matches:
+                # if match is not new, mark is as ignored
+                if match not in matches_delta:
+                    match.ignored = True
+                    ignored += 1
+            if ignored:
+                _logger.warning(
+                    "Marked %s previously known violation(s) as ignored due to"
+                    " progressive mode.", ignored)
+
+    _render_matches(matches, options, formatter, cwd)
+
+    if matches and not mark_as_success:
+        return report_outcome(matches, options=options)
+    else:
+        return 0
+
+
+def _render_matches(
+        matches: List,
+        options: "Namespace",
+        formatter: Any,
+        cwd: Union[str, pathlib.Path]):
+
+    ignored_matches = [match for match in matches if match.ignored]
+    fatal_matches = [match for match in matches if not match.ignored]
+    # Displayed ignored matches first
+    if ignored_matches:
+        _logger.warning(
+            "Listing %s violation(s) marked as ignored, likely already known",
+            len(ignored_matches))
+        for match in ignored_matches:
+            if match.ignored:
+                print(formatter.format(match, options.colored))
+    if fatal_matches:
+        _logger.warning("Listing %s violation(s) that are fatal", len(fatal_matches))
+        for match in fatal_matches:
+            if not match.ignored:
+                print(formatter.format(match, options.colored))
+
+    # If run under GitHub Actions we also want to emit output recognized by it.
+    if os.getenv('GITHUB_ACTIONS') == 'true' and os.getenv('GITHUB_WORKFLOW'):
+        formatter = formatters.AnnotationsFormatter(cwd, True)
+        for match in matches:
+            print(formatter.format(match))
+
+
+def _get_matches(rules: RulesCollection, options: "Namespace") -> list:
+
+    if not options.playbook:
+        # no args triggers auto-detection mode
+        playbooks = get_playbooks_and_roles(options=options)
+    else:
+        playbooks = sorted(set(options.playbook))
+
+    matches = list()
+    checked_files: Set[str] = set()
+    for playbook in playbooks:
+        runner = Runner(rules, playbook, options.tags,
+                        options.skip_list, options.exclude_paths,
+                        options.verbosity, checked_files)
+        matches.extend(runner.run())
+    return matches
+
+
+@contextmanager
+def _previous_revision():
+    """Create or update a temporary workdir containing the previous revision."""
+    worktree_dir = ".cache/old-rev"
+    revision = subprocess.run(
+        ["git", "rev-parse", "HEAD^1"],
+        check=True,
+        universal_newlines=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.DEVNULL,
+        ).stdout
+    p = pathlib.Path(worktree_dir)
+    p.mkdir(parents=True, exist_ok=True)
+    os.system(f"git worktree add -f {worktree_dir} 2>/dev/null")
+    with cwd(worktree_dir):
+        os.system(f"git checkout {revision}")
+        yield
+
+
+if __name__ == "__main__":
+    try:
+        sys.exit(main())
+    except IOError as exc:
+        if exc.errno != errno.EPIPE:
+            raise
+    except RuntimeError as e:
+        raise SystemExit(str(e))