# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, # You can obtain one at http://mozilla.org/MPL/2.0/.
import concurrent.futures
import json
import logging
import multiprocessing
import ntpath
import os
import pathlib
import posixpath
import re
import shutil
import subprocess
import sys
import tempfile
import xml.etree.ElementTree as ET
from types import SimpleNamespace

import mozpack.path as mozpath
import six
import yaml
from mach.decorators import Command, CommandArgument, SubCommand
from mach.main import Mach
from mozversioncontrol import get_repository_object
from six.moves import input

from mozbuild import build_commands
from mozbuild.controller.clobber import Clobberer
from mozbuild.nodeutil import find_node_executable
from mozbuild.util import memoize


# Function used to run clang-format on a batch of files. It is a helper function
# in order to integrate into the futures ecosystem clang-format.
def run_one_clang_format_batch(args):
    try:
        subprocess.check_output(args)
    except subprocess.CalledProcessError as e:
        return e


def build_repo_relative_path(abs_path, repo_path):
    """Build path relative to repository root"""

    if os.path.islink(abs_path):
        abs_path = mozpath.realpath(abs_path)

    return mozpath.relpath(abs_path, repo_path)


def prompt_bool(prompt, limit=5):
    """Prompts the user with prompt and requires a boolean value."""
    from distutils.util import strtobool

    for _ in range(limit):
        try:
            return strtobool(input(prompt + "[Y/N]\n"))
        except ValueError:
            print(
                "ERROR! Please enter a valid option! Please use any of the following:"
                " Y, N, True, False, 1, 0"
            )
    return False


class StaticAnalysisSubCommand(SubCommand):
    def __call__(self, func):
        after = SubCommand.__call__(self, func)
        args = [
            CommandArgument(
                "--verbose", "-v", action="store_true", help="Print verbose output."
            )
        ]
        for arg in args:
            after = arg(after)
        return after


class StaticAnalysisMonitor(object):
    def __init__(self, srcdir, objdir, checks, total):
        self._total = total
        self._processed = 0
        self._current = None
        self._srcdir = srcdir

        import copy

        self._checks = copy.deepcopy(checks)

        # Transform the configuration to support Regex
        for item in self._checks:
            if item["name"] == "-*":
                continue
            item["name"] = item["name"].replace("*", ".*")

        from mozbuild.compilation.warnings import WarningsCollector, WarningsDatabase

        self._warnings_database = WarningsDatabase()

        def on_warning(warning):

            # Output paths relative to repository root if the paths are under repo tree
            warning["filename"] = build_repo_relative_path(
                warning["filename"], self._srcdir
            )

            self._warnings_database.insert(warning)

        self._warnings_collector = WarningsCollector(on_warning, objdir=objdir)

    @property
    def num_files(self):
        return self._total

    @property
    def num_files_processed(self):
        return self._processed

    @property
    def current_file(self):
        return self._current

    @property
    def warnings_db(self):
        return self._warnings_database

    def on_line(self, line):
        warning = None

        try:
            warning = self._warnings_collector.process_line(line)
        except Exception:
            pass

        if line.find("clang-tidy") != -1:
            filename = line.split(" ")[-1]
            if os.path.isfile(filename):
                self._current = build_repo_relative_path(filename, self._srcdir)
            else:
                self._current = None
            self._processed = self._processed + 1
            return (warning, False)
        if warning is not None:

            def get_check_config(checker_name):
                # get the matcher from self._checks that is the 'name' field
                for item in self._checks:
                    if item["name"] == checker_name:
                        return item

                    # We are using a regex in order to also match 'mozilla-.* like checkers'
                    matcher = re.match(item["name"], checker_name)
                    if matcher is not None and matcher.group(0) == checker_name:
                        return item

            check_config = get_check_config(warning["flag"])
            if check_config is not None:
                warning["reliability"] = check_config.get("reliability", "low")
                warning["reason"] = check_config.get("reason")
                warning["publish"] = check_config.get("publish", True)
            elif warning["flag"] == "clang-diagnostic-error":
                # For a "warning" that is flagged as "clang-diagnostic-error"
                # set it as "publish"
                warning["publish"] = True

        return (warning, True)


# Utilities for running C++ static analysis checks and format.

# List of file extension to consider (should start with dot)
_format_include_extensions = (".cpp", ".c", ".cc", ".h", ".m", ".mm")
# File contaning all paths to exclude from formatting
_format_ignore_file = ".clang-format-ignore"

# (TOOLS) Function return codes
TOOLS_SUCCESS = 0
TOOLS_FAILED_DOWNLOAD = 1
TOOLS_UNSUPORTED_PLATFORM = 2
TOOLS_CHECKER_NO_TEST_FILE = 3
TOOLS_CHECKER_RETURNED_NO_ISSUES = 4
TOOLS_CHECKER_RESULT_FILE_NOT_FOUND = 5
TOOLS_CHECKER_DIFF_FAILED = 6
TOOLS_CHECKER_NOT_FOUND = 7
TOOLS_CHECKER_FAILED_FILE = 8
TOOLS_CHECKER_LIST_EMPTY = 9
TOOLS_GRADLE_FAILED = 10


@Command(
    "clang-tidy",
    category="devenv",
    description="Convenience alias for the static-analysis command",
)
def clang_tidy(command_context):
    # If no arguments are provided, just print a help message.
    """Detailed documentation:
    https://firefox-source-docs.mozilla.org/code-quality/static-analysis/index.html
    """
    mach = Mach(os.getcwd())

    def populate_context(key=None):
        if key == "topdir":
            return command_context.topsrcdir

    mach.populate_context_handler = populate_context
    mach.run(["static-analysis", "--help"])


@Command(
    "static-analysis",
    category="devenv",
    description="Run C++ static analysis checks using clang-tidy",
)
def static_analysis(command_context):
    # If no arguments are provided, just print a help message.
    """Detailed documentation:
    https://firefox-source-docs.mozilla.org/code-quality/static-analysis/index.html
    """
    mach = Mach(os.getcwd())

    def populate_context(key=None):
        if key == "topdir":
            return command_context.topsrcdir

    mach.populate_context_handler = populate_context
    mach.run(["static-analysis", "--help"])


@StaticAnalysisSubCommand(
    "static-analysis", "check", "Run the checks using the helper tool"
)
@CommandArgument(
    "source",
    nargs="*",
    default=[".*"],
    help="Source files to be analyzed (regex on path). "
    "Can be omitted, in which case the entire code base "
    "is analyzed.  The source argument is ignored if "
    "there is anything fed through stdin, in which case "
    "the analysis is only performed on the files changed "
    "in the patch streamed through stdin.  This is called "
    "the diff mode.",
)
@CommandArgument(
    "--checks",
    "-c",
    default="-*",
    metavar="checks",
    help="Static analysis checks to enable.  By default, this enables only "
    "checks that are published here: https://mzl.la/2DRHeTh, but can be any "
    "clang-tidy checks syntax.",
)
@CommandArgument(
    "--jobs",
    "-j",
    default="0",
    metavar="jobs",
    type=int,
    help="Number of concurrent jobs to run. Default is the number of CPUs.",
)
@CommandArgument(
    "--strip",
    "-p",
    default="1",
    metavar="NUM",
    help="Strip NUM leading components from file names in diff mode.",
)
@CommandArgument(
    "--fix",
    "-f",
    default=False,
    action="store_true",
    help="Try to autofix errors detected by clang-tidy checkers.",
)
@CommandArgument(
    "--header-filter",
    "-h-f",
    default="",
    metavar="header_filter",
    help="Regular expression matching the names of the headers to "
    "output diagnostics from. Diagnostics from the main file "
    "of each translation unit are always displayed",
)
@CommandArgument(
    "--output", "-o", default=None, help="Write clang-tidy output in a file"
)
@CommandArgument(
    "--format",
    default="text",
    choices=("text", "json"),
    help="Output format to write in a file",
)
@CommandArgument(
    "--outgoing",
    default=False,
    action="store_true",
    help="Run static analysis checks on outgoing files from mercurial repository",
)
def check(
    command_context,
    source=None,
    jobs=2,
    strip=1,
    verbose=False,
    checks="-*",
    fix=False,
    header_filter="",
    output=None,
    format="text",
    outgoing=False,
):
    from mozbuild.controller.building import (
        StaticAnalysisFooter,
        StaticAnalysisOutputManager,
    )

    command_context._set_log_level(verbose)
    command_context.activate_virtualenv()
    command_context.log_manager.enable_unstructured()

    rc, clang_paths = get_clang_tools(command_context, verbose=verbose)
    if rc != 0:
        return rc

    if not _is_version_eligible(command_context, clang_paths):
        return 1

    rc, _compile_db, compilation_commands_path = _build_compile_db(
        command_context, verbose=verbose
    )
    rc = rc or _build_export(command_context, jobs=jobs, verbose=verbose)
    if rc != 0:
        return rc

    # Use outgoing files instead of source files
    if outgoing:
        repo = get_repository_object(command_context.topsrcdir)
        files = repo.get_outgoing_files()
        source = get_abspath_files(command_context, files)

    # Split in several chunks to avoid hitting Python's limit of 100 groups in re
    compile_db = json.loads(open(_compile_db, "r").read())
    total = 0
    import re

    chunk_size = 50
    for offset in range(0, len(source), chunk_size):
        source_chunks = [
            re.escape(f) for f in source[offset : offset + chunk_size].copy()
        ]
        name_re = re.compile("(" + ")|(".join(source_chunks) + ")")
        for f in compile_db:
            if name_re.search(f["file"]):
                total = total + 1

    # Filter source to remove excluded files
    source = _generate_path_list(command_context, source, verbose=verbose)

    if not total or not source:
        command_context.log(
            logging.INFO,
            "static-analysis",
            {},
            "There are no files eligible for analysis. Please note that 'header' files "
            "cannot be used for analysis since they do not consist compilation units.",
        )
        return 0

    # Escape the files from source
    source = [re.escape(f) for f in source]

    cwd = command_context.topobjdir

    monitor = StaticAnalysisMonitor(
        command_context.topsrcdir,
        command_context.topobjdir,
        get_clang_tidy_config(command_context).checks_with_data,
        total,
    )

    footer = StaticAnalysisFooter(command_context.log_manager.terminal, monitor)

    with StaticAnalysisOutputManager(
        command_context.log_manager, monitor, footer
    ) as output_manager:
        import math

        batch_size = int(math.ceil(float(len(source)) / multiprocessing.cpu_count()))
        for i in range(0, len(source), batch_size):
            args = _get_clang_tidy_command(
                command_context,
                clang_paths,
                compilation_commands_path,
                checks=checks,
                header_filter=header_filter,
                sources=source[i : (i + batch_size)],
                jobs=jobs,
                fix=fix,
            )
            rc = command_context.run_process(
                args=args,
                ensure_exit_code=False,
                line_handler=output_manager.on_line,
                cwd=cwd,
            )

        command_context.log(
            logging.WARNING,
            "warning_summary",
            {"count": len(monitor.warnings_db)},
            "{count} warnings present.",
        )

        # Write output file
        if output is not None:
            output_manager.write(output, format)

    return rc


def get_abspath_files(command_context, files):
    return [mozpath.join(command_context.topsrcdir, f) for f in files]


def get_files_with_commands(command_context, compile_db, source):
    """
    Returns an array of dictionaries having file_path with build command
    """

    compile_db = json.load(open(compile_db, "r"))

    commands_list = []

    for f in source:
        # It must be a C/C++ file
        _, ext = os.path.splitext(f)

        if ext.lower() not in _format_include_extensions:
            command_context.log(
                logging.INFO, "static-analysis", {}, "Skipping {}".format(f)
            )
            continue
        file_with_abspath = os.path.join(command_context.topsrcdir, f)
        for f in compile_db:
            # Found for a file that we are looking
            if file_with_abspath == f["file"]:
                commands_list.append(f)

    return commands_list


@memoize
def get_clang_tidy_config(command_context):
    from mozbuild.code_analysis.utils import ClangTidyConfig

    return ClangTidyConfig(command_context.topsrcdir)


def _get_required_version(command_context):
    version = get_clang_tidy_config(command_context).version
    if version is None:
        command_context.log(
            logging.ERROR,
            "static-analysis",
            {},
            "ERROR: Unable to find 'package_version' in config.yml",
        )
    return version


def _get_current_version(command_context, clang_paths):
    # Because the fact that we ship together clang-tidy and clang-format
    # we are sure that these two will always share the same version.
    # Thus in order to determine that the version is compatible we only
    # need to check one of them, going with clang-format
    cmd = [clang_paths._clang_format_path, "--version"]
    version_info = None
    try:
        version_info = (
            subprocess.check_output(cmd, stderr=subprocess.STDOUT)
            .decode("utf-8")
            .strip()
        )

        if "MOZ_AUTOMATION" in os.environ:
            # Only show it in the CI
            command_context.log(
                logging.INFO,
                "static-analysis",
                {},
                "{} Version = {} ".format(clang_paths._clang_format_path, version_info),
            )

    except subprocess.CalledProcessError as e:
        command_context.log(
            logging.ERROR,
            "static-analysis",
            {},
            "Error determining the version clang-tidy/format binary, please see the "
            "attached exception: \n{}".format(e.output),
        )
    return version_info


def _is_version_eligible(command_context, clang_paths, log_error=True):
    version = _get_required_version(command_context)
    if version is None:
        return False

    current_version = _get_current_version(command_context, clang_paths)
    if current_version is None:
        return False
    version = "clang-format version " + version
    if version in current_version:
        return True

    if log_error:
        command_context.log(
            logging.ERROR,
            "static-analysis",
            {},
            "ERROR: You're using an old or incorrect version ({}) of clang-format binary. "
            "Please update to a more recent one (at least > {}) "
            "by running: './mach bootstrap' ".format(
                _get_current_version(command_context, clang_paths),
                _get_required_version(command_context),
            ),
        )

    return False


def _get_clang_tidy_command(
    command_context,
    clang_paths,
    compilation_commands_path,
    checks,
    header_filter,
    sources,
    jobs,
    fix,
):

    if checks == "-*":
        checks = ",".join(get_clang_tidy_config(command_context).checks)

    common_args = [
        "-clang-tidy-binary",
        clang_paths._clang_tidy_path,
        "-clang-apply-replacements-binary",
        clang_paths._clang_apply_replacements,
        "-checks=%s" % checks,
        "-extra-arg=-DMOZ_CLANG_PLUGIN",
    ]

    # Flag header-filter is passed in order to limit the diagnostic messages only
    # to the specified header files. When no value is specified the default value
    # is considered to be the source in order to limit the diagnostic message to
    # the source files or folders.
    common_args += [
        "-header-filter=%s"
        % (header_filter if len(header_filter) else "|".join(sources))
    ]

    # From our configuration file, config.yaml, we build the configuration list, for
    # the checkers that are used. These configuration options are used to better fit
    # the checkers to our code.
    cfg = get_clang_tidy_config(command_context).checks_config
    if cfg:
        common_args += ["-config=%s" % yaml.dump(cfg)]

    if fix:
        common_args += ["-fix"]

    return (
        [
            command_context.virtualenv_manager.python_path,
            clang_paths._run_clang_tidy_path,
            "-j",
            str(jobs),
            "-p",
            compilation_commands_path,
        ]
        + common_args
        + sources
    )


@StaticAnalysisSubCommand(
    "static-analysis",
    "autotest",
    "Run the auto-test suite in order to determine that"
    " the analysis did not regress.",
)
@CommandArgument(
    "--dump-results",
    "-d",
    default=False,
    action="store_true",
    help="Generate the baseline for the regression test. Based on"
    " this baseline we will test future results.",
)
@CommandArgument(
    "--intree-tool",
    "-i",
    default=False,
    action="store_true",
    help="Use a pre-aquired in-tree clang-tidy package from the automation env."
    " This option is only valid on automation environments.",
)
@CommandArgument(
    "checker_names",
    nargs="*",
    default=[],
    help="Checkers that are going to be auto-tested.",
)
def autotest(
    command_context,
    verbose=False,
    dump_results=False,
    intree_tool=False,
    checker_names=[],
):
    # If 'dump_results' is True than we just want to generate the issues files for each
    # checker in particulat and thus 'force_download' becomes 'False' since we want to
    # do this on a local trusted clang-tidy package.
    command_context._set_log_level(verbose)
    command_context.activate_virtualenv()
    dump_results = dump_results

    force_download = not dump_results

    # Configure the tree or download clang-tidy package, depending on the option that we choose
    if intree_tool:
        clang_paths = SimpleNamespace()
        if "MOZ_AUTOMATION" not in os.environ:
            command_context.log(
                logging.INFO,
                "static-analysis",
                {},
                "The `autotest` with `--intree-tool` can only be ran in automation.",
            )
            return 1
        if "MOZ_FETCHES_DIR" not in os.environ:
            command_context.log(
                logging.INFO,
                "static-analysis",
                {},
                "`MOZ_FETCHES_DIR` is missing from the environment variables.",
            )
            return 1

        _, config, _ = _get_config_environment(command_context)
        clang_tools_path = os.environ["MOZ_FETCHES_DIR"]
        clang_paths._clang_tidy_path = mozpath.join(
            clang_tools_path,
            "clang-tidy",
            "bin",
            "clang-tidy" + config.substs.get("HOST_BIN_SUFFIX", ""),
        )
        clang_paths._clang_format_path = mozpath.join(
            clang_tools_path,
            "clang-tidy",
            "bin",
            "clang-format" + config.substs.get("HOST_BIN_SUFFIX", ""),
        )
        clang_paths._clang_apply_replacements = mozpath.join(
            clang_tools_path,
            "clang-tidy",
            "bin",
            "clang-apply-replacements" + config.substs.get("HOST_BIN_SUFFIX", ""),
        )
        clang_paths._run_clang_tidy_path = mozpath.join(
            clang_tools_path, "clang-tidy", "bin", "run-clang-tidy"
        )
        clang_paths._clang_format_diff = mozpath.join(
            clang_tools_path, "clang-tidy", "share", "clang", "clang-format-diff.py"
        )

        # Ensure that clang-tidy is present
        rc = not os.path.exists(clang_paths._clang_tidy_path)
    else:
        rc, clang_paths = get_clang_tools(
            command_context, force=force_download, verbose=verbose
        )

    if rc != 0:
        command_context.log(
            logging.ERROR,
            "ERROR: static-analysis",
            {},
            "ERROR: clang-tidy unable to locate package.",
        )
        return TOOLS_FAILED_DOWNLOAD

    clang_paths._clang_tidy_base_path = mozpath.join(
        command_context.topsrcdir, "tools", "clang-tidy"
    )

    # For each checker run it
    platform, _ = command_context.platform

    if platform not in get_clang_tidy_config(command_context).platforms:
        command_context.log(
            logging.ERROR,
            "static-analysis",
            {},
            "ERROR: RUNNING: clang-tidy autotest for platform {} not supported.".format(
                platform
            ),
        )
        return TOOLS_UNSUPORTED_PLATFORM

    max_workers = multiprocessing.cpu_count()

    command_context.log(
        logging.INFO,
        "static-analysis",
        {},
        "RUNNING: clang-tidy autotest for platform {0} with {1} workers.".format(
            platform, max_workers
        ),
    )

    # List all available checkers
    cmd = [clang_paths._clang_tidy_path, "-list-checks", "-checks=*"]
    clang_output = subprocess.check_output(cmd, stderr=subprocess.STDOUT).decode(
        "utf-8"
    )
    available_checks = clang_output.split("\n")[1:]
    clang_tidy_checks = [c.strip() for c in available_checks if c]

    # Build the dummy compile_commands.json
    compilation_commands_path = _create_temp_compilation_db(command_context)
    checkers_test_batch = []
    checkers_results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = []
        for item in get_clang_tidy_config(command_context).checks_with_data:
            # Skip if any of the following statements is true:
            # 1. Checker attribute 'publish' is False.
            not_published = not bool(item.get("publish", True))
            # 2. Checker has restricted-platforms and current platform is not of them.
            ignored_platform = (
                "restricted-platforms" in item
                and platform not in item["restricted-platforms"]
            )
            # 3. Checker name is mozilla-* or -*.
            ignored_checker = item["name"] in ["mozilla-*", "-*"]
            # 4. List checker_names is passed and the current checker is not part of the
            #    list or 'publish' is False
            checker_not_in_list = checker_names and (
                item["name"] not in checker_names or not_published
            )
            if (
                not_published
                or ignored_platform
                or ignored_checker
                or checker_not_in_list
            ):
                continue
            checkers_test_batch.append(item["name"])
            futures.append(
                executor.submit(
                    _verify_checker,
                    command_context,
                    clang_paths,
                    compilation_commands_path,
                    dump_results,
                    clang_tidy_checks,
                    item,
                    checkers_results,
                )
            )

        error_code = TOOLS_SUCCESS
        for future in concurrent.futures.as_completed(futures):
            # Wait for every task to finish
            ret_val = future.result()
            if ret_val != TOOLS_SUCCESS:
                # We are interested only in one error and we don't break
                # the execution of for loop since we want to make sure that all
                # tasks finished.
                error_code = ret_val

        if error_code != TOOLS_SUCCESS:

            command_context.log(
                logging.INFO,
                "static-analysis",
                {},
                "FAIL: the following clang-tidy check(s) failed:",
            )
            for failure in checkers_results:
                checker_error = failure["checker-error"]
                checker_name = failure["checker-name"]
                info1 = failure["info1"]
                info2 = failure["info2"]
                info3 = failure["info3"]

                message_to_log = ""
                if checker_error == TOOLS_CHECKER_NOT_FOUND:
                    message_to_log = (
                        "\tChecker "
                        "{} not present in this clang-tidy version.".format(
                            checker_name
                        )
                    )
                elif checker_error == TOOLS_CHECKER_NO_TEST_FILE:
                    message_to_log = (
                        "\tChecker "
                        "{0} does not have a test file - {0}.cpp".format(checker_name)
                    )
                elif checker_error == TOOLS_CHECKER_RETURNED_NO_ISSUES:
                    message_to_log = (
                        "\tChecker {0} did not find any issues in its test file, "
                        "clang-tidy output for the run is:\n{1}"
                    ).format(checker_name, info1)
                elif checker_error == TOOLS_CHECKER_RESULT_FILE_NOT_FOUND:
                    message_to_log = (
                        "\tChecker {0} does not have a result file - {0}.json"
                    ).format(checker_name)
                elif checker_error == TOOLS_CHECKER_DIFF_FAILED:
                    message_to_log = (
                        "\tChecker {0}\nExpected: {1}\n"
                        "Got: {2}\n"
                        "clang-tidy output for the run is:\n"
                        "{3}"
                    ).format(checker_name, info1, info2, info3)

                print("\n" + message_to_log)

            # Also delete the tmp folder
            shutil.rmtree(compilation_commands_path)
            return error_code

        # Run the analysis on all checkers at the same time only if we don't dump results.
        if not dump_results:
            ret_val = _run_analysis_batch(
                command_context,
                clang_paths,
                compilation_commands_path,
                checkers_test_batch,
            )
            if ret_val != TOOLS_SUCCESS:
                shutil.rmtree(compilation_commands_path)
                return ret_val

    command_context.log(
        logging.INFO, "static-analysis", {}, "SUCCESS: clang-tidy all tests passed."
    )
    # Also delete the tmp folder
    shutil.rmtree(compilation_commands_path)


def _run_analysis(
    command_context,
    clang_paths,
    compilation_commands_path,
    checks,
    header_filter,
    sources,
    jobs=1,
    fix=False,
    print_out=False,
):
    cmd = _get_clang_tidy_command(
        command_context,
        clang_paths,
        compilation_commands_path,
        checks=checks,
        header_filter=header_filter,
        sources=sources,
        jobs=jobs,
        fix=fix,
    )

    try:
        clang_output = subprocess.check_output(cmd, stderr=subprocess.STDOUT).decode(
            "utf-8"
        )
    except subprocess.CalledProcessError as e:
        print(e.output)
        return None
    return _parse_issues(command_context, clang_output), clang_output


def _run_analysis_batch(command_context, clang_paths, compilation_commands_path, items):
    command_context.log(
        logging.INFO,
        "static-analysis",
        {},
        "RUNNING: clang-tidy checker batch analysis.",
    )
    if not len(items):
        command_context.log(
            logging.ERROR,
            "static-analysis",
            {},
            "ERROR: clang-tidy checker list is empty!",
        )
        return TOOLS_CHECKER_LIST_EMPTY

    issues, clang_output = _run_analysis(
        command_context,
        clang_paths,
        compilation_commands_path,
        checks="-*," + ",".join(items),
        header_filter="",
        sources=[
            mozpath.join(clang_paths._clang_tidy_base_path, "test", checker) + ".cpp"
            for checker in items
        ],
        print_out=True,
    )

    if issues is None:
        return TOOLS_CHECKER_FAILED_FILE

    failed_checks = []
    failed_checks_baseline = []
    for checker in items:
        test_file_path_json = (
            mozpath.join(clang_paths._clang_tidy_base_path, "test", checker) + ".json"
        )
        # Read the pre-determined issues
        baseline_issues = _get_autotest_stored_issues(test_file_path_json)

        # We also stored the 'reliability' index so strip that from the baseline_issues
        baseline_issues[:] = [
            item for item in baseline_issues if "reliability" not in item
        ]

        found = all([element_base in issues for element_base in baseline_issues])

        if not found:
            failed_checks.append(checker)
            failed_checks_baseline.append(baseline_issues)

    if len(failed_checks) > 0:
        command_context.log(
            logging.ERROR,
            "static-analysis",
            {},
            "ERROR: The following check(s) failed for bulk analysis: "
            + " ".join(failed_checks),
        )

        for failed_check, baseline_issue in zip(failed_checks, failed_checks_baseline):
            print(
                "\tChecker {0} expect following results: \n\t\t{1}".format(
                    failed_check, baseline_issue
                )
            )

        print(
            "This is the output generated by clang-tidy for the bulk build:\n{}".format(
                clang_output
            )
        )
        return TOOLS_CHECKER_DIFF_FAILED

    return TOOLS_SUCCESS


def _create_temp_compilation_db(command_context):
    directory = tempfile.mkdtemp(prefix="cc")
    with open(mozpath.join(directory, "compile_commands.json"), "w") as file_handler:
        compile_commands = []
        director = mozpath.join(
            command_context.topsrcdir, "tools", "clang-tidy", "test"
        )
        for item in get_clang_tidy_config(command_context).checks:
            if item in ["-*", "mozilla-*"]:
                continue
            file = item + ".cpp"
            element = {}
            element["directory"] = director
            element["command"] = "cpp -std=c++17 " + file
            element["file"] = mozpath.join(director, file)
            compile_commands.append(element)

        json.dump(compile_commands, file_handler)
        file_handler.flush()

        return directory


@StaticAnalysisSubCommand(
    "static-analysis", "install", "Install the static analysis helper tool"
)
@CommandArgument(
    "source",
    nargs="?",
    type=str,
    help="Where to fetch a local archive containing the static-analysis and "
    "format helper tool."
    "It will be installed in ~/.mozbuild/clang-tools."
    "Can be omitted, in which case the latest clang-tools "
    "helper for the platform would be automatically detected and installed.",
)
@CommandArgument(
    "--skip-cache",
    action="store_true",
    help="Skip all local caches to force re-fetching the helper tool.",
    default=False,
)
@CommandArgument(
    "--force",
    action="store_true",
    help="Force re-install even though the tool exists in mozbuild.",
    default=False,
)
def install(
    command_context,
    source=None,
    skip_cache=False,
    force=False,
    verbose=False,
):
    command_context._set_log_level(verbose)
    rc, _ = get_clang_tools(
        command_context,
        force=force,
        skip_cache=skip_cache,
        source=source,
        verbose=verbose,
    )
    return rc


@StaticAnalysisSubCommand(
    "static-analysis",
    "clear-cache",
    "Delete local helpers and reset static analysis helper tool cache",
)
def clear_cache(command_context, verbose=False):
    command_context._set_log_level(verbose)
    rc, _ = get_clang_tools(
        command_context,
        force=True,
        download_if_needed=True,
        skip_cache=True,
        verbose=verbose,
    )

    if rc != 0:
        return rc

    from mozbuild.artifact_commands import artifact_clear_cache

    return artifact_clear_cache(command_context)


@StaticAnalysisSubCommand(
    "static-analysis",
    "print-checks",
    "Print a list of the static analysis checks performed by default",
)
def print_checks(command_context, verbose=False):
    command_context._set_log_level(verbose)
    rc, clang_paths = get_clang_tools(command_context, verbose=verbose)

    if rc != 0:
        return rc

    args = [
        clang_paths._clang_tidy_path,
        "-list-checks",
        "-checks=%s" % get_clang_tidy_config(command_context).checks,
    ]

    return command_context.run_process(args=args, pass_thru=True)


@Command(
    "prettier-format",
    category="misc",
    description="Run prettier on current changes",
)
@CommandArgument(
    "--path",
    "-p",
    nargs=1,
    required=True,
    help="Specify the path to reformat to stdout.",
)
@CommandArgument(
    "--assume-filename",
    "-a",
    nargs=1,
    required=True,
    help="This option is usually used in the context of hg-formatsource."
    "When reading from stdin, Prettier assumes this "
    "filename to decide which style and parser to use.",
)
def prettier_format(command_context, path, assume_filename):
    # With assume_filename we want to have stdout clean since the result of the
    # format will be redirected to stdout.

    binary, _ = find_node_executable()
    prettier = os.path.join(
        command_context.topsrcdir, "node_modules", "prettier", "bin-prettier.js"
    )
    path = os.path.join(command_context.topsrcdir, path[0])

    # Bug 1564824. Prettier fails on patches with moved files where the
    # original directory also does not exist.
    assume_dir = os.path.dirname(
        os.path.join(command_context.topsrcdir, assume_filename[0])
    )
    assume_filename = assume_filename[0] if os.path.isdir(assume_dir) else path

    # We use --stdin-filepath in order to better determine the path for
    # the prettier formatter when it is ran outside of the repo, for example
    # by the extension hg-formatsource.
    args = [binary, prettier, "--stdin-filepath", assume_filename]

    process = subprocess.Popen(args, stdin=subprocess.PIPE)
    with open(path, "rb") as fin:
        process.stdin.write(fin.read())
        process.stdin.close()
        process.wait()
        return process.returncode


@Command(
    "clang-format",
    category="misc",
    description="Run clang-format on current changes",
)
@CommandArgument(
    "--show",
    "-s",
    action="store_const",
    const="stdout",
    dest="output_path",
    help="Show diff output on stdout instead of applying changes",
)
@CommandArgument(
    "--assume-filename",
    "-a",
    nargs=1,
    default=None,
    help="This option is usually used in the context of hg-formatsource."
    "When reading from stdin, clang-format assumes this "
    "filename to look for a style config file (with "
    "-style=file) and to determine the language. When "
    "specifying this option only one file should be used "
    "as an input and the output will be forwarded to stdin. "
    "This option also impairs the download of the clang-tools "
    "and assumes the package is already located in it's default "
    "location",
)
@CommandArgument(
    "--path", "-p", nargs="+", default=None, help="Specify the path(s) to reformat"
)
@CommandArgument(
    "--commit",
    "-c",
    default=None,
    help="Specify a commit to reformat from. "
    "For git you can also pass a range of commits (foo..bar) "
    "to format all of them at the same time.",
)
@CommandArgument(
    "--output",
    "-o",
    default=None,
    dest="output_path",
    help="Specify a file handle to write clang-format raw output instead of "
    "applying changes. This can be stdout or a file path.",
)
@CommandArgument(
    "--format",
    "-f",
    choices=("diff", "json"),
    default="diff",
    dest="output_format",
    help="Specify the output format used: diff is the raw patch provided by "
    "clang-format, json is a list of atomic changes to process.",
)
@CommandArgument(
    "--outgoing",
    default=False,
    action="store_true",
    help="Run clang-format on outgoing files from mercurial repository.",
)
def clang_format(
    command_context,
    assume_filename,
    path,
    commit,
    output_path=None,
    output_format="diff",
    verbose=False,
    outgoing=False,
):
    # Run clang-format or clang-format-diff on the local changes
    # or files/directories
    if path is None and outgoing:
        repo = get_repository_object(command_context.topsrcdir)
        path = repo.get_outgoing_files()

    if path:
        # Create the full path list
        def path_maker(f_name):
            return os.path.join(command_context.topsrcdir, f_name)

        path = map(path_maker, path)

    os.chdir(command_context.topsrcdir)

    # Load output file handle, either stdout or a file handle in write mode
    output = None
    if output_path is not None:
        output = sys.stdout if output_path == "stdout" else open(output_path, "w")

    # With assume_filename we want to have stdout clean since the result of the
    # format will be redirected to stdout. Only in case of errror we
    # write something to stdout.
    # We don't actually want to get the clang-tools here since we want in some
    # scenarios to do this in parallel so we relay on the fact that the tools
    # have already been downloaded via './mach bootstrap' or directly via
    # './mach static-analysis install'
    if assume_filename:
        rc, clang_paths = _set_clang_tools_paths(command_context)
        if rc != 0:
            print("clang-format: Unable to set path to clang-format tools.")
            return rc

        if not _do_clang_tools_exist(clang_paths):
            print("clang-format: Unable to set locate clang-format tools.")
            return 1

        if not _is_version_eligible(command_context, clang_paths):
            return 1
    else:
        rc, clang_paths = get_clang_tools(command_context, verbose=verbose)
        if rc != 0:
            return rc

    if path is None:
        return _run_clang_format_diff(
            command_context,
            clang_paths._clang_format_diff,
            clang_paths._clang_format_path,
            commit,
            output,
        )

    if assume_filename:
        return _run_clang_format_in_console(
            command_context, clang_paths._clang_format_path, path, assume_filename
        )

    return _run_clang_format_path(
        command_context, clang_paths._clang_format_path, path, output, output_format
    )


def _verify_checker(
    command_context,
    clang_paths,
    compilation_commands_path,
    dump_results,
    clang_tidy_checks,
    item,
    checkers_results,
):
    check = item["name"]
    test_file_path = mozpath.join(clang_paths._clang_tidy_base_path, "test", check)
    test_file_path_cpp = test_file_path + ".cpp"
    test_file_path_json = test_file_path + ".json"

    command_context.log(
        logging.INFO,
        "static-analysis",
        {},
        "RUNNING: clang-tidy checker {}.".format(check),
    )

    # Structured information in case a checker fails
    checker_error = {
        "checker-name": check,
        "checker-error": "",
        "info1": "",
        "info2": "",
        "info3": "",
    }

    # Verify if this checker actually exists
    if check not in clang_tidy_checks:
        checker_error["checker-error"] = TOOLS_CHECKER_NOT_FOUND
        checkers_results.append(checker_error)
        return TOOLS_CHECKER_NOT_FOUND

    # Verify if the test file exists for this checker
    if not os.path.exists(test_file_path_cpp):
        checker_error["checker-error"] = TOOLS_CHECKER_NO_TEST_FILE
        checkers_results.append(checker_error)
        return TOOLS_CHECKER_NO_TEST_FILE

    issues, clang_output = _run_analysis(
        command_context,
        clang_paths,
        compilation_commands_path,
        checks="-*," + check,
        header_filter="",
        sources=[test_file_path_cpp],
    )
    if issues is None:
        return TOOLS_CHECKER_FAILED_FILE

    # Verify to see if we got any issues, if not raise exception
    if not issues:
        checker_error["checker-error"] = TOOLS_CHECKER_RETURNED_NO_ISSUES
        checker_error["info1"] = clang_output
        checkers_results.append(checker_error)
        return TOOLS_CHECKER_RETURNED_NO_ISSUES

    # Also store the 'reliability' index for this checker
    issues.append({"reliability": item["reliability"]})

    if dump_results:
        _build_autotest_result(test_file_path_json, json.dumps(issues))
    else:
        if not os.path.exists(test_file_path_json):
            # Result file for test not found maybe regenerate it?
            checker_error["checker-error"] = TOOLS_CHECKER_RESULT_FILE_NOT_FOUND
            checkers_results.append(checker_error)
            return TOOLS_CHECKER_RESULT_FILE_NOT_FOUND

        # Read the pre-determined issues
        baseline_issues = _get_autotest_stored_issues(test_file_path_json)

        # Compare the two lists
        if issues != baseline_issues:
            checker_error["checker-error"] = TOOLS_CHECKER_DIFF_FAILED
            checker_error["info1"] = baseline_issues
            checker_error["info2"] = issues
            checker_error["info3"] = clang_output
            checkers_results.append(checker_error)
            return TOOLS_CHECKER_DIFF_FAILED

    return TOOLS_SUCCESS


def _build_autotest_result(file, issues):
    with open(file, "w") as f:
        f.write(issues)


def _get_autotest_stored_issues(file):
    with open(file) as f:
        return json.load(f)


def _parse_issues(command_context, clang_output):
    """
    Parse clang-tidy output into structured issues
    """

    # Limit clang output parsing to 'Enabled checks:'
    end = re.search(r"^Enabled checks:\n", clang_output, re.MULTILINE)
    if end is not None:
        clang_output = clang_output[: end.start() - 1]

    platform, _ = command_context.platform
    re_strip_colors = re.compile(r"\x1b\[[\d;]+m", re.MULTILINE)
    filtered = re_strip_colors.sub("", clang_output)
    # Starting with clang 8, for the diagnostic messages we have multiple `LF CR`
    # in order to be compatiable with msvc compiler format, and for this
    # we are not interested to match the end of line.
    regex_string = r"(.+):(\d+):(\d+): (warning|error): ([^\[\]\n]+)(?: \[([\.\w-]+)\])"

    # For non 'win' based platforms we also need the 'end of the line' regex
    if platform not in ("win64", "win32"):
        regex_string += "?$"

    regex_header = re.compile(regex_string, re.MULTILINE)

    # Sort headers by positions
    headers = sorted(regex_header.finditer(filtered), key=lambda h: h.start())
    issues = []
    for _, header in enumerate(headers):
        header_group = header.groups()
        element = [header_group[3], header_group[4], header_group[5]]
        issues.append(element)
    return issues


def _get_config_environment(command_context):
    ran_configure = False
    config = None

    try:
        config = command_context.config_environment
    except Exception:
        command_context.log(
            logging.WARNING,
            "static-analysis",
            {},
            "Looks like configure has not run yet, running it now...",
        )

        clobber = Clobberer(command_context.topsrcdir, command_context.topobjdir)

        if clobber.clobber_needed():
            choice = prompt_bool(
                "Configuration has changed and Clobber is needed. "
                "Do you want to proceed?"
            )
            if not choice:
                command_context.log(
                    logging.ERROR,
                    "static-analysis",
                    {},
                    "ERROR: Without Clobber we cannot continue execution!",
                )
                return (1, None, None)
            os.environ["AUTOCLOBBER"] = "1"

        rc = build_commands.configure(command_context)
        if rc != 0:
            return (rc, config, ran_configure)
        ran_configure = True
        try:
            config = command_context.config_environment
        except Exception:
            pass

    return (0, config, ran_configure)


def _build_compile_db(command_context, verbose=False):
    compilation_commands_path = mozpath.join(
        command_context.topobjdir, "static-analysis"
    )
    compile_db = mozpath.join(compilation_commands_path, "compile_commands.json")

    if os.path.exists(compile_db):
        return 0, compile_db, compilation_commands_path

    rc, config, ran_configure = _get_config_environment(command_context)
    if rc != 0:
        return rc, compile_db, compilation_commands_path

    if ran_configure:
        # Configure may have created the compilation database if the
        # mozconfig enables building the CompileDB backend by default,
        # So we recurse to see if the file exists once again.
        return _build_compile_db(command_context, verbose=verbose)

    if config:
        print(
            "Looks like a clang compilation database has not been "
            "created yet, creating it now..."
        )
        rc = build_commands.build_backend(
            command_context, ["StaticAnalysis"], verbose=verbose
        )
        if rc != 0:
            return rc, compile_db, compilation_commands_path
        assert os.path.exists(compile_db)
        return 0, compile_db, compilation_commands_path


def _build_export(command_context, jobs, verbose=False):
    def on_line(line):
        command_context.log(logging.INFO, "build_output", {"line": line}, "{line}")

    # First install what we can through install manifests.
    rc = command_context._run_make(
        directory=command_context.topobjdir,
        target="pre-export",
        line_handler=None,
        silent=not verbose,
    )
    if rc != 0:
        return rc

    # Then build the rest of the build dependencies by running the full
    # export target, because we can't do anything better.
    for target in ("export", "pre-compile"):
        rc = command_context._run_make(
            directory=command_context.topobjdir,
            target=target,
            line_handler=None,
            silent=not verbose,
            num_jobs=jobs,
        )
        if rc != 0:
            return rc

    return 0


def _set_clang_tools_paths(command_context):
    rc, config, _ = _get_config_environment(command_context)

    clang_paths = SimpleNamespace()

    if rc != 0:
        return rc, clang_paths

    clang_paths._clang_tools_path = mozpath.join(
        command_context._mach_context.state_dir, "clang-tools"
    )
    clang_paths._clang_tidy_path = mozpath.join(
        clang_paths._clang_tools_path,
        "clang-tidy",
        "bin",
        "clang-tidy" + config.substs.get("HOST_BIN_SUFFIX", ""),
    )
    clang_paths._clang_format_path = mozpath.join(
        clang_paths._clang_tools_path,
        "clang-tidy",
        "bin",
        "clang-format" + config.substs.get("HOST_BIN_SUFFIX", ""),
    )
    clang_paths._clang_apply_replacements = mozpath.join(
        clang_paths._clang_tools_path,
        "clang-tidy",
        "bin",
        "clang-apply-replacements" + config.substs.get("HOST_BIN_SUFFIX", ""),
    )
    clang_paths._run_clang_tidy_path = mozpath.join(
        clang_paths._clang_tools_path,
        "clang-tidy",
        "bin",
        "run-clang-tidy",
    )
    clang_paths._clang_format_diff = mozpath.join(
        clang_paths._clang_tools_path,
        "clang-tidy",
        "share",
        "clang",
        "clang-format-diff.py",
    )
    return 0, clang_paths


def _do_clang_tools_exist(clang_paths):
    return (
        os.path.exists(clang_paths._clang_tidy_path)
        and os.path.exists(clang_paths._clang_format_path)
        and os.path.exists(clang_paths._clang_apply_replacements)
        and os.path.exists(clang_paths._run_clang_tidy_path)
    )


def get_clang_tools(
    command_context,
    force=False,
    skip_cache=False,
    source=None,
    download_if_needed=True,
    verbose=False,
):

    rc, clang_paths = _set_clang_tools_paths(command_context)

    if rc != 0:
        return rc, clang_paths

    if (
        _do_clang_tools_exist(clang_paths)
        and _is_version_eligible(command_context, clang_paths, log_error=False)
        and not force
    ):
        return 0, clang_paths

    if os.path.isdir(clang_paths._clang_tools_path) and download_if_needed:
        # The directory exists, perhaps it's corrupted?  Delete it
        # and start from scratch.
        shutil.rmtree(clang_paths._clang_tools_path)
        return get_clang_tools(
            command_context,
            force=force,
            skip_cache=skip_cache,
            source=source,
            verbose=verbose,
            download_if_needed=download_if_needed,
        )

    # Create base directory where we store clang binary
    os.mkdir(clang_paths._clang_tools_path)

    if source:
        return _get_clang_tools_from_source(command_context, clang_paths, source)

    if not download_if_needed:
        return 0, clang_paths

    from mozbuild.bootstrap import bootstrap_toolchain

    bootstrap_toolchain("clang-tools/clang-tidy")

    return 0 if _is_version_eligible(command_context, clang_paths) else 1, clang_paths


def _get_clang_tools_from_source(command_context, clang_paths, filename):
    from mozbuild.action.tooltool import unpack_file

    clang_tidy_path = mozpath.join(
        command_context._mach_context.state_dir, "clang-tools"
    )

    currentWorkingDir = os.getcwd()
    os.chdir(clang_tidy_path)

    unpack_file(filename)

    # Change back the cwd
    os.chdir(currentWorkingDir)

    clang_path = mozpath.join(clang_tidy_path, "clang")

    if not os.path.isdir(clang_path):
        raise Exception("Extracted the archive but didn't find the expected output")

    assert os.path.exists(clang_paths._clang_tidy_path)
    assert os.path.exists(clang_paths._clang_format_path)
    assert os.path.exists(clang_paths._clang_apply_replacements)
    assert os.path.exists(clang_paths._run_clang_tidy_path)
    return 0, clang_paths


def _get_clang_format_diff_command(command_context, commit):
    if command_context.repository.name == "hg":
        args = ["hg", "diff", "-U0"]
        if commit:
            args += ["-c", commit]
        else:
            args += ["-r", ".^"]
        for dot_extension in _format_include_extensions:
            args += ["--include", "glob:**{0}".format(dot_extension)]
        args += ["--exclude", "listfile:{0}".format(_format_ignore_file)]
    else:
        commit_range = "HEAD"  # All uncommitted changes.
        if commit:
            commit_range = (
                commit if ".." in commit else "{}~..{}".format(commit, commit)
            )
        args = ["git", "diff", "--no-color", "-U0", commit_range, "--"]
        for dot_extension in _format_include_extensions:
            args += ["*{0}".format(dot_extension)]
        # git-diff doesn't support an 'exclude-from-files' param, but
        # allow to add individual exclude pattern since v1.9, see
        # https://git-scm.com/docs/gitglossary#gitglossary-aiddefpathspecapathspec
        with open(_format_ignore_file, "rb") as exclude_pattern_file:
            for pattern in exclude_pattern_file.readlines():
                pattern = six.ensure_str(pattern.rstrip())
                pattern = pattern.replace(".*", "**")
                if not pattern or pattern.startswith("#"):
                    continue  # empty or comment
                magics = ["exclude"]
                if pattern.startswith("^"):
                    magics += ["top"]
                    pattern = pattern[1:]
                args += [":({0}){1}".format(",".join(magics), pattern)]
    return args


def _run_clang_format_diff(
    command_context, clang_format_diff, clang_format, commit, output_file
):
    # Run clang-format on the diff
    # Note that this will potentially miss a lot things
    from subprocess import PIPE, CalledProcessError, Popen, check_output

    diff_process = Popen(
        _get_clang_format_diff_command(command_context, commit), stdout=PIPE
    )
    args = [sys.executable, clang_format_diff, "-p1", "-binary=%s" % clang_format]

    if not output_file:
        args.append("-i")
    try:
        output = check_output(args, stdin=diff_process.stdout)
        if output_file:
            # We want to print the diffs
            print(output, file=output_file)

        return 0
    except CalledProcessError as e:
        # Something wrong happend
        print("clang-format: An error occured while running clang-format-diff.")
        return e.returncode


def _is_ignored_path(command_context, ignored_dir_re, f):
    # path needs to be relative to the src root
    root_dir = command_context.topsrcdir + os.sep
    if f.startswith(root_dir):
        f = f[len(root_dir) :]
    # the ignored_dir_re regex uses / on all platforms
    return re.match(ignored_dir_re, f.replace(os.sep, "/"))


def _generate_path_list(command_context, paths, verbose=True):
    path_to_third_party = os.path.join(command_context.topsrcdir, _format_ignore_file)
    ignored_dir = []
    with open(path_to_third_party, "r") as fh:
        for line in fh:
            # Remove comments and empty lines
            if line.startswith("#") or len(line.strip()) == 0:
                continue
            # The regexp is to make sure we are managing relative paths
            ignored_dir.append(r"^[\./]*" + line.rstrip())

    # Generates the list of regexp
    ignored_dir_re = "(%s)" % "|".join(ignored_dir)
    extensions = _format_include_extensions

    path_list = []
    for f in paths:
        if _is_ignored_path(command_context, ignored_dir_re, f):
            # Early exit if we have provided an ignored directory
            if verbose:
                print("static-analysis: Ignored third party code '{0}'".format(f))
            continue

        if os.path.isdir(f):
            # Processing a directory, generate the file list
            for folder, subs, files in os.walk(f):
                subs.sort()
                for filename in sorted(files):
                    f_in_dir = posixpath.join(pathlib.Path(folder).as_posix(), filename)
                    if f_in_dir.endswith(extensions) and not _is_ignored_path(
                        command_context, ignored_dir_re, f_in_dir
                    ):
                        # Supported extension and accepted path
                        path_list.append(f_in_dir)
        else:
            # Make sure that the file exists and it has a supported extension
            if os.path.isfile(f) and f.endswith(extensions):
                path_list.append(f)

    return path_list


def _run_clang_format_in_console(command_context, clang_format, paths, assume_filename):
    path_list = _generate_path_list(command_context, assume_filename, False)

    if path_list == []:
        return 0

    # We use -assume-filename in order to better determine the path for
    # the .clang-format when it is ran outside of the repo, for example
    # by the extension hg-formatsource
    args = [clang_format, "-assume-filename={}".format(assume_filename[0])]

    process = subprocess.Popen(args, stdin=subprocess.PIPE)
    with open(paths[0], "r") as fin:
        process.stdin.write(fin.read())
        process.stdin.close()
        process.wait()
        return process.returncode


def _get_clang_format_cfg(command_context, current_dir):
    clang_format_cfg_path = mozpath.join(current_dir, ".clang-format")

    if os.path.exists(clang_format_cfg_path):
        # Return found path for .clang-format
        return clang_format_cfg_path

    if current_dir != command_context.topsrcdir:
        # Go to parent directory
        return _get_clang_format_cfg(command_context, os.path.split(current_dir)[0])
    # We have reached command_context.topsrcdir so return None
    return None


def _copy_clang_format_for_show_diff(
    command_context, current_dir, cached_clang_format_cfg, tmpdir
):
    # Lookup for .clang-format first in cache
    clang_format_cfg = cached_clang_format_cfg.get(current_dir, None)

    if clang_format_cfg is None:
        # Go through top directories
        clang_format_cfg = _get_clang_format_cfg(command_context, current_dir)

        # This is unlikely to happen since we must have .clang-format from
        # command_context.topsrcdir but in any case we should handle a potential error
        if clang_format_cfg is None:
            print("Cannot find corresponding .clang-format.")
            return 1

        # Cache clang_format_cfg for potential later usage
        cached_clang_format_cfg[current_dir] = clang_format_cfg

    # Copy .clang-format to the tmp dir where the formatted file is copied
    shutil.copy(clang_format_cfg, tmpdir)
    return 0


def _run_clang_format_path(
    command_context, clang_format, paths, output_file, output_format
):

    # Run clang-format on files or directories directly
    from subprocess import CalledProcessError, check_output

    if output_format == "json":
        # Get replacements in xml, then process to json
        args = [clang_format, "-output-replacements-xml"]
    else:
        args = [clang_format, "-i"]

    if output_file:
        # We just want to show the diff, we create the directory to copy it
        tmpdir = os.path.join(command_context.topobjdir, "tmp")
        if not os.path.exists(tmpdir):
            os.makedirs(tmpdir)

    path_list = _generate_path_list(command_context, paths)

    if path_list == []:
        return

    print("Processing %d file(s)..." % len(path_list))

    if output_file:
        patches = {}
        cached_clang_format_cfg = {}
        for i in range(0, len(path_list)):
            l = path_list[i : (i + 1)]

            # Copy the files into a temp directory
            # and run clang-format on the temp directory
            # and show the diff
            original_path = l[0]
            local_path = ntpath.basename(original_path)
            current_dir = ntpath.dirname(original_path)
            target_file = os.path.join(tmpdir, local_path)
            faketmpdir = os.path.dirname(target_file)
            if not os.path.isdir(faketmpdir):
                os.makedirs(faketmpdir)
            shutil.copy(l[0], faketmpdir)
            l[0] = target_file

            ret = _copy_clang_format_for_show_diff(
                command_context, current_dir, cached_clang_format_cfg, faketmpdir
            )
            if ret != 0:
                return ret

            # Run clang-format on the list
            try:
                output = check_output(args + l)
                if output and output_format == "json":
                    # Output a relative path in json patch list
                    relative_path = os.path.relpath(
                        original_path, command_context.topsrcdir
                    )
                    patches[relative_path] = _parse_xml_output(original_path, output)
            except CalledProcessError as e:
                # Something wrong happend
                print("clang-format: An error occured while running clang-format.")
                return e.returncode

            # show the diff
            if output_format == "diff":
                diff_command = ["diff", "-u", original_path, target_file]
                try:
                    output = check_output(diff_command)
                except CalledProcessError as e:
                    # diff -u returns 0 when no change
                    # here, we expect changes. if we are here, this means that
                    # there is a diff to show
                    if e.output:
                        # Replace the temp path by the path relative to the repository to
                        # display a valid patch
                        relative_path = os.path.relpath(
                            original_path, command_context.topsrcdir
                        )
                        # We must modify the paths in order to be compatible with the
                        # `diff` format.
                        original_path_diff = os.path.join("a", relative_path)
                        target_path_diff = os.path.join("b", relative_path)
                        e.output = e.output.decode("utf-8")
                        patch = e.output.replace(
                            "+++ {}".format(target_file),
                            "+++ {}".format(target_path_diff),
                        ).replace(
                            "-- {}".format(original_path),
                            "-- {}".format(original_path_diff),
                        )
                        patches[original_path] = patch

        if output_format == "json":
            output = json.dumps(patches, indent=4)
        else:
            # Display all the patches at once
            output = "\n".join(patches.values())

        # Output to specified file or stdout
        print(output, file=output_file)

        shutil.rmtree(tmpdir)
        return 0

    # Run clang-format in parallel trying to saturate all of the available cores.
    import math

    max_workers = multiprocessing.cpu_count()

    # To maximize CPU usage when there are few items to handle,
    # underestimate the number of items per batch, then dispatch
    # outstanding items across workers. Per definition, each worker will
    # handle at most one outstanding item.
    batch_size = int(math.floor(float(len(path_list)) / max_workers))
    outstanding_items = len(path_list) - batch_size * max_workers

    batches = []

    i = 0
    while i < len(path_list):
        num_items = batch_size + (1 if outstanding_items > 0 else 0)
        batches.append(args + path_list[i : (i + num_items)])

        outstanding_items -= 1
        i += num_items

    error_code = None

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = []
        for batch in batches:
            futures.append(executor.submit(run_one_clang_format_batch, batch))

        for future in concurrent.futures.as_completed(futures):
            # Wait for every task to finish
            ret_val = future.result()
            if ret_val is not None:
                error_code = ret_val

        if error_code is not None:
            return error_code
    return 0


def _parse_xml_output(path, clang_output):
    """
    Parse the clang-format XML output to convert it in a JSON compatible
    list of patches, and calculates line level informations from the
    character level provided changes.
    """
    content = six.ensure_str(open(path, "r").read())

    def _nb_of_lines(start, end):
        return len(content[start:end].splitlines())

    def _build(replacement):
        offset = int(replacement.attrib["offset"])
        length = int(replacement.attrib["length"])
        last_line = content.rfind("\n", 0, offset)
        return {
            "replacement": replacement.text,
            "char_offset": offset,
            "char_length": length,
            "line": _nb_of_lines(0, offset),
            "line_offset": last_line != -1 and (offset - last_line) or 0,
            "lines_modified": _nb_of_lines(offset, offset + length),
        }

    return [
        _build(replacement)
        for replacement in ET.fromstring(clang_output).findall("replacement")
    ]
