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))
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"))]

# 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


def bench(**kwargs):
    def repo_setup(repo_name):
        return hg.repository(ui.ui(), os.path.join(REPOS_DIR, repo_name))
    return _bench_with_repo(repo_setup, **kwargs)


PERF_RE = re.compile(r'! wall (\d+.\d+) comb (\d+.\d+) user (\d+.\d+) sys (\d+.\d+) \(best of (\d+)\)')


def perfbench(**kwargs):
    def repo_setup(repo_name):
        def perf(command, *args):
            venv = os.path.abspath(os.path.join(os.path.dirname(sys.executable), ".."))
            if os.path.isdir(os.path.join(venv, "project")):
                # Used with asv run (installed in virtualenv)
                perfpath = os.path.join(venv, "project", "contrib", "perf.py")
                hgpath = os.path.join(venv, "bin", "hg")
            else:
                # Used with asv dev (installed in ./mercurial)
                perfpath = os.path.join(BASEDIR, "mercurial", "contrib", "perf.py")
                hgpath = os.path.join(BASEDIR, "mercurial", "hg")
Boris Feld's avatar
Boris Feld committed
                hgpath, "--config", "extensions.perf=" + perfpath,
Boris Feld's avatar
Boris Feld committed
            os.environ["HGRCPATH"] = ""
            # test if the command exist in this perf.py version using 'hg help'
            cmd = basecmd + [
                "-R", os.path.join(REPOS_DIR, repo_name), command] + list(args)
            try:
                output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
            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:
                        output = subprocess.check_output(basecmd + ['help', command])
                    except subprocess.CalledProcessError as exc:
                        if exc.returncode == 255:
                            # command does not exist for this version of perf.py NaN
                            # result to n/a displayed in results which is a kind of
                            # "skipped" status
                            return float('nan')
                    raise
                raise
Boris Feld's avatar
Boris Feld committed
            match = PERF_RE.search(output)
            if not match:
                raise ValueError("Invalid output {0}".format(output))
            return float(match.group(1))
        return perf
    return _bench_with_repo(repo_setup, **kwargs)
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 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"]