lib/libscanbuild/__init__.py

# -*- coding: utf-8 -*-
# Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
# See https://llvm.org/LICENSE.txt for license information.
# SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
""" This module is a collection of methods commonly used in this project. """
import collections
import functools
import json
import logging
import os
import os.path
import re
import shlex
import subprocess
import sys

ENVIRONMENT_KEY = "INTERCEPT_BUILD"

Execution = collections.namedtuple("Execution", ["pid", "cwd", "cmd"])

CtuConfig = collections.namedtuple(
    "CtuConfig", ["collect", "analyze", "dir", "extdef_map_cmd"]
)


def duplicate_check(method):
    """Predicate to detect duplicated entries.

    Unique hash method can be use to detect duplicates. Entries are
    represented as dictionaries, which has no default hash method.
    This implementation uses a set datatype to store the unique hash values.

    This method returns a method which can detect the duplicate values."""

    def predicate(entry):
        entry_hash = predicate.unique(entry)
        if entry_hash not in predicate.state:
            predicate.state.add(entry_hash)
            return False
        return True

    predicate.unique = method
    predicate.state = set()
    return predicate


def run_build(command, *args, **kwargs):
    """Run and report build command execution

    :param command: array of tokens
    :return: exit code of the process
    """
    environment = kwargs.get("env", os.environ)
    logging.debug("run build %s, in environment: %s", command, environment)
    exit_code = subprocess.call(command, *args, **kwargs)
    logging.debug("build finished with exit code: %d", exit_code)
    return exit_code


def run_command(command, cwd=None):
    """Run a given command and report the execution.

    :param command: array of tokens
    :param cwd: the working directory where the command will be executed
    :return: output of the command
    """

    def decode_when_needed(result):
        """check_output returns bytes or string depend on python version"""
        return result.decode("utf-8") if isinstance(result, bytes) else result

    try:
        directory = os.path.abspath(cwd) if cwd else os.getcwd()
        logging.debug("exec command %s in %s", command, directory)
        output = subprocess.check_output(
            command, cwd=directory, stderr=subprocess.STDOUT
        )
        return decode_when_needed(output).splitlines()
    except subprocess.CalledProcessError as ex:
        ex.output = decode_when_needed(ex.output).splitlines()
        raise ex


def reconfigure_logging(verbose_level):
    """Reconfigure logging level and format based on the verbose flag.

    :param verbose_level: number of `-v` flags received by the command
    :return: no return value
    """
    # Exit when nothing to do.
    if verbose_level == 0:
        return

    root = logging.getLogger()
    # Tune logging level.
    level = logging.WARNING - min(logging.WARNING, (10 * verbose_level))
    root.setLevel(level)
    # Be verbose with messages.
    if verbose_level <= 3:
        fmt_string = "%(name)s: %(levelname)s: %(message)s"
    else:
        fmt_string = "%(name)s: %(levelname)s: %(funcName)s: %(message)s"
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(logging.Formatter(fmt=fmt_string))
    root.handlers = [handler]


def command_entry_point(function):
    """Decorator for command entry methods.

    The decorator initialize/shutdown logging and guard on programming
    errors (catch exceptions).

    The decorated method can have arbitrary parameters, the return value will
    be the exit code of the process."""

    @functools.wraps(function)
    def wrapper(*args, **kwargs):
        """Do housekeeping tasks and execute the wrapped method."""

        try:
            logging.basicConfig(
                format="%(name)s: %(message)s", level=logging.WARNING, stream=sys.stdout
            )
            # This hack to get the executable name as %(name).
            logging.getLogger().name = os.path.basename(sys.argv[0])
            return function(*args, **kwargs)
        except KeyboardInterrupt:
            logging.warning("Keyboard interrupt")
            return 130  # Signal received exit code for bash.
        except Exception:
            logging.exception("Internal error.")
            if logging.getLogger().isEnabledFor(logging.DEBUG):
                logging.error(
                    "Please report this bug and attach the output " "to the bug report"
                )
            else:
                logging.error(
                    "Please run this command again and turn on "
                    "verbose mode (add '-vvvv' as argument)."
                )
            return 64  # Some non used exit code for internal errors.
        finally:
            logging.shutdown()

    return wrapper


def compiler_wrapper(function):
    """Implements compiler wrapper base functionality.

    A compiler wrapper executes the real compiler, then implement some
    functionality, then returns with the real compiler exit code.

    :param function: the extra functionality what the wrapper want to
    do on top of the compiler call. If it throws exception, it will be
    caught and logged.
    :return: the exit code of the real compiler.

    The :param function: will receive the following arguments:

    :param result:       the exit code of the compilation.
    :param execution:    the command executed by the wrapper."""

    def is_cxx_compiler():
        """Find out was it a C++ compiler call. Compiler wrapper names
        contain the compiler type. C++ compiler wrappers ends with `c++`,
        but might have `.exe` extension on windows."""

        wrapper_command = os.path.basename(sys.argv[0])
        return re.match(r"(.+)c\+\+(.*)", wrapper_command)

    def run_compiler(executable):
        """Execute compilation with the real compiler."""

        command = executable + sys.argv[1:]
        logging.debug("compilation: %s", command)
        result = subprocess.call(command)
        logging.debug("compilation exit code: %d", result)
        return result

    # Get relevant parameters from environment.
    parameters = json.loads(os.environ[ENVIRONMENT_KEY])
    reconfigure_logging(parameters["verbose"])
    # Execute the requested compilation. Do crash if anything goes wrong.
    cxx = is_cxx_compiler()
    compiler = parameters["cxx"] if cxx else parameters["cc"]
    result = run_compiler(compiler)
    # Call the wrapped method and ignore it's return value.
    try:
        call = Execution(
            pid=os.getpid(),
            cwd=os.getcwd(),
            cmd=["c++" if cxx else "cc"] + sys.argv[1:],
        )
        function(result, call)
    except:
        logging.exception("Compiler wrapper failed complete.")
    finally:
        # Always return the real compiler exit code.
        return result


def wrapper_environment(args):
    """Set up environment for interpose compiler wrapper."""

    return {
        ENVIRONMENT_KEY: json.dumps(
            {
                "verbose": args.verbose,
                "cc": shlex.split(args.cc),
                "cxx": shlex.split(args.cxx),
            }
        )
    }