kallithea Changeset - 02cfb2197c63

Changeset - 02cfb2197c63

Parent rev.

Child rev.

[Not reviewed]

Merge default

0 5 0

Mads Kiilerich - 9 years ago 2016-06-29 16:47:47
madski@unity3d.com

Merge stable

5 files changed with 14 insertions and 2 deletions:

CONTRIBUTORS

kallithea/lib/base.py

kallithea/lib/middleware/simplehg.py

kallithea/templates/about.html

kallithea/tests/functional/test_admin_repos.py

0 comments (0 inline, 0 general)

CONTRIBUTORS

➞

Show inline comments

 List of contributors to Kallithea project:
     Mads Kiilerich <madski@unity3d.com> 2012-2016
     Takumi IINO <trot.thunder@gmail.com> 2012-2016
     Unity Technologies 2012-2016
     Andrew Shadura <andrew@shadura.me> 2012 2014-2016
     Dominik Ruf <dominikruf@gmail.com> 2012 2014-2016
     Thomas De Schampheleire <thomas.de.schampheleire@gmail.com> 2014-2016
     Étienne Gilli <etienne.gilli@gmail.com> 2015-2016
     Jan Heylen <heyleke@gmail.com> 2015-2016
     Robert Martinez <ntttq@inboxen.org> 2015-2016
     Robert Rauch <mail@robertrauch.de> 2015-2016
     Søren Løvborg <sorenl@unity3d.com> 2015-2016
     Angel Ezquerra <angel.ezquerra@gmail.com> 2016
     Asterios Dimitriou <steve@pci.gr> 2016
     Kateryna Musina <kateryna@unity3d.com> 2016
     Konstantin Veretennicov <kveretennicov@gmail.com> 2016
     Oscar Curero <oscar@naiandei.net> 2016
     Robert James Dennington <tinytimrob@googlemail.com> 2016
     timeless@gmail.com 2016
     YFdyh000 <yfdyh000@gmail.com> 2016
     Aras Pranckevičius <aras@unity3d.com> 2012-2013 2015
     Sean Farley <sean.michael.farley@gmail.com> 2013-2015
     Christian Oyarzun <oyarzun@gmail.com> 2014-2015
     Joseph Rivera <rivera.d.joseph@gmail.com> 2014-2015
     Michal Čihař <michal@cihar.com> 2014-2015
     Anatoly Bubenkov <bubenkoff@gmail.com> 2015
     Andrew Bartlett <abartlet@catalyst.net.nz> 2015
     Balázs Úr <urbalazs@gmail.com> 2015
     Ben Finney <ben@benfinney.id.au> 2015
     Branko Majic <branko@majic.rs> 2015
     Daniel Hobley <danielh@unity3d.com> 2015
     David Avigni <david.avigni@ankapi.com> 2015
     Denis Blanchette <dblanchette@coveo.com> 2015
     duanhongyi <duanhongyi@doopai.com> 2015
     EriCSN Chang <ericsning@gmail.com> 2015
     Grzegorz Krason <grzegorz.krason@gmail.com> 2015
     Jiří Suchan <yed@vanyli.net> 2015
     Kazunari Kobayashi <kobanari@nifty.com> 2015
     Kevin Bullock <kbullock@ringworld.org> 2015
     kobanari <kobanari@nifty.com> 2015
     Marc Abramowitz <marc@marc-abramowitz.com> 2015
     Marc Villetard <marc.villetard@gmail.com> 2015
     Matthias Zilk <matthias.zilk@gmail.com> 2015
     Michael Pohl <michael@mipapo.de> 2015
     Michael V. DePalatis <mike@depalatis.net> 2015
     Morten Skaaning <mortens@unity3d.com> 2015
     Nick High <nick@silverchip.org> 2015
     Niemand Jedermann <predatorix@web.de> 2015
     Peter Vitt <petervitt@web.de> 2015
     Ronny Pfannschmidt <opensource@ronnypfannschmidt.de> 2015
     Sam Jaques <sam.jaques@me.com> 2015
     Tuux <tuxa@galaxie.eu.org> 2015
     Viktar Palstsiuk <vipals@gmail.com> 2015
     Ante Ilic <ante@unity3d.com> 2014
     Bradley M. Kuhn <bkuhn@sfconservancy.org> 2014
     Calinou <calinou@opmbx.org> 2014
     Daniel Anderson <daniel@dattrix.com> 2014
     Henrik Stuart <hg@hstuart.dk> 2014
     Ingo von Borstel <kallithea@planetmaker.de> 2014
     Jelmer Vernooĳ <jelmer@samba.org> 2014
     Jim Hague <jim.hague@acm.org> 2014
     Matt Fellows <kallithea@matt-fellows.me.uk> 2014
     Max Roman <max@choloclos.se> 2014
     Na'Tosha Bard <natosha@unity3d.com> 2014
     Rasmus Selsmark <rasmuss@unity3d.com> 2014
     Tim Freund <tim@freunds.net> 2014
     Travis Burtrum <android@moparisthebest.com> 2014
     Zoltan Gyarmati <mr.zoltan.gyarmati@gmail.com> 2014
     Marcin Kuźmiński <marcin@python-works.com> 2010-2013
     xpol <xpolife@gmail.com> 2012-2013
     Aparkar <aparkar@icloud.com> 2013
     Dennis Brakhane <brakhane@googlemail.com> 2013
     Grzegorz Rożniecki <xaerxess@gmail.com> 2013
     Jonathan Sternberg <jonathansternberg@gmail.com> 2013
     Leonardo Carneiro <leonardo@unity3d.com> 2013
     Magnus Ericmats <magnus.ericmats@gmail.com> 2013
     Martin Vium <martinv@unity3d.com> 2013
     Simon Lopez <simon.lopez@slopez.org> 2013
     Ton Plomp <tcplomp@gmail.com> 2013
     Augusto Herrmann <augusto.herrmann@planejamento.gov.br> 2011-2012
     Dan Sheridan <djs@adelard.com> 2012
     Dies Koper <diesk@fast.au.fujitsu.com> 2012
     Erwin Kroon <e.kroon@smartmetersolutions.nl> 2012
     H Waldo G <gwaldo@gmail.com> 2012
     hppj <hppj@postmage.biz> 2012
     Indra Talip <indra.talip@gmail.com> 2012
     mikespook 2012
     nansenat16 <nansenat16@null.tw> 2012
     Philip Jameson <philip.j@hostdime.com> 2012
     Raoul Thill <raoul.thill@gmail.com> 2012
     Stefan Engel <mail@engel-stefan.de> 2012
     Tony Bussieres <t.bussieres@gmail.com> 2012
     Vincent Caron <vcaron@bearstech.com> 2012
     Vincent Duvert <vincent@duvert.net> 2012
     Vladislav Poluhin <nuklea@gmail.com> 2012
     Zachary Auclair <zach101@gmail.com> 2012
     Ankit Solanki <ankit.solanki@gmail.com> 2011
     Dmitri Kuznetsov 2011
     Jared Bunting <jared.bunting@peachjean.com> 2011
     Jason Harris <jason@jasonfharris.com> 2011
     Les Peabody <lpeabody@gmail.com> 2011
     Liad Shani <liadff@gmail.com> 2011
     Lorenzo M. Catucci <lorenzo@sancho.ccd.uniroma2.it> 2011
     Matt Zuba <matt.zuba@goodwillaz.org> 2011
     Nicolas VINOT <aeris@imirhil.fr> 2011
     Shawn K. O'Shea <shawn@eth0.net> 2011
     Thayne Harbaugh <thayne@fusionio.com> 2011
     Łukasz Balcerzak <lukaszbalcerzak@gmail.com> 2010
     Andrew Kesterson <andrew@aklabs.net>
     cejones
     David A. Sjøen <david.sjoen@westcon.no>
     James Rhodes <jrhodes@redpointsoftware.com.au>
     Jonas Oberschweiber <jonas.oberschweiber@d-velop.de>
     larikale
     RhodeCode GmbH
     Sebastian Kreutzberger <sebastian@rhodecode.com>
     Steve Romanow <slestak989@gmail.com>
     SteveCohen
     Thomas <thomas@rhodecode.com>
     Thomas Waldmann <tw-public@gmx.de>

kallithea/lib/base.py

➞

Show inline comments

@@ @@ -13,385 +13,385 @@ @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.lib.base
 ~~~~~~~~~~~~~~~~~~
 The base Controller API
 Provides the BaseController class for subclassing. And usage in different
 controllers
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Oct 06, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import datetime
 import logging
 import time
 import traceback
 import webob.exc
 import paste.httpexceptions
 import paste.auth.basic
 import paste.httpheaders
 from pylons import config, tmpl_context as c, request, session, url
 from pylons.controllers import WSGIController
 from pylons.templating import render_mako as render  # don't remove this import
 from pylons.i18n.translation import _
 from kallithea import __version__, BACKENDS
 from kallithea.lib.utils2 import str2bool, safe_unicode, AttributeDict, \
     safe_str, safe_int
 from kallithea.lib import auth_modules
 from kallithea.lib.auth import AuthUser, HasPermissionAnyMiddleware
 from kallithea.lib.utils import get_repo_slug
 from kallithea.lib.exceptions import UserCreationError
 from kallithea.lib.vcs.exceptions import RepositoryError, EmptyRepositoryError, ChangesetDoesNotExistError
 from kallithea.model import meta
 from kallithea.model.db import Repository, Ui, User, Setting
 from kallithea.model.notification import NotificationModel
 from kallithea.model.scm import ScmModel
 from kallithea.model.pull_request import PullRequestModel
 log = logging.getLogger(__name__)
 def _filter_proxy(ip):
     """
     HEADERS can have multiple ips inside the left-most being the original
     client, and each successive proxy that passed the request adding the IP
     address where it received the request from.
     :param ip:
     """
     if ',' in ip:
         _ips = ip.split(',')
         _first_ip = _ips[0].strip()
         log.debug('Got multiple IPs %s, using %s', ','.join(_ips), _first_ip)
         return _first_ip
     return ip
 def _get_ip_addr(environ):
     proxy_key = 'HTTP_X_REAL_IP'
     proxy_key2 = 'HTTP_X_FORWARDED_FOR'
     def_key = 'REMOTE_ADDR'
     ip = environ.get(proxy_key)
     if ip:
         return _filter_proxy(ip)
     ip = environ.get(proxy_key2)
     if ip:
         return _filter_proxy(ip)
     ip = environ.get(def_key, '0.0.0.0')
     return _filter_proxy(ip)
 def _get_access_path(environ):
     path = environ.get('PATH_INFO')
     org_req = environ.get('pylons.original_request')
     if org_req:
         path = org_req.environ.get('PATH_INFO')
     return path
 def log_in_user(user, remember, is_external_auth):
     """
     Log a `User` in and update session and cookies. If `remember` is True,
     the session cookie is set to expire in a year; otherwise, it expires at
     the end of the browser session.
     Returns populated `AuthUser` object.
     """
     user.update_lastlogin()
     meta.Session().commit()
     auth_user = AuthUser(dbuser=user,
                          is_external_auth=is_external_auth)
     # It should not be possible to explicitly log in as the default user.
     assert not auth_user.is_default_user
     auth_user.is_authenticated = True
     # Start new session to prevent session fixation attacks.
     session.invalidate()
     session['authuser'] = cookie = auth_user.to_cookie()
     # If they want to be remembered, update the cookie.
     # NOTE: Assumes that beaker defaults to browser session cookie.
     if remember:
         t = datetime.datetime.now() + datetime.timedelta(days=365)
         session._set_cookie_expires(t)
     session.save()
     log.info('user %s is now authenticated and stored in '
              'session, session attrs %s', user.username, cookie)
     # dumps session attrs back to cookie
     session._update_cookie_out()
     return auth_user
 class BasicAuth(paste.auth.basic.AuthBasicAuthenticator):
     def __init__(self, realm, authfunc, auth_http_code=None):
         self.realm = realm
         self.authfunc = authfunc
         self._rc_auth_http_code = auth_http_code
     def build_authentication(self):
         head = paste.httpheaders.WWW_AUTHENTICATE.tuples('Basic realm="%s"' % self.realm)
         if self._rc_auth_http_code and self._rc_auth_http_code == '403':
             # return 403 if alternative http return code is specified in
             # Kallithea config
             return paste.httpexceptions.HTTPForbidden(headers=head)
         return paste.httpexceptions.HTTPUnauthorized(headers=head)
     def authenticate(self, environ):
         authorization = paste.httpheaders.AUTHORIZATION(environ)
         if not authorization:
             return self.build_authentication()
         (authmeth, auth) = authorization.split(' ', 1)
         if 'basic' != authmeth.lower():
             return self.build_authentication()
         auth = auth.strip().decode('base64')
         _parts = auth.split(':', 1)
         if len(_parts) == 2:
             username, password = _parts
             if self.authfunc(username, password, environ) is not None:
                 return username
         return self.build_authentication()
     __call__ = authenticate
 class BaseVCSController(object):
     def __init__(self, application, config):
         self.application = application
         self.config = config
         # base path of repo locations
         self.basepath = self.config['base_path']
         # authenticate this VCS request using the authentication modules
         self.authenticate = BasicAuth('', auth_modules.authenticate,
                                       config.get('auth_ret_code'))
         self.ip_addr = '0.0.0.0'
     def _handle_request(self, environ, start_response):
         raise NotImplementedError()
     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> permanent URLs
         :param repo_name:
         """
         data = repo_name.split('/')
         if len(data) >= 2:
             from kallithea.lib.utils import get_repo_by_id
             by_id_match = get_repo_by_id(repo_name)
             if by_id_match:
                 data[1] = by_id_match
+                data[1] = safe_str(by_id_match)
         return '/'.join(data)
     def _invalidate_cache(self, repo_name):
         """
         Sets 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 _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
         ip_allowed = AuthUser.check_ip_allowed(user, ip_addr)
         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 _get_ip_addr(self, environ):
         return _get_ip_addr(environ)
     def _check_ssl(self, environ):
         """
         Checks the SSL check flag and returns False if SSL is not present
         and required True otherwise
         """
         #check if we have SSL required  ! if not it's a bad request !
         if str2bool(Ui.get_by_key('web', 'push_ssl').ui_value):
             org_proto = environ.get('wsgi._org_proto', environ['wsgi.url_scheme'])
             if org_proto != 'https':
                 log.debug('proto is %s and SSL is required BAD REQUEST !',
                           org_proto)
                 return False
         return True
     def _check_locking_state(self, environ, action, repo, user_id):
         """
         Checks locking on this repository, if locking is enabled and lock is
         present returns a tuple of make_lock, locked, locked_by.
         make_lock can have 3 states None (do nothing) True, make lock
         False release lock, This value is later propagated to hooks, which
         do the locking. Think about this as signals passed to hooks what to do.
         """
         locked = False  # defines that locked error should be thrown to user
         make_lock = None
         repo = Repository.get_by_repo_name(repo)
         user = User.get(user_id)
         # this is kind of hacky, but due to how mercurial handles client-server
         # server see all operation on changeset; bookmarks, phases and
         # obsolescence marker in different transaction, we don't want to check
         # locking on those
         obsolete_call = environ['QUERY_STRING'] in ['cmd=listkeys',]
         locked_by = repo.locked
         if repo and repo.enable_locking and not obsolete_call:
             if action == 'push':
                 #check if it's already locked !, if it is compare users
                 user_id, _date = repo.locked
                 if user.user_id == user_id:
                     log.debug('Got push from user %s, now unlocking', user)
                     # unlock if we have push from user who locked
                     make_lock = False
                 else:
                     # we're not the same user who locked, ban with 423 !
                     locked = True
             if action == 'pull':
                 if repo.locked[0] and repo.locked[1]:
                     locked = True
                 else:
                     log.debug('Setting lock on repo %s by %s', repo, user)
                     make_lock = True
         else:
             log.debug('Repository %s do not have locking enabled', repo)
         log.debug('FINAL locking values make_lock:%s,locked:%s,locked_by:%s',
                   make_lock, locked, locked_by)
         return make_lock, locked, locked_by
     def __call__(self, environ, start_response):
         start = time.time()
         try:
             return self._handle_request(environ, start_response)
         finally:
             log = logging.getLogger('kallithea.' + self.__class__.__name__)
             log.debug('Request time: %.3fs', time.time() - start)
             meta.Session.remove()
 class BaseController(WSGIController):
     def __before__(self):
         """
         __before__ is called before controller methods and after __call__
         """
         c.kallithea_version = __version__
         rc_config = Setting.get_app_settings()
         # Visual options
         c.visual = AttributeDict({})
         ## DB stored
         c.visual.show_public_icon = str2bool(rc_config.get('show_public_icon'))
         c.visual.show_private_icon = str2bool(rc_config.get('show_private_icon'))
         c.visual.stylify_metatags = str2bool(rc_config.get('stylify_metatags'))
         c.visual.dashboard_items = safe_int(rc_config.get('dashboard_items', 100))
         c.visual.admin_grid_items = safe_int(rc_config.get('admin_grid_items', 100))
         c.visual.repository_fields = str2bool(rc_config.get('repository_fields'))
         c.visual.show_version = str2bool(rc_config.get('show_version'))
         c.visual.use_gravatar = str2bool(rc_config.get('use_gravatar'))
         c.visual.gravatar_url = rc_config.get('gravatar_url')
         c.ga_code = rc_config.get('ga_code')
         # TODO: replace undocumented backwards compatibility hack with db upgrade and rename ga_code
         if c.ga_code and '<' not in c.ga_code:
             c.ga_code = '''<script type="text/javascript">
                 var _gaq = _gaq || [];
                 _gaq.push(['_setAccount', '%s']);
                 _gaq.push(['_trackPageview']);
                 (function() {
                     var ga = document.createElement('script'); ga.type = 'text/javascript'; ga.async = true;
                     ga.src = ('https:' == document.location.protocol ? 'https://ssl' : 'http://www') + '.google-analytics.com/ga.js';
                     var s = document.getElementsByTagName('script')[0]; s.parentNode.insertBefore(ga, s);
                     })();
             </script>''' % c.ga_code
         c.site_name = rc_config.get('title')
         c.clone_uri_tmpl = rc_config.get('clone_uri_tmpl')
         ## INI stored
         c.visual.allow_repo_location_change = str2bool(config.get('allow_repo_location_change', True))
         c.visual.allow_custom_hooks_settings = str2bool(config.get('allow_custom_hooks_settings', True))
         c.instance_id = config.get('instance_id')
         c.issues_url = config.get('bugtracker', url('issues_url'))
         # END CONFIG VARS
         c.repo_name = get_repo_slug(request)  # can be empty
         c.backends = BACKENDS.keys()
         c.unread_notifications = NotificationModel() \
                         .get_unread_cnt_for_user(c.authuser.user_id)
         self.cut_off_limit = safe_int(config.get('cut_off_limit'))
         c.my_pr_count = PullRequestModel().get_pullrequest_cnt_for_user(c.authuser.user_id)
         self.sa = meta.Session
         self.scm_model = ScmModel(self.sa)
     @staticmethod
     def _determine_auth_user(api_key, session_authuser):
         """
         Create an `AuthUser` object given the API key (if any) and the
         value of the authuser session cookie.
         """
         # Authenticate by API key
         if api_key:
             # when using API_KEY we are sure user exists.
             return AuthUser(dbuser=User.get_by_api_key(api_key),
                             is_external_auth=True)
         # Authenticate by session cookie
         # In ancient login sessions, 'authuser' may not be a dict.
         # In that case, the user will have to log in again.
         # v0.3 and earlier included an 'is_authenticated' key; if present,
         # this must be True.

kallithea/lib/middleware/simplehg.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # 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 General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.lib.middleware.simplehg
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 SimpleHg middleware for handling mercurial protocol request
 (push/clone etc.). It's implemented with basic auth function
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 28, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import logging
 import traceback
 from paste.httpheaders import REMOTE_USER, AUTH_TYPE
 from webob.exc import HTTPNotFound, HTTPForbidden, HTTPInternalServerError, \
     HTTPNotAcceptable
 from kallithea.model.db import User
 from kallithea.lib.utils2 import safe_str, safe_unicode, fix_PATH, get_server_url, \
     _set_extras
 from kallithea.lib.base import BaseVCSController, WSGIResultCloseCallback
 from kallithea.lib.utils import make_ui, is_valid_repo, ui_sections
 from kallithea.lib.vcs.utils.hgcompat import RepoError, hgweb_mod
 from kallithea.lib.exceptions import HTTPLockedRC
 from kallithea.lib import auth_modules
 log = logging.getLogger(__name__)
 def is_mercurial(environ):
     """
     Returns True if request's target is mercurial server - header
     ``HTTP_ACCEPT`` of such request would start with ``application/mercurial``.
     """
     http_accept = environ.get('HTTP_ACCEPT')
     path_info = environ['PATH_INFO']
     if http_accept and http_accept.startswith('application/mercurial'):
         ishg_path = True
     else:
         ishg_path = False
     log.debug('pathinfo: %s detected as Mercurial %s',
         path_info, ishg_path
+    )
     return ishg_path
 class SimpleHg(BaseVCSController):
     def _handle_request(self, environ, start_response):
         if not is_mercurial(environ):
             return self.application(environ, start_response)
         if not self._check_ssl(environ):
             return HTTPNotAcceptable('SSL REQUIRED !')(environ, start_response)
         ip_addr = self._get_ip_addr(environ)
         username = None
         # skip passing error to error controller
         environ['pylons.status_code_redirect'] = True
         #======================================================================
         # EXTRACT REPOSITORY NAME FROM ENV
         #======================================================================
         try:
             str_repo_name = environ['REPO_NAME'] = self.__get_repository(environ)
             assert isinstance(str_repo_name, str)
+            assert isinstance(str_repo_name, str), str_repo_name
             repo_name = safe_unicode(str_repo_name)
             assert safe_str(repo_name) == str_repo_name, (str_repo_name, repo_name)
             log.debug('Extracted repo name is %s', repo_name)
         except Exception as e:
             log.error('error extracting repo_name: %r', e)
             return HTTPInternalServerError()(environ, start_response)
         # quick check if that dir exists...
         if not is_valid_repo(repo_name, self.basepath, 'hg'):
             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(cache=True)
             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)
             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 = auth_modules.authenticate('', '', environ)
                 if pre_auth is not None 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 = \
                         safe_str(self.config['realm'])
                     result = self.authenticate(environ)
                     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
                 #==============================================================
                 try:
                     user = User.get_by_username_or_email(username)
                     if user is None or not user.active:
                         return HTTPForbidden()(environ, start_response)
                     username = user.username
                 except Exception:
                     log.error(traceback.format_exc())
                     return HTTPInternalServerError()(environ, start_response)
                 #check permissions for this repository
                 perm = self._check_permission(action, user, repo_name, ip_addr)
                 if not perm:
                     return HTTPForbidden()(environ, start_response)
         # extras are injected into mercurial UI object and later available
         # in hg hooks executed by kallithea
         from kallithea import CONFIG
         server_url = get_server_url(environ)
         extras = {
             'ip': ip_addr,
             'username': username,
             'action': action,
             'repository': repo_name,
             'scm': 'hg',
             'config': CONFIG['__file__'],
             'server_url': server_url,
             'make_lock': None,
             'locked_by': [None, None]
+        }
         #======================================================================
         # MERCURIAL REQUEST HANDLING
         #======================================================================
         repo_path = os.path.join(safe_str(self.basepath), str_repo_name)
         log.debug('Repository path is %s', repo_path)
         # CHECK LOCKING only if it's not ANONYMOUS USER
         if username != User.DEFAULT_USER:
             log.debug('Checking locking on repository')
             (make_lock,
              locked,
              locked_by) = self._check_locking_state(
                             environ=environ, action=action,
                             repo=repo_name, user_id=user.user_id
+                       )
             # store the make_lock for later evaluation in hooks
             extras.update({'make_lock': make_lock,
                            'locked_by': locked_by})
         fix_PATH()
         log.debug('HOOKS extras is %s', extras)
         baseui = make_ui('db')
         self.__inject_extras(repo_path, baseui, extras)
         try:
             log.info('%s action on Mercurial repo "%s" by "%s" from %s',
                      action, str_repo_name, safe_str(username), ip_addr)
             app = self.__make_app(repo_path, baseui, extras)
             result = app(environ, start_response)
             if action == 'push':
                 result = WSGIResultCloseCallback(result,
                     lambda: self._invalidate_cache(repo_name))
             return result
         except RepoError as e:
             if str(e).find('not found') != -1:
                 return HTTPNotFound()(environ, start_response)
         except HTTPLockedRC as e:
             # Before Mercurial 3.6, lock exceptions were caught here
             log.debug('Locked, response %s: %s', e.code, e.title)
             return e(environ, start_response)
         except Exception:
             log.error(traceback.format_exc())
             return HTTPInternalServerError()(environ, start_response)
     def __make_app(self, repo_name, baseui, extras):
         """
         Make an wsgi application using hgweb, and inject generated baseui
         instance, additionally inject some extras into ui object
         """
         class HgWebWrapper(hgweb_mod.hgweb):
             # Work-around for Mercurial 3.6+ causing lock exceptions to be
             # thrown late
             def _runwsgi(self, req, repo):
                 try:
                     return super(HgWebWrapper, self)._runwsgi(req, repo)
                 except HTTPLockedRC as e:
                     log.debug('Locked, response %s: %s', e.code, e.title)
                     req.respond(e.status, 'text/plain')
                     return ''
         return HgWebWrapper(repo_name, name=repo_name, baseui=baseui)
     def __get_repository(self, environ):
         """
         Gets repository name out of PATH_INFO header
         :param environ: environ where PATH_INFO is stored
         """
         try:
             environ['PATH_INFO'] = self._get_by_id(environ['PATH_INFO'])
             repo_name = '/'.join(environ['PATH_INFO'].split('/')[1:])
             if repo_name.endswith('/'):
                 repo_name = repo_name.rstrip('/')
         except Exception:
             log.error(traceback.format_exc())
             raise
         return repo_name
     def __get_action(self, environ):
         """
         Maps mercurial request commands into a clone,pull or push command.
         This should always return a valid command string
         :param environ:
         """
         mapping = {'changegroup': 'pull',
                    'changegroupsubset': 'pull',
                    'stream_out': 'pull',
                    'listkeys': 'pull',
                    'unbundle': 'push',
                    'pushkey': 'push', }
         for qry in environ['QUERY_STRING'].split('&'):
             if qry.startswith('cmd'):
                 cmd = qry.split('=')[-1]
                 if cmd in mapping:
                     return mapping[cmd]

kallithea/templates/about.html

➞

Show inline comments

 ## -*- coding: utf-8 -*-
 <%inherit file="/base/base.html"/>
 <%block name="title">
     ${_('About')}
 </%block>
 <%def name="breadcrumbs()">
     ${c.site_name}
 </%def>
 <%block name="header_menu">
     ${self.menu('about')}
 </%block>
 <%def name="main()">
 <div class="box">
   <!-- box / title -->
   <div class="title">
     <h5>${_('About')} Kallithea</h5>
   </div>
   <p><a href="https://kallithea-scm.org/">Kallithea</a> is a project of the
   <a href="http://sfconservancy.org/">Software Freedom Conservancy, Inc.</a>
   and is released under the terms of the
   <a href="http://www.gnu.org/copyleft/gpl.html">GNU General Public License,
   v 3.0 (GPLv3)</a>.</p>
   <p>Kallithea is copyrighted by various authors, including but not
   necessarily limited to the following:
   <ul style="margin: 0 0 0 50px;">
   <li>Copyright &copy; 2012&ndash;2016, Mads Kiilerich</li>
   <li>Copyright &copy; 2012&ndash;2016, Takumi IINO</li>
   <li>Copyright &copy; 2012&ndash;2016, Unity Technologies</li>
   <li>Copyright &copy; 2012, 2014&ndash;2016, Andrew Shadura</li>
   <li>Copyright &copy; 2012, 2014&ndash;2016, Dominik Ruf</li>
   <li>Copyright &copy; 2014&ndash;2016, Thomas De Schampheleire</li>
   <li>Copyright &copy; 2015&ndash;2016, Étienne Gilli</li>
   <li>Copyright &copy; 2015&ndash;2016, Jan Heylen</li>
   <li>Copyright &copy; 2015&ndash;2016, Robert Martinez</li>
   <li>Copyright &copy; 2015&ndash;2016, Robert Rauch</li>
   <li>Copyright &copy; 2015&ndash;2016, Søren Løvborg</li>
   <li>Copyright &copy; 2016, Angel Ezquerra</li>
   <li>Copyright &copy; 2016, Asterios Dimitriou</li>
   <li>Copyright &copy; 2016, Kateryna Musina</li>
   <li>Copyright &copy; 2016, Konstantin Veretennicov</li>
   <li>Copyright &copy; 2016, Oscar Curero</li>
   <li>Copyright &copy; 2016, Robert James Dennington</li>
   <li>Copyright &copy; 2016, timeless@gmail.com</li>
   <li>Copyright &copy; 2016, YFdyh000</li>
   <li>Copyright &copy; 2012&ndash;2013, 2015, Aras Pranckevičius</li>
   <li>Copyright &copy; 2014&ndash;2015, Christian Oyarzun</li>
   <li>Copyright &copy; 2014&ndash;2015, Joseph Rivera</li>
   <li>Copyright &copy; 2014&ndash;2015, Michal Čihař</li>
   <li>Copyright &copy; 2014&ndash;2015, Sean Farley</li>
   <li>Copyright &copy; 2015, Anatoly Bubenkov</li>
   <li>Copyright &copy; 2015, Andrew Bartlett</li>
   <li>Copyright &copy; 2015, Balázs Úr</li>
   <li>Copyright &copy; 2015, Ben Finney</li>
   <li>Copyright &copy; 2015, Branko Majic</li>
   <li>Copyright &copy; 2015, Daniel Hobley</li>
   <li>Copyright &copy; 2015, David Avigni</li>
   <li>Copyright &copy; 2015, Denis Blanchette</li>
   <li>Copyright &copy; 2015, duanhongyi</li>
   <li>Copyright &copy; 2015, EriCSN Chang</li>
   <li>Copyright &copy; 2015, Grzegorz Krason</li>
   <li>Copyright &copy; 2015, Jiří Suchan</li>
   <li>Copyright &copy; 2015, Kazunari Kobayashi</li>
   <li>Copyright &copy; 2015, Kevin Bullock</li>
   <li>Copyright &copy; 2015, kobanari</li>
   <li>Copyright &copy; 2015, Marc Abramowitz</li>
   <li>Copyright &copy; 2015, Marc Villetard</li>
   <li>Copyright &copy; 2015, Matthias Zilk</li>
   <li>Copyright &copy; 2015, Michael Pohl</li>
   <li>Copyright &copy; 2015, Michael V. DePalatis</li>
   <li>Copyright &copy; 2015, Morten Skaaning</li>
   <li>Copyright &copy; 2015, Nick High</li>
   <li>Copyright &copy; 2015, Niemand Jedermann</li>
   <li>Copyright &copy; 2015, Peter Vitt</li>
   <li>Copyright &copy; 2015, Ronny Pfannschmidt</li>
   <li>Copyright &copy; 2015, Sam Jaques</li>
   <li>Copyright &copy; 2015, Tuux</li>
   <li>Copyright &copy; 2015, Viktar Palstsiuk</li>
   <li>Copyright &copy; 2014, Ante Ilic</li>
   <li>Copyright &copy; 2014, Bradley M. Kuhn</li>
   <li>Copyright &copy; 2014, Calinou</li>
   <li>Copyright &copy; 2014, Daniel Anderson</li>
   <li>Copyright &copy; 2014, Henrik Stuart</li>
   <li>Copyright &copy; 2014, Ingo von Borstel</li>
   <li>Copyright &copy; 2014, Jelmer Vernooĳ</li>
   <li>Copyright &copy; 2014, Jim Hague</li>
   <li>Copyright &copy; 2014, Matt Fellows</li>
   <li>Copyright &copy; 2014, Max Roman</li>
   <li>Copyright &copy; 2014, Na'Tosha Bard</li>
   <li>Copyright &copy; 2014, Rasmus Selsmark</li>
   <li>Copyright &copy; 2014, Tim Freund</li>
   <li>Copyright &copy; 2014, Travis Burtrum</li>
   <li>Copyright &copy; 2014, Zoltan Gyarmati</li>
   <li>Copyright &copy; 2010&ndash;2013, Marcin Kuźmiński</li>
   <li>Copyright &copy; 2010&ndash;2013, RhodeCode GmbH</li>
   <li>Copyright &copy; 2011, 2013, Aparkar</li>
   <li>Copyright &copy; 2012&ndash;2013, xpol</li>
   <li>Copyright &copy; 2013, Dennis Brakhane</li>
   <li>Copyright &copy; 2013, Grzegorz Rożniecki</li>
   <li>Copyright &copy; 2013, Jonathan Sternberg</li>
   <li>Copyright &copy; 2013, Leonardo Carneiro</li>
   <li>Copyright &copy; 2013, Magnus Ericmats</li>
   <li>Copyright &copy; 2013, Martin Vium</li>
   <li>Copyright &copy; 2013, Simon Lopez</li>
   <li>Copyright &copy; 2011&ndash;2012, Augusto Herrmann</li>
   <li>Copyright &copy; 2012, Dan Sheridan</li>
   <li>Copyright &copy; 2012, H Waldo G</li>
   <li>Copyright &copy; 2012, hppj</li>
   <li>Copyright &copy; 2012, Indra Talip</li>
   <li>Copyright &copy; 2012, mikespook</li>
   <li>Copyright &copy; 2012, nansenat16</li>
   <li>Copyright &copy; 2012, Philip Jameson</li>
   <li>Copyright &copy; 2012, Raoul Thill</li>
   <li>Copyright &copy; 2012, Tony Bussieres</li>
   <li>Copyright &copy; 2012, Vincent Duvert</li>
   <li>Copyright &copy; 2012, Vladislav Poluhin</li>
   <li>Copyright &copy; 2012, Zachary Auclair</li>
   <li>Copyright &copy; 2011, Ankit Solanki</li>
   <li>Copyright &copy; 2011, Dmitri Kuznetsov</li>
   <li>Copyright &copy; 2011, Jared Bunting</li>
   <li>Copyright &copy; 2011, Jason Harris</li>
   <li>Copyright &copy; 2011, Les Peabody</li>
   <li>Copyright &copy; 2011, Liad Shani</li>
   <li>Copyright &copy; 2011, Lorenzo M. Catucci</li>
   <li>Copyright &copy; 2011, Matt Zuba</li>
   <li>Copyright &copy; 2011, Nicolas VINOT</li>
   <li>Copyright &copy; 2011, Shawn K. O'Shea</li>
   <li>Copyright &copy; 2010, Łukasz Balcerzak</li>
 ## We did not list the following copyright holders, given that they appeared
 ## to use for-profit company affiliations in their contribution in the
 ## Mercurial log and therefore I didn't know if copyright was theirs or
 ## their company's.
 ## Copyright &copy; 2011 Thayne Harbaugh <thayne@fusionio.com>
 ## Copyright &copy; 2012 Dies Koper <diesk@fast.au.fujitsu.com>
 ## Copyright &copy; 2012 Erwin Kroon <e.kroon@smartmetersolutions.nl>
 ## Copyright &copy; 2012 Vincent Caron <vcaron@bearstech.com>
 ##
 ## These contributors' contributions may not be copyrightable:
 ## philip.j@hostdime.com in 2012
 ## Stefan Engel <mail@engel-stefan.de> in 2012
 ## Ton Plomp <tcplomp@gmail.com> in 2013
 ##
 </ul></p>
 <p>The above are the copyright holders who have submitted direct
   contributions to the Kallithea repository.  In
   the <a href="https://kallithea-scm.org/repos/kallithea">Kallithea source
   code</a>, there is
   a <a href="https://kallithea-scm.org/repos/kallithea/files/tip/LICENSE.md">list
   of third-party libraries and code that Kallithea incorporates</a>.</p>
 </div>
 </%def>

kallithea/tests/functional/test_admin_repos.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import os
 import mock
 import urllib
 import pytest
 from kallithea.lib import vcs
 from kallithea.lib.utils2 import safe_str, safe_unicode
 from kallithea.model.db import Repository, RepoGroup, UserRepoToPerm, User, \
     Permission, Ui
 from kallithea.model.user import UserModel
 from kallithea.tests import *
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.repo import RepoModel
 from kallithea.model.meta import Session
 from kallithea.tests.fixture import Fixture, error_function
 fixture = Fixture()
 def _get_permission_for_user(user, repo):
     perm = UserRepoToPerm.query() \
                 .filter(UserRepoToPerm.repository ==
                         Repository.get_by_repo_name(repo)) \
                 .filter(UserRepoToPerm.user == User.get_by_username(user)) \
                 .all()
     return perm
 class _BaseTestCase(TestController):
     """
     Write all tests here
     """
     REPO = None
     REPO_TYPE = None
     NEW_REPO = None
     OTHER_TYPE_REPO = None
     OTHER_TYPE = None
     def test_index(self):
         self.log_user()
         response = self.app.get(url('repos'))
     def test_create(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
         assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name, repo_name))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name).one()
         assert new_repo.repo_name == repo_name
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)))
         except vcs.exceptions.VCSError:
             pytest.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name)
         Session().commit()
     def test_create_in_group(self):
         self.log_user()
         ## create GROUP
         group_name = u'sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description=u'test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
         repo_name = u'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr.group_id,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
         assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name_full, repo_name_full))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_full).one()
         new_repo_id = new_repo.repo_id
         assert new_repo.repo_name == repo_name_full
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name_full))
         response.mustcontain(repo_name_full)
         response.mustcontain(self.REPO_TYPE)
         inherited_perms = UserRepoToPerm.query() \
             .filter(UserRepoToPerm.repository_id == new_repo_id).all()
         assert len(inherited_perms) == 1
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full)))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
             Session().commit()
             pytest.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         Session().commit()
     def test_create_in_group_without_needed_permissions(self):
         usr = self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         # avoid spurious RepoGroup DetachedInstanceError ...
         authentication_token = self.authentication_token()
         # revoke
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
         ## create GROUP
         group_name = u'reg_sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description=u'test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
         group_name_allowed = u'reg_sometest_allowed_%s' % self.REPO_TYPE
         gr_allowed = RepoGroupModel().create(group_name=group_name_allowed,
                                      group_description=u'test',
                                      owner=TEST_USER_REGULAR_LOGIN)
         Session().commit()
         repo_name = u'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr.group_id,
                                                 _authentication_token=authentication_token))
         response.mustcontain('Invalid value')
         # user is allowed to create in this group
         repo_name = u'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name_allowed, repo_name])
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr_allowed.group_id,
                                                 _authentication_token=authentication_token))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
         assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
@@ @@ -405,206 +407,212 @@ class _BaseTestCase(TestController): @@
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)))
         except vcs.exceptions.VCSError:
             pytest.fail('no repo %s in filesystem' % repo_name)
         response = self.app.post(url('delete_repo', repo_name=repo_name),
             params={'_method': 'delete', '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name_unicode))
         response.follow()
         #check if repo was deleted from db
         deleted_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_unicode).scalar()
         assert deleted_repo == None
         assert os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)) == False
     def test_delete_repo_with_group(self):
         #TODO:
         pass
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('delete_repo', repo_name=self.REPO),
                                  params=dict(_method='delete', _authentication_token=self.authentication_token()))
     def test_show(self):
         self.log_user()
         response = self.app.get(url('summary_home', repo_name=self.REPO))
     def test_edit(self):
         response = self.app.get(url('edit_repo', repo_name=self.REPO))
     def test_set_private_flag_sets_default_to_none(self):
         self.log_user()
         #initially repository perm should be read
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         assert len(perm), 1
         assert perm[0].permission.permission_name == 'repository.read'
         assert Repository.get_by_repo_name(self.REPO).private == False
         response = self.app.put(url('put_repo', repo_name=self.REPO),
                         fixture._get_repo_create_params(repo_private=1,
                                                 repo_name=self.REPO,
                                                 repo_type=self.REPO_TYPE,
                                                 user=TEST_USER_ADMIN_LOGIN,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                msg='Repository %s updated successfully' % (self.REPO))
         assert Repository.get_by_repo_name(self.REPO).private == True
         #now the repo default permission should be None
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         assert len(perm), 1
         assert perm[0].permission.permission_name == 'repository.none'
         response = self.app.put(url('put_repo', repo_name=self.REPO),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=self.REPO,
                                                 repo_type=self.REPO_TYPE,
                                                 user=TEST_USER_ADMIN_LOGIN,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                msg='Repository %s updated successfully' % (self.REPO))
         assert Repository.get_by_repo_name(self.REPO).private == False
         #we turn off private now the repo default permission should stay None
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         assert len(perm), 1
         assert perm[0].permission.permission_name == 'repository.none'
         #update this permission back
         perm[0].permission = Permission.get_by_key('repository.read')
         Session().add(perm[0])
         Session().commit()
     def test_set_repo_fork_has_no_self_id(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         response = self.app.get(url('edit_repo_advanced', repo_name=self.REPO))
         opt = """<option value="%s">%s</option>""" % (repo.repo_id, self.REPO)
         response.mustcontain(no=[opt])
     def test_set_fork_of_other_repo(self):
         self.log_user()
         other_repo = u'other_%s' % self.REPO_TYPE
         fixture.create_repo(other_repo, repo_type=self.REPO_TYPE)
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(other_repo)
         response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(other_repo)
         self.checkSessionFlash(response,
             'Marked repository %s as fork of %s' % (repo.repo_name, repo2.repo_name))
         assert repo.fork == repo2
         response = response.follow()
         # check if given repo is selected
         opt = """<option value="%s" selected="selected">%s</option>""" % (
                     repo2.repo_id, repo2.repo_name)
         response.mustcontain(opt)
         fixture.destroy_repo(other_repo, forks='detach')
     def test_set_fork_of_other_type_repo(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
         response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
         self.checkSessionFlash(response,
             'Cannot set repository as fork of repository with other type')
     def test_set_fork_of_none(self):
         self.log_user()
         ## mark it as None
         response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=None, _authentication_token=self.authentication_token()))
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
         self.checkSessionFlash(response,
                                'Marked repository %s as fork of %s'
                                % (repo.repo_name, "Nothing"))
         assert repo.fork is None
     def test_set_fork_of_same_repo(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=repo.repo_id, _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                'An error occurred during this operation')
     def test_create_on_top_level_without_permissions(self):
         usr = self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         # revoke
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
         user = User.get(usr['user_id'])
         repo_name = self.NEW_REPO + u'no_perms'
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         response.mustcontain('<span class="error-message">Invalid value</span>')
         RepoModel().delete(repo_name)
         Session().commit()
     @mock.patch.object(RepoModel, '_create_filesystem_repo', error_function)
     def test_create_repo_when_filesystem_op_fails(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                'Error creating repository %s' % repo_name)
         # repo must not be in db
         repo = Repository.get_by_repo_name(repo_name)
         assert repo == None
         # repo must not be in filesystem !
         assert not os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name))
 class TestAdminReposControllerGIT(_BaseTestCase):
     REPO = GIT_REPO
     REPO_TYPE = 'git'
     NEW_REPO = NEW_GIT_REPO
     OTHER_TYPE_REPO = HG_REPO
     OTHER_TYPE = 'hg'
 class TestAdminReposControllerHG(_BaseTestCase):
     REPO = HG_REPO
     REPO_TYPE = 'hg'
     NEW_REPO = NEW_HG_REPO
     OTHER_TYPE_REPO = GIT_REPO
     OTHER_TYPE = 'git'
     def test_permanent_url_protocol_access(self):
         with pytest.raises(Exception) as e:
             self.app.get(url('summary_home', repo_name='_1'), extra_environ={'HTTP_ACCEPT': 'application/mercurial'})
         assert 'Unable to detect pull/push action' in str(e)

0 comments (0 inline, 0 general)