Skip to content
Snippets Groups Projects
basic_commands.py 7.82 KiB
Newer Older
Boris Feld's avatar
Boris Feld committed
import os
import os.path
import time
import timeit
import tempfile
import shutil
import subprocess
Boris Feld's avatar
Boris Feld committed

from .utils import (BaseNChangesetsTestSuite, BaseTimeTestSuite,
                    BaseTrackTestSuite, median, REPOS_DIR)
Boris Feld's avatar
Boris Feld committed


class TestSuite(BaseTrackTestSuite):
Boris Feld's avatar
Boris Feld committed

Boris Feld's avatar
Boris Feld committed

    def track_commit(self, *args, **kwargs):
Boris Feld's avatar
Boris Feld committed
        timings = []

        args = ["-A", "-m", "My commit message", "-u", "<test@octobus.net>"]

        cmd = self._prepare_cmd("commit", *args)
        rollback_cmd = self._prepare_cmd("rollback", "--config",
                                         "ui.rollback=true")
Boris Feld's avatar
Boris Feld committed
        clean_cmd = self._prepare_cmd("update", "-C", ".")

        with open(os.path.join(self.repo_path, 'BABAR'), 'w') as f:
            f.write("BABAR")

        # Do the commits N time
        for i in range(3):
            try:
                before = time.time()
                self._single_execute(cmd)
                after = time.time()
                timings.append(after - before)
            finally:
                # Rollback and clean
                self._single_execute(rollback_cmd)
                self._single_execute(clean_cmd)

        return median(timings)


class TimeTestSuite(BaseTimeTestSuite):
Boris Feld's avatar
Boris Feld committed

    def __init__(self):
        super(TimeTestSuite, self).__init__()

        self.timer = timeit.default_timer
        self.goal_time = 10

    def time_emptystatus(self, *args, **kwargs):
        return self._execute("status")

    def time_status_tip(self, *args, **kwargs):
        time = self._execute("status", "--change", "tip")
        return time

Boris Feld's avatar
Boris Feld committed
    def bench_command_status(self, *args, **kwargs):
        return "status --change tip"

Boris Feld's avatar
Boris Feld committed
    def time_emptydiff(self, *args, **kwargs):
        time = self._execute("diff")
        return time

    def time_diff_tip(self, *args, **kwargs):
        time = self._execute("diff", "-c", "tip")
        return time

    def time_log_tip(self, *args, **kwargs):
Boris Feld's avatar
Boris Feld committed
        self._execute("log", "-r", "tip")
Boris Feld's avatar
Boris Feld committed

    def time_summary(self, *args, **kwargs):
        time = self._execute("summary")
        return time

    def time_version(self, *args, **kwargs):
        time = self._execute("--version")
        return time

    def time_bookmarks(self, *args, **kwargs):
        time = self._execute("bookmarks")
        return time

    def time_id(self, *args, **kwargs):
        time = self._execute("id")
        return time

    def time_id_current(self, *args, **kwargs):
        time = self._execute("id", "-r", ". ")
        return time
class LogTimeTestSuite(BaseNChangesetsTestSuite):
Boris Feld's avatar
Boris Feld committed

    def __init__(self):
        super(LogTimeTestSuite, self).__init__()

        self.timer = timeit.default_timer
        self.goal_time = 10

    def time_log_history(self, repo, n):
        self._execute("log", "-r", "tip~%d::" % n)
class UpdateTimeTestSuite(BaseNChangesetsTestSuite):
    timeout = 500

    def __init__(self):
        super(UpdateTimeTestSuite, self).__init__()

        self.timer = timeit.default_timer
        self.goal_time = 10

    def time_up_tip(self, repo, n):
        self._execute("up", "-r", "tip~%d" % n)
        self._execute("up", "-r", "tip")
class BundleTimeTestSuite(BaseNChangesetsTestSuite):

    timeout = 500

    def __init__(self):
        super(BundleTimeTestSuite, self).__init__()

        self.timer = timeit.default_timer
        self.goal_time = 10

    def time_bundle(self, repo, n):
        self._execute("bundle", "--base", ":(-%d)" % (n+1), "/tmp/bundle.bundle")


class ExchangeTrackSuite(BaseTrackTestSuite):
    param_names = BaseTrackTestSuite.param_names + ['single_rev', 'revset']
    params = BaseTrackTestSuite.params + [[True, False]] + [['default~10']]

    def setup(self, repo_name, single_rev, revset, *args, **kwargs):
        super(ExchangeTrackSuite, self).setup(repo_name, *args, **kwargs)

        self.single_rev = single_rev

        from base64 import b64encode

        self.safe_range = b64encode(revset)
        self.tempdir = tempfile.mkdtemp(dir=os.environ.get('ASVTEMPDIR'))

        if single_rev is False:
            self.cache_name = "%s.tar" % repo_name
            self.cache_path = os.path.join(self.tempdir, self.cache_name)

            if not os.path.isfile(self.cache_path):
                tar_cmd = "tar cf %s -C  %s ." % (self.cache_path, self.repo_path)
                self._single_execute(shlex.split(tar_cmd))
        else:
            self.cache_name = "%s-%s-%s.tar" % (repo_name, single_rev, self.safe_range)
            self.cache_path = os.path.join(self.tempdir, self.cache_name)

            if not os.path.isfile(self.cache_path):
                # First clone it
                clone_path = os.path.join(self.tempdir, "%s-%s-clone" % (repo_name, self.safe_range))

                rev = self._identify_id(revset)

                cmd = [self.hgpath, '--cwd', self.repo_path,
                       'clone', '--noupdate', '.', clone_path,
                       '-r', rev]
                self._single_execute(cmd)

                tar_cmd = "tar cf %s -C  %s ." % (self.cache_path, clone_path)
                self._single_execute(shlex.split(tar_cmd))

                shutil.rmtree(clone_path)

    def teardown(self, *args, **kwargs):
        try:
            shutil.rmtree(self.tempdir)
        except Exception:
            pass

    def _identify_id(self, revset):
        rev = subprocess.check_output([
            self.hgpath, '--cwd', self.repo_path, 'identify', '-i', "-r", revset], env=self.environ)
        return rev.strip()

    @contextlib.contextmanager
    def clone(self, revset=None):
        try:
            clone_path = os.path.join(self.tempdir, 'clone')
            cmd = [self.hgpath, '--cwd', self.repo_path,
                   'clone', '--noupdate', '.', clone_path]
            if revset is not None:
                rev = self._identify_id(revset)
                cmd.extend(['-r', rev])
            clone_time = self._single_execute(cmd)
            yield clone_path, clone_time
        finally:
            shutil.rmtree(clone_path)

    @contextlib.contextmanager
    def clone_from_cache(self):
        try:
            import time
            clone_path = os.path.join(self.tempdir, 'clone-%s' % time.time())

            os.mkdir(clone_path)
            cmd = "tar xf %s -C %s" % (self.cache_path, clone_path)
            clone_time = self._single_execute(shlex.split(cmd))

            yield clone_path

        finally:
            shutil.rmtree(clone_path)

    def track_incoming(self, repo_name, single_rev, revset):
        with self.clone_from_cache() as clone_path:
            cmd = [self.hgpath, '--cwd', clone_path, 'incoming', self.repo_path]
            if single_rev:
                cmd.extend(['-r', 'default'])

            # Ignore the return code
            return self._timeit(self._single_execute, (cmd, False))

    def track_outgoing(self, repo_name, single_rev, revset):
        with self.clone_from_cache() as clone_path:
            cmd = [self.hgpath, '--cwd', self.repo_path, 'outgoing',
                   clone_path]
            if single_rev:
                cmd.extend(['-r', 'default'])

            # Ignore the return code
            return self._timeit(self._single_execute, (cmd, False))
    def track_push(self, repo_name, single_rev, revset):
            with self.clone_from_cache() as clone_path:
                cmd = [self.hgpath, '--cwd', self.repo_path, 'push', '-f',
                       clone_path]
                if single_rev:
                    cmd.extend(['-r', 'default'])
                return self._single_execute(cmd, False)
        return self._timeit(clone_and_push)

    def track_clone(self, repo_name, single_rev, revset):
        revset = 'default' if single_rev else None

        def clone():
            with self.clone(revset=revset) as (_, clone_time):
                return clone_time
        return self._timeit(clone)