-
Boris Feld authoredBoris Feld authored
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)