import contextlib 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 ExchangeTrackSuite(BaseTrackTestSuite): 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)