vcs: register repo_name as specialized middleware variables instead

of relying on env variables.

- use order of defined vcs.backends to detect middleware for handling requests

rhodecode/lib/middleware/simplevcs.py

@@ -1,449 +1,444 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2014-2016  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/
 
 """
 SimpleVCS middleware for handling protocol request (push/clone etc.)
 It's implemented with basic auth function
 """
 
 import os
 import logging
 import importlib
 from functools import wraps
 
 from paste.httpheaders import REMOTE_USER, AUTH_TYPE
 from webob.exc import (
     HTTPNotFound, HTTPForbidden, HTTPNotAcceptable, HTTPInternalServerError)
 
 import rhodecode
 from rhodecode.authentication.base import authenticate, VCS_TYPE
 from rhodecode.lib.auth import AuthUser, HasPermissionAnyMiddleware
 from rhodecode.lib.base import BasicAuth, get_ip_addr, vcs_operation_context
 from rhodecode.lib.exceptions import (
     HTTPLockedRC, HTTPRequirementError, UserCreationError,
     NotAllowedToCreateUserError)
 from rhodecode.lib.hooks_daemon import prepare_callback_daemon
 from rhodecode.lib.middleware import appenlight
 from rhodecode.lib.middleware.utils import scm_app
 from rhodecode.lib.utils import (
     is_valid_repo, get_rhodecode_realm, get_rhodecode_base_path)
 from rhodecode.lib.utils2 import safe_str, fix_PATH, str2bool
 from rhodecode.lib.vcs.conf import settings as vcs_settings
 from rhodecode.lib.vcs.backends import base
 from rhodecode.model import meta
 from rhodecode.model.db import User, Repository
 from rhodecode.model.scm import ScmModel
-from rhodecode.model.settings import SettingsModel
 
 
 log = logging.getLogger(__name__)
 
 
 def initialize_generator(factory):
     """
     Initializes the returned generator by draining its first element.
 
     This can be used to give a generator an initializer, which is the code
     up to the first yield statement. This decorator enforces that the first
     produced element has the value ``"__init__"`` to make its special
     purpose very explicit in the using code.
     """
 
     @wraps(factory)
     def wrapper(*args, **kwargs):
         gen = factory(*args, **kwargs)
         try:
             init = gen.next()
         except StopIteration:
             raise ValueError('Generator must yield at least one element.')
         if init != "__init__":
             raise ValueError('First yielded element must be "__init__".')
         return gen
     return wrapper
 
 
 class SimpleVCS(object):
     """Common functionality for SCM HTTP handlers."""
 
     SCM = 'unknown'
 
     def __init__(self, application, config, registry):
         self.registry = registry
         self.application = application
         self.config = config
-        # re-populated by specialized middlewares
+        # re-populated by specialized middleware
+        self.repo_name = None
         self.repo_vcs_config = base.Config()
+
         # base path of repo locations
         self.basepath = get_rhodecode_base_path()
         # authenticate this VCS request using authfunc
         auth_ret_code_detection = \
             str2bool(self.config.get('auth_ret_code_detection', False))
         self.authenticate = BasicAuth(
             '', authenticate, registry, config.get('auth_ret_code'),
             auth_ret_code_detection)
         self.ip_addr = '0.0.0.0'
 
     @property
     def scm_app(self):
         custom_implementation = self.config.get('vcs.scm_app_implementation')
         if custom_implementation and custom_implementation != 'pyro4':
             log.info(
                 "Using custom implementation of scm_app: %s",
                 custom_implementation)
             scm_app_impl = importlib.import_module(custom_implementation)
         else:
             scm_app_impl = scm_app
         return scm_app_impl
 
     def _get_by_id(self, repo_name):
         """
         Gets a special pattern _<ID> from clone url and tries to replace it
-        with a repository_name for support of _<ID> non changable urls
-
-        :param repo_name:
+        with a repository_name for support of _<ID> non changeable urls
         """
 
         data = repo_name.split('/')
         if len(data) >= 2:
             from rhodecode.model.repo import RepoModel
             by_id_match = RepoModel().get_repo_by_id(repo_name)
             if by_id_match:
                 data[1] = by_id_match.repo_name
 
         return safe_str('/'.join(data))
 
     def _invalidate_cache(self, repo_name):
         """
         Set's cache for this repository for invalidation on next access
 
         :param repo_name: full repo name, also a cache key
         """
         ScmModel().mark_for_invalidation(repo_name)
 
     def is_valid_and_existing_repo(self, repo_name, base_path, scm_type):
         db_repo = Repository.get_by_repo_name(repo_name)
         if not db_repo:
             log.debug('Repository `%s` not found inside the database.',
                       repo_name)
             return False
 
         if db_repo.repo_type != scm_type:
             log.warning(
                 'Repository `%s` have incorrect scm_type, expected %s got %s',
                 repo_name, db_repo.repo_type, scm_type)
             return False
 
         return is_valid_repo(repo_name, base_path, expect_scm=scm_type)
 
     def valid_and_active_user(self, user):
         """
         Checks if that user is not empty, and if it's actually object it checks
         if he's active.
 
         :param user: user object or None
         :return: boolean
         """
         if user is None:
             return False
 
         elif user.active:
             return True
 
         return False
 
     def _check_permission(self, action, user, repo_name, ip_addr=None):
         """
         Checks permissions using action (push/pull) user and repository
         name
 
         :param action: push or pull action
         :param user: user instance
         :param repo_name: repository name
         """
         # check IP
         inherit = user.inherit_default_permissions
         ip_allowed = AuthUser.check_ip_allowed(user.user_id, ip_addr,
                                                inherit_from_default=inherit)
         if ip_allowed:
             log.info('Access for IP:%s allowed', ip_addr)
         else:
             return False
 
         if action == 'push':
             if not HasPermissionAnyMiddleware('repository.write',
                                               'repository.admin')(user,
                                                                   repo_name):
                 return False
 
         else:
             # any other action need at least read permission
             if not HasPermissionAnyMiddleware('repository.read',
                                               'repository.write',
                                               'repository.admin')(user,
                                                                   repo_name):
                 return False
 
         return True
 
     def _check_ssl(self, environ, start_response):
         """
         Checks the SSL check flag and returns False if SSL is not present
         and required True otherwise
         """
         org_proto = environ['wsgi._org_proto']
         # check if we have SSL required  ! if not it's a bad request !
         require_ssl = str2bool(self.repo_vcs_config.get('web', 'push_ssl'))
         if require_ssl and org_proto == 'http':
             log.debug('proto is %s and SSL is required BAD REQUEST !',
                       org_proto)
             return False
         return True
 
     def __call__(self, environ, start_response):
         try:
             return self._handle_request(environ, start_response)
         except Exception:
             log.exception("Exception while handling request")
             appenlight.track_exception(environ)
             return HTTPInternalServerError()(environ, start_response)
         finally:
             meta.Session.remove()
 
     def _handle_request(self, environ, start_response):
 
         if not self._check_ssl(environ, start_response):
             reason = ('SSL required, while RhodeCode was unable '
                       'to detect this as SSL request')
             log.debug('User not allowed to proceed, %s', reason)
             return HTTPNotAcceptable(reason)(environ, start_response)
 
+        if not self.repo_name:
+            log.warning('Repository name is empty: %s', self.repo_name)
+            # failed to get repo name, we fail now
+            return HTTPNotFound()(environ, start_response)
+        log.debug('Extracted repo name is %s', self.repo_name)
+
         ip_addr = get_ip_addr(environ)
         username = None
 
         # skip passing error to error controller
         environ['pylons.status_code_redirect'] = True
 
         # ======================================================================
-        # EXTRACT REPOSITORY NAME FROM ENV SET IN `class VCSMiddleware`
-        # ======================================================================
-        repo_name = environ['REPO_NAME']
-        log.debug('Extracted repo name is %s', repo_name)
-
-        # check for type, presence in database and on filesystem
-        if not self.is_valid_and_existing_repo(
-                repo_name, self.basepath, self.SCM):
-            return HTTPNotFound()(environ, start_response)
-
-        # ======================================================================
         # GET ACTION PULL or PUSH
         # ======================================================================
         action = self._get_action(environ)
 
         # ======================================================================
         # CHECK ANONYMOUS PERMISSION
         # ======================================================================
         if action in ['pull', 'push']:
             anonymous_user = User.get_default_user()
             username = anonymous_user.username
             if anonymous_user.active:
                 # ONLY check permissions if the user is activated
                 anonymous_perm = self._check_permission(
-                    action, anonymous_user, repo_name, ip_addr)
+                    action, anonymous_user, self.repo_name, ip_addr)
             else:
                 anonymous_perm = False
 
             if not anonymous_user.active or not anonymous_perm:
                 if not anonymous_user.active:
                     log.debug('Anonymous access is disabled, running '
                               'authentication')
 
                 if not anonymous_perm:
                     log.debug('Not enough credentials to access this '
                               'repository as anonymous user')
 
                 username = None
                 # ==============================================================
                 # DEFAULT PERM FAILED OR ANONYMOUS ACCESS IS DISABLED SO WE
                 # NEED TO AUTHENTICATE AND ASK FOR AUTH USER PERMISSIONS
                 # ==============================================================
 
                 # try to auth based on environ, container auth methods
                 log.debug('Running PRE-AUTH for container based authentication')
                 pre_auth = authenticate(
                     '', '', environ, VCS_TYPE, registry=self.registry)
                 if pre_auth and pre_auth.get('username'):
                     username = pre_auth['username']
                 log.debug('PRE-AUTH got %s as username', username)
 
                 # If not authenticated by the container, running basic auth
                 if not username:
                     self.authenticate.realm = get_rhodecode_realm()
 
                     try:
                         result = self.authenticate(environ)
                     except (UserCreationError, NotAllowedToCreateUserError) as e:
                         log.error(e)
                         reason = safe_str(e)
                         return HTTPNotAcceptable(reason)(environ, start_response)
 
                     if isinstance(result, str):
                         AUTH_TYPE.update(environ, 'basic')
                         REMOTE_USER.update(environ, result)
                         username = result
                     else:
                         return result.wsgi_application(environ, start_response)
 
                 # ==============================================================
                 # CHECK PERMISSIONS FOR THIS REQUEST USING GIVEN USERNAME
                 # ==============================================================
                 user = User.get_by_username(username)
                 if not self.valid_and_active_user(user):
                     return HTTPForbidden()(environ, start_response)
                 username = user.username
                 user.update_lastactivity()
                 meta.Session().commit()
 
                 # check user attributes for password change flag
                 user_obj = user
                 if user_obj and user_obj.username != User.DEFAULT_USER and \
                         user_obj.user_data.get('force_password_change'):
                     reason = 'password change required'
                     log.debug('User not allowed to authenticate, %s', reason)
                     return HTTPNotAcceptable(reason)(environ, start_response)
 
                 # check permissions for this repository
-                perm = self._check_permission(action, user, repo_name, ip_addr)
+                perm = self._check_permission(
+                    action, user, self.repo_name, ip_addr)
                 if not perm:
                     return HTTPForbidden()(environ, start_response)
 
         # extras are injected into UI object and later available
         # in hooks executed by rhodecode
         check_locking = _should_check_locking(environ.get('QUERY_STRING'))
         extras = vcs_operation_context(
-            environ, repo_name=repo_name, username=username,
+            environ, repo_name=self.repo_name, username=username,
             action=action, scm=self.SCM,
             check_locking=check_locking)
 
         # ======================================================================
         # REQUEST HANDLING
         # ======================================================================
-        str_repo_name = safe_str(repo_name)
+        str_repo_name = safe_str(self.repo_name)
         repo_path = os.path.join(safe_str(self.basepath), str_repo_name)
         log.debug('Repository path is %s', repo_path)
 
         fix_PATH()
 
         log.info(
             '%s action on %s repo "%s" by "%s" from %s',
             action, self.SCM, str_repo_name, safe_str(username), ip_addr)
 
         return self._generate_vcs_response(
-            environ, start_response, repo_path, repo_name, extras, action)
+            environ, start_response, repo_path, self.repo_name, extras, action)
 
     @initialize_generator
     def _generate_vcs_response(
             self, environ, start_response, repo_path, repo_name, extras,
             action):
         """
         Returns a generator for the response content.
 
         This method is implemented as a generator, so that it can trigger
         the cache validation after all content sent back to the client. It
         also handles the locking exceptions which will be triggered when
         the first chunk is produced by the underlying WSGI application.
         """
         callback_daemon, extras = self._prepare_callback_daemon(extras)
         config = self._create_config(extras, repo_name)
         log.debug('HOOKS extras is %s', extras)
         app = self._create_wsgi_app(repo_path, repo_name, config)
 
         try:
             with callback_daemon:
                 try:
                     response = app(environ, start_response)
                 finally:
                     # This statement works together with the decorator
                     # "initialize_generator" above. The decorator ensures that
                     # we hit the first yield statement before the generator is
                     # returned back to the WSGI server. This is needed to
                     # ensure that the call to "app" above triggers the
                     # needed callback to "start_response" before the
                     # generator is actually used.
                     yield "__init__"
 
                 for chunk in response:
                     yield chunk
         except Exception as exc:
             # TODO: johbo: Improve "translating" back the exception.
             if getattr(exc, '_vcs_kind', None) == 'repo_locked':
                 exc = HTTPLockedRC(*exc.args)
                 _code = rhodecode.CONFIG.get('lock_ret_code')
                 log.debug('Repository LOCKED ret code %s!', (_code,))
             elif getattr(exc, '_vcs_kind', None) == 'requirement':
                 log.debug(
                     'Repository requires features unknown to this Mercurial')
                 exc = HTTPRequirementError(*exc.args)
             else:
                 raise
 
             for chunk in exc(environ, start_response):
                 yield chunk
         finally:
             # invalidate cache on push
             try:
                 if action == 'push':
                     self._invalidate_cache(repo_name)
             finally:
                 meta.Session.remove()
 
     def _get_repository_name(self, environ):
         """Get repository name out of the environmnent
 
         :param environ: WSGI environment
         """
         raise NotImplementedError()
 
     def _get_action(self, environ):
         """Map request commands into a pull or push command.
 
         :param environ: WSGI environment
         """
         raise NotImplementedError()
 
     def _create_wsgi_app(self, repo_path, repo_name, config):
         """Return the WSGI app that will finally handle the request."""
         raise NotImplementedError()
 
     def _create_config(self, extras, repo_name):
         """Create a Pyro safe config representation."""
         raise NotImplementedError()
 
     def _prepare_callback_daemon(self, extras):
         return prepare_callback_daemon(
             extras, protocol=vcs_settings.HOOKS_PROTOCOL,
             use_direct_calls=vcs_settings.HOOKS_DIRECT_CALLS)
 
 
 def _should_check_locking(query_string):
     # this is kind of hacky, but due to how mercurial handles client-server
     # server see all operation on commit; bookmarks, phases and
     # obsolescence marker in different transaction, we don't want to check
     # locking on those
     return query_string not in ['cmd=listkeys']

rhodecode/lib/middleware/vcs.py

@@ -1,183 +1,197 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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/
 
 import gzip
 import shutil
 import logging
 import tempfile
 import urlparse
 
+from webob.exc import HTTPNotFound
+
 import rhodecode
 from rhodecode.lib.middleware.appenlight import wrap_in_appenlight_if_enabled
 from rhodecode.lib.middleware.simplegit import SimpleGit, GIT_PROTO_PAT
 from rhodecode.lib.middleware.simplehg import SimpleHg
 from rhodecode.lib.middleware.simplesvn import SimpleSvn
-from rhodecode.lib.vcs.backends import base
 from rhodecode.model.settings import VcsSettingsModel
 
 log = logging.getLogger(__name__)
 
 
 def is_git(environ):
     """
     Returns True if requests should be handled by GIT wsgi middleware
     """
     is_git_path = GIT_PROTO_PAT.match(environ['PATH_INFO'])
     log.debug(
         'request path: `%s` detected as GIT PROTOCOL %s', environ['PATH_INFO'],
         is_git_path is not None)
 
     return is_git_path
 
 
 def is_hg(environ):
     """
     Returns True if requests target is mercurial server - header
     ``HTTP_ACCEPT`` of such request would start with ``application/mercurial``.
     """
     is_hg_path = False
 
     http_accept = environ.get('HTTP_ACCEPT')
 
     if http_accept and http_accept.startswith('application/mercurial'):
         query = urlparse.parse_qs(environ['QUERY_STRING'])
         if 'cmd' in query:
             is_hg_path = True
 
     log.debug(
         'request path: `%s` detected as HG PROTOCOL %s', environ['PATH_INFO'],
         is_hg_path)
 
     return is_hg_path
 
 
 def is_svn(environ):
     """
     Returns True if requests target is Subversion server
     """
     http_dav = environ.get('HTTP_DAV', '')
     magic_path_segment = rhodecode.CONFIG.get(
         'rhodecode_subversion_magic_path', '/!svn')
     is_svn_path = (
         'subversion' in http_dav or
         magic_path_segment in environ['PATH_INFO'])
     log.debug(
         'request path: `%s` detected as SVN PROTOCOL %s', environ['PATH_INFO'],
         is_svn_path)
 
     return is_svn_path
 
 
 class GunzipMiddleware(object):
     """
     WSGI middleware that unzips gzip-encoded requests before
     passing on to the underlying application.
     """
 
     def __init__(self, application):
         self.app = application
 
     def __call__(self, environ, start_response):
         accepts_encoding_header = environ.get('HTTP_CONTENT_ENCODING', b'')
 
         if b'gzip' in accepts_encoding_header:
             log.debug('gzip detected, now running gunzip wrapper')
             wsgi_input = environ['wsgi.input']
 
             if not hasattr(environ['wsgi.input'], 'seek'):
                 # The gzip implementation in the standard library of Python 2.x
                 # requires the '.seek()' and '.tell()' methods to be available
                 # on the input stream.  Read the data into a temporary file to
                 # work around this limitation.
 
                 wsgi_input = tempfile.SpooledTemporaryFile(64 * 1024 * 1024)
                 shutil.copyfileobj(environ['wsgi.input'], wsgi_input)
                 wsgi_input.seek(0)
 
             environ['wsgi.input'] = gzip.GzipFile(fileobj=wsgi_input, mode='r')
             # since we "Ungzipped" the content we say now it's no longer gzip
             # content encoding
             del environ['HTTP_CONTENT_ENCODING']
 
             # content length has changes ? or i'm not sure
             if 'CONTENT_LENGTH' in environ:
                 del environ['CONTENT_LENGTH']
         else:
             log.debug('content not gzipped, gzipMiddleware passing '
                       'request further')
         return self.app(environ, start_response)
 
 
 class VCSMiddleware(object):
 
     def __init__(self, app, config, appenlight_client, registry):
         self.application = app
         self.config = config
         self.appenlight_client = appenlight_client
         self.registry = registry
-        self.repo_vcs_config = base.Config()
         self.use_gzip = True
+        # order in which we check the middlewares, based on vcs.backends config
+        self.check_middlewares = config['vcs.backends']
+        self.checks = {
+            'hg': (is_hg, SimpleHg),
+            'git': (is_git, SimpleGit),
+            'svn': (is_svn, SimpleSvn),
+        }
 
     def vcs_config(self, repo_name=None):
         """
         returns serialized VcsSettings
         """
         return VcsSettingsModel(repo=repo_name).get_ui_settings_as_config_obj()
 
-    def wrap_in_gzip_if_enabled(self, app):
+    def wrap_in_gzip_if_enabled(self, app, config):
         if self.use_gzip:
             app = GunzipMiddleware(app)
         return app
 
     def _get_handler_app(self, environ):
         app = None
-
-        if is_hg(environ):
-            app = SimpleHg(self.application, self.config, self.registry)
-
-        if is_git(environ):
-            app = SimpleGit(self.application, self.config, self.registry)
-
-        if is_svn(environ):
-            app = SimpleSvn(self.application, self.config, self.registry)
-
-        if app:
-            # translate the _REPO_ID into real repo NAME for usage
-            # in midddleware
-            environ['PATH_INFO'] = app._get_by_id(environ['PATH_INFO'])
-            repo_name = app._get_repository_name(environ)
-            environ['REPO_NAME'] = repo_name
-
-            self.repo_vcs_config = self.vcs_config(repo_name)
-            app.repo_vcs_config = self.repo_vcs_config
-
-            app = self.wrap_in_gzip_if_enabled(app)
-            app, _ = wrap_in_appenlight_if_enabled(
-                app, self.config, self.appenlight_client)
+        log.debug('Checking vcs types in order: %r', self.check_middlewares)
+        for vcs_type in self.check_middlewares:
+            vcs_check, handler = self.checks[vcs_type]
+            if vcs_check(environ):
+                log.debug(
+                    'Found VCS Middleware to handle the request %s', handler)
+                app = handler(self.application, self.config, self.registry)
+                break
 
         return app
 
     def __call__(self, environ, start_response):
-        # check if we handle one of interesting protocols ?
+        # check if we handle one of interesting protocols, optionally extract
+        # specific vcsSettings and allow changes of how things are wrapped
         vcs_handler = self._get_handler_app(environ)
         if vcs_handler:
+            # translate the _REPO_ID into real repo NAME for usage
+            # in middleware
+            environ['PATH_INFO'] = vcs_handler._get_by_id(environ['PATH_INFO'])
+            repo_name = vcs_handler._get_repository_name(environ)
+
+            # check for type, presence in database and on filesystem
+            if not vcs_handler.is_valid_and_existing_repo(
+                    repo_name, vcs_handler.basepath, vcs_handler.SCM):
+                return HTTPNotFound()(environ, start_response)
+
+            # TODO(marcink): this is probably not needed anymore
+            environ['REPO_NAME'] = repo_name
+
+            # register repo_name and it's config back to the handler
+            vcs_handler.repo_name = repo_name
+            vcs_handler.repo_vcs_config = self.vcs_config(repo_name)
+
+            vcs_handler = self.wrap_in_gzip_if_enabled(
+                vcs_handler, self.config)
+            vcs_handler, _ = wrap_in_appenlight_if_enabled(
+                vcs_handler, self.config, self.appenlight_client)
             return vcs_handler(environ, start_response)
 
         return self.application(environ, start_response)

rhodecode/tests/lib/middleware/test_simplevcs.py

@@ -1,304 +1,308 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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/
 
 import base64
 
 import mock
 import pytest
 import webtest.app
 
 from rhodecode.lib.caching_query import FromCache
 from rhodecode.lib.hooks_daemon import DummyHooksCallbackDaemon
 from rhodecode.lib.middleware import simplevcs
 from rhodecode.lib.middleware.https_fixup import HttpsFixup
 from rhodecode.lib.middleware.utils import scm_app
 from rhodecode.model.db import User, _hash_key
 from rhodecode.model.meta import Session
 from rhodecode.tests import (
     HG_REPO, TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS)
 from rhodecode.tests.lib.middleware import mock_scm_app
 from rhodecode.tests.utils import set_anonymous_access
 
 
 class StubVCSController(simplevcs.SimpleVCS):
 
     SCM = 'hg'
     stub_response_body = tuple()
 
+    def __init__(self, *args, **kwargs):
+        super(StubVCSController, self).__init__(*args, **kwargs)
+        self.repo_name = HG_REPO
+
     def _get_repository_name(self, environ):
         return HG_REPO
 
     def _get_action(self, environ):
         return "pull"
 
     def _create_wsgi_app(self, repo_path, repo_name, config):
         def fake_app(environ, start_response):
             start_response('200 OK', [])
             return self.stub_response_body
         return fake_app
 
     def _create_config(self, extras, repo_name):
         return None
 
 
 @pytest.fixture
 def vcscontroller(pylonsapp, config_stub):
     config_stub.testing_securitypolicy()
     config_stub.include('rhodecode.authentication')
 
     set_anonymous_access(True)
     controller = StubVCSController(pylonsapp, pylonsapp.config, None)
     app = HttpsFixup(controller, pylonsapp.config)
     app = webtest.app.TestApp(app)
 
     _remove_default_user_from_query_cache()
 
     # Sanity checks that things are set up correctly
     app.get('/' + HG_REPO, status=200)
 
     app.controller = controller
     return app
 
 
 def _remove_default_user_from_query_cache():
     user = User.get_default_user(cache=True)
     query = Session().query(User).filter(User.username == user.username)
     query = query.options(FromCache(
         "sql_cache_short", "get_user_%s" % _hash_key(user.username)))
     query.invalidate()
     Session().expire(user)
 
 
 @pytest.fixture
 def disable_anonymous_user(request, pylonsapp):
     set_anonymous_access(False)
 
     @request.addfinalizer
     def cleanup():
         set_anonymous_access(True)
 
 
 def test_handles_exceptions_during_permissions_checks(
         vcscontroller, disable_anonymous_user):
     user_and_pass = '%s:%s' % (TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS)
     auth_password = base64.encodestring(user_and_pass).strip()
     extra_environ = {
         'AUTH_TYPE': 'Basic',
         'HTTP_AUTHORIZATION': 'Basic %s' % auth_password,
         'REMOTE_USER': TEST_USER_ADMIN_LOGIN,
     }
 
     # Verify that things are hooked up correctly
     vcscontroller.get('/', status=200, extra_environ=extra_environ)
 
     # Simulate trouble during permission checks
     with mock.patch('rhodecode.model.db.User.get_by_username',
                     side_effect=Exception) as get_user:
         # Verify that a correct 500 is returned and check that the expected
         # code path was hit.
         vcscontroller.get('/', status=500, extra_environ=extra_environ)
         assert get_user.called
 
 
 def test_returns_forbidden_if_no_anonymous_access(
         vcscontroller, disable_anonymous_user):
     vcscontroller.get('/', status=401)
 
 
 class StubFailVCSController(simplevcs.SimpleVCS):
     def _handle_request(self, environ, start_response):
         raise Exception("BOOM")
 
 
 @pytest.fixture(scope='module')
 def fail_controller(pylonsapp):
     controller = StubFailVCSController(pylonsapp, pylonsapp.config, None)
     controller = HttpsFixup(controller, pylonsapp.config)
     controller = webtest.app.TestApp(controller)
     return controller
 
 
 def test_handles_exceptions_as_internal_server_error(fail_controller):
     fail_controller.get('/', status=500)
 
 
 def test_provides_traceback_for_appenlight(fail_controller):
     response = fail_controller.get(
         '/', status=500, extra_environ={'appenlight.client': 'fake'})
     assert 'appenlight.__traceback' in response.request.environ
 
 
 def test_provides_utils_scm_app_as_scm_app_by_default(pylonsapp):
     controller = StubVCSController(pylonsapp, pylonsapp.config, None)
     assert controller.scm_app is scm_app
 
 
 def test_allows_to_override_scm_app_via_config(pylonsapp):
     config = pylonsapp.config.copy()
     config['vcs.scm_app_implementation'] = (
         'rhodecode.tests.lib.middleware.mock_scm_app')
     controller = StubVCSController(pylonsapp, config, None)
     assert controller.scm_app is mock_scm_app
 
 
 @pytest.mark.parametrize('query_string, expected', [
     ('cmd=stub_command', True),
     ('cmd=listkeys', False),
 ])
 def test_should_check_locking(query_string, expected):
     result = simplevcs._should_check_locking(query_string)
     assert result == expected
 
 
 @mock.patch.multiple(
     'Pyro4.config', SERVERTYPE='multiplex', POLLTIMEOUT=0.01)
 class TestGenerateVcsResponse:
 
     def test_ensures_that_start_response_is_called_early_enough(self):
         self.call_controller_with_response_body(iter(['a', 'b']))
         assert self.start_response.called
 
     def test_invalidates_cache_after_body_is_consumed(self):
         result = self.call_controller_with_response_body(iter(['a', 'b']))
         assert not self.was_cache_invalidated()
         # Consume the result
         list(result)
         assert self.was_cache_invalidated()
 
     @mock.patch('rhodecode.lib.middleware.simplevcs.HTTPLockedRC')
     def test_handles_locking_exception(self, http_locked_rc):
         result = self.call_controller_with_response_body(
             self.raise_result_iter(vcs_kind='repo_locked'))
         assert not http_locked_rc.called
         # Consume the result
         list(result)
         assert http_locked_rc.called
 
     @mock.patch('rhodecode.lib.middleware.simplevcs.HTTPRequirementError')
     def test_handles_requirement_exception(self, http_requirement):
         result = self.call_controller_with_response_body(
             self.raise_result_iter(vcs_kind='requirement'))
         assert not http_requirement.called
         # Consume the result
         list(result)
         assert http_requirement.called
 
     @mock.patch('rhodecode.lib.middleware.simplevcs.HTTPLockedRC')
     def test_handles_locking_exception_in_app_call(self, http_locked_rc):
         app_factory_patcher = mock.patch.object(
             StubVCSController, '_create_wsgi_app')
         with app_factory_patcher as app_factory:
             app_factory().side_effect = self.vcs_exception()
             result = self.call_controller_with_response_body(['a'])
             list(result)
         assert http_locked_rc.called
 
     def test_raises_unknown_exceptions(self):
         result = self.call_controller_with_response_body(
             self.raise_result_iter(vcs_kind='unknown'))
         with pytest.raises(Exception):
             list(result)
 
     def test_prepare_callback_daemon_is_called(self):
         def side_effect(extras):
             return DummyHooksCallbackDaemon(), extras
 
         prepare_patcher = mock.patch.object(
             StubVCSController, '_prepare_callback_daemon')
         with prepare_patcher as prepare_mock:
             prepare_mock.side_effect = side_effect
             self.call_controller_with_response_body(iter(['a', 'b']))
         assert prepare_mock.called
         assert prepare_mock.call_count == 1
 
     def call_controller_with_response_body(self, response_body):
         settings = {
             'base_path': 'fake_base_path',
             'vcs.hooks.protocol': 'http',
             'vcs.hooks.direct_calls': False,
         }
         controller = StubVCSController(None, settings, None)
         controller._invalidate_cache = mock.Mock()
         controller.stub_response_body = response_body
         self.start_response = mock.Mock()
         result = controller._generate_vcs_response(
             environ={}, start_response=self.start_response,
             repo_path='fake_repo_path',
             repo_name='fake_repo_name',
             extras={}, action='push')
         self.controller = controller
         return result
 
     def raise_result_iter(self, vcs_kind='repo_locked'):
         """
         Simulates an exception due to a vcs raised exception if kind vcs_kind
         """
         raise self.vcs_exception(vcs_kind=vcs_kind)
         yield "never_reached"
 
     def vcs_exception(self, vcs_kind='repo_locked'):
         locked_exception = Exception('TEST_MESSAGE')
         locked_exception._vcs_kind = vcs_kind
         return locked_exception
 
     def was_cache_invalidated(self):
         return self.controller._invalidate_cache.called
 
 
 class TestInitializeGenerator:
 
     def test_drains_first_element(self):
         gen = self.factory(['__init__', 1, 2])
         result = list(gen)
         assert result == [1, 2]
 
     @pytest.mark.parametrize('values', [
         [],
         [1, 2],
     ])
     def test_raises_value_error(self, values):
         with pytest.raises(ValueError):
             self.factory(values)
 
     @simplevcs.initialize_generator
     def factory(self, iterable):
         for elem in iterable:
             yield elem
 
 
 class TestPrepareHooksDaemon(object):
     def test_calls_imported_prepare_callback_daemon(self, app_settings):
         expected_extras = {'extra1': 'value1'}
         daemon = DummyHooksCallbackDaemon()
 
         controller = StubVCSController(None, app_settings, None)
         prepare_patcher = mock.patch.object(
             simplevcs, 'prepare_callback_daemon',
             return_value=(daemon, expected_extras))
         with prepare_patcher as prepare_mock:
             callback_daemon, extras = controller._prepare_callback_daemon(
                 expected_extras.copy())
         prepare_mock.assert_called_once_with(
             expected_extras,
             protocol=app_settings['vcs.hooks.protocol'],
             use_direct_calls=app_settings['vcs.hooks.direct_calls'])
 
         assert callback_daemon == daemon
         assert extras == extras

rhodecode/tests/lib/middleware/test_vcs.py

@@ -1,140 +1,140 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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/
 
 from mock import patch, Mock
 
 import rhodecode
 from rhodecode.lib.middleware import vcs
 from rhodecode.lib.middleware.simplesvn import (
     SimpleSvn, DisabledSimpleSvnApp, SimpleSvnApp)
 from rhodecode.tests import SVN_REPO
 
 svn_repo_path = '/'+ SVN_REPO
 
 def test_is_hg():
     environ = {
         'PATH_INFO': svn_repo_path,
         'QUERY_STRING': 'cmd=changegroup',
         'HTTP_ACCEPT': 'application/mercurial'
     }
     assert vcs.is_hg(environ)
 
 
 def test_is_hg_no_cmd():
     environ = {
         'PATH_INFO': svn_repo_path,
         'QUERY_STRING': '',
         'HTTP_ACCEPT': 'application/mercurial'
     }
     assert not vcs.is_hg(environ)
 
 
 def test_is_hg_empty_cmd():
     environ = {
         'PATH_INFO': svn_repo_path,
         'QUERY_STRING': 'cmd=',
         'HTTP_ACCEPT': 'application/mercurial'
     }
     assert not vcs.is_hg(environ)
 
 
 def test_is_svn_returns_true_if_subversion_is_in_a_dav_header():
     environ = {
         'PATH_INFO': svn_repo_path,
         'HTTP_DAV': 'http://subversion.tigris.org/xmlns/dav/svn/log-revprops'
     }
     assert vcs.is_svn(environ) is True
 
 
 def test_is_svn_returns_false_if_subversion_is_not_in_a_dav_header():
     environ = {
         'PATH_INFO': svn_repo_path,
         'HTTP_DAV': 'http://stuff.tigris.org/xmlns/dav/svn/log-revprops'
     }
     assert vcs.is_svn(environ) is False
 
 
 def test_is_svn_returns_false_if_no_dav_header():
     environ = {
         'PATH_INFO': svn_repo_path,
     }
     assert vcs.is_svn(environ) is False
 
 
 def test_is_svn_returns_true_if_magic_path_segment():
     environ = {
         'PATH_INFO': '/stub-repository/!svn/rev/4',
     }
     assert vcs.is_svn(environ)
 
 
 def test_is_svn_allows_to_configure_the_magic_path(monkeypatch):
     """
     This is intended as a fallback in case someone has configured his
     Subversion server with a different magic path segment.
     """
     monkeypatch.setitem(
         rhodecode.CONFIG, 'rhodecode_subversion_magic_path', '/!my-magic')
     environ = {
         'PATH_INFO': '/stub-repository/!my-magic/rev/4',
     }
     assert vcs.is_svn(environ)
 
 
 class TestVCSMiddleware(object):
     def test_get_handler_app_retuns_svn_app_when_proxy_enabled(self, app):
         environ = {
             'PATH_INFO': SVN_REPO,
             'HTTP_DAV': 'http://subversion.tigris.org/xmlns/dav/svn/log'
         }
         application = Mock()
-        config = {'appenlight': False}
+        config = {'appenlight': False, 'vcs.backends': ['svn']}
         registry = Mock()
         middleware = vcs.VCSMiddleware(
             application, config=config,
             appenlight_client=None, registry=registry)
         middleware.use_gzip = False
 
         with patch.object(SimpleSvn, '_is_svn_enabled') as mock_method:
             mock_method.return_value = True
             application = middleware._get_handler_app(environ)
             assert isinstance(application, SimpleSvn)
             assert isinstance(application._create_wsgi_app(
                 Mock(), Mock(), Mock()), SimpleSvnApp)
 
     def test_get_handler_app_retuns_dummy_svn_app_when_proxy_disabled(self, app):
         environ = {
             'PATH_INFO': SVN_REPO,
             'HTTP_DAV': 'http://subversion.tigris.org/xmlns/dav/svn/log'
         }
         application = Mock()
-        config = {'appenlight': False}
+        config = {'appenlight': False, 'vcs.backends': ['svn']}
         registry = Mock()
         middleware = vcs.VCSMiddleware(
             application, config=config,
             appenlight_client=None, registry=registry)
         middleware.use_gzip = False
 
         with patch.object(SimpleSvn, '_is_svn_enabled') as mock_method:
             mock_method.return_value = False
             application = middleware._get_handler_app(environ)
             assert isinstance(application, SimpleSvn)
             assert isinstance(application._create_wsgi_app(
                 Mock(), Mock(), Mock()), DisabledSimpleSvnApp)