Newer
Older
import contextlib
import tempfile
import shutil
import subprocess
from .utils import (BaseNChangesetsTestSuite, BaseTimeTestSuite,
BaseTrackTestSuite, median)
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")
137
138
139
140
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
191
class ExchangeTrackSuite(BaseTrackTestSuite):
param_names = BaseTrackTestSuite.param_names + ['single_rev']
params = BaseTrackTestSuite.params + [[True, False]]
@contextlib.contextmanager
def clone(self, revset=None):
tempdir = tempfile.mkdtemp(dir=os.environ.get('ASVTEMPDIR'))
try:
clone_path = os.path.join(tempdir, 'clone')
cmd = [self.hgpath, '--cwd', self.repo_path,
'clone', '--noupdate', '.', clone_path]
if revset is not None:
rev = subprocess.check_output([
self.hgpath, '--cwd', self.repo_path, 'log', '-T',
'{node}', '-r', revset, '--limit', '1'], env=self.environ)
cmd.extend(['-r', rev])
clone_time = self._single_execute(cmd)
yield clone_path, clone_time
finally:
shutil.rmtree(tempdir)
def track_incoming(self, repo_name, single_rev):
with self.clone(revset='default~10') as (clone_path, _):
cmd = [self.hgpath, '--cwd', clone_path, 'incoming']
if single_rev:
cmd.extend(['-r', 'default'])
return self._timeit(self._single_execute, (cmd,))
def track_outgoing(self, repo_name, single_rev):
with self.clone(revset='default~10') as (clone_path, _):
cmd = [self.hgpath, '--cwd', self.repo_path, 'outgoing',
clone_path]
if single_rev:
cmd.extend(['-r', 'default'])
return self._timeit(self._single_execute, (cmd,))
def track_push(self, repo_name, single_rev):
def clone_and_push():
with self.clone(revset='default~10') 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)
return self._timeit(clone_and_push)
def track_clone(self, repo_name, single_rev):
revset = 'default' if single_rev else None
def clone():
with self.clone(revset=revset) as (_, clone_time):
return clone_time
return self._timeit(clone)