Newer
Older
import errno
import shutil
import subprocess
from .utils import (BaseNChangesetsTestSuite, BaseTimeTestSuite,
BaseTrackTestSuite, median, REPOS_DIR)
class TestSuite(BaseTrackTestSuite):
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):
timer = timeit.default_timer
sample_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):
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):
timer = timeit.default_timer
sample_time = 10
def time_log_history(self, repo, n):
self._execute("log", "-r", "tip~%d::" % n)
class UpdateTimeTestSuite(BaseNChangesetsTestSuite):
timer = timeit.default_timer
sample_time = 10
def time_up_tip(self, repo, n):
self._execute("up", "-r", "tip~%d" % n)
self._execute("up", "-r", "tip")
class BundleTimeTestSuite(BaseNChangesetsTestSuite):
timer = timeit.default_timer
sample_time = 10
timeout = 500
def time_bundle(self, repo, n):
self._execute("bundle", "--base", ":(-%d)" % (n+1), "/tmp/bundle.bundle")
class BaseExchangeTimeSuite(BaseTimeTestSuite):
param_names = BaseTimeTestSuite.param_names + [
'repo_type', 'strip', 'revset']
params = BaseTimeTestSuite.params + [
['same', 'last(all(), 10)', 'last(all(), 100)', 'last(all(), 1000)'],
[None, 'default']]
timer = timeit.default_timer
sample_time = 10
def _remote_path_cmd(self, path):
if self.repo_type == 'local':
return [path]
elif self.repo_type == 'ssh':
with open('hg_wrapper', 'wb') as f:
f.write('#!/bin/sh\nexec env -i HGRCPATH= {} $*\n'.format(
os.path.abspath(self.hgpath)))
st = os.stat('hg_wrapper')
os.chmod('hg_wrapper', st.st_mode | stat.S_IEXEC)
'--remotecmd', os.path.abspath('hg_wrapper'),
'ssh://localhost/{}'.format(os.path.abspath(path))]
else:
raise NotImplementedError
def run(self, local_repo, command, remote_repo, expected_return_code=None):
if not isinstance(command, (list, tuple)):
command = [command]
cmd = [self.hgpath, '--cwd', local_repo] + command
cmd.extend(self._remote_path_cmd(remote_repo))
if self.rev:
cmd.extend(['-r', self.rev])
if expected_return_code is None:
expected_return_code = 1 if self.strip == "same" else 0
self._single_execute(cmd, expected_return_code=expected_return_code)
def setup(self, repo_name, repo_type, strip, revset):
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
self.strip = strip
class ExchangeTimeSuite(BaseExchangeTimeSuite):
def time_incoming(self, *args, **kwargs):
self.run(self.clone_path, 'incoming', self.repo_path)
def time_outgoing(self, *args, **kwargs):
self.run(self.repo_path, 'outgoing', self.clone_path)
class DiscoveryTimeSuite(BaseExchangeTimeSuite):
# debugdiscovery does not support revset argument
params = BaseTimeTestSuite.params + [
['local', 'ssh'],
['same', 'last(all(), 10)', 'last(all(), 100)', 'last(all(), 1000)'],
[None]]
def time_debugdiscovery(self, *args, **kwargs):
self.run(self.clone_path, 'debugdiscovery', self.repo_path,
expected_return_code=0)
class PushPullTimeSuite(BaseExchangeTimeSuite):
timeout = 300
# Force setup to be called between two push or pull
number = 1
repeat = 20
warmup_time = 0
def setup(self, *args, **kwargs):
super(PushPullTimeSuite, self).setup(*args, **kwargs)
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])
# Wait for everything to be written on disk to avoid Disk IO wait to
# impact performances metrics
self._single_execute(['sync'])
def time_push(self, *args, **kwargs):
self.run(self.repo_path, ['push', '-f'], self.tmp_clone_path)
def time_pull(self, *args, **kwargs):
self.run(self.tmp_clone_path, 'pull', self.repo_path,
expected_return_code=0)
class CloneTimeSuite(BaseExchangeTimeSuite):
# skip other repos since they are too big
params = [['mercurial-2017'], ['local', 'ssh'], ['same'], [None, 'default']]
def setup(self, *args, **kwargs):
super(CloneTimeSuite, self).setup(*args, **kwargs)
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, *args, **kwargs):
shutil.rmtree(self.tmp_clone_path, ignore_errors=True)
def time_clone(self, *args, **kwargs):
cmd = [self.hgpath, 'clone', '--pull']
cmd.extend(self._remote_path_cmd(self.clone_path))
cmd.append(self.tmp_clone_path)
if self.rev:
cmd.extend(['-r', self.rev])
self._single_execute(cmd)