@@ -12,385 +12,385 @@
# 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.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.controllers.util import redirect
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)
ip = environ.get(def_key, '0.0.0.0')
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)
auth_user.set_authenticated()
# 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():
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
__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):
#any other action need at least read permission
if not HasPermissionAnyMiddleware('repository.read',
'repository.write',
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('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)
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:
#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
# we're not the same user who locked, ban with 423 !
locked = True
if action == 'pull':
if repo.locked[0] and repo.locked[1]:
log.debug('Setting lock on repo %s by %s', repo, user)
make_lock = True
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.
if isinstance(session_authuser, dict):
# -*- 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.
kallithea.lib.middleware.simplehg
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
SimpleHg middleware for handling mercurial protocol request
(push/clone etc.). It's implemented with basic auth function
:created_on: Apr 28, 2010
import os
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
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
ishg_path = False
log.debug('pathinfo: %s detected as Mercurial %s',
path_info, ishg_path
)
return ishg_path
class SimpleHg(BaseVCSController):
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
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 = self.__get_user('default')
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)
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')
#==============================================================
# 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
return result.wsgi_application(environ, start_response)
# CHECK PERMISSIONS FOR THIS REQUEST USING GIVEN USERNAME
user = self.__get_user(username)
if user is None or not user.active:
return HTTPForbidden()(environ, start_response)
username = user.username
except Exception:
log.error(traceback.format_exc())
#check permissions for this repository
perm = self._check_permission(action, user, repo_name, ip_addr)
if not perm:
# 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)
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)
result = WSGIResultCloseCallback(result,
lambda: self._invalidate_cache(repo_name))
return result
except RepoError as e:
if str(e).find('not found') != -1:
except HTTPLockedRC as e:
_code = CONFIG.get('lock_ret_code')
log.debug('Repository LOCKED ret code %s!', _code)
return e(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
return hgweb_mod.hgweb(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
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('/')
raise
return repo_name
def __get_user(self, username):
return User.get_by_username(username)
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]
return 'pull'
raise Exception('Unable to detect pull/push action !!'
'Are you using non standard command or client ?')
def __inject_extras(self, repo_path, baseui, extras={}):
import mock
import urllib
import pytest
from kallithea.lib import vcs
from kallithea.model.db import Repository, RepoGroup, UserRepoToPerm, User,\
Permission
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 _BaseTest(object):
Write all tests here
REPO = None
REPO_TYPE = None
NEW_REPO = None
OTHER_TYPE_REPO = None
OTHER_TYPE = None
@classmethod
def setup_class(cls):
pass
def teardown_class(cls):
def test_index(self):
self.log_user()
response = self.app.get(url('repos'))
def test_create(self):
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()))
## run the check page that triggers the flash message
response = self.app.get(url('repo_check_home', repo_name=repo_name))
self.assertEqual(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()
self.assertEqual(new_repo.repo_name, repo_name)
self.assertEqual(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
vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name))
except vcs.exceptions.VCSError:
self.fail('no repo %s in filesystem' % repo_name)
RepoModel().delete(repo_name)
Session().commit()
def test_create_non_ascii(self):
non_ascii = "ąęł"
repo_name = "%s%s" % (self.NEW_REPO, non_ascii)
repo_name_unicode = repo_name.decode('utf8')
description = 'description for newly created repo' + non_ascii
description_unicode = description.decode('utf8')
u'Created repository <a href="/%s">%s</a>'
% (urllib.quote(repo_name), repo_name_unicode))
.filter(Repository.repo_name == repo_name_unicode).one()
self.assertEqual(new_repo.repo_name, repo_name_unicode)
self.assertEqual(new_repo.description, description_unicode)
def test_create_in_group(self):
## create GROUP
group_name = 'sometest_%s' % self.REPO_TYPE
gr = RepoGroupModel().create(group_name=group_name,
group_description='test',
owner=TEST_USER_ADMIN_LOGIN)
repo_name = 'ingroup'
repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
repo_group=gr.group_id,
response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
% (repo_name_full, repo_name_full))
.filter(Repository.repo_name == repo_name_full).one()
new_repo_id = new_repo.repo_id
self.assertEqual(new_repo.repo_name, repo_name_full)
response = self.app.get(url('summary_home', repo_name=repo_name_full))
response.mustcontain(repo_name_full)
inherited_perms = UserRepoToPerm.query()\
.filter(UserRepoToPerm.repository_id == new_repo_id).all()
self.assertEqual(len(inherited_perms), 1)
vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name_full))
RepoGroupModel().delete(group_name)
RepoModel().delete(repo_name_full)
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')
@@ -465,192 +467,197 @@ class _BaseTest(object):
self.assertEqual(os.path.isdir(os.path.join(TESTS_TMP_PATH, repo_name)),
False)
def test_delete_repo_with_group(self):
#TODO:
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):
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):
#initially repository perm should be read
perm = _get_permission_for_user(user='default', repo=self.REPO)
self.assertTrue(len(perm), 1)
self.assertEqual(perm[0].permission.permission_name, 'repository.read')
self.assertEqual(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,
user=TEST_USER_ADMIN_LOGIN,
msg='Repository %s updated successfully' % (self.REPO))
self.assertEqual(Repository.get_by_repo_name(self.REPO).private, True)
#now the repo default permission should be None
self.assertEqual(perm[0].permission.permission_name, 'repository.none')
#we turn off private now the repo default permission should stay None
#update this permission back
perm[0].permission = Permission.get_by_key('repository.read')
Session().add(perm[0])
def test_set_repo_fork_has_no_self_id(self):
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):
other_repo = 'other_%s' % self.REPO_TYPE
fixture.create_repo(other_repo, repo_type=self.REPO_TYPE)
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()))
'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):
repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
'Cannot set repository as fork of repository with other type')
def test_set_fork_of_none(self):
## mark it as None
params=dict(id_fork_of=None, _authentication_token=self.authentication_token()))
'Marked repository %s as fork of %s'
% (repo.repo_name, "Nothing"))
assert repo.fork is None
def test_set_fork_of_same_repo(self):
params=dict(id_fork_of=repo.repo_id, _authentication_token=self.authentication_token()))
'An error occurred during this operation')
def test_create_on_top_level_without_permissions(self):
user = User.get(usr['user_id'])
repo_name = self.NEW_REPO+'no_perms'
response.mustcontain('<span class="error-message">Invalid value</span>')
@mock.patch.object(RepoModel, '_create_filesystem_repo', error_function)
def test_create_repo_when_filesystem_op_fails(self):
'Error creating repository %s' % repo_name)
# repo must not be in db
repo = Repository.get_by_repo_name(repo_name)
self.assertEqual(repo, None)
# repo must not be in filesystem !
self.assertFalse(os.path.isdir(os.path.join(TESTS_TMP_PATH, repo_name)))
class TestAdminReposControllerGIT(TestController, _BaseTest):
REPO = GIT_REPO
REPO_TYPE = 'git'
NEW_REPO = NEW_GIT_REPO
OTHER_TYPE_REPO = HG_REPO
OTHER_TYPE = 'hg'
class TestAdminReposControllerHG(TestController, _BaseTest):
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)
Status change: