Skip to content
Snippets Groups Projects
basic_commands.py 7.51 KiB
import base64
import contextlib
import errno
import os
import os.path
import time
import timeit
import tempfile
import shutil
import subprocess
import shlex

from .utils import (BaseNChangesetsTestSuite, BaseTimeTestSuite,
                    BaseTrackTestSuite, median, REPOS_DIR)


class TestSuite(BaseTrackTestSuite):

    timeout = 120

    def track_commit(self, *args, **kwargs):
        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")
        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):

    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

    def bench_command_status(self, *args, **kwargs):
        return "status --change tip"

    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):
        self._execute("log", "-r", "tip")

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

    timeout = 300

    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 BaseExchangeTimeSuite(BaseTimeTestSuite):
    param_names = BaseTimeTestSuite.param_names + ['strip', 'revset']
    params = BaseTimeTestSuite.params + [['same', 'last(all(), 10)', 'last(all(), 100)', 'last(all(), 1000)']] + [[None, 'default']]

    def setup(self, repo_name, strip, revset):
        self.timer = timeit.default_timer
        self.sample_time = 10
        super(BaseExchangeTimeSuite, self).setup(repo_name)
        self.clone_path = os.path.join(REPOS_DIR, '.cache', 'partial-{}-{}'.format(
            repo_name, base64.b64encode(strip)))
        if revset is not None:
            self.rev = subprocess.check_output([
                self.hgpath, '--cwd', self.repo_path, 'identify', '-i', "-r", revset
            ], env=self.environ).strip()
        else:
            self.rev = None


class ExchangeTimeSuite(BaseExchangeTimeSuite):

    def time_incoming(self, repo_name, strip, revset):
        cmd = [self.hgpath, '--cwd', self.clone_path, 'incoming', self.repo_path]
        if self.rev:
            cmd.extend(['-r', self.rev])
        self._single_execute(cmd, expected_return_code=1 if strip == "same" else 0)

    def time_outgoing(self, repo_name, strip, revset):
        cmd = [self.hgpath, '--cwd', self.repo_path, 'outgoing', self.clone_path]
        if self.rev:
            cmd.extend(['-r', self.rev])
        self._single_execute(cmd, expected_return_code=1 if strip == "same" else 0)

    def time_debugdiscovery(self, repo_name, strip, revset):
        cmd = [self.hgpath, '--cwd', self.clone_path, 'debugdiscovery', self.repo_path]
        # if self.rev:
        #     cmd.extend(['--rev', self.rev])
        self._single_execute(cmd, expected_return_code=1 if strip == "same" else 0)


class PushPullTimeSuite(BaseExchangeTimeSuite):

    timeout = 300

    def setup(self, repo_name, strip, revset):
        super(PushPullTimeSuite, self).setup(repo_name, strip, revset)
        tmpdir = os.path.join(REPOS_DIR, '.tmp')
        try:
            os.makedirs(tmpdir)
        except OSError as exc:
            if exc.errno != errno.EEXIST:
                raise
        self.tmp_clone_path = os.path.join(tmpdir, 'clone-{}'.format(
            os.path.basename(self.clone_path)))
        # XXX: This should be deleted at the end but teardown, like setup, is
        # called for each repeat...
        self._single_execute(['rsync', '-aH', '--delete', '{}/'.format(self.clone_path),
                              self.tmp_clone_path])

    def time_push(self, repo_name, strip, revset):
        cmd = [self.hgpath, '--cwd', self.repo_path, 'push', '-f',
               self.tmp_clone_path]
        if self.rev:
            cmd.extend(['-r', self.rev])
        self._single_execute(cmd, expected_return_code=1 if strip == "same" else 0)

    def time_pull(self, repo_name, strip, revset):
        cmd = [self.hgpath, '--cwd', self.tmp_clone_path, 'pull',
               self.repo_path]
        if self.rev:
            cmd.extend(['-r', self.rev])
        self._single_execute(cmd, expected_return_code=1 if strip == "same" else 0)


class CloneTimeSuite(BaseExchangeTimeSuite):
    # skip other repos since they are too big
    params = [['mercurial-2017'], ['same'], [None, 'default']]

    timeout = 300

    def setup(self, repo_name, strip, revset):
        super(CloneTimeSuite, self).setup(repo_name, strip, revset)
        tmpdir = os.path.join(REPOS_DIR, '.tmp')
        try:
            os.makedirs(tmpdir)
        except OSError as exc:
            if exc.errno != errno.EEXIST:
                raise
        self.tmp_clone_path = os.path.join(tmpdir, 'clone-{}'.format(
            os.path.basename(self.clone_path)))
        shutil.rmtree(self.tmp_clone_path, ignore_errors=True)

    def teardown(self, repo_name, strip, revset):
        shutil.rmtree(self.tmp_clone_path, ignore_errors=True)

    def time_clone(self, repo_name, strip, revset):
        cmd = [self.hgpath, 'clone', '--pull', self.clone_path,
               self.tmp_clone_path]
        if self.rev:
            cmd.extend(['-r', self.rev])
        self._single_execute(cmd)