Skip to content
Snippets Groups Projects
utils.py 8.38 KiB
Newer Older
from __future__ import print_function
Boris Feld's avatar
Boris Feld committed
import os
Boris Feld's avatar
Boris Feld committed
import re
Boris Feld's avatar
Boris Feld committed
import sys
from functools import wraps

BASEDIR = os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir))
STRIP_VARIANTS_PATH = os.path.join(BASEDIR, "partial-sets.yaml")
Boris Feld's avatar
Boris Feld committed
REPOS_DIR = os.path.join(BASEDIR, "repos")
REPOS = [d for d in os.listdir(REPOS_DIR)
         if os.path.isdir(os.path.join(REPOS_DIR, d, ".hg"))]
# TODO fix
# with open(STRIP_VARIANTS_PATH) as f:
#     STRIP_VARIANTS = yaml.load(f.read())["partial-sets"]
Boris Feld's avatar
Boris Feld committed


class SkipResult(Exception):
    pass


Boris Feld's avatar
Boris Feld committed
# Backward compatibility for python 2.6
if not hasattr(subprocess, 'check_output'):
    STDOUT = subprocess.STDOUT

    def check_output(*popenargs, **kwargs):
        if 'stdout' in kwargs:  # pragma: no cover
            raise ValueError('stdout argument not allowed, '
                             'it will be overridden.')
        process = subprocess.Popen(stdout=subprocess.PIPE,
                                   *popenargs, **kwargs)
        output, _ = process.communicate()
        retcode = process.poll()
        if retcode:
            cmd = kwargs.get("args")
            if cmd is None:
                cmd = popenargs[0]
            raise subprocess.CalledProcessError(retcode, cmd,
                                                output=output)
        return output
    subprocess.check_output = check_output

    # overwrite CalledProcessError due to `output`
    # keyword not being available (in 2.6)
    class CalledProcessError(Exception):

        def __init__(self, returncode, cmd, output=None):
            self.returncode = returncode
            self.cmd = cmd
            self.output = output

        def __str__(self):
            return "Command '%s' returned non-zero exit status %d\n%s" % (
                self.cmd, self.returncode, self.output)
    subprocess.CalledProcessError = CalledProcessError


def _bench_with_repo(
    repo_setup,
    repos=REPOS, setup=None, params=None, param_names=None,
    pretty_name=None,
):
    if params is None:
        assert param_names is None
        params = [repos]
        param_names = ["repo"]
    else:
        params = params[:]
        params.insert(0, repos)
        if param_names is None:
            param_names = ["repo"]
        else:
            param_names = param_names[:]
            param_names.insert(0, "repo")

    def decorator(func):

        _setup = {'args': None}

        @wraps(func)
        def wrapper(*args):
            return func(*(_setup['args'] + args[1:]))

        def bench_setup(repo_name, *args):
            repo = repo_setup(repo_name)
            if setup is not None:
                value = setup(repo, *args)
                _setup['args'] = (repo, value)
            else:
                _setup['args'] = (repo,)

        wrapper.setup = bench_setup
        wrapper.params = params
        wrapper.param_names = param_names
        wrapper.pretty_name = pretty_name
        return wrapper

    return decorator


Philippe Pepiot's avatar
Philippe Pepiot committed
PERF_RE = re.compile(r'! wall (\d+.\d+) comb (\d+.\d+) user (\d+.\d+) sys (\d+.\d+) \(best of (\d+)\)')  # noqa: E501
Boris Feld's avatar
Boris Feld committed


def median(lst):
    quotient, remainder = divmod(len(lst), 2)
    if remainder:
        return sorted(lst)[quotient]
    return sum(sorted(lst)[quotient - 1:quotient + 1]) / 2.


class BaseTestSuite(object):

    params = [REPOS]
    param_names = ["repo"]

    @staticmethod
    def get_skip():
        with open(os.path.join(REPOS_DIR, 'skip.json'), 'r') as f:
            return json.load(f)

    def get_asv_rev(self):
        '''Return currently benchmarked mercurial revision'''
        return self.hg('log', '--cwd', self.project_dir, '-T',
                       '{node|short}', '-r', '.').strip()
    def check_output(self, *args, **kwargs):
        """Helper to run commands

        Run given command in a subprocess
        Optional expected_return_code (default 0) is used to control whenever
        we expect the command should exit.

        If the command succeeded with expected_return_code = 0, return the output
        If the command succeeded with expected_return_code != 0, raise RuntimeError
        If the command fail with expected_return_code, return None, else raise
        original subprocess.CalledProcessError exception.
        """
        env = kwargs.pop('env', self.environ)
        expected_return_code = kwargs.pop('expected_return_code', 0)
        cmd = list(args)
        try:
            output = subprocess.check_output(cmd, env=env, **kwargs)
        except subprocess.CalledProcessError as exc:
            if exc.returncode == expected_return_code:
                # failed as we expected
                return None
            raise
        else:
            if expected_return_code != 0:
                raise RuntimeError('unexpected return code 0 for {}'.format(cmd))
            return output

    def hg(self, *args, **kwargs):
        """Run given command arguments with hg

        When there is no '--cwd' in arguments, use the benchmarked repo with
        'hg --cwd /path/to/repo'
        """
        if '--cwd' not in args:
            # use self.repo_path as repo
            cmd = [self.hgpath, '--cwd', self.repo_path] + list(args)
        else:
            cmd = [self.hgpath] + list(args)
        return self.check_output(*cmd, **kwargs)

    def safe_hg(self, command, *args, **kwargs):
        """Run given command argument with hg and ignore unknown commands

        This is to be used for commands that may not exist in earlier mercurial
        versions.

        When hg exit code is 255, test the command existence with 'hg help CMD'
        to test command existence, if it also return 255, raise SkipResult
        exception.

        We expect `command` to be the hg command we want to run, either a list
        of arguments required to run 'hg help' (for extensions etc).
        """
        if isinstance(command, (list, tuple)):
            cmd = command
        else:
            cmd = [command]
        try:
            return self.hg(*(cmd + list(args)), **kwargs)
        except subprocess.CalledProcessError as exc:
            if exc.returncode == 255:
                # test if it return 255 because the command does not exist
                # or if it's another issue
                try:
                    self.hg(*(['help'] + command))
                except subprocess.CalledProcessError as exc:
                    if exc.returncode == 255:
                        # command does not exist in this version of mercurial
                        raise SkipResult()
                    raise
            raise

    def perfext(self, command, *args, **kwargs):
        """Use contrib/perf.py extension from mercurial to get a benchmark result"""
        perfpath = os.path.join(self.project_dir, 'contrib', 'perf.py')
        kwargs.setdefault('stderr', subprocess.STDOUT)
        try:
            output = self.safe_hg(
                [command, '--config', 'extensions.perf={0}'.format(perfpath)],
                *args, **kwargs)
        except SkipResult:
            # command does not exist for this version of perf.py
            # return NaN which is a "n/a" status for asv
            return float('nan')
        match = PERF_RE.search(output)
        if not match:
            raise ValueError("Invalid output {0}".format(output))
        return float(match.group(1))

    def setup(self, repo_name, *args, **kwargs):
        venv = os.path.abspath(os.path.join(os.path.dirname(sys.executable), ".."))
        self.repo_name = repo_name
        self.project_dir = os.path.join(venv, 'project')
        if os.path.isdir(self.project_dir):
            # use hg in virtualenv for "asv run"
            self.hgpath = os.path.join(venv, "bin", "hg")
        else:
            # use local hg for "asv dev"
            self.project_dir = os.path.join(BASEDIR, 'mercurial')
            sys.path.insert(0, self.project_dir)
            self.hgpath = os.path.join(os.path.join(self.project_dir, 'hg'))
        self.repo_path = os.path.join(REPOS_DIR, self.repo_name)

        # Use a clean environ to run command
        self.environ = {'HGRCPATH': ''}
        # keep some environment variables
        # SSH_AUTH_SOCK for hg over ssh
        for key in ('SSH_AUTH_SOCK',):
            if key in os.environ:
                self.environ[key] = os.environ[key]
class BaseNChangesetsTestSuite(BaseTestSuite):

    params = [REPOS, [10, 100, 1000, 10000]]
    param_names = ["repo", "changesets"]