-
Pierre-Yves David authoredPierre-Yves David authored
exchange.py 7.09 KiB
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)