from __future__ import print_function import os import os.path import pipes import threading import re import subprocess import stat import sys import urllib from . import ( BaseTestSuite, params_as_kwargs, REPOS_DIR, STRIP_VARIANTS_LIST, ROLES, BASEDIR, not_compatible_with, ) if sys.version_info[0] == 2: import Queue as queue else: import queue class classproperty(object): def __init__(self, f): self.f = f def __get__(self, obj, owner): return self.f(owner) def setup_role(cls): """install the right partial variants for the configured action""" cls.params = cls.params[:] cls.params[cls._partials_idx] = list(sorted(cls.role_data)) return cls class HgWeb(object): def __init__(self): super(HgWeb, self).__init__() self.queue = queue.Queue() self.proc = None self.thread = None def start(self, hgpath, environ): config = os.path.abspath(os.path.join(BASEDIR, 'hgweb.config')) hgweb_cmd = [ hgpath, 'serve', '--cwd', REPOS_DIR, '-a', 'localhost', '-p', '0', '--config', 'web.push_ssl=False', '--config', 'web.allow_push=*', '--webdir-conf', config] self.proc = subprocess.Popen(hgweb_cmd, env=environ, stdout=subprocess.PIPE) # we have to read output in a thread to avoid deadlocks self.thread = threading.Thread( target=self._enqueue, args=(self.queue, self.proc.stdout)) self.thread.daemon = True self.thread.start() # wait the server to be started statusline = self.queue.get() if not statusline: self.stop() raise RuntimeError('hg serve has crashed') return re.search(':(\d+)', statusline).groups()[0] @staticmethod def _enqueue(queue, fd): while True: data = fd.readline() if not data: break queue.put(data) queue.put(None) def stop(self): self.proc.kill() self.proc.wait() self.thread.join() class BaseExchangeMixin(object): # this help recovering from failure during setup _hgserve = None 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(b'#!/bin/sh\nexec env -i {} {} $*\n'.format( ' '.join(['{}={}'.format(k, pipes.quote(v)) for k, v in self.environ.items()]), os.path.abspath(self.hgpath))) st = os.stat('hg_wrapper') os.chmod('hg_wrapper', st.st_mode | stat.S_IEXEC) return [ '--remotecmd', os.path.abspath('hg_wrapper'), 'ssh://localhost/{}'.format(os.path.abspath(path))] elif self.repo_type == 'http': path = os.path.abspath(path) repo_dir = os.path.abspath(REPOS_DIR) assert path.startswith(REPOS_DIR), path return ['http://localhost:{}/{}'.format( self.hgport, path[len(repo_dir) + 1:])] else: raise NotImplementedError def _setup_repo_type(self, repo_type): """setup a hgweb server if we have to""" self._hgserve = None self.repo_type = repo_type if repo_type == 'http': ### skip hgweb for some broken instance: # we used to skip for f0a851542a05::877185de62^ # pointing at https://bz.mercurial-scm.org/show_bug.cgi?id=5851 # but there is futher issue so 6b0275e5f276^ it is. # # should be a BaseTestSuite object currentrev = self.get_asv_rev() badhttp = b"::(6b0275e5f27696226595c114c9bf034519aaad45^)" if self._matchrevset(badhttp, currentrev): msg = "http server may hang for %s" % currentrev raise NotImplementedError(msg) else: self._hgserve = HgWeb() self.hgport = self._hgserve.start(self.hgpath, self.environ) def _teardown_repo_type(self): if self._hgserve is not None: self._hgserve.stop() self._hgserve = None def _setup_revset(self, revset): """If the operation target a specific revision, resolve it beforehand""" if revset is not None: self.rev = self.util_hg('identify', '-i', '-r', revset).strip() else: self.rev = None def finalize_teardown(self, exc_info, current_params): self._teardown_repo_type() class BaseExchangeTimeSuite(BaseExchangeMixin, BaseTestSuite): # the exchange action we measure role_action = None # the subtypes of this action we measure role_subtype = None param_names = BaseTestSuite.param_names + [ 'repo_type', 'strip', 'revset'] params = BaseTestSuite.params + [['local', 'ssh', 'http']] _partials_idx = len(params) params += [STRIP_VARIANTS_LIST] params += [[None, 'tip']] timeout = 1800 @classproperty def role_data(cls): """{"partial-key" -> {"repo-key" -> {data}} map for the current role""" if cls.role_action is None: return None if cls.role_subtype is None: return None return ROLES.get(cls.role_action, {}).get(cls.role_subtype, {}) def run(self, local_repo, command, remote_repo, expected_return_code=None): if not isinstance(command, (list, tuple)): command = [command] cmd = ['--cwd', pipes.quote(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.partial_key == "same" else 0 self.hg(*cmd, expected_return_code=expected_return_code) @params_as_kwargs def setup(self, repo, repo_type, strip, revset, **kwargs): self.partial_key = strip super(BaseExchangeTimeSuite, self).setup(repo, **kwargs) self._setup_repo_type(repo_type) self._setup_revset(revset) @property def clone_path(self): return self.repo_path_from_id(self.partial_key) def repo_path_from_id(self, partial_id): """Return the absolute path for a given partial "reference" is a special value that means the original repository. """ if partial_id == 'reference': return self.repo_path suffix = urllib.quote_plus(partial_id) # We need to use the repo name here because the repo doesn't contains # the hash partial_name = '{}-partial-{}'.format(self.repo_name, suffix) return os.path.join(REPOS_DIR, 'partial-references', partial_name) def teardown(self, *args, **kwargs): self._teardown_repo_type() def _rsync(self, src, dst): cmd = [ 'rsync', '--inplace', '--no-whole-file', '-aH', '--delete', '{}/'.format(src), dst, ] self.check_output(*cmd)