# -*- coding: utf-8 -*-
"""
rhodecode.controllers.admin.repos
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Admin controller for RhodeCode
:created_on: Apr 7, 2010
:author: marcink
:copyright: (C) 2009-2011 Marcin Kuzminski <marcin@python-works.com>
:license: GPLv3, see COPYING for more details.
# 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/>.
import logging
import traceback
import formencode
from operator import itemgetter
from formencode import htmlfill
from paste.httpexceptions import HTTPInternalServerError
from pylons import request, response, session, tmpl_context as c, url
from pylons.controllers.util import abort, redirect
from pylons.i18n.translation import _
from rhodecode.lib import helpers as h
from rhodecode.lib.auth import LoginRequired, HasPermissionAllDecorator, \
HasPermissionAnyDecorator
from rhodecode.lib.base import BaseController, render
from rhodecode.lib.utils import invalidate_cache, action_logger, repo_name_slug
from rhodecode.lib.helpers import get_token
from rhodecode.model.db import User, Repository, UserFollowing, Group
from rhodecode.model.forms import RepoForm
from rhodecode.model.scm import ScmModel
from rhodecode.model.repo import RepoModel
from sqlalchemy.exc import IntegrityError
log = logging.getLogger(__name__)
class ReposController(BaseController):
REST Controller styled on the Atom Publishing Protocol"""
# To properly map this controller, ensure your config/routing.py
# file has a resource setup:
# map.resource('repo', 'repos')
@LoginRequired()
@HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
def __before__(self):
c.admin_user = session.get('admin_user')
c.admin_username = session.get('admin_username')
super(ReposController, self).__before__()
def __load_defaults(self):
repo_model = RepoModel()
c.repo_groups = [('', '')]
parents_link = lambda k: h.literal('»'.join(
map(lambda k: k.group_name,
k.parents + [k])
)
c.repo_groups.extend([(x.group_id, parents_link(x)) for \
x in self.sa.query(Group).all()])
c.repo_groups = sorted(c.repo_groups,
key=lambda t: t[1].split('»')[0])
c.repo_groups_choices = map(lambda k: unicode(k[0]), c.repo_groups)
c.users_array = repo_model.get_users_js()
c.users_groups_array = repo_model.get_users_groups_js()
def __load_data(self, repo_name=None):
Load defaults settings for edit, and update
:param repo_name:
self.__load_defaults()
repo, dbrepo = ScmModel().get(repo_name, retval='repo')
c.repo_info = repo_model.get_by_repo_name(repo_name)
if c.repo_info is None:
h.flash(_('%s repository is not mapped to db perhaps'
' it was created or renamed from the filesystem'
' please run the application again'
' in order to rescan repositories') % repo_name,
category='error')
return redirect(url('repos'))
c.default_user_id = User.by_username('default').user_id
c.in_public_journal = self.sa.query(UserFollowing)\
.filter(UserFollowing.user_id == c.default_user_id)\
.filter(UserFollowing.follows_repository == c.repo_info).scalar()
if c.repo_info.stats:
last_rev = c.repo_info.stats.stat_on_revision
else:
last_rev = 0
c.stats_revision = last_rev
c.repo_last_rev = repo.count() - 1 if repo.revisions else 0
if last_rev == 0 or c.repo_last_rev == 0:
c.stats_percentage = 0
c.stats_percentage = '%.2f' % ((float((last_rev)) /
c.repo_last_rev) * 100)
defaults = c.repo_info.get_dict()
group, repo_name = c.repo_info.groups_and_repo
defaults['repo_name'] = repo_name
defaults['repo_group'] = getattr(group[-1] if group else None,
'group_id', None)
#fill owner
if c.repo_info.user:
defaults.update({'user': c.repo_info.user.username})
replacement_user = self.sa.query(User)\
.filter(User.admin == True).first().username
defaults.update({'user': replacement_user})
#fill repository users
for p in c.repo_info.repo_to_perm:
defaults.update({'u_perm_%s' % p.user.username:
p.permission.permission_name})
#fill repository groups
for p in c.repo_info.users_group_to_perm:
defaults.update({'g_perm_%s' % p.users_group.users_group_name:
return defaults
@HasPermissionAllDecorator('hg.admin')
def index(self, format='html'):
"""GET /repos: All items in the collection"""
# url('repos')
all_repos = [r.repo_name for r in Repository.query().all()]
cached_repo_list = ScmModel().get_repos(all_repos)
c.repos_list = sorted(cached_repo_list, key=itemgetter('name_sort'))
return render('admin/repos/repos.html')
def create(self):
POST /repos: Create a new item"""
form_result = {}
try:
form_result = RepoForm(repo_groups=c.repo_groups_choices)()\
.to_python(dict(request.POST))
repo_model.create(form_result, self.rhodecode_user)
if form_result['clone_uri']:
h.flash(_('created repository %s from %s') \
% (form_result['repo_name'], form_result['clone_uri']),
category='success')
h.flash(_('created repository %s') % form_result['repo_name'],
if request.POST.get('user_created'):
#created by regular non admin user
action_logger(self.rhodecode_user, 'user_created_repo',
form_result['repo_name'], '', self.sa)
form_result['repo_name_full'], '', self.sa)
action_logger(self.rhodecode_user, 'admin_created_repo',
except formencode.Invalid, errors:
c.new_repo = errors.value['repo_name']
r = render('admin/repos/repo_add_create_repository.html')
r = render('admin/repos/repo_add.html')
return htmlfill.render(
r,
defaults=errors.value,
errors=errors.error_dict or {},
prefix_error=False,
encoding="UTF-8")
except Exception:
log.error(traceback.format_exc())
msg = _('error occurred during creation of repository %s') \
% form_result.get('repo_name')
h.flash(msg, category='error')
return redirect(url('home'))
def new(self, format='html'):
"""GET /repos/new: Form to create a new item"""
new_repo = request.GET.get('repo', '')
c.new_repo = repo_name_slug(new_repo)
return render('admin/repos/repo_add.html')
def update(self, repo_name):
PUT /repos/repo_name: Update an existing item"""
# Forms posted to this method should contain a hidden field:
# <input type="hidden" name="_method" value="PUT" />
# Or using helpers:
# h.form(url('repo', repo_name=ID),
# method='put')
# url('repo', repo_name=ID)
changed_name = repo_name
_form = RepoForm(edit=True, old_data={'repo_name': repo_name},
repo_groups=c.repo_groups_choices)()
form_result = _form.to_python(dict(request.POST))
repo_model.update(repo_name, form_result)
invalidate_cache('get_repo_cached_%s' % repo_name)
h.flash(_('Repository %s updated successfully' % repo_name),
changed_name = form_result['repo_name_full']
action_logger(self.rhodecode_user, 'admin_updated_repo',
changed_name, '', self.sa)
defaults = self.__load_data(repo_name)
defaults.update(errors.value)
render('admin/repos/repo_edit.html'),
defaults=defaults,
h.flash(_('error occurred during update of repository %s') \
% repo_name, category='error')
return redirect(url('edit_repo', repo_name=changed_name))
def delete(self, repo_name):
DELETE /repos/repo_name: Delete an existing item"""
# <input type="hidden" name="_method" value="DELETE" />
# method='delete')
repo = repo_model.get_by_repo_name(repo_name)
if not repo:
' it was moved or renamed from the filesystem'
action_logger(self.rhodecode_user, 'admin_deleted_repo',
repo_name, '', self.sa)
repo_model.delete(repo)
h.flash(_('deleted repository %s') % repo_name, category='success')
except IntegrityError, e:
if e.message.find('repositories_fork_id_fkey'):
h.flash(_('Cannot delete %s it still contains attached '
'forks') % repo_name,
category='warning')
h.flash(_('An error occurred during '
'deletion of %s') % repo_name,
except Exception, e:
h.flash(_('An error occurred during deletion of %s') % repo_name,
def delete_perm_user(self, repo_name):
DELETE an existing repository permission user
repo_model.delete_perm_user(request.POST, repo_name)
h.flash(_('An error occurred during deletion of repository user'),
raise HTTPInternalServerError()
def delete_perm_users_group(self, repo_name):
DELETE an existing repository permission users group
repo_model.delete_perm_users_group(request.POST, repo_name)
h.flash(_('An error occurred during deletion of repository'
' users groups'),
def repo_stats(self, repo_name):
DELETE an existing repository statistics
repo_model.delete_stats(repo_name)
h.flash(_('An error occurred during deletion of repository stats'),
return redirect(url('edit_repo', repo_name=repo_name))
def repo_cache(self, repo_name):
INVALIDATE existing repository cache
ScmModel().mark_for_invalidation(repo_name)
h.flash(_('An error occurred during cache invalidation'),
def repo_public_journal(self, repo_name):
Set's this repository to be visible in public journal,
in other words assing default user to follow this repo
cur_token = request.POST.get('auth_token')
token = get_token()
if cur_token == token:
repo_id = Repository.by_repo_name(repo_name).repo_id
user_id = User.by_username('default').user_id
self.scm_model.toggle_following_repo(repo_id, user_id)
h.flash(_('Updated repository visibility in public journal'),
except:
h.flash(_('An error occurred during setting this'
' repository in public journal'),
h.flash(_('Token mismatch'), category='error')
def repo_pull(self, repo_name):
Runs task to update given repository with remote changes,
ie. make pull on remote location
ScmModel().pull_changes(repo_name, self.rhodecode_user.username)
h.flash(_('Pulled from remote location'), category='success')
h.flash(_('An error occurred during pull from remote location'),
def show(self, repo_name, format='html'):
"""GET /repos/repo_name: Show a specific item"""
def edit(self, repo_name, format='html'):
"""GET /repos/repo_name/edit: Form to edit an existing item"""
# url('edit_repo', repo_name=ID)
encoding="UTF-8",
force_defaults=False
@@ -197,384 +197,385 @@ def check_repo_fast(repo_name, base_path
def check_repo(repo_name, base_path, verify=True):
repo_path = os.path.join(base_path, repo_name)
if not check_repo_fast(repo_name, base_path):
return False
r = hg.repository(ui.ui(), repo_path)
if verify:
hg.verify(r)
#here we hnow that repo exists it was verified
log.info('%s repo is already created', repo_name)
except RepoError:
#it means that there is no valid repo there...
log.info('%s repo is free for creation', repo_name)
return True
def ask_ok(prompt, retries=4, complaint='Yes or no, please!'):
while True:
ok = raw_input(prompt)
if ok in ('y', 'ye', 'yes'):
if ok in ('n', 'no', 'nop', 'nope'):
retries = retries - 1
if retries < 0:
raise IOError
print complaint
#propagated from mercurial documentation
ui_sections = ['alias', 'auth',
'decode/encode', 'defaults',
'diff', 'email',
'extensions', 'format',
'merge-patterns', 'merge-tools',
'hooks', 'http_proxy',
'smtp', 'patch',
'paths', 'profiling',
'server', 'trusted',
'ui', 'web', ]
def make_ui(read_from='file', path=None, checkpaths=True):
"""A function that will read python rc files or database
and make an mercurial ui object from read options
:param path: path to mercurial config file
:param checkpaths: check the path
:param read_from: read from 'file' or 'db'
baseui = ui.ui()
#clean the baseui object
baseui._ocfg = config.config()
baseui._ucfg = config.config()
baseui._tcfg = config.config()
if read_from == 'file':
if not os.path.isfile(path):
log.warning('Unable to read config file %s' % path)
log.debug('reading hgrc from %s', path)
cfg = config.config()
cfg.read(path)
for section in ui_sections:
for k, v in cfg.items(section):
log.debug('settings ui from file[%s]%s:%s', section, k, v)
baseui.setconfig(section, k, v)
elif read_from == 'db':
sa = meta.Session()
ret = sa.query(RhodeCodeUi)\
.options(FromCache("sql_cache_short",
"get_hg_ui_settings")).all()
hg_ui = ret
for ui_ in hg_ui:
if ui_.ui_active:
log.debug('settings ui from db[%s]%s:%s', ui_.ui_section,
ui_.ui_key, ui_.ui_value)
baseui.setconfig(ui_.ui_section, ui_.ui_key, ui_.ui_value)
meta.Session.remove()
return baseui
def set_rhodecode_config(config):
"""Updates pylons config with new settings from database
:param config:
hgsettings = RhodeCodeSettings.get_app_settings()
for k, v in hgsettings.items():
config[k] = v
def invalidate_cache(cache_key, *args):
"""Puts cache invalidation task into db for
further global cache invalidation
if cache_key.startswith('get_repo_cached_'):
name = cache_key.split('get_repo_cached_')[-1]
ScmModel().mark_for_invalidation(name)
class EmptyChangeset(BaseChangeset):
An dummy empty changeset. It's possible to pass hash when creating
an EmptyChangeset
def __init__(self, cs='0' * 40, repo=None):
self._empty_cs = cs
self.revision = -1
self.message = ''
self.author = ''
self.date = ''
self.repository = repo
@LazyProperty
def raw_id(self):
"""Returns raw string identifying this changeset, useful for web
representation.
return self._empty_cs
def short_id(self):
return self.raw_id[:12]
def get_file_changeset(self, path):
return self
def get_file_content(self, path):
return u''
def get_file_size(self, path):
return 0
def map_groups(groups):
"""Checks for groups existence, and creates groups structures.
It returns last group in structure
:param groups: list of groups structure
parent = None
group = None
for lvl, group_name in enumerate(groups[:-1]):
group = sa.query(Group).filter(Group.group_name == group_name).scalar()
if group is None:
group = Group(group_name, parent)
sa.add(group)
sa.commit()
parent = group
return group
def repo2db_mapper(initial_repo_list, remove_obsolete=False):
"""maps all repos given in initial_repo_list, non existing repositories
are created, if remove_obsolete is True it also check for db entries
that are not in initial_repo_list and removes them.
:param initial_repo_list: list of repositories found by scanning methods
:param remove_obsolete: check for obsolete entries in database
rm = RepoModel()
user = sa.query(User).filter(User.admin == True).first()
added = []
for name, repo in initial_repo_list.items():
group = map_groups(name.split('/'))
if not rm.get_by_repo_name(name, cache=False):
log.info('repository %s not found creating default', name)
added.append(name)
form_data = {
'repo_name': name,
'repo_name_full': name,
'repo_type': repo.alias,
'description': repo.description \
if repo.description != 'unknown' else \
'%s repository' % name,
'private': False,
'group_id': getattr(group, 'group_id', None)
}
rm.create(form_data, user, just_db=True)
removed = []
if remove_obsolete:
#remove from database those repositories that are not in the filesystem
for repo in sa.query(Repository).all():
if repo.repo_name not in initial_repo_list.keys():
removed.append(repo.repo_name)
sa.delete(repo)
return added, removed
#set cache regions for beaker so celery can utilise it
def add_cache(settings):
cache_settings = {'regions': None}
for key in settings.keys():
for prefix in ['beaker.cache.', 'cache.']:
if key.startswith(prefix):
name = key.split(prefix)[1].strip()
cache_settings[name] = settings[key].strip()
if cache_settings['regions']:
for region in cache_settings['regions'].split(','):
region = region.strip()
region_settings = {}
for key, value in cache_settings.items():
if key.startswith(region):
region_settings[key.split('.')[1]] = value
region_settings['expire'] = int(region_settings.get('expire',
60))
region_settings.setdefault('lock_dir',
cache_settings.get('lock_dir'))
region_settings.setdefault('data_dir',
cache_settings.get('data_dir'))
if 'type' not in region_settings:
region_settings['type'] = cache_settings.get('type',
'memory')
beaker.cache.cache_regions[region] = region_settings
def get_current_revision():
"""Returns tuple of (number, id) from repository containing this package
or None if repository could not be found.
from vcs import get_repo
from vcs.utils.helpers import get_scm
from vcs.exceptions import RepositoryError, VCSError
repopath = os.path.join(os.path.dirname(__file__), '..', '..')
scm = get_scm(repopath)[0]
repo = get_repo(path=repopath, alias=scm)
tip = repo.get_changeset()
return (tip.revision, tip.short_id)
except (ImportError, RepositoryError, VCSError), err:
logging.debug("Cannot retrieve rhodecode's revision. Original error "
"was: %s" % err)
return None
#==============================================================================
# TEST FUNCTIONS AND CREATORS
def create_test_index(repo_location, full_index):
"""Makes default test index
:param repo_location:
:param full_index:
from rhodecode.lib.indexers.daemon import WhooshIndexingDaemon
from rhodecode.lib.pidlock import DaemonLock, LockHeld
import shutil
index_location = os.path.join(repo_location, 'index')
if os.path.exists(index_location):
shutil.rmtree(index_location)
l = DaemonLock(file=jn(dn(dn(index_location)), 'make_index.lock'))
WhooshIndexingDaemon(index_location=index_location,
repo_location=repo_location)\
.run(full_index=full_index)
l.release()
except LockHeld:
pass
def create_test_env(repos_test_path, config):
"""Makes a fresh database and
install test repository into tmp dir
from rhodecode.lib.db_manage import DbManage
from rhodecode.tests import HG_REPO, GIT_REPO, NEW_HG_REPO, NEW_GIT_REPO, \
HG_FORK, GIT_FORK, TESTS_TMP_PATH
import tarfile
from os.path import dirname as dn, join as jn, abspath
log = logging.getLogger('TestEnvCreator')
# create logger
log.setLevel(logging.DEBUG)
log.propagate = True
# create console handler and set level to debug
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
# create formatter
formatter = logging.Formatter("%(asctime)s - %(name)s -"
" %(levelname)s - %(message)s")
# add formatter to ch
ch.setFormatter(formatter)
# add ch to logger
log.addHandler(ch)
#PART ONE create db
dbconf = config['sqlalchemy.db1.url']
log.debug('making test db %s', dbconf)
dbmanage = DbManage(log_sql=True, dbconf=dbconf, root=config['here'],
tests=True)
dbmanage.create_tables(override=True)
dbmanage.create_settings(dbmanage.config_prompt(repos_test_path))
dbmanage.create_default_user()
dbmanage.admin_prompt()
dbmanage.create_permissions()
dbmanage.populate_default_permissions()
#PART TWO make test repo
log.debug('making test vcs repositories')
#remove old one from previos tests
for r in [HG_REPO, GIT_REPO, NEW_HG_REPO, NEW_GIT_REPO, HG_FORK, GIT_FORK]:
if os.path.isdir(jn(TESTS_TMP_PATH, r)):
log.debug('removing %s', r)
shutil.rmtree(jn(TESTS_TMP_PATH, r))
#CREATE DEFAULT HG REPOSITORY
cur_dir = dn(dn(abspath(__file__)))
tar = tarfile.open(jn(cur_dir, 'tests', "vcs_test_hg.tar.gz"))
tar.extractall(jn(TESTS_TMP_PATH, HG_REPO))
tar.close()
# PASTER COMMANDS
class BasePasterCommand(Command):
Abstract Base Class for paster commands.
The celery commands are somewhat aggressive about loading
celery.conf, and since our module sets the `CELERY_LOADER`
environment variable to our loader, we have to bootstrap a bit and
make sure we've had a chance to load the pylons config off of the
command line, otherwise everything fails.
min_args = 1
min_args_error = "Please provide a paster config file as an argument."
takes_config_file = 1
requires_config_file = True
def notify_msg(self, msg, log=False):
"""Make a notification to user, additionally if logger is passed
it logs this action using given logger
:param msg: message that will be printed to user
:param log: logging instance, to use to additionally log this message
if log and isinstance(log, logging):
log(msg)
def run(self, args):
Overrides Command.run
Checks for a config file argument and loads it.
if len(args) < self.min_args:
raise BadCommand(
self.min_args_error % {'min_args': self.min_args,
'actual_args': len(args)})
@@ -38,409 +38,409 @@ from rhodecode.lib.exceptions import Lda
from rhodecode.model.user import UserModel
from rhodecode.model.db import User, UsersGroup, Group
from rhodecode import BACKENDS
#this is needed to translate the messages using _() in validators
class State_obj(object):
_ = staticmethod(_)
# VALIDATORS
class ValidAuthToken(formencode.validators.FancyValidator):
messages = {'invalid_token':_('Token mismatch')}
def validate_python(self, value, state):
if value != authentication_token():
raise formencode.Invalid(self.message('invalid_token', state,
search_number=value), value, state)
def ValidUsername(edit, old_data):
class _ValidUsername(formencode.validators.FancyValidator):
if value in ['default', 'new_user']:
raise formencode.Invalid(_('Invalid username'), value, state)
#check if user is unique
old_un = None
if edit:
old_un = UserModel().get(old_data.get('user_id')).username
if old_un != value or not edit:
if UserModel().get_by_username(value, cache=False,
case_insensitive=True):
raise formencode.Invalid(_('This username already '
'exists') , value, state)
if re.match(r'^[a-zA-Z0-9]{1}[a-zA-Z0-9\-\_\.]+$', value) is None:
raise formencode.Invalid(_('Username may only contain '
'alphanumeric characters '
'underscores, periods or dashes '
'and must begin with alphanumeric '
'character'), value, state)
return _ValidUsername
def ValidUsersGroup(edit, old_data):
class _ValidUsersGroup(formencode.validators.FancyValidator):
if value in ['default']:
raise formencode.Invalid(_('Invalid group name'), value, state)
#check if group is unique
old_ugname = None
old_ugname = UsersGroup.get(
old_data.get('users_group_id')).users_group_name
if old_ugname != value or not edit:
if UsersGroup.get_by_group_name(value, cache=False,
raise formencode.Invalid(_('This users group '
'already exists') , value,
state)
raise formencode.Invalid(_('Group name may only contain '
return _ValidUsersGroup
def ValidReposGroup(edit, old_data):
class _ValidReposGroup(formencode.validators.FancyValidator):
#TODO WRITE VALIDATIONS
group_name = value.get('group_name')
group_parent_id = int(value.get('group_parent_id') or - 1)
# slugify repo group just in case :)
slug = repo_name_slug(group_name)
# check for parent of self
if old_data['group_id'] == group_parent_id:
e_dict = {'group_parent_id':_('Cannot assign this group '
'as parent')}
raise formencode.Invalid('', value, state,
error_dict=e_dict)
old_gname = None
old_gname = Group.get(
old_data.get('group_id')).group_name
if old_gname != group_name or not edit:
# check filesystem
gr = Group.query().filter(Group.group_name == slug)\
.filter(Group.group_parent_id == group_parent_id).scalar()
if gr:
e_dict = {'group_name':_('This group already exists')}
return _ValidReposGroup
class ValidPassword(formencode.validators.FancyValidator):
def to_python(self, value, state):
if value:
if value.get('password'):
value['password'] = get_crypt_password(value['password'])
except UnicodeEncodeError:
e_dict = {'password':_('Invalid characters in password')}
raise formencode.Invalid('', value, state, error_dict=e_dict)
if value.get('password_confirmation'):
value['password_confirmation'] = \
get_crypt_password(value['password_confirmation'])
e_dict = {'password_confirmation':_('Invalid characters in password')}
if value.get('new_password'):
value['new_password'] = \
get_crypt_password(value['new_password'])
e_dict = {'new_password':_('Invalid characters in password')}
return value
class ValidPasswordsMatch(formencode.validators.FancyValidator):
if value['password'] != value['password_confirmation']:
e_dict = {'password_confirmation':
_('Password do not match')}
class ValidAuth(formencode.validators.FancyValidator):
messages = {
'invalid_password':_('invalid password'),
'invalid_login':_('invalid user name'),
'disabled_account':_('Your account is disabled')
#error mapping
e_dict = {'username':messages['invalid_login'],
'password':messages['invalid_password']}
e_dict_disable = {'username':messages['disabled_account']}
password = value['password']
username = value['username']
user = UserModel().get_by_username(username)
if authenticate(username, password):
if user and user.active is False:
log.warning('user %s is disabled', username)
raise formencode.Invalid(self.message('disabled_account',
state=State_obj),
value, state,
error_dict=self.e_dict_disable)
log.warning('user %s not authenticated', username)
raise formencode.Invalid(self.message('invalid_password',
state=State_obj), value, state,
error_dict=self.e_dict)
class ValidRepoUser(formencode.validators.FancyValidator):
self.user_db = User.query()\
.filter(User.active == True)\
User.query().filter(User.active == True)\
.filter(User.username == value).one()
raise formencode.Invalid(_('This username is not valid'),
value, state)
def ValidRepoName(edit, old_data):
class _ValidRepoName(formencode.validators.FancyValidator):
repo_name = value.get('repo_name')
slug = repo_name_slug(repo_name)
if slug in ['_admin', '']:
e_dict = {'repo_name': _('This repository name is disallowed')}
if value.get('repo_group'):
gr = Group.get(value.get('repo_group'))
group_path = gr.full_path
# value needs to be aware of group name in order to check
# db key
# db key This is an actuall just the name to store in the
# database
repo_name_full = group_path + Group.url_sep() + repo_name
group_path = ''
repo_name_full = repo_name
value['repo_name_full'] = repo_name_full
if old_data.get('repo_name') != repo_name_full or not edit:
if group_path != '':
if RepoModel().get_by_repo_name(repo_name_full,):
e_dict = {'repo_name':_('This repository already '
'exists in group "%s"') %
gr.group_name}
if RepoModel().get_by_repo_name(repo_name_full):
e_dict = {'repo_name':_('This repository '
'already exists')}
return _ValidRepoName
def SlugifyName():
class _SlugifyName(formencode.validators.FancyValidator):
return repo_name_slug(value)
return _SlugifyName
def ValidCloneUri():
from mercurial.httprepo import httprepository, httpsrepository
from rhodecode.lib.utils import make_ui
class _ValidCloneUri(formencode.validators.FancyValidator):
if not value:
elif value.startswith('https'):
httpsrepository(make_ui('db'), value).capabilities
raise formencode.Invalid(_('invalid clone url'), value,
elif value.startswith('http'):
httprepository(make_ui('db'), value).capabilities
raise formencode.Invalid(_('Invalid clone url, provide a '
'valid clone http\s url'), value,
return _ValidCloneUri
def ValidForkType(old_data):
class _ValidForkType(formencode.validators.FancyValidator):
if old_data['repo_type'] != value:
raise formencode.Invalid(_('Fork have to be the same '
'type as original'), value, state)
return _ValidForkType
class ValidPerms(formencode.validators.FancyValidator):
messages = {'perm_new_member_name':_('This username or users group name'
' is not valid')}
perms_update = []
perms_new = []
#build a list of permission to update and new permission to create
for k, v in value.items():
#means new added member to permissions
if k.startswith('perm_new_member'):
new_perm = value.get('perm_new_member', False)
new_member = value.get('perm_new_member_name', False)
new_type = value.get('perm_new_member_type')
if new_member and new_perm:
if (new_member, new_perm, new_type) not in perms_new:
perms_new.append((new_member, new_perm, new_type))
elif k.startswith('u_perm_') or k.startswith('g_perm_'):
member = k[7:]
t = {'u':'user',
'g':'users_group'}[k[0]]
if member == 'default':
if value['private']:
#set none for default when updating to private repo
v = 'repository.none'
perms_update.append((member, v, t))
value['perms_updates'] = perms_update
value['perms_new'] = perms_new
#update permissions
for k, v, t in perms_new:
if t is 'user':
.filter(User.username == k).one()
if t is 'users_group':
self.user_db = UsersGroup.query()\
.filter(UsersGroup.users_group_active == True)\
.filter(UsersGroup.users_group_name == k).one()
msg = self.message('perm_new_member_name',
state=State_obj)
raise formencode.Invalid(msg, value, state,
error_dict={'perm_new_member_name':msg})
class ValidSettings(formencode.validators.FancyValidator):
#settings form can't edit user
if value.has_key('user'):
del['value']['user']
class ValidPath(formencode.validators.FancyValidator):
if not os.path.isdir(value):
msg = _('This is not a valid path')
error_dict={'paths_root_path':msg})
def UniqSystemEmail(old_data):
class _UniqSystemEmail(formencode.validators.FancyValidator):
value = value.lower()
if old_data.get('email') != value:
user = User.query().filter(User.email == value).scalar()
if user:
raise formencode.Invalid(
_("This e-mail address is already taken"),
return _UniqSystemEmail
class ValidSystemEmail(formencode.validators.FancyValidator):
if user is None:
raise formencode.Invalid(_("This e-mail address doesn't exist.") ,
class LdapLibValidator(formencode.validators.FancyValidator):
import ldap
except ImportError:
raise LdapImportError
class AttrLoginValidator(formencode.validators.FancyValidator):
if not value or not isinstance(value, (str, unicode)):
raise formencode.Invalid(_("The LDAP Login attribute of the CN "
"must be specified - this is the name "
"of the attribute that is equivalent "
"to 'username'"),
@@ -9,363 +9,368 @@
import os
from datetime import datetime
from sqlalchemy.orm import joinedload, make_transient
from vcs.utils.lazy import LazyProperty
from vcs.backends import get_backend
from rhodecode.model import BaseModel
from rhodecode.model.caching_query import FromCache
from rhodecode.model.db import Repository, RepoToPerm, User, Permission, \
Statistics, UsersGroup, UsersGroupRepoToPerm, RhodeCodeUi, Group
class RepoModel(BaseModel):
def repos_path(self):
"""Get's the repositories root path from database
q = self.sa.query(RhodeCodeUi).filter(RhodeCodeUi.ui_key == '/').one()
return q.ui_value
def get(self, repo_id, cache=False):
repo = self.sa.query(Repository)\
.filter(Repository.repo_id == repo_id)
if cache:
repo = repo.options(FromCache("sql_cache_short",
"get_repo_%s" % repo_id))
return repo.scalar()
def get_by_repo_name(self, repo_name, cache=False):
.filter(Repository.repo_name == repo_name)
"get_repo_%s" % repo_name))
def get_full(self, repo_name, cache=False, invalidate=False):
.options(joinedload(Repository.fork))\
.options(joinedload(Repository.user))\
.options(joinedload(Repository.group))\
.filter(Repository.repo_name == repo_name)\
repo = repo.options(FromCache("sql_cache_long",
"get_repo_full_%s" % repo_name))
if invalidate and cache:
repo.invalidate()
ret = repo.scalar()
#make transient for sake of errors
make_transient(ret)
for k in ['fork', 'user', 'group']:
attr = getattr(ret, k, False)
if attr:
make_transient(attr)
return ret
def get_users_js(self):
users = self.sa.query(User).filter(User.active == True).all()
u_tmpl = '''{id:%s, fname:"%s", lname:"%s", nname:"%s"},'''
users_array = '[%s]' % '\n'.join([u_tmpl % (u.user_id, u.name,
u.lastname, u.username)
for u in users])
return users_array
def get_users_groups_js(self):
users_groups = self.sa.query(UsersGroup)\
.filter(UsersGroup.users_group_active == True).all()
g_tmpl = '''{id:%s, grname:"%s",grmembers:"%s"},'''
users_groups_array = '[%s]' % '\n'.join([g_tmpl % \
(gr.users_group_id, gr.users_group_name,
len(gr.members))
for gr in users_groups])
return users_groups_array
def update(self, repo_name, form_data):
cur_repo = self.get_by_repo_name(repo_name, cache=False)
user_model = UserModel(self.sa)
for member, perm, member_type in form_data['perms_updates']:
if member_type == 'user':
r2p = self.sa.query(RepoToPerm)\
.filter(RepoToPerm.user == user_model.
get_by_username(member))\
.filter(RepoToPerm.repository == cur_repo)\
.one()
r2p.permission = self.sa.query(Permission)\
.filter(Permission.permission_name ==
perm).scalar()
self.sa.add(r2p)
g2p = self.sa.query(UsersGroupRepoToPerm)\
.filter(UsersGroupRepoToPerm.users_group ==
UsersGroup.get_by_group_name(member))\
.filter(UsersGroupRepoToPerm.repository ==
cur_repo).one()
g2p.permission = self.sa.query(Permission)\
self.sa.add(g2p)
#set new permissions
for member, perm, member_type in form_data['perms_new']:
r2p = RepoToPerm()
r2p.repository = cur_repo
r2p.user = user_model.get_by_username(member)
.filter(Permission.
permission_name == perm)\
.scalar()
g2p = UsersGroupRepoToPerm()
g2p.repository = cur_repo
g2p.users_group = UsersGroup.get_by_group_name(member)
#update current repo
for k, v in form_data.items():
if k == 'user':
cur_repo.user = user_model.get_by_username(v)
elif k == 'repo_name':
cur_repo.repo_name = form_data['repo_name_full']
elif k == 'repo_group' and v:
cur_repo.group_id = v
setattr(cur_repo, k, v)
self.sa.add(cur_repo)
if repo_name != form_data['repo_name_full']:
# rename repository
self.__rename_repo(old=repo_name,
new=form_data['repo_name_full'])
self.sa.commit()
self.sa.rollback()
raise
def create(self, form_data, cur_user, just_db=False, fork=False):
if fork:
#force str since hg doesn't go with unicode
repo_name = str(form_data['fork_name'])
org_name = str(form_data['repo_name'])
org_full_name = str(form_data['repo_name_full'])
org_name = repo_name = str(form_data['repo_name'])
repo_name_full = form_data['repo_name_full']
new_repo = Repository()
new_repo.enable_statistics = False
if k == 'repo_name':
v = repo_name
v = repo_name_full
if k == 'repo_group':
k = 'group_id'
setattr(new_repo, k, v)
parent_repo = self.sa.query(Repository)\
.filter(Repository.repo_name == org_name).scalar()
.filter(Repository.repo_name == org_full_name).scalar()
new_repo.fork = parent_repo
new_repo.user_id = cur_user.user_id
self.sa.add(new_repo)
#create default permission
repo_to_perm = RepoToPerm()
default = 'repository.read'
for p in UserModel(self.sa).get_by_username('default',
cache=False).user_perms:
if p.permission.permission_name.startswith('repository.'):
default = p.permission.permission_name
break
default_perm = 'repository.none' if form_data['private'] else default
repo_to_perm.permission_id = self.sa.query(Permission)\
.filter(Permission.permission_name == default_perm)\
.one().permission_id
repo_to_perm.repository = new_repo
repo_to_perm.user_id = UserModel(self.sa)\
.get_by_username('default', cache=False).user_id
self.sa.add(repo_to_perm)
if not just_db:
self.__create_repo(repo_name, form_data['repo_type'],
form_data['repo_group'],
form_data['clone_uri'])
#now automatically start following this repository as owner
ScmModel(self.sa).toggle_following_repo(new_repo.repo_id,
cur_user.user_id)
def create_fork(self, form_data, cur_user):
from rhodecode.lib.celerylib import tasks, run_task
run_task(tasks.create_repo_fork, form_data, cur_user)
def delete(self, repo):
self.sa.delete(repo)
self.__delete_repo(repo)
def delete_perm_user(self, form_data, repo_name):
self.sa.query(RepoToPerm)\
.filter(RepoToPerm.repository \
== self.get_by_repo_name(repo_name))\
.filter(RepoToPerm.user_id == form_data['user_id']).delete()
def delete_perm_users_group(self, form_data, repo_name):
self.sa.query(UsersGroupRepoToPerm)\
.filter(UsersGroupRepoToPerm.repository \
.filter(UsersGroupRepoToPerm.users_group_id \
== form_data['users_group_id']).delete()
def delete_stats(self, repo_name):
self.sa.query(Statistics)\
.filter(Statistics.repository == \
self.get_by_repo_name(repo_name)).delete()
def __create_repo(self, repo_name, alias, new_parent_id, clone_uri=False):
makes repository on filesystem it's group aware
makes repository on filesystem. It's group aware means it'll create
a repository within a group, and alter the paths accordingly of
group location
:param alias:
:param parent_id:
:param clone_uri:
from rhodecode.lib.utils import check_repo
if new_parent_id:
paths = Group.get(new_parent_id).full_path.split(Group.url_sep())
new_parent_path = os.sep.join(paths)
new_parent_path = ''
repo_path = os.path.join(self.repos_path, new_parent_path, repo_name)
if check_repo(repo_name, self.repos_path):
log.info('creating repo %s in %s @ %s', repo_name, repo_path,
clone_uri)
backend = get_backend(alias)
backend(repo_path, create=True, src_url=clone_uri)
def __rename_repo(self, old, new):
renames repository on filesystem
:param old: old name
:param new: new name
log.info('renaming repo from %s to %s', old, new)
old_path = os.path.join(self.repos_path, old)
new_path = os.path.join(self.repos_path, new)
if os.path.isdir(new_path):
raise Exception('Was trying to rename to already existing dir %s',
new_path)
shutil.move(old_path, new_path)
def __delete_repo(self, repo):
removes repo from filesystem, the removal is acctually made by
added rm__ prefix into dir, and rename internat .hg/.git dirs so this
repository is no longer valid for rhodecode, can be undeleted later on
by reverting the renames on this repository
:param repo: repo object
rm_path = os.path.join(self.repos_path, repo.repo_name)
log.info("Removing %s", rm_path)
#disable hg/git
alias = repo.repo_type
shutil.move(os.path.join(rm_path, '.%s' % alias),
os.path.join(rm_path, 'rm__.%s' % alias))
#disable repo
shutil.move(rm_path, os.path.join(self.repos_path, 'rm__%s__%s' \
% (datetime.today()\
.strftime('%Y%m%d_%H%M%S_%f'),
repo.repo_name)))
Status change: