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)