Newer
Older
import contextlib
import tempfile
import shutil
import subprocess
import shlex
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):
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):
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):
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):
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):
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
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):
def clone_and_push():
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)