file-nodes: added streaming remote attributes for vcsserver.
- this will enable doing a streaming raw content or raw downloads of large files without transfering them over to enterprise for pack & repack using msgpack - msgpack has a limit of 4gb and generally pack+repack for 2gb is very slow
This commit is contained in:
parent
0ae89b9907
commit
de8ff5bcfa
11 changed files with 252 additions and 433 deletions
|
|
@ -844,10 +844,9 @@ class RepoFilesView(RepoAppView):
|
|||
if disposition == 'attachment':
|
||||
disposition = self._get_attachement_headers(f_path)
|
||||
|
||||
def stream_node():
|
||||
yield file_node.raw_bytes
|
||||
stream_content = file_node.stream_bytes()
|
||||
|
||||
response = Response(app_iter=stream_node())
|
||||
response = Response(app_iter=stream_content)
|
||||
response.content_disposition = disposition
|
||||
response.content_type = mimetype
|
||||
|
||||
|
|
@ -883,10 +882,9 @@ class RepoFilesView(RepoAppView):
|
|||
|
||||
disposition = self._get_attachement_headers(f_path)
|
||||
|
||||
def stream_node():
|
||||
yield file_node.raw_bytes
|
||||
stream_content = file_node.stream_bytes()
|
||||
|
||||
response = Response(app_iter=stream_node())
|
||||
response = Response(app_iter=stream_content)
|
||||
response.content_disposition = disposition
|
||||
response.content_type = file_node.mimetype
|
||||
|
||||
|
|
|
|||
|
|
@ -72,11 +72,11 @@ def connect_http(server_and_port):
|
|||
|
||||
session_factory = client_http.ThreadlocalSessionFactory()
|
||||
|
||||
connection.Git = client_http.RepoMaker(
|
||||
connection.Git = client_http.RemoteVCSMaker(
|
||||
server_and_port, '/git', 'git', session_factory)
|
||||
connection.Hg = client_http.RepoMaker(
|
||||
connection.Hg = client_http.RemoteVCSMaker(
|
||||
server_and_port, '/hg', 'hg', session_factory)
|
||||
connection.Svn = client_http.RepoMaker(
|
||||
connection.Svn = client_http.RemoteVCSMaker(
|
||||
server_and_port, '/svn', 'svn', session_factory)
|
||||
connection.Service = client_http.ServiceConnection(
|
||||
server_and_port, '/_service', session_factory)
|
||||
|
|
@ -107,21 +107,6 @@ def connect_vcs(server_and_port, protocol):
|
|||
raise Exception('Invalid vcs server protocol "{}"'.format(protocol))
|
||||
|
||||
|
||||
def create_vcsserver_proxy(server_and_port, protocol):
|
||||
if protocol == 'http':
|
||||
return _create_vcsserver_proxy_http(server_and_port)
|
||||
else:
|
||||
raise Exception('Invalid vcs server protocol "{}"'.format(protocol))
|
||||
|
||||
|
||||
def _create_vcsserver_proxy_http(server_and_port):
|
||||
from rhodecode.lib.vcs import client_http
|
||||
|
||||
session = _create_http_rpc_session()
|
||||
url = urlparse.urljoin('http://%s' % server_and_port, '/server')
|
||||
return client_http.RemoteObject(url, session)
|
||||
|
||||
|
||||
class CurlSession(object):
|
||||
"""
|
||||
Modeled so that it provides a subset of the requests interface.
|
||||
|
|
@ -176,12 +161,23 @@ class CurlResponse(object):
|
|||
|
||||
@property
|
||||
def content(self):
|
||||
return self._response_buffer.getvalue()
|
||||
try:
|
||||
return self._response_buffer.getvalue()
|
||||
finally:
|
||||
self._response_buffer.close()
|
||||
|
||||
@property
|
||||
def status_code(self):
|
||||
return self._status_code
|
||||
|
||||
def iter_content(self, chunk_size):
|
||||
self._response_buffer.seek(0)
|
||||
while 1:
|
||||
chunk = self._response_buffer.read(chunk_size)
|
||||
if not chunk:
|
||||
break
|
||||
yield chunk
|
||||
|
||||
|
||||
def _create_http_rpc_session():
|
||||
session = CurlSession()
|
||||
|
|
|
|||
|
|
@ -1060,6 +1060,12 @@ class BaseCommit(object):
|
|||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def get_file_content_streamed(self, path):
|
||||
"""
|
||||
returns a streaming response from vcsserver with file content
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def get_file_size(self, path):
|
||||
"""
|
||||
Returns size of the file at the given `path`.
|
||||
|
|
@ -1631,6 +1637,9 @@ class EmptyCommit(BaseCommit):
|
|||
def get_file_content(self, path):
|
||||
return u''
|
||||
|
||||
def get_file_content_streamed(self, path):
|
||||
yield self.get_file_content()
|
||||
|
||||
def get_file_size(self, path):
|
||||
return 0
|
||||
|
||||
|
|
|
|||
|
|
@ -252,6 +252,11 @@ class GitCommit(base.BaseCommit):
|
|||
tree_id, _ = self._get_tree_id_for_path(path)
|
||||
return self._remote.blob_as_pretty_string(tree_id)
|
||||
|
||||
def get_file_content_streamed(self, path):
|
||||
tree_id, _ = self._get_tree_id_for_path(path)
|
||||
stream_method = getattr(self._remote, 'stream:blob_as_pretty_string')
|
||||
return stream_method(tree_id)
|
||||
|
||||
def get_file_size(self, path):
|
||||
"""
|
||||
Returns size of the file at given `path`.
|
||||
|
|
|
|||
|
|
@ -238,6 +238,11 @@ class MercurialCommit(base.BaseCommit):
|
|||
path = self._get_filectx(path)
|
||||
return self._remote.fctx_node_data(self.raw_id, path)
|
||||
|
||||
def get_file_content_streamed(self, path):
|
||||
path = self._get_filectx(path)
|
||||
stream_method = getattr(self._remote, 'stream:fctx_node_data')
|
||||
return stream_method(self.raw_id, path)
|
||||
|
||||
def get_file_size(self, path):
|
||||
"""
|
||||
Returns size of the file at given ``path``.
|
||||
|
|
|
|||
|
|
@ -117,6 +117,11 @@ class SubversionCommit(base.BaseCommit):
|
|||
path = self._fix_path(path)
|
||||
return self._remote.get_file_content(safe_str(path), self._svn_rev)
|
||||
|
||||
def get_file_content_streamed(self, path):
|
||||
path = self._fix_path(path)
|
||||
stream_method = getattr(self._remote, 'stream:get_file_content')
|
||||
return stream_method(safe_str(path), self._svn_rev)
|
||||
|
||||
def get_file_size(self, path):
|
||||
path = self._fix_path(path)
|
||||
return self._remote.get_file_size(safe_str(path), self._svn_rev)
|
||||
|
|
|
|||
|
|
@ -51,163 +51,6 @@ EXCEPTIONS_MAP = {
|
|||
}
|
||||
|
||||
|
||||
class RepoMaker(object):
|
||||
|
||||
def __init__(self, server_and_port, backend_endpoint, backend_type, session_factory):
|
||||
self.url = urlparse.urljoin('http://%s' % server_and_port, backend_endpoint)
|
||||
self._session_factory = session_factory
|
||||
self.backend_type = backend_type
|
||||
|
||||
def __call__(self, path, repo_id, config, with_wire=None):
|
||||
log.debug('%s RepoMaker call on %s', self.backend_type.upper(), path)
|
||||
return RemoteRepo(path, repo_id, config, self.url, self._session_factory(),
|
||||
with_wire=with_wire)
|
||||
|
||||
def __getattr__(self, name):
|
||||
def f(*args, **kwargs):
|
||||
return self._call(name, *args, **kwargs)
|
||||
return f
|
||||
|
||||
@exceptions.map_vcs_exceptions
|
||||
def _call(self, name, *args, **kwargs):
|
||||
payload = {
|
||||
'id': str(uuid.uuid4()),
|
||||
'method': name,
|
||||
'backend': self.backend_type,
|
||||
'params': {'args': args, 'kwargs': kwargs}
|
||||
}
|
||||
return _remote_call(
|
||||
self.url, payload, EXCEPTIONS_MAP, self._session_factory())
|
||||
|
||||
|
||||
class ServiceConnection(object):
|
||||
def __init__(self, server_and_port, backend_endpoint, session_factory):
|
||||
self.url = urlparse.urljoin('http://%s' % server_and_port, backend_endpoint)
|
||||
self._session_factory = session_factory
|
||||
|
||||
def __getattr__(self, name):
|
||||
def f(*args, **kwargs):
|
||||
return self._call(name, *args, **kwargs)
|
||||
|
||||
return f
|
||||
|
||||
@exceptions.map_vcs_exceptions
|
||||
def _call(self, name, *args, **kwargs):
|
||||
payload = {
|
||||
'id': str(uuid.uuid4()),
|
||||
'method': name,
|
||||
'params': {'args': args, 'kwargs': kwargs}
|
||||
}
|
||||
return _remote_call(
|
||||
self.url, payload, EXCEPTIONS_MAP, self._session_factory())
|
||||
|
||||
|
||||
class RemoteRepo(object):
|
||||
|
||||
def __init__(self, path, repo_id, config, url, session, with_wire=None):
|
||||
self.url = url
|
||||
self._session = session
|
||||
with_wire = with_wire or {}
|
||||
|
||||
repo_state_uid = with_wire.get('repo_state_uid') or 'state'
|
||||
self._wire = {
|
||||
"path": path, # repo path
|
||||
"repo_id": repo_id,
|
||||
"config": config,
|
||||
"repo_state_uid": repo_state_uid,
|
||||
"context": self._create_vcs_cache_context(path, repo_state_uid)
|
||||
}
|
||||
|
||||
if with_wire:
|
||||
self._wire.update(with_wire)
|
||||
|
||||
# NOTE(johbo): Trading complexity for performance. Avoiding the call to
|
||||
# log.debug brings a few percent gain even if is is not active.
|
||||
if log.isEnabledFor(logging.DEBUG):
|
||||
self._call_with_logging = True
|
||||
|
||||
self.cert_dir = get_cert_path(rhodecode.CONFIG.get('__file__'))
|
||||
|
||||
def __getattr__(self, name):
|
||||
def f(*args, **kwargs):
|
||||
return self._call(name, *args, **kwargs)
|
||||
return f
|
||||
|
||||
@exceptions.map_vcs_exceptions
|
||||
def _call(self, name, *args, **kwargs):
|
||||
# TODO: oliver: This is currently necessary pre-call since the
|
||||
# config object is being changed for hooking scenarios
|
||||
wire = copy.deepcopy(self._wire)
|
||||
wire["config"] = wire["config"].serialize()
|
||||
wire["config"].append(('vcs', 'ssl_dir', self.cert_dir))
|
||||
|
||||
payload = {
|
||||
'id': str(uuid.uuid4()),
|
||||
'method': name,
|
||||
'params': {'wire': wire, 'args': args, 'kwargs': kwargs}
|
||||
}
|
||||
|
||||
if self._call_with_logging:
|
||||
start = time.time()
|
||||
context_uid = wire.get('context')
|
||||
log.debug('Calling %s@%s with args:%.10240r. wire_context: %s',
|
||||
self.url, name, args, context_uid)
|
||||
result = _remote_call(self.url, payload, EXCEPTIONS_MAP, self._session)
|
||||
if self._call_with_logging:
|
||||
log.debug('Call %s@%s took: %.4fs. wire_context: %s',
|
||||
self.url, name, time.time()-start, context_uid)
|
||||
return result
|
||||
|
||||
def __getitem__(self, key):
|
||||
return self.revision(key)
|
||||
|
||||
def _create_vcs_cache_context(self, *args):
|
||||
"""
|
||||
Creates a unique string which is passed to the VCSServer on every
|
||||
remote call. It is used as cache key in the VCSServer.
|
||||
"""
|
||||
hash_key = '-'.join(map(str, args))
|
||||
return str(uuid.uuid5(uuid.NAMESPACE_URL, hash_key))
|
||||
|
||||
def invalidate_vcs_cache(self):
|
||||
"""
|
||||
This invalidates the context which is sent to the VCSServer on every
|
||||
call to a remote method. It forces the VCSServer to create a fresh
|
||||
repository instance on the next call to a remote method.
|
||||
"""
|
||||
self._wire['context'] = str(uuid.uuid4())
|
||||
|
||||
|
||||
class RemoteObject(object):
|
||||
|
||||
def __init__(self, url, session):
|
||||
self._url = url
|
||||
self._session = session
|
||||
|
||||
# johbo: Trading complexity for performance. Avoiding the call to
|
||||
# log.debug brings a few percent gain even if is is not active.
|
||||
if log.isEnabledFor(logging.DEBUG):
|
||||
self._call = self._call_with_logging
|
||||
|
||||
def __getattr__(self, name):
|
||||
def f(*args, **kwargs):
|
||||
return self._call(name, *args, **kwargs)
|
||||
return f
|
||||
|
||||
@exceptions.map_vcs_exceptions
|
||||
def _call(self, name, *args, **kwargs):
|
||||
payload = {
|
||||
'id': str(uuid.uuid4()),
|
||||
'method': name,
|
||||
'params': {'args': args, 'kwargs': kwargs}
|
||||
}
|
||||
return _remote_call(self._url, payload, EXCEPTIONS_MAP, self._session)
|
||||
|
||||
def _call_with_logging(self, name, *args, **kwargs):
|
||||
log.debug('Calling %s@%s', self._url, name)
|
||||
return RemoteObject._call(self, name, *args, **kwargs)
|
||||
|
||||
|
||||
def _remote_call(url, payload, exceptions_map, session):
|
||||
try:
|
||||
response = session.post(url, data=msgpack.packb(payload))
|
||||
|
|
@ -254,6 +97,191 @@ def _remote_call(url, payload, exceptions_map, session):
|
|||
return response.get('result')
|
||||
|
||||
|
||||
def _streaming_remote_call(url, payload, exceptions_map, session, chunk_size):
|
||||
try:
|
||||
response = session.post(url, data=msgpack.packb(payload))
|
||||
except pycurl.error as e:
|
||||
msg = '{}. \npycurl traceback: {}'.format(e, traceback.format_exc())
|
||||
raise exceptions.HttpVCSCommunicationError(msg)
|
||||
except Exception as e:
|
||||
message = getattr(e, 'message', '')
|
||||
if 'Failed to connect' in message:
|
||||
# gevent doesn't return proper pycurl errors
|
||||
raise exceptions.HttpVCSCommunicationError(e)
|
||||
else:
|
||||
raise
|
||||
|
||||
if response.status_code >= 400:
|
||||
log.error('Call to %s returned non 200 HTTP code: %s',
|
||||
url, response.status_code)
|
||||
raise exceptions.HttpVCSCommunicationError(repr(response.content))
|
||||
|
||||
return response.iter_content(chunk_size=chunk_size)
|
||||
|
||||
|
||||
class ServiceConnection(object):
|
||||
def __init__(self, server_and_port, backend_endpoint, session_factory):
|
||||
self.url = urlparse.urljoin('http://%s' % server_and_port, backend_endpoint)
|
||||
self._session_factory = session_factory
|
||||
|
||||
def __getattr__(self, name):
|
||||
def f(*args, **kwargs):
|
||||
return self._call(name, *args, **kwargs)
|
||||
|
||||
return f
|
||||
|
||||
@exceptions.map_vcs_exceptions
|
||||
def _call(self, name, *args, **kwargs):
|
||||
payload = {
|
||||
'id': str(uuid.uuid4()),
|
||||
'method': name,
|
||||
'params': {'args': args, 'kwargs': kwargs}
|
||||
}
|
||||
return _remote_call(
|
||||
self.url, payload, EXCEPTIONS_MAP, self._session_factory())
|
||||
|
||||
|
||||
class RemoteVCSMaker(object):
|
||||
|
||||
def __init__(self, server_and_port, backend_endpoint, backend_type, session_factory):
|
||||
self.url = urlparse.urljoin('http://%s' % server_and_port, backend_endpoint)
|
||||
self.stream_url = urlparse.urljoin('http://%s' % server_and_port, backend_endpoint+'/stream')
|
||||
|
||||
self._session_factory = session_factory
|
||||
self.backend_type = backend_type
|
||||
|
||||
def __call__(self, path, repo_id, config, with_wire=None):
|
||||
log.debug('%s RepoMaker call on %s', self.backend_type.upper(), path)
|
||||
return RemoteRepo(path, repo_id, config, self, with_wire=with_wire)
|
||||
|
||||
def __getattr__(self, name):
|
||||
def remote_attr(*args, **kwargs):
|
||||
return self._call(name, *args, **kwargs)
|
||||
return remote_attr
|
||||
|
||||
@exceptions.map_vcs_exceptions
|
||||
def _call(self, func_name, *args, **kwargs):
|
||||
payload = {
|
||||
'id': str(uuid.uuid4()),
|
||||
'method': func_name,
|
||||
'backend': self.backend_type,
|
||||
'params': {'args': args, 'kwargs': kwargs}
|
||||
}
|
||||
url = self.url
|
||||
return _remote_call(url, payload, EXCEPTIONS_MAP, self._session_factory())
|
||||
|
||||
|
||||
class RemoteRepo(object):
|
||||
CHUNK_SIZE = 16384
|
||||
|
||||
def __init__(self, path, repo_id, config, remote_maker, with_wire=None):
|
||||
self.url = remote_maker.url
|
||||
self.stream_url = remote_maker.stream_url
|
||||
self._session = remote_maker._session_factory()
|
||||
|
||||
with_wire = with_wire or {}
|
||||
|
||||
repo_state_uid = with_wire.get('repo_state_uid') or 'state'
|
||||
self._wire = {
|
||||
"path": path, # repo path
|
||||
"repo_id": repo_id,
|
||||
"config": config,
|
||||
"repo_state_uid": repo_state_uid,
|
||||
"context": self._create_vcs_cache_context(path, repo_state_uid)
|
||||
}
|
||||
|
||||
if with_wire:
|
||||
self._wire.update(with_wire)
|
||||
|
||||
# NOTE(johbo): Trading complexity for performance. Avoiding the call to
|
||||
# log.debug brings a few percent gain even if is is not active.
|
||||
if log.isEnabledFor(logging.DEBUG):
|
||||
self._call_with_logging = True
|
||||
|
||||
self.cert_dir = get_cert_path(rhodecode.CONFIG.get('__file__'))
|
||||
|
||||
def __getattr__(self, name):
|
||||
|
||||
if name.startswith('stream:'):
|
||||
def repo_remote_attr(*args, **kwargs):
|
||||
return self._call_stream(name, *args, **kwargs)
|
||||
else:
|
||||
def repo_remote_attr(*args, **kwargs):
|
||||
return self._call(name, *args, **kwargs)
|
||||
|
||||
return repo_remote_attr
|
||||
|
||||
def _base_call(self, name, *args, **kwargs):
|
||||
# TODO: oliver: This is currently necessary pre-call since the
|
||||
# config object is being changed for hooking scenarios
|
||||
wire = copy.deepcopy(self._wire)
|
||||
wire["config"] = wire["config"].serialize()
|
||||
wire["config"].append(('vcs', 'ssl_dir', self.cert_dir))
|
||||
|
||||
payload = {
|
||||
'id': str(uuid.uuid4()),
|
||||
'method': name,
|
||||
'params': {'wire': wire, 'args': args, 'kwargs': kwargs}
|
||||
}
|
||||
|
||||
context_uid = wire.get('context')
|
||||
return context_uid, payload
|
||||
|
||||
@exceptions.map_vcs_exceptions
|
||||
def _call(self, name, *args, **kwargs):
|
||||
context_uid, payload = self._base_call(name, *args, **kwargs)
|
||||
url = self.url
|
||||
|
||||
start = time.time()
|
||||
if self._call_with_logging:
|
||||
log.debug('Calling %s@%s with args:%.10240r. wire_context: %s',
|
||||
url, name, args, context_uid)
|
||||
|
||||
result = _remote_call(url, payload, EXCEPTIONS_MAP, self._session)
|
||||
if self._call_with_logging:
|
||||
log.debug('Call %s@%s took: %.4fs. wire_context: %s',
|
||||
url, name, time.time()-start, context_uid)
|
||||
return result
|
||||
|
||||
@exceptions.map_vcs_exceptions
|
||||
def _call_stream(self, name, *args, **kwargs):
|
||||
context_uid, payload = self._base_call(name, *args, **kwargs)
|
||||
payload['chunk_size'] = self.CHUNK_SIZE
|
||||
url = self.stream_url
|
||||
|
||||
start = time.time()
|
||||
if self._call_with_logging:
|
||||
log.debug('Calling %s@%s with args:%.10240r. wire_context: %s',
|
||||
url, name, args, context_uid)
|
||||
|
||||
result = _streaming_remote_call(url, payload, EXCEPTIONS_MAP, self._session,
|
||||
self.CHUNK_SIZE)
|
||||
|
||||
if self._call_with_logging:
|
||||
log.debug('Call %s@%s took: %.4fs. wire_context: %s',
|
||||
url, name, time.time()-start, context_uid)
|
||||
return result
|
||||
|
||||
def __getitem__(self, key):
|
||||
return self.revision(key)
|
||||
|
||||
def _create_vcs_cache_context(self, *args):
|
||||
"""
|
||||
Creates a unique string which is passed to the VCSServer on every
|
||||
remote call. It is used as cache key in the VCSServer.
|
||||
"""
|
||||
hash_key = '-'.join(map(str, args))
|
||||
return str(uuid.uuid5(uuid.NAMESPACE_URL, hash_key))
|
||||
|
||||
def invalidate_vcs_cache(self):
|
||||
"""
|
||||
This invalidates the context which is sent to the VCSServer on every
|
||||
call to a remote method. It forces the VCSServer to create a fresh
|
||||
repository instance on the next call to a remote method.
|
||||
"""
|
||||
self._wire['context'] = str(uuid.uuid4())
|
||||
|
||||
|
||||
class VcsHttpProxy(object):
|
||||
|
||||
CHUNK_SIZE = 16384
|
||||
|
|
|
|||
|
|
@ -27,6 +27,7 @@ import stat
|
|||
|
||||
from zope.cachedescriptors.property import Lazy as LazyProperty
|
||||
|
||||
import rhodecode
|
||||
from rhodecode.config.conf import LANGUAGES_EXTENSIONS_MAP
|
||||
from rhodecode.lib.utils import safe_unicode, safe_str
|
||||
from rhodecode.lib.utils2 import md5
|
||||
|
|
@ -369,6 +370,17 @@ class FileNode(Node):
|
|||
content = self._content
|
||||
return content
|
||||
|
||||
def stream_bytes(self):
|
||||
"""
|
||||
Returns an iterator that will stream the content of the file directly from
|
||||
vcsserver without loading it to memory.
|
||||
"""
|
||||
if self.commit:
|
||||
return self.commit.get_file_content_streamed(self.path)
|
||||
raise NodeError(
|
||||
"Cannot retrieve message of the file without related "
|
||||
"commit attribute")
|
||||
|
||||
@LazyProperty
|
||||
def md5(self):
|
||||
"""
|
||||
|
|
@ -848,3 +860,11 @@ class LargeFileNode(FileNode):
|
|||
Overwrites name to be the org lf path
|
||||
"""
|
||||
return self.org_path
|
||||
|
||||
def stream_bytes(self):
|
||||
with open(self.path, 'rb') as stream:
|
||||
while True:
|
||||
data = stream.read(16 * 1024)
|
||||
if not data:
|
||||
break
|
||||
yield data
|
||||
|
|
|
|||
|
|
@ -57,7 +57,6 @@ from rhodecode.model.integration import IntegrationModel
|
|||
from rhodecode.integrations import integration_type_registry
|
||||
from rhodecode.integrations.types.base import IntegrationTypeBase
|
||||
from rhodecode.lib.utils import repo2db_mapper
|
||||
from rhodecode.lib.vcs import create_vcsserver_proxy
|
||||
from rhodecode.lib.vcs.backends import get_backend
|
||||
from rhodecode.lib.vcs.nodes import FileNode
|
||||
from rhodecode.tests import (
|
||||
|
|
@ -1398,82 +1397,7 @@ def testrun():
|
|||
}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def collect_appenlight_stats(request, testrun):
|
||||
"""
|
||||
This fixture reports memory consumtion of single tests.
|
||||
|
||||
It gathers data based on `psutil` and sends them to Appenlight. The option
|
||||
``--ae`` has te be used to enable this fixture and the API key for your
|
||||
application has to be provided in ``--ae-key``.
|
||||
"""
|
||||
try:
|
||||
# cygwin cannot have yet psutil support.
|
||||
import psutil
|
||||
except ImportError:
|
||||
return
|
||||
|
||||
if not request.config.getoption('--appenlight'):
|
||||
return
|
||||
else:
|
||||
# Only request the baseapp fixture if appenlight tracking is
|
||||
# enabled. This will speed up a test run of unit tests by 2 to 3
|
||||
# seconds if appenlight is not enabled.
|
||||
baseapp = request.getfuncargvalue("baseapp")
|
||||
url = '{}/api/logs'.format(request.config.getoption('--appenlight-url'))
|
||||
client = AppenlightClient(
|
||||
url=url,
|
||||
api_key=request.config.getoption('--appenlight-api-key'),
|
||||
namespace=request.node.nodeid,
|
||||
request=str(testrun['uuid']),
|
||||
testrun=testrun)
|
||||
|
||||
client.collect({
|
||||
'message': "Starting",
|
||||
})
|
||||
|
||||
server_and_port = baseapp.config.get_settings()['vcs.server']
|
||||
protocol = baseapp.config.get_settings()['vcs.server.protocol']
|
||||
server = create_vcsserver_proxy(server_and_port, protocol)
|
||||
with server:
|
||||
vcs_pid = server.get_pid()
|
||||
server.run_gc()
|
||||
vcs_process = psutil.Process(vcs_pid)
|
||||
mem = vcs_process.memory_info()
|
||||
client.tag_before('vcsserver.rss', mem.rss)
|
||||
client.tag_before('vcsserver.vms', mem.vms)
|
||||
|
||||
test_process = psutil.Process()
|
||||
mem = test_process.memory_info()
|
||||
client.tag_before('test.rss', mem.rss)
|
||||
client.tag_before('test.vms', mem.vms)
|
||||
|
||||
client.tag_before('time', time.time())
|
||||
|
||||
@request.addfinalizer
|
||||
def send_stats():
|
||||
client.tag_after('time', time.time())
|
||||
with server:
|
||||
gc_stats = server.run_gc()
|
||||
for tag, value in gc_stats.items():
|
||||
client.tag_after(tag, value)
|
||||
mem = vcs_process.memory_info()
|
||||
client.tag_after('vcsserver.rss', mem.rss)
|
||||
client.tag_after('vcsserver.vms', mem.vms)
|
||||
|
||||
mem = test_process.memory_info()
|
||||
client.tag_after('test.rss', mem.rss)
|
||||
client.tag_after('test.vms', mem.vms)
|
||||
|
||||
client.collect({
|
||||
'message': "Finished",
|
||||
})
|
||||
client.send_stats()
|
||||
|
||||
return client
|
||||
|
||||
|
||||
class AppenlightClient():
|
||||
class AppenlightClient(object):
|
||||
|
||||
url_template = '{url}?protocol_version=0.5'
|
||||
|
||||
|
|
|
|||
|
|
@ -96,7 +96,7 @@ def test_uses_persistent_http_connections(caplog, vcsbackend_hg):
|
|||
|
||||
|
||||
def test_repo_maker_uses_session_for_classmethods(stub_session_factory):
|
||||
repo_maker = client_http.RepoMaker(
|
||||
repo_maker = client_http.RemoteVCSMaker(
|
||||
'server_and_port', 'endpoint', 'test_dummy_scm', stub_session_factory)
|
||||
repo_maker.example_call()
|
||||
stub_session_factory().post.assert_called_with(
|
||||
|
|
@ -105,7 +105,7 @@ def test_repo_maker_uses_session_for_classmethods(stub_session_factory):
|
|||
|
||||
def test_repo_maker_uses_session_for_instance_methods(
|
||||
stub_session_factory, config):
|
||||
repo_maker = client_http.RepoMaker(
|
||||
repo_maker = client_http.RemoteVCSMaker(
|
||||
'server_and_port', 'endpoint', 'test_dummy_scm', stub_session_factory)
|
||||
repo = repo_maker('stub_path', 'stub_repo_id', config)
|
||||
repo.example_call()
|
||||
|
|
@ -125,7 +125,7 @@ def test_connect_passes_in_the_same_session(
|
|||
|
||||
def test_repo_maker_uses_session_that_throws_error(
|
||||
stub_session_failing_factory, config):
|
||||
repo_maker = client_http.RepoMaker(
|
||||
repo_maker = client_http.RemoteVCSMaker(
|
||||
'server_and_port', 'endpoint', 'test_dummy_scm', stub_session_failing_factory)
|
||||
repo = repo_maker('stub_path', 'stub_repo_id', config)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,171 +0,0 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright (C) 2010-2019 RhodeCode GmbH
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License, version 3
|
||||
# (only), as published by the Free Software Foundation.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
#
|
||||
# This program is dual-licensed. If you wish to learn more about the
|
||||
# RhodeCode Enterprise Edition, including its added features, Support services,
|
||||
# and proprietary license terms, please see https://rhodecode.com/licenses/
|
||||
|
||||
"""
|
||||
Load tests of certain vcs operations which can be executed by the test runner.
|
||||
|
||||
To gather timing information, run like this::
|
||||
|
||||
py.test rhodecode/tests/vcs/test_load.py --duration=0
|
||||
|
||||
To use this file with an old codebase which does not provide a compatible
|
||||
fixture setup, make sure that py.test is installed inside of the environment
|
||||
and copy this file in a place outside of the current repository::
|
||||
|
||||
TEST_HG_REPO=~/tmp/repos/vcs-hg TEST_GIT_REPO=~/tmp/repos/vcs-git \
|
||||
py.test test_load.py --duration=0
|
||||
|
||||
"""
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from rhodecode.lib.vcs import create_vcsserver_proxy
|
||||
from rhodecode.lib.vcs.backends import get_backend, get_vcs_instance
|
||||
from rhodecode.tests import TEST_HG_REPO, TEST_GIT_REPO
|
||||
|
||||
|
||||
# Allows to inject different repository paths. Used to run this
|
||||
# file with an pre-pytest codebase of rhodecode.
|
||||
TEST_HG_REPO = os.environ.get('TEST_HG_REPO', TEST_HG_REPO)
|
||||
TEST_GIT_REPO = os.environ.get('TEST_GIT_REPO', TEST_GIT_REPO)
|
||||
|
||||
|
||||
@pytest.fixture(params=('hg', 'git'))
|
||||
def repo(request, baseapp):
|
||||
repos = {
|
||||
'hg': TEST_HG_REPO,
|
||||
'git': TEST_GIT_REPO,
|
||||
}
|
||||
repo = get_vcs_instance(repos[request.param])
|
||||
return repo
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def server(baseapp):
|
||||
"""
|
||||
Returns a proxy of the server object.
|
||||
"""
|
||||
server_and_port = baseapp.config.get_settings()['vcs.server']
|
||||
protocol = baseapp.config.get_settings()['vcs.server.protocol']
|
||||
server = create_vcsserver_proxy(server_and_port, protocol)
|
||||
return server
|
||||
|
||||
|
||||
def test_server_echo(server):
|
||||
resp = server.echo('a')
|
||||
assert resp == 'a'
|
||||
|
||||
|
||||
def test_server_echo_no_data(server, repeat):
|
||||
for x in xrange(repeat):
|
||||
server.echo(None)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("payload", [
|
||||
{'a': 'dict', 'with': 'values'},
|
||||
[1, 2, 3, 4, 5] * 5,
|
||||
['a', 1, 1.2, None, {}] * 5,
|
||||
], ids=['dict', 'list-int', 'list-mix'])
|
||||
def test_server_echo_small_payload(server, repeat, payload):
|
||||
for x in xrange(repeat):
|
||||
server.echo(payload)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("payload", [
|
||||
[{'a': 'dict', 'with': 'values'}] * 100,
|
||||
[1, 2, 3, 4, 5] * 100,
|
||||
['a', 1, 1.2, None, {}] * 100,
|
||||
], ids=['dict', 'list-int', 'list-mix'])
|
||||
def test_server_echo_middle_payload(server, repeat, payload):
|
||||
for x in xrange(repeat):
|
||||
server.echo(payload)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("payload", [
|
||||
[{'a': 'dict', 'with': 'values'}] * 1000,
|
||||
[1, 2, 3, 4, 5] * 1000,
|
||||
['a', 1, 1.2, None, {}] * 1000,
|
||||
], ids=['dict', 'list-int', 'list-mix'])
|
||||
def test_server_echo_large_payload(server, repeat, payload):
|
||||
for x in xrange(repeat):
|
||||
server.echo(payload)
|
||||
|
||||
|
||||
def test_create_repo_object(repo, repeat):
|
||||
backend = get_backend(repo.alias)
|
||||
for x in xrange(repeat):
|
||||
repo = backend(repo.path)
|
||||
|
||||
|
||||
def test_get_first_commit_of_repository(repo, repeat):
|
||||
for x in xrange(repeat):
|
||||
repo.get_commit(commit_idx=1)
|
||||
|
||||
|
||||
def test_get_first_commits_slicing(repo, repeat):
|
||||
count_commits = repeat / 10
|
||||
commit = repo[0:count_commits]
|
||||
commit = list(commit)
|
||||
|
||||
|
||||
def test_get_first_commits(repo, repeat):
|
||||
end_idx = repeat / 10
|
||||
start = repo.commit_ids[0]
|
||||
end = repo.commit_ids[end_idx]
|
||||
commit = repo.get_commits(start_id=start, end_id=end)
|
||||
commit = list(commit)
|
||||
|
||||
|
||||
def test_fetch_file(repo, repeat):
|
||||
path = 'vcs/cli.py'
|
||||
tip = repo.get_commit()
|
||||
for x in xrange(repeat):
|
||||
tip.get_file_content(path=path)
|
||||
|
||||
|
||||
def test_annotate_file(repo, repeat):
|
||||
path = 'vcs/cli.py'
|
||||
tip = repo.get_commit()
|
||||
for x in xrange(repeat / 10):
|
||||
annotation_generator = tip.get_file_annotate(path=path)
|
||||
list(annotation_generator)
|
||||
|
||||
|
||||
def test_read_full_file_tree_using_walk(repo):
|
||||
tip = repo.get_commit()
|
||||
|
||||
for topnode, dirs, files in tip.walk():
|
||||
for f in files:
|
||||
len(f.content)
|
||||
|
||||
|
||||
def test_commit_diff(repo, repeat):
|
||||
tip = repo.get_commit()
|
||||
for x in xrange(repeat / 10):
|
||||
tip.diff()
|
||||
|
||||
|
||||
def test_walk_changelog(repo, repeat):
|
||||
page_size = 20
|
||||
for page in xrange(repeat / 50):
|
||||
start = page * page_size
|
||||
end = start + page_size - 1
|
||||
list(repo[start:end])
|
||||
Loading…
Add table
Add a link
Reference in a new issue