path: root/src/ansiblelint/cli.py

"""CLI parser setup and helpers."""
from __future__ import annotations

import argparse
import logging
import os
import sys
from argparse import Namespace
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable

from ansiblelint.config import (
    DEFAULT_KINDS,
    DEFAULT_WARN_LIST,
    PROFILES,
    Options,
    log_entries,
)
from ansiblelint.constants import CUSTOM_RULESDIR_ENVVAR, DEFAULT_RULESDIR, RC
from ansiblelint.file_utils import (
    Lintable,
    abspath,
    expand_path_vars,
    find_project_root,
    normpath,
)
from ansiblelint.loaders import IGNORE_FILE
from ansiblelint.schemas.main import validate_file_schema
from ansiblelint.yaml_utils import clean_json

if TYPE_CHECKING:
    from collections.abc import Sequence


_logger = logging.getLogger(__name__)
_PATH_VARS = [
    "rulesdir",
]


def expand_to_normalized_paths(
    config: dict[str, Any],
    base_dir: str | None = None,
) -> None:
    """Mutate given config normalizing any path values in it."""
    # config can be None (-c /dev/null)
    if not config:
        return
    base_dir = base_dir or os.getcwd()
    for paths_var in _PATH_VARS:
        if paths_var not in config:
            continue  # Cause we don't want to add a variable not present

        normalized_paths = []
        for path in config.pop(paths_var):
            normalized_path = abspath(expand_path_vars(path), base_dir=base_dir)

            normalized_paths.append(normalized_path)

        config[paths_var] = normalized_paths


def load_config(config_file: str | None) -> tuple[dict[Any, Any], str | None]:
    """Load configuration from disk."""
    config_path = None

    if config_file == "/dev/null":
        _logger.debug("Skipping config file as it was set to /dev/null")
        return {}, config_file

    if config_file:
        config_path = os.path.abspath(config_file)
        if not os.path.exists(config_path):
            _logger.error("Config file not found '%s'", config_path)
            sys.exit(RC.INVALID_CONFIG)
    config_path = config_path or get_config_path()
    if not config_path or not os.path.exists(config_path):
        # a missing default config file should not trigger an error
        return {}, None

    config_lintable = Lintable(
        config_path,
        kind="ansible-lint-config",
        base_kind="text/yaml",
    )

    for error in validate_file_schema(config_lintable):
        _logger.error("Invalid configuration file %s. %s", config_path, error)
        sys.exit(RC.INVALID_CONFIG)

    config = clean_json(config_lintable.data)
    if not isinstance(config, dict):
        msg = "Schema failed to properly validate the config file."
        raise RuntimeError(msg)
    config["config_file"] = config_path
    config_dir = os.path.dirname(config_path)
    expand_to_normalized_paths(config, config_dir)

    return config, config_path


def get_config_path(config_file: str | None = None) -> str | None:
    """Return local config file."""
    if config_file:
        project_filenames = [config_file]
    else:
        project_filenames = [
            ".ansible-lint",
            ".config/ansible-lint.yml",
            ".config/ansible-lint.yaml",
        ]
    parent = tail = os.getcwd()
    while tail:
        for project_filename in project_filenames:
            filename = os.path.abspath(os.path.join(parent, project_filename))
            if os.path.exists(filename):
                return filename
        if os.path.exists(os.path.abspath(os.path.join(parent, ".git"))):
            # Avoid looking outside .git folders as we do not want end-up
            # picking config files from upper level projects if current
            # project has no config.
            return None
        (parent, tail) = os.path.split(parent)
    return None


class AbspathArgAction(argparse.Action):
    """Argparse action to convert relative paths to absolute paths."""

    def __call__(
        self,
        parser: argparse.ArgumentParser,
        namespace: Namespace,
        values: str | Sequence[Any] | None,
        option_string: str | None = None,
    ) -> None:
        if isinstance(values, (str, Path)):
            values = [values]
        if values:
            normalized_values = [
                Path(expand_path_vars(str(path))).resolve() for path in values
            ]
            previous_values = getattr(namespace, self.dest, [])
            setattr(namespace, self.dest, previous_values + normalized_values)


class WriteArgAction(argparse.Action):
    """Argparse action to handle the --write flag with optional args."""

    _default = "__default__"

    # noinspection PyShadowingBuiltins
    def __init__(  # pylint: disable=too-many-arguments,redefined-builtin
        self,
        option_strings: list[str],
        dest: str,
        nargs: int | str | None = None,
        const: Any = None,
        default: Any = None,
        type: Callable[[str], Any] | None = None,  # noqa: A002
        choices: list[Any] | None = None,
        *,
        required: bool = False,
        help: str | None = None,  # noqa: A002
        metavar: str | None = None,
    ) -> None:
        """Create the argparse action with WriteArg-specific defaults."""
        if nargs is not None:
            msg = "nargs for WriteArgAction must not be set."
            raise ValueError(msg)
        if const is not None:
            msg = "const for WriteArgAction must not be set."
            raise ValueError(msg)
        super().__init__(
            option_strings=option_strings,
            dest=dest,
            nargs="?",  # either 0 (--write) or 1 (--write=a,b,c) argument
            const=self._default,  # --write (no option) implicitly stores this
            default=default,
            type=type,
            choices=choices,
            required=required,
            help=help,
            metavar=metavar,
        )

    def __call__(
        self,
        parser: argparse.ArgumentParser,
        namespace: Namespace,
        values: str | Sequence[Any] | None,
        option_string: str | None = None,
    ) -> None:
        lintables = getattr(namespace, "lintables", None)
        if not lintables and isinstance(values, str):
            # args are processed in order.
            # If --write is after lintables, then that is not ambiguous.
            # But if --write comes first, then it might actually be a lintable.
            maybe_lintable = Path(values)
            if maybe_lintable.exists():
                namespace.lintables = [values]
                values = []
        if isinstance(values, str):
            values = values.split(",")
        default = [self.const] if isinstance(self.const, str) else self.const
        previous_values = getattr(namespace, self.dest, default) or default
        if not values:
            values = previous_values
        elif previous_values != default:
            values = previous_values + values
        setattr(namespace, self.dest, values)

    @classmethod
    def merge_write_list_config(
        cls,
        from_file: list[str],
        from_cli: list[str],
    ) -> list[str]:
        """Combine the write_list from file config with --write CLI arg.

        Handles the implicit "all" when "__default__" is present and file config is empty.
        """
        if not from_file or "none" in from_cli:
            # --write is the same as --write=all
            return ["all" if value == cls._default else value for value in from_cli]
        # --write means use the config from the config file
        from_cli = [value for value in from_cli if value != cls._default]
        return from_file + from_cli


def get_cli_parser() -> argparse.ArgumentParser:
    """Initialize an argument parser."""
    parser = argparse.ArgumentParser()

    listing_group = parser.add_mutually_exclusive_group()
    listing_group.add_argument(
        "-P",
        "--list-profiles",
        dest="list_profiles",
        default=False,
        action="store_true",
        help="List all profiles, no formatting options available.",
    )
    listing_group.add_argument(
        "-L",
        "--list-rules",
        dest="list_rules",
        default=False,
        action="store_true",
        help="List all the rules. For listing rules only the following formats "
        "for argument -f are supported: {brief, full, md} with 'brief' as default.",
    )
    listing_group.add_argument(
        "-T",
        "--list-tags",
        dest="list_tags",
        action="store_true",
        help="List all the tags and the rules they cover. Increase the verbosity level "
        "with `-v` to include 'opt-in' tag and its rules.",
    )
    parser.add_argument(
        "-f",
        "--format",
        dest="format",
        default=None,
        choices=[
            "brief",
            # "plain",
            "full",
            "md",
            "json",
            "codeclimate",
            "quiet",
            "pep8",
            "sarif",
        ],
        help="stdout formatting, json being an alias for codeclimate. (default: %(default)s)",
    )
    parser.add_argument(
        "--sarif-file",
        default=None,
        type=Path,
        help="SARIF output file",
    )
    parser.add_argument(
        "-q",
        dest="quiet",
        default=0,
        action="count",
        help="quieter, reduce verbosity, can be specified twice.",
    )
    parser.add_argument(
        "--profile",
        dest="profile",
        default=None,
        action="store",
        choices=PROFILES.keys(),
        help="Specify which rules profile to be used.",
    )
    parser.add_argument(
        "-p",
        "--parseable",
        dest="parseable",
        default=False,
        action="store_true",
        help="parseable output, same as '-f pep8'",
    )
    parser.add_argument(
        "--project-dir",
        dest="project_dir",
        default=None,
        help="Location of project/repository, autodetected based on location "
        "of configuration file.",
    )
    parser.add_argument(
        "-r",
        "--rules-dir",
        action=AbspathArgAction,
        dest="rulesdir",
        default=[],
        type=Path,
        help="Specify custom rule directories. Add -R "
        f"to keep using embedded rules from {DEFAULT_RULESDIR}",
    )
    parser.add_argument(
        "-R",
        action="store_true",
        default=False,
        dest="use_default_rules",
        help="Keep default rules when using -r",
    )
    parser.add_argument(
        "-s",
        "--strict",
        action="store_true",
        default=False,
        dest="strict",
        help="Return non-zero exit code on warnings as well as errors",
    )
    parser.add_argument(
        "--write",
        dest="write_list",
        # this is a tri-state argument that takes an optional comma separated list:
        action=WriteArgAction,
        help="Allow ansible-lint to reformat YAML files and run rule transforms "
        "(Reformatting YAML files standardizes spacing, quotes, etc. "
        "A rule transform can fix or simplify fixing issues identified by that rule). "
        "You can limit the effective rule transforms (the 'write_list') by passing a "
        "keywords 'all' or 'none' or a comma separated list of rule ids or rule tags. "
        "YAML reformatting happens whenever '--write' or '--write=' is used. "
        "'--write' and '--write=all' are equivalent: they allow all transforms to run. "
        "The effective list of transforms comes from 'write_list' in the config file, "
        "followed whatever '--write' args are provided on the commandline. "
        "'--write=none' resets the list of transforms to allow reformatting YAML "
        "without running any of the transforms (ie '--write=none,rule-id' will "
        "ignore write_list in the config file and only run the rule-id transform).",
    )
    parser.add_argument(
        "--show-relpath",
        dest="display_relative_path",
        action="store_false",
        default=True,
        help="Display path relative to CWD",
    )
    parser.add_argument(
        "-t",
        "--tags",
        dest="tags",
        action="append",
        default=[],
        help="only check rules whose id/tags match these values",
    )
    parser.add_argument(
        "-v",
        dest="verbosity",
        action="count",
        help="Increase verbosity level (-vv for more)",
        default=0,
    )
    parser.add_argument(
        "-x",
        "--skip-list",
        dest="skip_list",
        default=[],
        action="append",
        help="only check rules whose id/tags do not match these values. \
            e.g: --skip-list=name,run-once",
    )
    parser.add_argument(
        "--generate-ignore",
        dest="generate_ignore",
        action="store_true",
        default=False,
        help="Generate a text file '.ansible-lint-ignore' that ignores all found violations. Each line contains filename and rule id separated by a space.",
    )
    parser.add_argument(
        "-w",
        "--warn-list",
        dest="warn_list",
        default=[],
        action="append",
        help="only warn about these rules, unless overridden in "
        f"config file. Current version default value is: {', '.join(DEFAULT_WARN_LIST)}",
    )
    parser.add_argument(
        "--enable-list",
        dest="enable_list",
        default=[],
        action="append",
        help="activate optional rules by their tag name",
    )
    # Do not use store_true/store_false because they create opposite defaults.
    parser.add_argument(
        "--nocolor",
        dest="colored",
        action="store_const",
        const=False,
        help="disable colored output, same as NO_COLOR=1",
    )
    parser.add_argument(
        "--force-color",
        dest="colored",
        action="store_const",
        const=True,
        help="Force colored output, same as FORCE_COLOR=1",
    )
    parser.add_argument(
        "--exclude",
        dest="exclude_paths",
        action="extend",
        nargs="+",
        type=str,
        default=[],
        help="path to directories or files to skip. This option is repeatable.",
    )
    parser.add_argument(
        "-c",
        "--config-file",
        dest="config_file",
        help="Specify configuration file to use. By default it will look for '.ansible-lint', '.config/ansible-lint.yml', or '.config/ansible-lint.yaml'",
    )
    parser.add_argument(
        "-i",
        "--ignore-file",
        dest="ignore_file",
        type=Path,
        default=None,
        help=f"Specify ignore file to use. By default it will look for '{IGNORE_FILE.default}' or '{IGNORE_FILE.alternative}'",
    )
    parser.add_argument(
        "--offline",
        dest="offline",
        action="store_const",
        const=True,
        help="Disable installation of requirements.yml and schema refreshing",
    )
    parser.add_argument(
        "--version",
        action="store_true",
    )
    parser.add_argument(
        dest="lintables",
        nargs="*",
        action="extend",
        help="One or more files or paths. When missing it will enable auto-detection mode.",
    )

    return parser


def merge_config(file_config: dict[Any, Any], cli_config: Options) -> Options:
    """Combine the file config with the CLI args."""
    bools = (
        "display_relative_path",
        "parseable",
        "quiet",
        "strict",
        "use_default_rules",
        "offline",
    )
    # maps lists to their default config values
    lists_map = {
        "exclude_paths": [".cache", ".git", ".hg", ".svn", ".tox"],
        "rulesdir": [],
        "skip_list": [],
        "tags": [],
        "warn_list": DEFAULT_WARN_LIST,
        "mock_modules": [],
        "mock_roles": [],
        "enable_list": [],
        "only_builtins_allow_collections": [],
        "only_builtins_allow_modules": [],
        # do not include "write_list" here. See special logic below.
    }

    scalar_map = {
        "loop_var_prefix": None,
        "project_dir": None,
        "profile": None,
        "sarif_file": None,
    }

    if not file_config:
        # use defaults if we don't have a config file and the commandline
        # parameter is not set
        for entry, default in lists_map.items():
            if not getattr(cli_config, entry, None):
                setattr(cli_config, entry, default)
        return cli_config

    for entry in bools:
        file_value = file_config.pop(entry, False)
        v = getattr(cli_config, entry) or file_value
        setattr(cli_config, entry, v)

    for entry, default in scalar_map.items():
        file_value = file_config.pop(entry, default)
        v = getattr(cli_config, entry, None) or file_value
        setattr(cli_config, entry, v)

    # if either commandline parameter or config file option is set merge
    # with the other, if neither is set use the default
    for entry, default in lists_map.items():
        if getattr(cli_config, entry, None) or entry in file_config:
            value = getattr(cli_config, entry, [])
            value.extend(file_config.pop(entry, []))
        else:
            value = default
        setattr(cli_config, entry, value)

    # "write_list" config has special merge rules
    entry = "write_list"
    setattr(
        cli_config,
        entry,
        WriteArgAction.merge_write_list_config(
            from_file=file_config.pop(entry, []),
            from_cli=getattr(cli_config, entry, []) or [],
        ),
    )

    if "verbosity" in file_config:
        cli_config.verbosity = cli_config.verbosity + file_config.pop("verbosity")

    # merge options that can be set only via a file config
    for entry, value in file_config.items():
        setattr(cli_config, entry, value)

    # append default kinds to the custom list
    kinds = file_config.get("kinds", [])
    kinds.extend(DEFAULT_KINDS)
    cli_config.kinds = kinds

    return cli_config


def get_config(arguments: list[str]) -> Options:
    """Extract the config based on given args."""
    parser = get_cli_parser()
    options = Options(**vars(parser.parse_args(arguments)))

    # docs is not document, being used for internal documentation building
    if options.list_rules and options.format not in [
        None,
        "brief",
        "full",
        "md",
    ]:
        parser.error(
            f"argument -f: invalid choice: '{options.format}'. "
            f"In combination with argument -L only 'brief', "
            f"'rich' or 'md' are supported with -f.",
        )

    # save info about custom config file, as options.config_file may be modified by merge_config
    file_config, options.config_file = load_config(options.config_file)
    config = merge_config(file_config, options)

    options.rulesdirs = get_rules_dirs(
        options.rulesdir,
        use_default=options.use_default_rules,
    )

    if not options.project_dir:
        project_dir, method = find_project_root(
            srcs=options.lintables,
            config_file=options.config_file,
        )
        options.project_dir = os.path.expanduser(normpath(project_dir))
        log_entries.append(
            (
                logging.INFO,
                f"Identified [filename]{project_dir}[/] as project root due [bold]{method}[/].",
            ),
        )

    if not options.project_dir or not os.path.exists(options.project_dir):
        msg = f"Failed to determine a valid project_dir: {options.project_dir}"
        raise RuntimeError(msg)

    # expand user home dir in exclude_paths
    options.exclude_paths = [
        os.path.expandvars(os.path.expanduser(p)) for p in options.exclude_paths
    ]

    # Compute final verbosity level by subtracting -q counter.
    options.verbosity -= options.quiet
    return config


def print_help(file: Any = sys.stdout) -> None:
    """Print help test to the given stream."""
    get_cli_parser().print_help(file=file)


def get_rules_dirs(rulesdir: list[Path], *, use_default: bool = True) -> list[Path]:
    """Return a list of rules dirs."""
    default_ruledirs = [DEFAULT_RULESDIR]
    default_custom_rulesdir = os.environ.get(
        CUSTOM_RULESDIR_ENVVAR,
        os.path.join(DEFAULT_RULESDIR, "custom"),
    )
    custom_ruledirs = sorted(
        str(x.resolve())
        for x in Path(default_custom_rulesdir).iterdir()
        if x.is_dir() and (x / "__init__.py").exists()
    )

    result: list[Any] = []
    if use_default:
        result = rulesdir + custom_ruledirs + default_ruledirs
    elif rulesdir:
        result = rulesdir
    else:
        result = custom_ruledirs + default_ruledirs
    return [Path(p) for p in result]