clang/tools/scan-build-py/libscanbuild/__init__.py

# -*- coding: utf-8 -*-
#                     The LLVM Compiler Infrastructure
#
# This file is distributed under the University of Illinois Open Source
# License. See LICENSE.TXT for details.
""" 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'])


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)
        })
    }