+ }
+ error : null
+ """
+
+ if not has_superadmin_permission(apiuser):
+ raise JSONRPCForbidden()
+
+ path = VcsSettingsModel().get_repos_location()
+ return {"path": path}
+
+
+@jsonrpc_method()
def get_ip(request, apiuser, userid=Optional(OAttr('apiuser'))):
"""
Displays the IP Address as seen from the |RCE| server.
diff --git a/rhodecode/api/views/user_group_api.py b/rhodecode/api/views/user_group_api.py
--- a/rhodecode/api/views/user_group_api.py
+++ b/rhodecode/api/views/user_group_api.py
@@ -464,6 +464,7 @@ def add_user_to_user_group(request, apiu
raise JSONRPCError('user group `%s` does not exist' % (
usergroupid,))
+ old_values = user_group.get_api_data()
try:
ugm = UserGroupModel().add_user_to_group(user_group, user)
success = True if ugm is not True else False
@@ -474,7 +475,8 @@ def add_user_to_user_group(request, apiu
if success:
user_data = user.get_api_data()
audit_logger.store_api(
- 'user_group.edit.member.add', action_data={'user': user_data},
+ 'user_group.edit.member.add',
+ action_data={'user': user_data, 'old_data': old_values},
user=apiuser)
Session().commit()
@@ -534,6 +536,7 @@ def remove_user_from_user_group(request,
raise JSONRPCError(
'user group `%s` does not exist' % (usergroupid,))
+ old_values = user_group.get_api_data()
try:
success = UserGroupModel().remove_user_from_group(user_group, user)
msg = 'removed member `%s` from user group `%s`' % (
@@ -543,7 +546,8 @@ def remove_user_from_user_group(request,
if success:
user_data = user.get_api_data()
audit_logger.store_api(
- 'user_group.edit.member.delete', action_data={'user': user_data},
+ 'user_group.edit.member.delete',
+ action_data={'user': user_data, 'old_data': old_values},
user=apiuser)
Session().commit()
diff --git a/rhodecode/apps/_base/__init__.py b/rhodecode/apps/_base/__init__.py
--- a/rhodecode/apps/_base/__init__.py
+++ b/rhodecode/apps/_base/__init__.py
@@ -20,16 +20,17 @@
import time
import logging
-from pylons import tmpl_context as c
+import operator
+
from pyramid.httpexceptions import HTTPFound
from rhodecode.lib import helpers as h
-from rhodecode.lib.utils import PartialRenderer
from rhodecode.lib.utils2 import StrictAttributeDict, safe_int, datetime_to_time
from rhodecode.lib.vcs.exceptions import RepositoryRequirementError
-from rhodecode.lib.ext_json import json
from rhodecode.model import repo
from rhodecode.model import repo_group
+from rhodecode.model import user_group
+from rhodecode.model import user
from rhodecode.model.db import User
from rhodecode.model.scm import ScmModel
@@ -39,6 +40,19 @@ log = logging.getLogger(__name__)
ADMIN_PREFIX = '/_admin'
STATIC_FILE_PREFIX = '/_static'
+URL_NAME_REQUIREMENTS = {
+ # group name can have a slash in them, but they must not end with a slash
+ 'group_name': r'.*?[^/]',
+ 'repo_group_name': r'.*?[^/]',
+ # repo names can have a slash in them, but they must not end with a slash
+ 'repo_name': r'.*?[^/]',
+ # file path eats up everything at the end
+ 'f_path': r'.*',
+ # reference types
+ 'source_ref_type': '(branch|book|tag|rev|\%\(source_ref_type\)s)',
+ 'target_ref_type': '(branch|book|tag|rev|\%\(target_ref_type\)s)',
+}
+
def add_route_with_slash(config,name, pattern, **kw):
config.add_route(name, pattern, **kw)
@@ -46,6 +60,17 @@ def add_route_with_slash(config,name, pa
config.add_route(name + '_slash', pattern + '/', **kw)
+def add_route_requirements(route_path, requirements=URL_NAME_REQUIREMENTS):
+ """
+ Adds regex requirements to pyramid routes using a mapping dict
+ e.g::
+ add_route_requirements('{repo_name}/settings')
+ """
+ for key, regex in requirements.items():
+ route_path = route_path.replace('{%s}' % key, '{%s:%s}' % (key, regex))
+ return route_path
+
+
def get_format_ref_id(repo):
"""Returns a `repo` specific reference formatter function"""
if h.is_svn(repo):
@@ -105,23 +130,50 @@ class BaseAppView(object):
raise HTTPFound(
self.request.route_path('my_account_password'))
+ def _log_creation_exception(self, e, repo_name):
+ _ = self.request.translate
+ reason = None
+ if len(e.args) == 2:
+ reason = e.args[1]
+
+ if reason == 'INVALID_CERTIFICATE':
+ log.exception(
+ 'Exception creating a repository: invalid certificate')
+ msg = (_('Error creating repository %s: invalid certificate')
+ % repo_name)
+ else:
+ log.exception("Exception creating a repository")
+ msg = (_('Error creating repository %s')
+ % repo_name)
+ return msg
+
def _get_local_tmpl_context(self, include_app_defaults=False):
c = TemplateArgs()
c.auth_user = self.request.user
+ # TODO(marcink): migrate the usage of c.rhodecode_user to c.auth_user
+ c.rhodecode_user = self.request.user
+
if include_app_defaults:
# NOTE(marcink): after full pyramid migration include_app_defaults
# should be turned on by default
from rhodecode.lib.base import attach_context_attributes
attach_context_attributes(c, self.request, self.request.user.user_id)
+
return c
def _register_global_c(self, tmpl_args):
"""
Registers attributes to pylons global `c`
"""
+
# TODO(marcink): remove once pyramid migration is finished
- for k, v in tmpl_args.items():
- setattr(c, k, v)
+ from pylons import tmpl_context as c
+ try:
+ for k, v in tmpl_args.items():
+ setattr(c, k, v)
+ except TypeError:
+ log.exception('Failed to register pylons C')
+ pass
def _get_template_context(self, tmpl_args):
self._register_global_c(tmpl_args)
@@ -129,6 +181,10 @@ class BaseAppView(object):
local_tmpl_args = {
'defaults': {},
'errors': {},
+ # register a fake 'c' to be used in templates instead of global
+ # pylons c, after migration to pyramid we should rename it to 'c'
+ # make sure we replace usage of _c in templates too
+ '_c': tmpl_args
}
local_tmpl_args.update(tmpl_args)
return local_tmpl_args
@@ -160,6 +216,7 @@ class RepoAppView(BaseAppView):
self.db_repo_name, error.message)
def _get_local_tmpl_context(self, include_app_defaults=False):
+ _ = self.request.translate
c = super(RepoAppView, self)._get_local_tmpl_context(
include_app_defaults=include_app_defaults)
@@ -174,9 +231,70 @@ class RepoAppView(BaseAppView):
except RepositoryRequirementError as e:
c.repository_requirements_missing = True
self._handle_missing_requirements(e)
+ self.rhodecode_vcs_repo = None
+
+ if (not c.repository_requirements_missing
+ and self.rhodecode_vcs_repo is None):
+ # unable to fetch this repo as vcs instance, report back to user
+ h.flash(_(
+ "The repository `%(repo_name)s` cannot be loaded in filesystem. "
+ "Please check if it exist, or is not damaged.") %
+ {'repo_name': c.repo_name},
+ category='error', ignore_duplicate=True)
+ raise HTTPFound(h.route_path('home'))
return c
+ def _get_f_path(self, matchdict, default=None):
+ f_path = matchdict.get('f_path')
+ if f_path:
+ # fix for multiple initial slashes that causes errors for GIT
+ return f_path.lstrip('/')
+
+ return default
+
+
+class RepoGroupAppView(BaseAppView):
+ def __init__(self, context, request):
+ super(RepoGroupAppView, self).__init__(context, request)
+ self.db_repo_group = request.db_repo_group
+ self.db_repo_group_name = self.db_repo_group.group_name
+
+ def _revoke_perms_on_yourself(self, form_result):
+ _updates = filter(lambda u: self._rhodecode_user.user_id == int(u[0]),
+ form_result['perm_updates'])
+ _additions = filter(lambda u: self._rhodecode_user.user_id == int(u[0]),
+ form_result['perm_additions'])
+ _deletions = filter(lambda u: self._rhodecode_user.user_id == int(u[0]),
+ form_result['perm_deletions'])
+ admin_perm = 'group.admin'
+ if _updates and _updates[0][1] != admin_perm or \
+ _additions and _additions[0][1] != admin_perm or \
+ _deletions and _deletions[0][1] != admin_perm:
+ return True
+ return False
+
+
+class UserGroupAppView(BaseAppView):
+ def __init__(self, context, request):
+ super(UserGroupAppView, self).__init__(context, request)
+ self.db_user_group = request.db_user_group
+ self.db_user_group_name = self.db_user_group.users_group_name
+
+
+class UserAppView(BaseAppView):
+ def __init__(self, context, request):
+ super(UserAppView, self).__init__(context, request)
+ self.db_user = request.db_user
+ self.db_user_id = self.db_user.user_id
+
+ _ = self.request.translate
+ if not request.db_user_supports_default:
+ if self.db_user.username == User.DEFAULT_USER:
+ h.flash(_("Editing user `{}` is disabled.".format(
+ User.DEFAULT_USER)), category='warning')
+ raise HTTPFound(h.route_path('users'))
+
class DataGridAppView(object):
"""
@@ -203,6 +321,15 @@ class DataGridAppView(object):
draw = safe_int(request.GET.get('draw'))
return draw, start, length
+ def _get_order_col(self, order_by, model):
+ if isinstance(order_by, basestring):
+ try:
+ return operator.attrgetter(order_by)(model)
+ except AttributeError:
+ return None
+ else:
+ return order_by
+
class BaseReferencesView(RepoAppView):
"""
@@ -211,35 +338,48 @@ class BaseReferencesView(RepoAppView):
def load_default_context(self):
c = self._get_local_tmpl_context()
- # TODO(marcink): remove repo_info and use c.rhodecode_db_repo instead
- c.repo_info = self.db_repo
-
self._register_global_c(c)
return c
def load_refs_context(self, ref_items, partials_template):
- _render = PartialRenderer(partials_template)
- _data = []
+ _render = self.request.get_partial_renderer(partials_template)
pre_load = ["author", "date", "message"]
is_svn = h.is_svn(self.rhodecode_vcs_repo)
+ is_hg = h.is_hg(self.rhodecode_vcs_repo)
+
format_ref_id = get_format_ref_id(self.rhodecode_vcs_repo)
+ closed_refs = {}
+ if is_hg:
+ closed_refs = self.rhodecode_vcs_repo.branches_closed
+
+ data = []
for ref_name, commit_id in ref_items:
commit = self.rhodecode_vcs_repo.get_commit(
commit_id=commit_id, pre_load=pre_load)
+ closed = ref_name in closed_refs
# TODO: johbo: Unify generation of reference links
use_commit_id = '/' in ref_name or is_svn
- files_url = h.url(
- 'files_home',
- repo_name=c.repo_name,
- f_path=ref_name if is_svn else '',
- revision=commit_id if use_commit_id else ref_name,
- at=ref_name)
+
+ if use_commit_id:
+ files_url = h.route_path(
+ 'repo_files',
+ repo_name=self.db_repo_name,
+ f_path=ref_name if is_svn else '',
+ commit_id=commit_id)
- _data.append({
- "name": _render('name', ref_name, files_url),
+ else:
+ files_url = h.route_path(
+ 'repo_files',
+ repo_name=self.db_repo_name,
+ f_path=ref_name if is_svn else '',
+ commit_id=ref_name,
+ _query=dict(at=ref_name))
+
+ data.append({
+ "name": _render('name', ref_name, files_url, closed),
"name_raw": ref_name,
"date": _render('date', commit.date),
"date_raw": datetime_to_time(commit.date),
@@ -250,8 +390,8 @@ class BaseReferencesView(RepoAppView):
"compare": _render(
'compare', format_ref_id(ref_name, commit.raw_id)),
})
- c.has_references = bool(_data)
- c.data = json.dumps(_data)
+
+ return data
class RepoRoutePredicate(object):
@@ -273,14 +413,22 @@ class RepoRoutePredicate(object):
repo_model = repo.RepoModel()
by_name_match = repo_model.get_by_repo_name(repo_name, cache=True)
+ def redirect_if_creating(db_repo):
+ if db_repo.repo_state in [repo.Repository.STATE_PENDING]:
+ raise HTTPFound(
+ request.route_path('repo_creating',
+ repo_name=db_repo.repo_name))
+
if by_name_match:
# register this as request object we can re-use later
request.db_repo = by_name_match
+ redirect_if_creating(by_name_match)
return True
by_id_match = repo_model.get_repo_by_id(repo_name)
if by_id_match:
request.db_repo = by_id_match
+ redirect_if_creating(by_id_match)
return True
return False
@@ -348,6 +496,79 @@ class RepoGroupRoutePredicate(object):
return False
+class UserGroupRoutePredicate(object):
+ def __init__(self, val, config):
+ self.val = val
+
+ def text(self):
+ return 'user_group_route = %s' % self.val
+
+ phash = text
+
+ def __call__(self, info, request):
+ if hasattr(request, 'vcs_call'):
+ # skip vcs calls
+ return
+
+ user_group_id = info['match']['user_group_id']
+ user_group_model = user_group.UserGroup()
+ by_id_match = user_group_model.get(
+ user_group_id, cache=True)
+
+ if by_id_match:
+ # register this as request object we can re-use later
+ request.db_user_group = by_id_match
+ return True
+
+ return False
+
+
+class UserRoutePredicateBase(object):
+ supports_default = None
+
+ def __init__(self, val, config):
+ self.val = val
+
+ def text(self):
+ raise NotImplementedError()
+
+ def __call__(self, info, request):
+ if hasattr(request, 'vcs_call'):
+ # skip vcs calls
+ return
+
+ user_id = info['match']['user_id']
+ user_model = user.User()
+ by_id_match = user_model.get(
+ user_id, cache=True)
+
+ if by_id_match:
+ # register this as request object we can re-use later
+ request.db_user = by_id_match
+ request.db_user_supports_default = self.supports_default
+ return True
+
+ return False
+
+
+class UserRoutePredicate(UserRoutePredicateBase):
+ supports_default = False
+
+ def text(self):
+ return 'user_route = %s' % self.val
+
+ phash = text
+
+
+class UserRouteWithDefaultPredicate(UserRoutePredicateBase):
+ supports_default = True
+
+ def text(self):
+ return 'user_with_default_route = %s' % self.val
+
+ phash = text
+
+
def includeme(config):
config.add_route_predicate(
'repo_route', RepoRoutePredicate)
@@ -355,3 +576,9 @@ def includeme(config):
'repo_accepted_types', RepoTypeRoutePredicate)
config.add_route_predicate(
'repo_group_route', RepoGroupRoutePredicate)
+ config.add_route_predicate(
+ 'user_group_route', UserGroupRoutePredicate)
+ config.add_route_predicate(
+ 'user_route_with_default', UserRouteWithDefaultPredicate)
+ config.add_route_predicate(
+ 'user_route', UserRoutePredicate)
\ No newline at end of file
diff --git a/rhodecode/apps/admin/__init__.py b/rhodecode/apps/admin/__init__.py
--- a/rhodecode/apps/admin/__init__.py
+++ b/rhodecode/apps/admin/__init__.py
@@ -34,14 +34,18 @@ def admin_routes(config):
pattern='/audit_logs')
config.add_route(
+ name='admin_audit_log_entry',
+ pattern='/audit_logs/{audit_log_id}')
+
+ config.add_route(
name='pull_requests_global_0', # backward compat
- pattern='/pull_requests/{pull_request_id:[0-9]+}')
+ pattern='/pull_requests/{pull_request_id:\d+}')
config.add_route(
name='pull_requests_global_1', # backward compat
- pattern='/pull-requests/{pull_request_id:[0-9]+}')
+ pattern='/pull-requests/{pull_request_id:\d+}')
config.add_route(
name='pull_requests_global',
- pattern='/pull-request/{pull_request_id:[0-9]+}')
+ pattern='/pull-request/{pull_request_id:\d+}')
config.add_route(
name='admin_settings_open_source',
@@ -64,11 +68,66 @@ def admin_routes(config):
name='admin_settings_sessions_cleanup',
pattern='/settings/sessions/cleanup')
+ config.add_route(
+ name='admin_settings_process_management',
+ pattern='/settings/process_management')
+ config.add_route(
+ name='admin_settings_process_management_signal',
+ pattern='/settings/process_management/signal')
+
+ # default settings
+ config.add_route(
+ name='admin_defaults_repositories',
+ pattern='/defaults/repositories')
+ config.add_route(
+ name='admin_defaults_repositories_update',
+ pattern='/defaults/repositories/update')
+
# global permissions
+
+ config.add_route(
+ name='admin_permissions_application',
+ pattern='/permissions/application')
+ config.add_route(
+ name='admin_permissions_application_update',
+ pattern='/permissions/application/update')
+
+ config.add_route(
+ name='admin_permissions_global',
+ pattern='/permissions/global')
+ config.add_route(
+ name='admin_permissions_global_update',
+ pattern='/permissions/global/update')
+
+ config.add_route(
+ name='admin_permissions_object',
+ pattern='/permissions/object')
+ config.add_route(
+ name='admin_permissions_object_update',
+ pattern='/permissions/object/update')
+
config.add_route(
name='admin_permissions_ips',
pattern='/permissions/ips')
+ config.add_route(
+ name='admin_permissions_overview',
+ pattern='/permissions/overview')
+
+ config.add_route(
+ name='admin_permissions_auth_token_access',
+ pattern='/permissions/auth_token_access')
+
+ config.add_route(
+ name='admin_permissions_ssh_keys',
+ pattern='/permissions/ssh_keys')
+ config.add_route(
+ name='admin_permissions_ssh_keys_data',
+ pattern='/permissions/ssh_keys/data')
+ config.add_route(
+ name='admin_permissions_ssh_keys_update',
+ pattern='/permissions/ssh_keys/update')
+
# users admin
config.add_route(
name='users',
@@ -78,52 +137,176 @@ def admin_routes(config):
name='users_data',
pattern='/users_data')
+ config.add_route(
+ name='users_create',
+ pattern='/users/create')
+
+ config.add_route(
+ name='users_new',
+ pattern='/users/new')
+
+ # user management
+ config.add_route(
+ name='user_edit',
+ pattern='/users/{user_id:\d+}/edit',
+ user_route=True)
+ config.add_route(
+ name='user_edit_advanced',
+ pattern='/users/{user_id:\d+}/edit/advanced',
+ user_route=True)
+ config.add_route(
+ name='user_edit_global_perms',
+ pattern='/users/{user_id:\d+}/edit/global_permissions',
+ user_route=True)
+ config.add_route(
+ name='user_edit_global_perms_update',
+ pattern='/users/{user_id:\d+}/edit/global_permissions/update',
+ user_route=True)
+ config.add_route(
+ name='user_update',
+ pattern='/users/{user_id:\d+}/update',
+ user_route=True)
+ config.add_route(
+ name='user_delete',
+ pattern='/users/{user_id:\d+}/delete',
+ user_route=True)
+ config.add_route(
+ name='user_force_password_reset',
+ pattern='/users/{user_id:\d+}/password_reset',
+ user_route=True)
+ config.add_route(
+ name='user_create_personal_repo_group',
+ pattern='/users/{user_id:\d+}/create_repo_group',
+ user_route=True)
+
# user auth tokens
config.add_route(
name='edit_user_auth_tokens',
- pattern='/users/{user_id:\d+}/edit/auth_tokens')
+ pattern='/users/{user_id:\d+}/edit/auth_tokens',
+ user_route=True)
config.add_route(
name='edit_user_auth_tokens_add',
- pattern='/users/{user_id:\d+}/edit/auth_tokens/new')
+ pattern='/users/{user_id:\d+}/edit/auth_tokens/new',
+ user_route=True)
config.add_route(
name='edit_user_auth_tokens_delete',
- pattern='/users/{user_id:\d+}/edit/auth_tokens/delete')
+ pattern='/users/{user_id:\d+}/edit/auth_tokens/delete',
+ user_route=True)
+
+ # user ssh keys
+ config.add_route(
+ name='edit_user_ssh_keys',
+ pattern='/users/{user_id:\d+}/edit/ssh_keys',
+ user_route=True)
+ config.add_route(
+ name='edit_user_ssh_keys_generate_keypair',
+ pattern='/users/{user_id:\d+}/edit/ssh_keys/generate',
+ user_route=True)
+ config.add_route(
+ name='edit_user_ssh_keys_add',
+ pattern='/users/{user_id:\d+}/edit/ssh_keys/new',
+ user_route=True)
+ config.add_route(
+ name='edit_user_ssh_keys_delete',
+ pattern='/users/{user_id:\d+}/edit/ssh_keys/delete',
+ user_route=True)
# user emails
config.add_route(
name='edit_user_emails',
- pattern='/users/{user_id:\d+}/edit/emails')
+ pattern='/users/{user_id:\d+}/edit/emails',
+ user_route=True)
config.add_route(
name='edit_user_emails_add',
- pattern='/users/{user_id:\d+}/edit/emails/new')
+ pattern='/users/{user_id:\d+}/edit/emails/new',
+ user_route=True)
config.add_route(
name='edit_user_emails_delete',
- pattern='/users/{user_id:\d+}/edit/emails/delete')
+ pattern='/users/{user_id:\d+}/edit/emails/delete',
+ user_route=True)
# user IPs
config.add_route(
name='edit_user_ips',
- pattern='/users/{user_id:\d+}/edit/ips')
+ pattern='/users/{user_id:\d+}/edit/ips',
+ user_route=True)
config.add_route(
name='edit_user_ips_add',
- pattern='/users/{user_id:\d+}/edit/ips/new')
+ pattern='/users/{user_id:\d+}/edit/ips/new',
+ user_route_with_default=True) # enabled for default user too
config.add_route(
name='edit_user_ips_delete',
- pattern='/users/{user_id:\d+}/edit/ips/delete')
+ pattern='/users/{user_id:\d+}/edit/ips/delete',
+ user_route_with_default=True) # enabled for default user too
- # user groups management
+ # user perms
+ config.add_route(
+ name='edit_user_perms_summary',
+ pattern='/users/{user_id:\d+}/edit/permissions_summary',
+ user_route=True)
+ config.add_route(
+ name='edit_user_perms_summary_json',
+ pattern='/users/{user_id:\d+}/edit/permissions_summary/json',
+ user_route=True)
+
+ # user user groups management
config.add_route(
name='edit_user_groups_management',
- pattern='/users/{user_id:\d+}/edit/groups_management')
+ pattern='/users/{user_id:\d+}/edit/groups_management',
+ user_route=True)
config.add_route(
name='edit_user_groups_management_updates',
- pattern='/users/{user_id:\d+}/edit/edit_user_groups_management/updates')
+ pattern='/users/{user_id:\d+}/edit/edit_user_groups_management/updates',
+ user_route=True)
# user audit logs
config.add_route(
name='edit_user_audit_logs',
- pattern='/users/{user_id:\d+}/edit/audit')
+ pattern='/users/{user_id:\d+}/edit/audit', user_route=True)
+
+ # user-groups admin
+ config.add_route(
+ name='user_groups',
+ pattern='/user_groups')
+
+ config.add_route(
+ name='user_groups_data',
+ pattern='/user_groups_data')
+
+ config.add_route(
+ name='user_groups_new',
+ pattern='/user_groups/new')
+
+ config.add_route(
+ name='user_groups_create',
+ pattern='/user_groups/create')
+
+ # repos admin
+ config.add_route(
+ name='repos',
+ pattern='/repos')
+
+ config.add_route(
+ name='repo_new',
+ pattern='/repos/new')
+
+ config.add_route(
+ name='repo_create',
+ pattern='/repos/create')
+
+ # repo groups admin
+ config.add_route(
+ name='repo_groups',
+ pattern='/repo_groups')
+
+ config.add_route(
+ name='repo_group_new',
+ pattern='/repo_group/new')
+
+ config.add_route(
+ name='repo_group_create',
+ pattern='/repo_group/create')
def includeme(config):
@@ -139,4 +322,4 @@ def includeme(config):
config.include(admin_routes, route_prefix=ADMIN_PREFIX)
# Scan module for configuration decorators.
- config.scan()
+ config.scan('.views', ignore='.tests')
diff --git a/rhodecode/apps/admin/navigation.py b/rhodecode/apps/admin/navigation.py
--- a/rhodecode/apps/admin/navigation.py
+++ b/rhodecode/apps/admin/navigation.py
@@ -22,7 +22,6 @@
import logging
import collections
-from pylons import url
from zope.interface import implementer
from rhodecode.apps.admin.interfaces import IAdminNavigationRegistry
@@ -64,6 +63,7 @@ class NavEntry(object):
pyramid_request = get_current_request()
return pyramid_request.route_path(self.view_name)
else:
+ from pylons import url
return url(self.view_name)
def get_localized_name(self, request):
@@ -94,6 +94,8 @@ class NavigationRegistry(object):
'global_integrations_home', pyramid=True),
NavEntry('system', _('System Info'),
'admin_settings_system', pyramid=True),
+ NavEntry('process_management', _('Processes'),
+ 'admin_settings_process_management', pyramid=True),
NavEntry('sessions', _('User Sessions'),
'admin_settings_sessions', pyramid=True),
NavEntry('open_source', _('Open Source Licenses'),
diff --git a/rhodecode/tests/functional/test_admin_defaults.py b/rhodecode/apps/admin/tests/test_admin_defaults.py
rename from rhodecode/tests/functional/test_admin_defaults.py
rename to rhodecode/apps/admin/tests/test_admin_defaults.py
--- a/rhodecode/tests/functional/test_admin_defaults.py
+++ b/rhodecode/apps/admin/tests/test_admin_defaults.py
@@ -20,15 +20,31 @@
import pytest
-from rhodecode.tests import assert_session_flash, url
+from rhodecode.tests import assert_session_flash
from rhodecode.model.settings import SettingsModel
+def route_path(name, params=None, **kwargs):
+ import urllib
+ from rhodecode.apps._base import ADMIN_PREFIX
+
+ base_url = {
+ 'admin_defaults_repositories':
+ ADMIN_PREFIX + '/defaults/repositories',
+ 'admin_defaults_repositories_update':
+ ADMIN_PREFIX + '/defaults/repositories/update',
+ }[name].format(**kwargs)
+
+ if params:
+ base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
+ return base_url
+
+
@pytest.mark.usefixtures("app")
-class TestDefaultsController:
+class TestDefaultsView(object):
def test_index(self, autologin_user):
- response = self.app.get(url('admin_defaults_repositories'))
+ response = self.app.get(route_path('admin_defaults_repositories'))
response.mustcontain('default_repo_private')
response.mustcontain('default_repo_enable_statistics')
response.mustcontain('default_repo_enable_downloads')
@@ -44,7 +60,7 @@ class TestDefaultsController:
'csrf_token': csrf_token,
}
response = self.app.post(
- url('admin_defaults_repositories'), params=params)
+ route_path('admin_defaults_repositories_update'), params=params)
assert_session_flash(response, 'Default settings updated successfully')
defs = SettingsModel().get_default_repo_settings()
@@ -61,8 +77,9 @@ class TestDefaultsController:
'csrf_token': csrf_token,
}
response = self.app.post(
- url('admin_defaults_repositories'), params=params)
+ route_path('admin_defaults_repositories_update'), params=params)
assert_session_flash(response, 'Default settings updated successfully')
+
defs = SettingsModel().get_default_repo_settings()
del params['csrf_token']
assert params == defs
diff --git a/rhodecode/tests/functional/test_admin_permissions.py b/rhodecode/apps/admin/tests/test_admin_permissions.py
rename from rhodecode/tests/functional/test_admin_permissions.py
rename to rhodecode/apps/admin/tests/test_admin_permissions.py
--- a/rhodecode/tests/functional/test_admin_permissions.py
+++ b/rhodecode/apps/admin/tests/test_admin_permissions.py
@@ -18,11 +18,14 @@
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
+import mock
import pytest
from rhodecode.model.db import User, UserIpMap
+from rhodecode.model.meta import Session
from rhodecode.model.permission import PermissionModel
+from rhodecode.model.ssh_key import SshKeyModel
from rhodecode.tests import (
- TestController, url, clear_all_caches, assert_session_flash)
+ TestController, clear_all_caches, assert_session_flash)
def route_path(name, params=None, **kwargs):
@@ -36,6 +39,34 @@ def route_path(name, params=None, **kwar
ADMIN_PREFIX + '/users/{user_id}/edit/ips/new',
'edit_user_ips_delete':
ADMIN_PREFIX + '/users/{user_id}/edit/ips/delete',
+
+ 'admin_permissions_application':
+ ADMIN_PREFIX + '/permissions/application',
+ 'admin_permissions_application_update':
+ ADMIN_PREFIX + '/permissions/application/update',
+
+ 'admin_permissions_global':
+ ADMIN_PREFIX + '/permissions/global',
+ 'admin_permissions_global_update':
+ ADMIN_PREFIX + '/permissions/global/update',
+
+ 'admin_permissions_object':
+ ADMIN_PREFIX + '/permissions/object',
+ 'admin_permissions_object_update':
+ ADMIN_PREFIX + '/permissions/object/update',
+
+ 'admin_permissions_ips':
+ ADMIN_PREFIX + '/permissions/ips',
+ 'admin_permissions_overview':
+ ADMIN_PREFIX + '/permissions/overview',
+
+ 'admin_permissions_ssh_keys':
+ ADMIN_PREFIX + '/permissions/ssh_keys',
+ 'admin_permissions_ssh_keys_data':
+ ADMIN_PREFIX + '/permissions/ssh_keys/data',
+ 'admin_permissions_ssh_keys_update':
+ ADMIN_PREFIX + '/permissions/ssh_keys/update'
+
}[name].format(**kwargs)
if params:
@@ -55,7 +86,7 @@ class TestAdminPermissionsController(Tes
def test_index_application(self):
self.log_user()
- self.app.get(url('admin_permissions_application'))
+ self.app.get(route_path('admin_permissions_application'))
@pytest.mark.parametrize(
'anonymous, default_register, default_register_message, default_password_reset,'
@@ -87,7 +118,7 @@ class TestAdminPermissionsController(Tes
'default_password_reset': default_password_reset,
'default_extern_activate': default_extern_activate,
}
- response = self.app.post(url('admin_permissions_application'),
+ response = self.app.post(route_path('admin_permissions_application_update'),
params=params)
if expect_form_error:
assert response.status_int == 200
@@ -101,7 +132,7 @@ class TestAdminPermissionsController(Tes
def test_index_object(self):
self.log_user()
- self.app.get(url('admin_permissions_object'))
+ self.app.get(route_path('admin_permissions_object'))
@pytest.mark.parametrize(
'repo, repo_group, user_group, expect_error, expect_form_error', [
@@ -127,7 +158,7 @@ class TestAdminPermissionsController(Tes
'default_user_group_perm': user_group,
'overwrite_default_user_group': False,
}
- response = self.app.post(url('admin_permissions_object'),
+ response = self.app.post(route_path('admin_permissions_object_update'),
params=params)
if expect_form_error:
assert response.status_int == 200
@@ -141,7 +172,7 @@ class TestAdminPermissionsController(Tes
def test_index_global(self):
self.log_user()
- self.app.get(url('admin_permissions_global'))
+ self.app.get(route_path('admin_permissions_global'))
@pytest.mark.parametrize(
'repo_create, repo_create_write, user_group_create, repo_group_create,'
@@ -175,7 +206,7 @@ class TestAdminPermissionsController(Tes
'default_fork_create': fork_create,
'default_inherit_default_permissions': inherit_default_permissions
}
- response = self.app.post(url('admin_permissions_global'),
+ response = self.app.post(route_path('admin_permissions_global_update'),
params=params)
if expect_form_error:
assert response.status_int == 200
@@ -189,7 +220,7 @@ class TestAdminPermissionsController(Tes
def test_index_ips(self):
self.log_user()
- response = self.app.get(url('admin_permissions_ips'))
+ response = self.app.get(route_path('admin_permissions_ips'))
# TODO: Test response...
response.mustcontain('All IP addresses are allowed')
@@ -203,7 +234,7 @@ class TestAdminPermissionsController(Tes
route_path('edit_user_ips_add', user_id=default_user_id),
params={'new_ip': '127.0.0.0/24', 'csrf_token': self.csrf_token})
- response = self.app.get(url('admin_permissions_ips'))
+ response = self.app.get(route_path('admin_permissions_ips'))
response.mustcontain('127.0.0.0/24')
response.mustcontain('127.0.0.0 - 127.0.0.255')
@@ -219,11 +250,51 @@ class TestAdminPermissionsController(Tes
assert_session_flash(response, 'Removed ip address from user whitelist')
clear_all_caches()
- response = self.app.get(url('admin_permissions_ips'))
+ response = self.app.get(route_path('admin_permissions_ips'))
response.mustcontain('All IP addresses are allowed')
response.mustcontain(no=['127.0.0.0/24'])
response.mustcontain(no=['127.0.0.0 - 127.0.0.255'])
def test_index_overview(self):
self.log_user()
- self.app.get(url('admin_permissions_overview'))
+ self.app.get(route_path('admin_permissions_overview'))
+
+ def test_ssh_keys(self):
+ self.log_user()
+ self.app.get(route_path('admin_permissions_ssh_keys'), status=200)
+
+ def test_ssh_keys_data(self, user_util, xhr_header):
+ self.log_user()
+ response = self.app.get(route_path('admin_permissions_ssh_keys_data'),
+ extra_environ=xhr_header)
+ assert response.json == {u'data': [], u'draw': None,
+ u'recordsFiltered': 0, u'recordsTotal': 0}
+
+ dummy_user = user_util.create_user()
+ SshKeyModel().create(dummy_user, 'ab:cd:ef', 'KEYKEY', 'test_key')
+ Session().commit()
+ response = self.app.get(route_path('admin_permissions_ssh_keys_data'),
+ extra_environ=xhr_header)
+ assert response.json['data'][0]['fingerprint'] == 'ab:cd:ef'
+
+ def test_ssh_keys_update(self):
+ self.log_user()
+ response = self.app.post(
+ route_path('admin_permissions_ssh_keys_update'),
+ dict(csrf_token=self.csrf_token), status=302)
+
+ assert_session_flash(
+ response, 'Updated SSH keys file')
+
+ def test_ssh_keys_update_disabled(self):
+ self.log_user()
+
+ from rhodecode.apps.admin.views.permissions import AdminPermissionsView
+ with mock.patch.object(AdminPermissionsView, 'ssh_enabled',
+ return_value=False):
+ response = self.app.post(
+ route_path('admin_permissions_ssh_keys_update'),
+ dict(csrf_token=self.csrf_token), status=302)
+
+ assert_session_flash(
+ response, 'SSH key support is disabled in .ini file')
\ No newline at end of file
diff --git a/rhodecode/tests/functional/test_admin_repos.py b/rhodecode/apps/admin/tests/test_admin_repos.py
rename from rhodecode/tests/functional/test_admin_repos.py
rename to rhodecode/apps/admin/tests/test_admin_repos.py
--- a/rhodecode/tests/functional/test_admin_repos.py
+++ b/rhodecode/apps/admin/tests/test_admin_repos.py
@@ -23,55 +23,83 @@ import urllib
import mock
import pytest
+from rhodecode.apps._base import ADMIN_PREFIX
from rhodecode.lib import auth
-from rhodecode.lib.utils2 import safe_str, str2bool
+from rhodecode.lib.utils2 import safe_str
from rhodecode.lib import helpers as h
from rhodecode.model.db import (
Repository, RepoGroup, UserRepoToPerm, User, Permission)
from rhodecode.model.meta import Session
from rhodecode.model.repo import RepoModel
from rhodecode.model.repo_group import RepoGroupModel
-from rhodecode.model.settings import SettingsModel, VcsSettingsModel
from rhodecode.model.user import UserModel
from rhodecode.tests import (
- login_user_session, url, assert_session_flash, TEST_USER_ADMIN_LOGIN,
- TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS, HG_REPO, GIT_REPO,
- logout_user_session)
+ login_user_session, assert_session_flash, TEST_USER_ADMIN_LOGIN,
+ TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
from rhodecode.tests.fixture import Fixture, error_function
from rhodecode.tests.utils import AssertResponse, repo_on_filesystem
fixture = Fixture()
+def route_path(name, params=None, **kwargs):
+ import urllib
+
+ base_url = {
+ 'repos': ADMIN_PREFIX + '/repos',
+ 'repo_new': ADMIN_PREFIX + '/repos/new',
+ 'repo_create': ADMIN_PREFIX + '/repos/create',
+
+ 'repo_creating_check': '/{repo_name}/repo_creating_check',
+ }[name].format(**kwargs)
+
+ if params:
+ base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
+ return base_url
+
+
+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
+
+
@pytest.mark.usefixtures("app")
class TestAdminRepos(object):
- def test_index(self):
- self.app.get(url('repos'))
+ def test_repo_list(self, autologin_user, user_util):
+ repo = user_util.create_repo()
+ response = self.app.get(
+ route_path('repos'), status=200)
- def test_create_page_restricted(self, autologin_user, backend):
+ response.mustcontain(repo.repo_name)
+
+ def test_create_page_restricted_to_single_backend(self, autologin_user, backend):
with mock.patch('rhodecode.BACKENDS', {'git': 'git'}):
- response = self.app.get(url('new_repo'), status=200)
+ response = self.app.get(route_path('repo_new'), status=200)
assert_response = AssertResponse(response)
element = assert_response.get_element('#repo_type')
assert element.text_content() == '\ngit\n'
- def test_create_page_non_restricted(self, autologin_user, backend):
- response = self.app.get(url('new_repo'), status=200)
+ def test_create_page_non_restricted_backends(self, autologin_user, backend):
+ response = self.app.get(route_path('repo_new'), status=200)
assert_response = AssertResponse(response)
assert_response.element_contains('#repo_type', 'git')
assert_response.element_contains('#repo_type', 'svn')
assert_response.element_contains('#repo_type', 'hg')
- @pytest.mark.parametrize("suffix",
- [u'', u'xxa'], ids=['', 'non-ascii'])
+ @pytest.mark.parametrize(
+ "suffix", [u'', u'xxa'], ids=['', 'non-ascii'])
def test_create(self, autologin_user, backend, suffix, csrf_token):
repo_name_unicode = backend.new_repo_name(suffix=suffix)
repo_name = repo_name_unicode.encode('utf8')
description_unicode = u'description for newly created repo' + suffix
description = description_unicode.encode('utf8')
response = self.app.post(
- url('repos'),
+ route_path('repo_create'),
fixture._get_repo_create_params(
repo_private=False,
repo_name=repo_name,
@@ -83,12 +111,12 @@ class TestAdminRepos(object):
self.assert_repository_is_created_correctly(
repo_name, description, backend)
- def test_create_numeric(self, autologin_user, backend, csrf_token):
+ def test_create_numeric_name(self, autologin_user, backend, csrf_token):
numeric_repo = '1234'
repo_name = numeric_repo
description = 'description for newly created repo' + numeric_repo
self.app.post(
- url('repos'),
+ route_path('repo_create'),
fixture._get_repo_create_params(
repo_private=False,
repo_name=repo_name,
@@ -114,7 +142,7 @@ class TestAdminRepos(object):
[group_name, repo_name])
description = u'description for newly created repo'
self.app.post(
- url('repos'),
+ route_path('repo_create'),
fixture._get_repo_create_params(
repo_private=False,
repo_name=safe_str(repo_name),
@@ -137,7 +165,7 @@ class TestAdminRepos(object):
RepoGroupModel().delete(group_name)
Session().commit()
- def test_create_in_group_numeric(
+ def test_create_in_group_numeric_name(
self, autologin_user, backend, csrf_token):
# create GROUP
group_name = 'sometest_%s' % backend.alias
@@ -150,7 +178,7 @@ class TestAdminRepos(object):
repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
description = 'description for newly created repo'
self.app.post(
- url('repos'),
+ route_path('repo_create'),
fixture._get_repo_create_params(
repo_private=False,
repo_name=repo_name,
@@ -209,7 +237,7 @@ class TestAdminRepos(object):
repo_name = 'ingroup'
description = 'description for newly created repo'
response = self.app.post(
- url('repos'),
+ route_path('repo_create'),
fixture._get_repo_create_params(
repo_private=False,
repo_name=repo_name,
@@ -226,7 +254,7 @@ class TestAdminRepos(object):
[group_name_allowed, repo_name])
description = 'description for newly created repo'
response = self.app.post(
- url('repos'),
+ route_path('repo_create'),
fixture._get_repo_create_params(
repo_private=False,
repo_name=repo_name,
@@ -270,7 +298,7 @@ class TestAdminRepos(object):
repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
description = 'description for newly created repo'
self.app.post(
- url('repos'),
+ route_path('repo_create'),
fixture._get_repo_create_params(
repo_private=False,
repo_name=repo_name,
@@ -314,7 +342,7 @@ class TestAdminRepos(object):
repo_name = backend.new_repo_name()
response = self.app.post(
- url('repos'),
+ route_path('repo_create'),
fixture._get_repo_create_params(
repo_private=False,
repo_name=repo_name,
@@ -342,7 +370,7 @@ class TestAdminRepos(object):
repo_name = backend.new_repo_name()
description = 'description for newly created repo'
response = self.app.post(
- url('repos'),
+ route_path('repo_create'),
fixture._get_repo_create_params(
repo_private=False,
repo_name=repo_name,
@@ -358,7 +386,7 @@ class TestAdminRepos(object):
repo_name = backend.new_repo_name()
description = 'description for newly created repo'
response = self.app.post(
- url('repos'),
+ route_path('repo_create'),
fixture._get_repo_create_params(
repo_private=False,
repo_name=repo_name,
@@ -373,7 +401,7 @@ class TestAdminRepos(object):
repo_name = backend.new_repo_name() + ".git"
description = 'description for newly created repo'
response = self.app.post(
- url('repos'),
+ route_path('repo_create'),
fixture._get_repo_create_params(
repo_private=False,
repo_name=repo_name,
@@ -382,11 +410,8 @@ class TestAdminRepos(object):
csrf_token=csrf_token))
response.mustcontain('Repository name cannot end with .git')
- def test_show(self, autologin_user, backend):
- self.app.get(url('repo', repo_name=backend.repo_name))
-
def test_default_user_cannot_access_private_repo_in_a_group(
- self, autologin_user, user_util, backend, csrf_token):
+ self, autologin_user, user_util, backend):
group = user_util.create_repo_group()
@@ -422,7 +447,7 @@ class TestAdminRepos(object):
repo_name = backend.new_repo_name()
description = 'description for newly created repo'
response = self.app.post(
- url('repos'),
+ route_path('repo_create'),
fixture._get_repo_create_params(
repo_private=False,
repo_name=repo_name,
@@ -441,7 +466,7 @@ class TestAdminRepos(object):
description = 'description for newly created repo'
response = self.app.post(
- url('repos'),
+ route_path('repo_create'),
fixture._get_repo_create_params(
repo_private=False,
repo_name=repo_name,
@@ -461,7 +486,8 @@ class TestAdminRepos(object):
repo_name_utf8 = safe_str(repo_name)
# run the check page that triggers the flash message
- response = self.app.get(url('repo_check_home', repo_name=repo_name))
+ response = self.app.get(
+ route_path('repo_creating_check', repo_name=safe_str(repo_name)))
assert response.json == {u'result': True}
flash_msg = u'Created repository {} '.format(
@@ -475,643 +501,9 @@ class TestAdminRepos(object):
assert new_repo.description == description
# test if the repository is visible in the list ?
- response = self.app.get(h.route_path('repo_summary', repo_name=repo_name))
+ response = self.app.get(
+ h.route_path('repo_summary', repo_name=safe_str(repo_name)))
response.mustcontain(repo_name)
response.mustcontain(backend.alias)
assert repo_on_filesystem(repo_name)
-
-
-@pytest.mark.usefixtures("app")
-class TestVcsSettings(object):
- FORM_DATA = {
- 'inherit_global_settings': False,
- 'hooks_changegroup_repo_size': False,
- 'hooks_changegroup_push_logger': False,
- 'hooks_outgoing_pull_logger': False,
- 'extensions_largefiles': False,
- 'extensions_evolve': False,
- 'phases_publish': 'False',
- 'rhodecode_pr_merge_enabled': False,
- 'rhodecode_use_outdated_comments': False,
- 'new_svn_branch': '',
- 'new_svn_tag': ''
- }
-
- @pytest.mark.skip_backends('svn')
- def test_global_settings_initial_values(self, autologin_user, backend):
- repo_name = backend.repo_name
- response = self.app.get(url('repo_vcs_settings', repo_name=repo_name))
-
- expected_settings = (
- 'rhodecode_use_outdated_comments', 'rhodecode_pr_merge_enabled',
- 'hooks_changegroup_repo_size', 'hooks_changegroup_push_logger',
- 'hooks_outgoing_pull_logger'
- )
- for setting in expected_settings:
- self.assert_repo_value_equals_global_value(response, setting)
-
- def test_show_settings_requires_repo_admin_permission(
- self, backend, user_util, settings_util):
- repo = backend.create_repo()
- repo_name = repo.repo_name
- user = UserModel().get_by_username(TEST_USER_REGULAR_LOGIN)
- user_util.grant_user_permission_to_repo(repo, user, 'repository.admin')
- login_user_session(
- self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
- self.app.get(url('repo_vcs_settings', repo_name=repo_name), status=200)
-
- def test_inherit_global_settings_flag_is_true_by_default(
- self, autologin_user, backend):
- repo_name = backend.repo_name
- response = self.app.get(url('repo_vcs_settings', repo_name=repo_name))
-
- assert_response = AssertResponse(response)
- element = assert_response.get_element('#inherit_global_settings')
- assert element.checked
-
- @pytest.mark.parametrize('checked_value', [True, False])
- def test_inherit_global_settings_value(
- self, autologin_user, backend, checked_value, settings_util):
- repo = backend.create_repo()
- repo_name = repo.repo_name
- settings_util.create_repo_rhodecode_setting(
- repo, 'inherit_vcs_settings', checked_value, 'bool')
- response = self.app.get(url('repo_vcs_settings', repo_name=repo_name))
-
- assert_response = AssertResponse(response)
- element = assert_response.get_element('#inherit_global_settings')
- assert element.checked == checked_value
-
- @pytest.mark.skip_backends('svn')
- def test_hooks_settings_are_created(
- self, autologin_user, backend, csrf_token):
- repo_name = backend.repo_name
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- settings = SettingsModel(repo=repo_name)
- try:
- for section, key in VcsSettingsModel.HOOKS_SETTINGS:
- ui = settings.get_ui_by_section_and_key(section, key)
- assert ui.ui_active is False
- finally:
- self._cleanup_repo_settings(settings)
-
- def test_hooks_settings_are_not_created_for_svn(
- self, autologin_user, backend_svn, csrf_token):
- repo_name = backend_svn.repo_name
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- settings = SettingsModel(repo=repo_name)
- try:
- for section, key in VcsSettingsModel.HOOKS_SETTINGS:
- ui = settings.get_ui_by_section_and_key(section, key)
- assert ui is None
- finally:
- self._cleanup_repo_settings(settings)
-
- @pytest.mark.skip_backends('svn')
- def test_hooks_settings_are_updated(
- self, autologin_user, backend, csrf_token):
- repo_name = backend.repo_name
- settings = SettingsModel(repo=repo_name)
- for section, key in VcsSettingsModel.HOOKS_SETTINGS:
- settings.create_ui_section_value(section, '', key=key, active=True)
-
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- try:
- for section, key in VcsSettingsModel.HOOKS_SETTINGS:
- ui = settings.get_ui_by_section_and_key(section, key)
- assert ui.ui_active is False
- finally:
- self._cleanup_repo_settings(settings)
-
- def test_hooks_settings_are_not_updated_for_svn(
- self, autologin_user, backend_svn, csrf_token):
- repo_name = backend_svn.repo_name
- settings = SettingsModel(repo=repo_name)
- for section, key in VcsSettingsModel.HOOKS_SETTINGS:
- settings.create_ui_section_value(section, '', key=key, active=True)
-
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- try:
- for section, key in VcsSettingsModel.HOOKS_SETTINGS:
- ui = settings.get_ui_by_section_and_key(section, key)
- assert ui.ui_active is True
- finally:
- self._cleanup_repo_settings(settings)
-
- @pytest.mark.skip_backends('svn')
- def test_pr_settings_are_created(
- self, autologin_user, backend, csrf_token):
- repo_name = backend.repo_name
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- settings = SettingsModel(repo=repo_name)
- try:
- for name in VcsSettingsModel.GENERAL_SETTINGS:
- setting = settings.get_setting_by_name(name)
- assert setting.app_settings_value is False
- finally:
- self._cleanup_repo_settings(settings)
-
- def test_pr_settings_are_not_created_for_svn(
- self, autologin_user, backend_svn, csrf_token):
- repo_name = backend_svn.repo_name
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- settings = SettingsModel(repo=repo_name)
- try:
- for name in VcsSettingsModel.GENERAL_SETTINGS:
- setting = settings.get_setting_by_name(name)
- assert setting is None
- finally:
- self._cleanup_repo_settings(settings)
-
- def test_pr_settings_creation_requires_repo_admin_permission(
- self, backend, user_util, settings_util, csrf_token):
- repo = backend.create_repo()
- repo_name = repo.repo_name
-
- logout_user_session(self.app, csrf_token)
- session = login_user_session(
- self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
- new_csrf_token = auth.get_csrf_token(session)
-
- user = UserModel().get_by_username(TEST_USER_REGULAR_LOGIN)
- repo = Repository.get_by_repo_name(repo_name)
- user_util.grant_user_permission_to_repo(repo, user, 'repository.admin')
- data = self.FORM_DATA.copy()
- data['csrf_token'] = new_csrf_token
- settings = SettingsModel(repo=repo_name)
-
- try:
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data,
- status=302)
- finally:
- self._cleanup_repo_settings(settings)
-
- @pytest.mark.skip_backends('svn')
- def test_pr_settings_are_updated(
- self, autologin_user, backend, csrf_token):
- repo_name = backend.repo_name
- settings = SettingsModel(repo=repo_name)
- for name in VcsSettingsModel.GENERAL_SETTINGS:
- settings.create_or_update_setting(name, True, 'bool')
-
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- try:
- for name in VcsSettingsModel.GENERAL_SETTINGS:
- setting = settings.get_setting_by_name(name)
- assert setting.app_settings_value is False
- finally:
- self._cleanup_repo_settings(settings)
-
- def test_pr_settings_are_not_updated_for_svn(
- self, autologin_user, backend_svn, csrf_token):
- repo_name = backend_svn.repo_name
- settings = SettingsModel(repo=repo_name)
- for name in VcsSettingsModel.GENERAL_SETTINGS:
- settings.create_or_update_setting(name, True, 'bool')
-
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- try:
- for name in VcsSettingsModel.GENERAL_SETTINGS:
- setting = settings.get_setting_by_name(name)
- assert setting.app_settings_value is True
- finally:
- self._cleanup_repo_settings(settings)
-
- def test_svn_settings_are_created(
- self, autologin_user, backend_svn, csrf_token, settings_util):
- repo_name = backend_svn.repo_name
- data = self.FORM_DATA.copy()
- data['new_svn_tag'] = 'svn-tag'
- data['new_svn_branch'] = 'svn-branch'
- data['csrf_token'] = csrf_token
-
- # Create few global settings to make sure that uniqueness validators
- # are not triggered
- settings_util.create_rhodecode_ui(
- VcsSettingsModel.SVN_BRANCH_SECTION, 'svn-branch')
- settings_util.create_rhodecode_ui(
- VcsSettingsModel.SVN_TAG_SECTION, 'svn-tag')
-
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- settings = SettingsModel(repo=repo_name)
- try:
- svn_branches = settings.get_ui_by_section(
- VcsSettingsModel.SVN_BRANCH_SECTION)
- svn_branch_names = [b.ui_value for b in svn_branches]
- svn_tags = settings.get_ui_by_section(
- VcsSettingsModel.SVN_TAG_SECTION)
- svn_tag_names = [b.ui_value for b in svn_tags]
- assert 'svn-branch' in svn_branch_names
- assert 'svn-tag' in svn_tag_names
- finally:
- self._cleanup_repo_settings(settings)
-
- def test_svn_settings_are_unique(
- self, autologin_user, backend_svn, csrf_token, settings_util):
- repo = backend_svn.repo
- repo_name = repo.repo_name
- data = self.FORM_DATA.copy()
- data['new_svn_tag'] = 'test_tag'
- data['new_svn_branch'] = 'test_branch'
- data['csrf_token'] = csrf_token
- settings_util.create_repo_rhodecode_ui(
- repo, VcsSettingsModel.SVN_BRANCH_SECTION, 'test_branch')
- settings_util.create_repo_rhodecode_ui(
- repo, VcsSettingsModel.SVN_TAG_SECTION, 'test_tag')
-
- response = self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=200)
- response.mustcontain('Pattern already exists')
-
- def test_svn_settings_with_empty_values_are_not_created(
- self, autologin_user, backend_svn, csrf_token):
- repo_name = backend_svn.repo_name
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- settings = SettingsModel(repo=repo_name)
- try:
- svn_branches = settings.get_ui_by_section(
- VcsSettingsModel.SVN_BRANCH_SECTION)
- svn_tags = settings.get_ui_by_section(
- VcsSettingsModel.SVN_TAG_SECTION)
- assert len(svn_branches) == 0
- assert len(svn_tags) == 0
- finally:
- self._cleanup_repo_settings(settings)
-
- def test_svn_settings_are_shown_for_svn_repository(
- self, autologin_user, backend_svn, csrf_token):
- repo_name = backend_svn.repo_name
- response = self.app.get(
- url('repo_vcs_settings', repo_name=repo_name), status=200)
- response.mustcontain('Subversion Settings')
-
- @pytest.mark.skip_backends('svn')
- def test_svn_settings_are_not_created_for_not_svn_repository(
- self, autologin_user, backend, csrf_token):
- repo_name = backend.repo_name
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- settings = SettingsModel(repo=repo_name)
- try:
- svn_branches = settings.get_ui_by_section(
- VcsSettingsModel.SVN_BRANCH_SECTION)
- svn_tags = settings.get_ui_by_section(
- VcsSettingsModel.SVN_TAG_SECTION)
- assert len(svn_branches) == 0
- assert len(svn_tags) == 0
- finally:
- self._cleanup_repo_settings(settings)
-
- @pytest.mark.skip_backends('svn')
- def test_svn_settings_are_shown_only_for_svn_repository(
- self, autologin_user, backend, csrf_token):
- repo_name = backend.repo_name
- response = self.app.get(
- url('repo_vcs_settings', repo_name=repo_name), status=200)
- response.mustcontain(no='Subversion Settings')
-
- def test_hg_settings_are_created(
- self, autologin_user, backend_hg, csrf_token):
- repo_name = backend_hg.repo_name
- data = self.FORM_DATA.copy()
- data['new_svn_tag'] = 'svn-tag'
- data['new_svn_branch'] = 'svn-branch'
- data['csrf_token'] = csrf_token
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- settings = SettingsModel(repo=repo_name)
- try:
- largefiles_ui = settings.get_ui_by_section_and_key(
- 'extensions', 'largefiles')
- assert largefiles_ui.ui_active is False
- phases_ui = settings.get_ui_by_section_and_key(
- 'phases', 'publish')
- assert str2bool(phases_ui.ui_value) is False
- finally:
- self._cleanup_repo_settings(settings)
-
- def test_hg_settings_are_updated(
- self, autologin_user, backend_hg, csrf_token):
- repo_name = backend_hg.repo_name
- settings = SettingsModel(repo=repo_name)
- settings.create_ui_section_value(
- 'extensions', '', key='largefiles', active=True)
- settings.create_ui_section_value(
- 'phases', '1', key='publish', active=True)
-
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- try:
- largefiles_ui = settings.get_ui_by_section_and_key(
- 'extensions', 'largefiles')
- assert largefiles_ui.ui_active is False
- phases_ui = settings.get_ui_by_section_and_key(
- 'phases', 'publish')
- assert str2bool(phases_ui.ui_value) is False
- finally:
- self._cleanup_repo_settings(settings)
-
- def test_hg_settings_are_shown_for_hg_repository(
- self, autologin_user, backend_hg, csrf_token):
- repo_name = backend_hg.repo_name
- response = self.app.get(
- url('repo_vcs_settings', repo_name=repo_name), status=200)
- response.mustcontain('Mercurial Settings')
-
- @pytest.mark.skip_backends('hg')
- def test_hg_settings_are_created_only_for_hg_repository(
- self, autologin_user, backend, csrf_token):
- repo_name = backend.repo_name
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- settings = SettingsModel(repo=repo_name)
- try:
- largefiles_ui = settings.get_ui_by_section_and_key(
- 'extensions', 'largefiles')
- assert largefiles_ui is None
- phases_ui = settings.get_ui_by_section_and_key(
- 'phases', 'publish')
- assert phases_ui is None
- finally:
- self._cleanup_repo_settings(settings)
-
- @pytest.mark.skip_backends('hg')
- def test_hg_settings_are_shown_only_for_hg_repository(
- self, autologin_user, backend, csrf_token):
- repo_name = backend.repo_name
- response = self.app.get(
- url('repo_vcs_settings', repo_name=repo_name), status=200)
- response.mustcontain(no='Mercurial Settings')
-
- @pytest.mark.skip_backends('hg')
- def test_hg_settings_are_updated_only_for_hg_repository(
- self, autologin_user, backend, csrf_token):
- repo_name = backend.repo_name
- settings = SettingsModel(repo=repo_name)
- settings.create_ui_section_value(
- 'extensions', '', key='largefiles', active=True)
- settings.create_ui_section_value(
- 'phases', '1', key='publish', active=True)
-
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
- try:
- largefiles_ui = settings.get_ui_by_section_and_key(
- 'extensions', 'largefiles')
- assert largefiles_ui.ui_active is True
- phases_ui = settings.get_ui_by_section_and_key(
- 'phases', 'publish')
- assert phases_ui.ui_value == '1'
- finally:
- self._cleanup_repo_settings(settings)
-
- def test_per_repo_svn_settings_are_displayed(
- self, autologin_user, backend_svn, settings_util):
- repo = backend_svn.create_repo()
- repo_name = repo.repo_name
- branches = [
- settings_util.create_repo_rhodecode_ui(
- repo, VcsSettingsModel.SVN_BRANCH_SECTION,
- 'branch_{}'.format(i))
- for i in range(10)]
- tags = [
- settings_util.create_repo_rhodecode_ui(
- repo, VcsSettingsModel.SVN_TAG_SECTION, 'tag_{}'.format(i))
- for i in range(10)]
-
- response = self.app.get(
- url('repo_vcs_settings', repo_name=repo_name), status=200)
- assert_response = AssertResponse(response)
- for branch in branches:
- css_selector = '[name=branch_value_{}]'.format(branch.ui_id)
- element = assert_response.get_element(css_selector)
- assert element.value == branch.ui_value
- for tag in tags:
- css_selector = '[name=tag_ui_value_new_{}]'.format(tag.ui_id)
- element = assert_response.get_element(css_selector)
- assert element.value == tag.ui_value
-
- def test_per_repo_hg_and_pr_settings_are_not_displayed_for_svn(
- self, autologin_user, backend_svn, settings_util):
- repo = backend_svn.create_repo()
- repo_name = repo.repo_name
- response = self.app.get(
- url('repo_vcs_settings', repo_name=repo_name), status=200)
- response.mustcontain(no='Hooks: ')
- response.mustcontain(no='Pull Request Settings: ')
-
- def test_inherit_global_settings_value_is_saved(
- self, autologin_user, backend, csrf_token):
- repo_name = backend.repo_name
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- data['inherit_global_settings'] = True
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
-
- settings = SettingsModel(repo=repo_name)
- vcs_settings = VcsSettingsModel(repo=repo_name)
- try:
- assert vcs_settings.inherit_global_settings is True
- finally:
- self._cleanup_repo_settings(settings)
-
- def test_repo_cache_is_invalidated_when_settings_are_updated(
- self, autologin_user, backend, csrf_token):
- repo_name = backend.repo_name
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- data['inherit_global_settings'] = True
- settings = SettingsModel(repo=repo_name)
-
- invalidation_patcher = mock.patch(
- 'rhodecode.controllers.admin.repos.ScmModel.mark_for_invalidation')
- with invalidation_patcher as invalidation_mock:
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data,
- status=302)
- try:
- invalidation_mock.assert_called_once_with(repo_name, delete=True)
- finally:
- self._cleanup_repo_settings(settings)
-
- def test_other_settings_not_saved_inherit_global_settings_is_true(
- self, autologin_user, backend, csrf_token):
- repo_name = backend.repo_name
- data = self.FORM_DATA.copy()
- data['csrf_token'] = csrf_token
- data['inherit_global_settings'] = True
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data, status=302)
-
- settings = SettingsModel(repo=repo_name)
- ui_settings = (
- VcsSettingsModel.HOOKS_SETTINGS + VcsSettingsModel.HG_SETTINGS)
-
- vcs_settings = []
- try:
- for section, key in ui_settings:
- ui = settings.get_ui_by_section_and_key(section, key)
- if ui:
- vcs_settings.append(ui)
- vcs_settings.extend(settings.get_ui_by_section(
- VcsSettingsModel.SVN_BRANCH_SECTION))
- vcs_settings.extend(settings.get_ui_by_section(
- VcsSettingsModel.SVN_TAG_SECTION))
- for name in VcsSettingsModel.GENERAL_SETTINGS:
- setting = settings.get_setting_by_name(name)
- if setting:
- vcs_settings.append(setting)
- assert vcs_settings == []
- finally:
- self._cleanup_repo_settings(settings)
-
- def test_delete_svn_branch_and_tag_patterns(
- self, autologin_user, backend_svn, settings_util, csrf_token):
- repo = backend_svn.create_repo()
- repo_name = repo.repo_name
- branch = settings_util.create_repo_rhodecode_ui(
- repo, VcsSettingsModel.SVN_BRANCH_SECTION, 'test_branch',
- cleanup=False)
- tag = settings_util.create_repo_rhodecode_ui(
- repo, VcsSettingsModel.SVN_TAG_SECTION, 'test_tag', cleanup=False)
- data = {
- '_method': 'delete',
- 'csrf_token': csrf_token
- }
- for id_ in (branch.ui_id, tag.ui_id):
- data['delete_svn_pattern'] = id_,
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data,
- headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=200)
- settings = VcsSettingsModel(repo=repo_name)
- assert settings.get_repo_svn_branch_patterns() == []
-
- def test_delete_svn_branch_requires_repo_admin_permission(
- self, backend_svn, user_util, settings_util, csrf_token):
- repo = backend_svn.create_repo()
- repo_name = repo.repo_name
-
- logout_user_session(self.app, csrf_token)
- session = login_user_session(
- self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
- csrf_token = auth.get_csrf_token(session)
-
- repo = Repository.get_by_repo_name(repo_name)
- user = UserModel().get_by_username(TEST_USER_REGULAR_LOGIN)
- user_util.grant_user_permission_to_repo(repo, user, 'repository.admin')
- branch = settings_util.create_repo_rhodecode_ui(
- repo, VcsSettingsModel.SVN_BRANCH_SECTION, 'test_branch',
- cleanup=False)
- data = {
- '_method': 'delete',
- 'csrf_token': csrf_token,
- 'delete_svn_pattern': branch.ui_id
- }
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data,
- headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=200)
-
- def test_delete_svn_branch_raises_400_when_not_found(
- self, autologin_user, backend_svn, settings_util, csrf_token):
- repo_name = backend_svn.repo_name
- data = {
- '_method': 'delete',
- 'delete_svn_pattern': 123,
- 'csrf_token': csrf_token
- }
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data,
- headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=400)
-
- def test_delete_svn_branch_raises_400_when_no_id_specified(
- self, autologin_user, backend_svn, settings_util, csrf_token):
- repo_name = backend_svn.repo_name
- data = {
- '_method': 'delete',
- 'csrf_token': csrf_token
- }
- self.app.post(
- url('repo_vcs_settings', repo_name=repo_name), data,
- headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=400)
-
- def _cleanup_repo_settings(self, settings_model):
- cleanup = []
- ui_settings = (
- VcsSettingsModel.HOOKS_SETTINGS + VcsSettingsModel.HG_SETTINGS)
-
- for section, key in ui_settings:
- ui = settings_model.get_ui_by_section_and_key(section, key)
- if ui:
- cleanup.append(ui)
-
- cleanup.extend(settings_model.get_ui_by_section(
- VcsSettingsModel.INHERIT_SETTINGS))
- cleanup.extend(settings_model.get_ui_by_section(
- VcsSettingsModel.SVN_BRANCH_SECTION))
- cleanup.extend(settings_model.get_ui_by_section(
- VcsSettingsModel.SVN_TAG_SECTION))
-
- for name in VcsSettingsModel.GENERAL_SETTINGS:
- setting = settings_model.get_setting_by_name(name)
- if setting:
- cleanup.append(setting)
-
- for object_ in cleanup:
- Session().delete(object_)
- Session().commit()
-
- def assert_repo_value_equals_global_value(self, response, setting):
- assert_response = AssertResponse(response)
- global_css_selector = '[name={}_inherited]'.format(setting)
- repo_css_selector = '[name={}]'.format(setting)
- repo_element = assert_response.get_element(repo_css_selector)
- global_element = assert_response.get_element(global_css_selector)
- assert repo_element.value == global_element.value
-
-
-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
diff --git a/rhodecode/apps/admin/tests/test_admin_repository_groups.py b/rhodecode/apps/admin/tests/test_admin_repository_groups.py
new file mode 100644
--- /dev/null
+++ b/rhodecode/apps/admin/tests/test_admin_repository_groups.py
@@ -0,0 +1,176 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2010-2017 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see %s ' % (
+ repo_gr_url, repo_group_name_unicode))
+
+ # # test if the repo group was created in the database
+ new_repo_group = RepoGroupModel()._get_repo_group(
+ repo_group_name_unicode)
+ assert new_repo_group is not None
+
+ assert new_repo_group.group_name == repo_group_name_unicode
+ assert new_repo_group.group_description == description
+
+ # test if the repository is visible in the list ?
+ response = self.app.get(repo_gr_url)
+ response.mustcontain(repo_group_name)
+
+ # test if the repository group was created on filesystem
+ is_on_filesystem = os.path.isdir(
+ os.path.join(TESTS_TMP_PATH, repo_group_name))
+ if not is_on_filesystem:
+ self.fail('no repo group %s in filesystem' % repo_group_name)
+
+ RepoGroupModel().delete(repo_group_name_unicode)
+ Session().commit()
+
+ @pytest.mark.parametrize('repo_group_name', [
+ 'git_repo',
+ 'git_repo_ąć',
+ 'hg_repo',
+ '12345',
+ 'hg_repo_ąć',
+ ])
+ def test_create_subgroup(self, autologin_user, user_util, repo_group_name, csrf_token):
+ parent_group = user_util.create_repo_group()
+ parent_group_name = parent_group.group_name
+
+ expected_group_name = '{}/{}'.format(
+ parent_group_name, repo_group_name)
+ expected_group_name_unicode = expected_group_name.decode('utf8')
+
+ try:
+ response = self.app.post(
+ route_path('repo_group_create'),
+ fixture._get_group_create_params(
+ group_name=repo_group_name,
+ group_parent_id=parent_group.group_id,
+ group_description='Test desciption',
+ csrf_token=csrf_token))
+
+ assert_session_flash(
+ response,
+ u'Created repository group %s ' % (
+ h.route_path('repo_group_home',
+ repo_group_name=expected_group_name),
+ expected_group_name_unicode))
+ finally:
+ RepoGroupModel().delete(expected_group_name_unicode)
+ Session().commit()
+
+ def test_user_with_creation_permissions_cannot_create_subgroups(
+ self, autologin_regular_user, user_util):
+
+ user_util.grant_user_permission(
+ TEST_USER_REGULAR_LOGIN, 'hg.repogroup.create.true')
+ parent_group = user_util.create_repo_group()
+ parent_group_id = parent_group.group_id
+ self.app.get(
+ route_path('repo_group_new',
+ params=dict(parent_group=parent_group_id), ),
+ status=403)
diff --git a/rhodecode/apps/admin/tests/test_admin_user_groups.py b/rhodecode/apps/admin/tests/test_admin_user_groups.py
new file mode 100644
--- /dev/null
+++ b/rhodecode/apps/admin/tests/test_admin_user_groups.py
@@ -0,0 +1,170 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2010-2017 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see DESC '
+
+ response = self.app.get(
+ route_path('user_groups_data'), extra_environ=xhr_header)
+
+ response.mustcontain(
+ '<strong onload="alert();">DESC</strong>')
+ response.mustcontain(
+ '<img src="/image1" onload="'
+ 'alert('Hello, World!');">')
+
+ def test_edit_user_group_autocomplete_empty_members(self, xhr_header, user_util):
+ self.log_user()
+ ug = user_util.create_user_group()
+ response = self.app.get(
+ route_path('user_group_members_data', user_group_id=ug.users_group_id),
+ extra_environ=xhr_header)
+
+ assert response.json == {'members': []}
+
+ def test_edit_user_group_autocomplete_members(self, xhr_header, user_util):
+ self.log_user()
+ members = [u.user_id for u in User.get_all()]
+ ug = user_util.create_user_group(members=members)
+ response = self.app.get(
+ route_path('user_group_members_data',
+ user_group_id=ug.users_group_id),
+ extra_environ=xhr_header)
+
+ assert len(response.json['members']) == len(members)
+
+ def test_creation_page(self):
+ self.log_user()
+ self.app.get(route_path('user_groups_new'), status=200)
+
+ def test_create(self):
+ from rhodecode.lib import helpers as h
+
+ self.log_user()
+ users_group_name = 'test_user_group'
+ response = self.app.post(route_path('user_groups_create'), {
+ 'users_group_name': users_group_name,
+ 'user_group_description': 'DESC',
+ 'active': True,
+ 'csrf_token': self.csrf_token})
+
+ user_group_id = UserGroup.get_by_group_name(
+ users_group_name).users_group_id
+
+ user_group_link = h.link_to(
+ users_group_name,
+ route_path('edit_user_group', user_group_id=user_group_id))
+
+ assert_session_flash(
+ response,
+ 'Created user group %s' % user_group_link)
+
+ fixture.destroy_user_group(users_group_name)
+
+ def test_create_with_empty_name(self):
+ self.log_user()
+
+ response = self.app.post(route_path('user_groups_create'), {
+ 'users_group_name': '',
+ 'user_group_description': 'DESC',
+ 'active': True,
+ 'csrf_token': self.csrf_token}, status=200)
+
+ response.mustcontain('Please enter a value')
+
+ def test_create_duplicate(self, user_util):
+ self.log_user()
+
+ user_group = user_util.create_user_group()
+ duplicate_name = user_group.users_group_name
+ response = self.app.post(route_path('user_groups_create'), {
+ 'users_group_name': duplicate_name,
+ 'user_group_description': 'DESC',
+ 'active': True,
+ 'csrf_token': self.csrf_token}, status=200)
+
+ response.mustcontain(
+ 'User group `{}` already exists'.format(user_group.users_group_name))
diff --git a/rhodecode/apps/admin/tests/test_admin_users.py b/rhodecode/apps/admin/tests/test_admin_users.py
--- a/rhodecode/apps/admin/tests/test_admin_users.py
+++ b/rhodecode/apps/admin/tests/test_admin_users.py
@@ -19,8 +19,12 @@
# and proprietary license terms, please see https://rhodecode.com/licenses/
import pytest
+from sqlalchemy.orm.exc import NoResultFound
-from rhodecode.model.db import User, UserApiKeys, UserEmailMap
+from rhodecode.lib import auth
+from rhodecode.lib import helpers as h
+from rhodecode.model import validators
+from rhodecode.model.db import User, UserApiKeys, UserEmailMap, Repository
from rhodecode.model.meta import Session
from rhodecode.model.user import UserModel
@@ -40,6 +44,27 @@ def route_path(name, params=None, **kwar
ADMIN_PREFIX + '/users',
'users_data':
ADMIN_PREFIX + '/users_data',
+ 'users_create':
+ ADMIN_PREFIX + '/users/create',
+ 'users_new':
+ ADMIN_PREFIX + '/users/new',
+ 'user_edit':
+ ADMIN_PREFIX + '/users/{user_id}/edit',
+ 'user_edit_advanced':
+ ADMIN_PREFIX + '/users/{user_id}/edit/advanced',
+ 'user_edit_global_perms':
+ ADMIN_PREFIX + '/users/{user_id}/edit/global_permissions',
+ 'user_edit_global_perms_update':
+ ADMIN_PREFIX + '/users/{user_id}/edit/global_permissions/update',
+ 'user_update':
+ ADMIN_PREFIX + '/users/{user_id}/update',
+ 'user_delete':
+ ADMIN_PREFIX + '/users/{user_id}/delete',
+ 'user_force_password_reset':
+ ADMIN_PREFIX + '/users/{user_id}/password_reset',
+ 'user_create_personal_repo_group':
+ ADMIN_PREFIX + '/users/{user_id}/create_repo_group',
+
'edit_user_auth_tokens':
ADMIN_PREFIX + '/users/{user_id}/edit/auth_tokens',
'edit_user_auth_tokens_add':
@@ -60,6 +85,15 @@ def route_path(name, params=None, **kwar
ADMIN_PREFIX + '/users/{user_id}/edit/ips/new',
'edit_user_ips_delete':
ADMIN_PREFIX + '/users/{user_id}/edit/ips/delete',
+
+ 'edit_user_perms_summary':
+ ADMIN_PREFIX + '/users/{user_id}/edit/permissions_summary',
+ 'edit_user_perms_summary_json':
+ ADMIN_PREFIX + '/users/{user_id}/edit/permissions_summary/json',
+
+ 'edit_user_audit_logs':
+ ADMIN_PREFIX + '/users/{user_id}/edit/audit',
+
}[name].format(**kwargs)
if params:
@@ -135,7 +169,7 @@ class TestAdminUsersView(TestController)
self.log_user()
user = user_util.create_user()
user_id = user.user_id
- keys = user.extra_auth_tokens
+ keys = user.auth_tokens
assert 2 == len(keys)
response = self.app.post(
@@ -220,7 +254,8 @@ class TestAdminUsersView(TestController)
def test_emails(self):
self.log_user()
user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
- response = self.app.get(route_path('edit_user_emails', user_id=user.user_id))
+ response = self.app.get(
+ route_path('edit_user_emails', user_id=user.user_id))
response.mustcontain('No additional emails specified')
def test_emails_add(self, user_util):
@@ -233,7 +268,8 @@ class TestAdminUsersView(TestController)
params={'new_email': 'example@rhodecode.com',
'csrf_token': self.csrf_token})
- response = self.app.get(route_path('edit_user_emails', user_id=user_id))
+ response = self.app.get(
+ route_path('edit_user_emails', user_id=user_id))
response.mustcontain('example@rhodecode.com')
def test_emails_add_existing_email(self, user_util, user_regular):
@@ -250,7 +286,8 @@ class TestAdminUsersView(TestController)
assert_session_flash(
response, 'This e-mail address is already taken')
- response = self.app.get(route_path('edit_user_emails', user_id=user_id))
+ response = self.app.get(
+ route_path('edit_user_emails', user_id=user_id))
response.mustcontain(no=[existing_email])
def test_emails_delete(self, user_util):
@@ -263,7 +300,8 @@ class TestAdminUsersView(TestController)
params={'new_email': 'example@rhodecode.com',
'csrf_token': self.csrf_token})
- response = self.app.get(route_path('edit_user_emails', user_id=user_id))
+ response = self.app.get(
+ route_path('edit_user_emails', user_id=user_id))
response.mustcontain('example@rhodecode.com')
user_email = UserEmailMap.query()\
@@ -277,5 +315,469 @@ class TestAdminUsersView(TestController)
params={'del_email_id': del_email_id,
'csrf_token': self.csrf_token})
- response = self.app.get(route_path('edit_user_emails', user_id=user_id))
- response.mustcontain(no=['example@rhodecode.com'])
\ No newline at end of file
+ response = self.app.get(
+ route_path('edit_user_emails', user_id=user_id))
+ response.mustcontain(no=['example@rhodecode.com'])
+
+
+ def test_create(self, request, xhr_header):
+ self.log_user()
+ username = 'newtestuser'
+ password = 'test12'
+ password_confirmation = password
+ name = 'name'
+ lastname = 'lastname'
+ email = 'mail@mail.com'
+
+ self.app.get(route_path('users_new'))
+
+ response = self.app.post(route_path('users_create'), params={
+ 'username': username,
+ 'password': password,
+ 'password_confirmation': password_confirmation,
+ 'firstname': name,
+ 'active': True,
+ 'lastname': lastname,
+ 'extern_name': 'rhodecode',
+ 'extern_type': 'rhodecode',
+ 'email': email,
+ 'csrf_token': self.csrf_token,
+ })
+ user_link = h.link_to(
+ username,
+ route_path(
+ 'user_edit', user_id=User.get_by_username(username).user_id))
+ assert_session_flash(response, 'Created user %s' % (user_link,))
+
+ @request.addfinalizer
+ def cleanup():
+ fixture.destroy_user(username)
+ Session().commit()
+
+ new_user = User.query().filter(User.username == username).one()
+
+ assert new_user.username == username
+ assert auth.check_password(password, new_user.password)
+ assert new_user.name == name
+ assert new_user.lastname == lastname
+ assert new_user.email == email
+
+ response = self.app.get(route_path('users_data'),
+ extra_environ=xhr_header)
+ response.mustcontain(username)
+
+ def test_create_err(self):
+ self.log_user()
+ username = 'new_user'
+ password = ''
+ name = 'name'
+ lastname = 'lastname'
+ email = 'errmail.com'
+
+ self.app.get(route_path('users_new'))
+
+ response = self.app.post(route_path('users_create'), params={
+ 'username': username,
+ 'password': password,
+ 'name': name,
+ 'active': False,
+ 'lastname': lastname,
+ 'email': email,
+ 'csrf_token': self.csrf_token,
+ })
+
+ msg = validators.ValidUsername(
+ False, {})._messages['system_invalid_username']
+ msg = h.html_escape(msg % {'username': 'new_user'})
+ response.mustcontain('%s ' % msg)
+ response.mustcontain(
+ 'Please enter a value ')
+ response.mustcontain(
+ 'An email address must contain a'
+ ' single @ ')
+
+ def get_user():
+ Session().query(User).filter(User.username == username).one()
+
+ with pytest.raises(NoResultFound):
+ get_user()
+
+ def test_new(self):
+ self.log_user()
+ self.app.get(route_path('users_new'))
+
+ @pytest.mark.parametrize("name, attrs", [
+ ('firstname', {'firstname': 'new_username'}),
+ ('lastname', {'lastname': 'new_username'}),
+ ('admin', {'admin': True}),
+ ('admin', {'admin': False}),
+ ('extern_type', {'extern_type': 'ldap'}),
+ ('extern_type', {'extern_type': None}),
+ ('extern_name', {'extern_name': 'test'}),
+ ('extern_name', {'extern_name': None}),
+ ('active', {'active': False}),
+ ('active', {'active': True}),
+ ('email', {'email': 'some@email.com'}),
+ ('language', {'language': 'de'}),
+ ('language', {'language': 'en'}),
+ # ('new_password', {'new_password': 'foobar123',
+ # 'password_confirmation': 'foobar123'})
+ ])
+ def test_update(self, name, attrs, user_util):
+ self.log_user()
+ usr = user_util.create_user(
+ password='qweqwe',
+ email='testme@rhodecode.org',
+ extern_type='rhodecode',
+ extern_name='xxx',
+ )
+ user_id = usr.user_id
+ Session().commit()
+
+ params = usr.get_api_data()
+ cur_lang = params['language'] or 'en'
+ params.update({
+ 'password_confirmation': '',
+ 'new_password': '',
+ 'language': cur_lang,
+ 'csrf_token': self.csrf_token,
+ })
+ params.update({'new_password': ''})
+ params.update(attrs)
+ if name == 'email':
+ params['emails'] = [attrs['email']]
+ elif name == 'extern_type':
+ # cannot update this via form, expected value is original one
+ params['extern_type'] = "rhodecode"
+ elif name == 'extern_name':
+ # cannot update this via form, expected value is original one
+ params['extern_name'] = 'xxx'
+ # special case since this user is not
+ # logged in yet his data is not filled
+ # so we use creation data
+
+ response = self.app.post(
+ route_path('user_update', user_id=usr.user_id), params)
+ assert response.status_int == 302
+ assert_session_flash(response, 'User updated successfully')
+
+ updated_user = User.get(user_id)
+ updated_params = updated_user.get_api_data()
+ updated_params.update({'password_confirmation': ''})
+ updated_params.update({'new_password': ''})
+
+ del params['csrf_token']
+ assert params == updated_params
+
+ def test_update_and_migrate_password(
+ self, autologin_user, real_crypto_backend, user_util):
+
+ user = user_util.create_user()
+ temp_user = user.username
+ user.password = auth._RhodeCodeCryptoSha256().hash_create(
+ b'test123')
+ Session().add(user)
+ Session().commit()
+
+ params = user.get_api_data()
+
+ params.update({
+ 'password_confirmation': 'qweqwe123',
+ 'new_password': 'qweqwe123',
+ 'language': 'en',
+ 'csrf_token': autologin_user.csrf_token,
+ })
+
+ response = self.app.post(
+ route_path('user_update', user_id=user.user_id), params)
+ assert response.status_int == 302
+ assert_session_flash(response, 'User updated successfully')
+
+ # new password should be bcrypted, after log-in and transfer
+ user = User.get_by_username(temp_user)
+ assert user.password.startswith('$')
+
+ updated_user = User.get_by_username(temp_user)
+ updated_params = updated_user.get_api_data()
+ updated_params.update({'password_confirmation': 'qweqwe123'})
+ updated_params.update({'new_password': 'qweqwe123'})
+
+ del params['csrf_token']
+ assert params == updated_params
+
+ def test_delete(self):
+ self.log_user()
+ username = 'newtestuserdeleteme'
+
+ fixture.create_user(name=username)
+
+ new_user = Session().query(User)\
+ .filter(User.username == username).one()
+ response = self.app.post(
+ route_path('user_delete', user_id=new_user.user_id),
+ params={'csrf_token': self.csrf_token})
+
+ assert_session_flash(response, 'Successfully deleted user')
+
+ def test_delete_owner_of_repository(self, request, user_util):
+ self.log_user()
+ obj_name = 'test_repo'
+ usr = user_util.create_user()
+ username = usr.username
+ fixture.create_repo(obj_name, cur_user=usr.username)
+
+ new_user = Session().query(User)\
+ .filter(User.username == username).one()
+ response = self.app.post(
+ route_path('user_delete', user_id=new_user.user_id),
+ params={'csrf_token': self.csrf_token})
+
+ msg = 'user "%s" still owns 1 repositories and cannot be removed. ' \
+ 'Switch owners or remove those repositories:%s' % (username,
+ obj_name)
+ assert_session_flash(response, msg)
+ fixture.destroy_repo(obj_name)
+
+ def test_delete_owner_of_repository_detaching(self, request, user_util):
+ self.log_user()
+ obj_name = 'test_repo'
+ usr = user_util.create_user(auto_cleanup=False)
+ username = usr.username
+ fixture.create_repo(obj_name, cur_user=usr.username)
+
+ new_user = Session().query(User)\
+ .filter(User.username == username).one()
+ response = self.app.post(
+ route_path('user_delete', user_id=new_user.user_id),
+ params={'user_repos': 'detach', 'csrf_token': self.csrf_token})
+
+ msg = 'Detached 1 repositories'
+ assert_session_flash(response, msg)
+ fixture.destroy_repo(obj_name)
+
+ def test_delete_owner_of_repository_deleting(self, request, user_util):
+ self.log_user()
+ obj_name = 'test_repo'
+ usr = user_util.create_user(auto_cleanup=False)
+ username = usr.username
+ fixture.create_repo(obj_name, cur_user=usr.username)
+
+ new_user = Session().query(User)\
+ .filter(User.username == username).one()
+ response = self.app.post(
+ route_path('user_delete', user_id=new_user.user_id),
+ params={'user_repos': 'delete', 'csrf_token': self.csrf_token})
+
+ msg = 'Deleted 1 repositories'
+ assert_session_flash(response, msg)
+
+ def test_delete_owner_of_repository_group(self, request, user_util):
+ self.log_user()
+ obj_name = 'test_group'
+ usr = user_util.create_user()
+ username = usr.username
+ fixture.create_repo_group(obj_name, cur_user=usr.username)
+
+ new_user = Session().query(User)\
+ .filter(User.username == username).one()
+ response = self.app.post(
+ route_path('user_delete', user_id=new_user.user_id),
+ params={'csrf_token': self.csrf_token})
+
+ msg = 'user "%s" still owns 1 repository groups and cannot be removed. ' \
+ 'Switch owners or remove those repository groups:%s' % (username,
+ obj_name)
+ assert_session_flash(response, msg)
+ fixture.destroy_repo_group(obj_name)
+
+ def test_delete_owner_of_repository_group_detaching(self, request, user_util):
+ self.log_user()
+ obj_name = 'test_group'
+ usr = user_util.create_user(auto_cleanup=False)
+ username = usr.username
+ fixture.create_repo_group(obj_name, cur_user=usr.username)
+
+ new_user = Session().query(User)\
+ .filter(User.username == username).one()
+ response = self.app.post(
+ route_path('user_delete', user_id=new_user.user_id),
+ params={'user_repo_groups': 'delete', 'csrf_token': self.csrf_token})
+
+ msg = 'Deleted 1 repository groups'
+ assert_session_flash(response, msg)
+
+ def test_delete_owner_of_repository_group_deleting(self, request, user_util):
+ self.log_user()
+ obj_name = 'test_group'
+ usr = user_util.create_user(auto_cleanup=False)
+ username = usr.username
+ fixture.create_repo_group(obj_name, cur_user=usr.username)
+
+ new_user = Session().query(User)\
+ .filter(User.username == username).one()
+ response = self.app.post(
+ route_path('user_delete', user_id=new_user.user_id),
+ params={'user_repo_groups': 'detach', 'csrf_token': self.csrf_token})
+
+ msg = 'Detached 1 repository groups'
+ assert_session_flash(response, msg)
+ fixture.destroy_repo_group(obj_name)
+
+ def test_delete_owner_of_user_group(self, request, user_util):
+ self.log_user()
+ obj_name = 'test_user_group'
+ usr = user_util.create_user()
+ username = usr.username
+ fixture.create_user_group(obj_name, cur_user=usr.username)
+
+ new_user = Session().query(User)\
+ .filter(User.username == username).one()
+ response = self.app.post(
+ route_path('user_delete', user_id=new_user.user_id),
+ params={'csrf_token': self.csrf_token})
+
+ msg = 'user "%s" still owns 1 user groups and cannot be removed. ' \
+ 'Switch owners or remove those user groups:%s' % (username,
+ obj_name)
+ assert_session_flash(response, msg)
+ fixture.destroy_user_group(obj_name)
+
+ def test_delete_owner_of_user_group_detaching(self, request, user_util):
+ self.log_user()
+ obj_name = 'test_user_group'
+ usr = user_util.create_user(auto_cleanup=False)
+ username = usr.username
+ fixture.create_user_group(obj_name, cur_user=usr.username)
+
+ new_user = Session().query(User)\
+ .filter(User.username == username).one()
+ try:
+ response = self.app.post(
+ route_path('user_delete', user_id=new_user.user_id),
+ params={'user_user_groups': 'detach',
+ 'csrf_token': self.csrf_token})
+
+ msg = 'Detached 1 user groups'
+ assert_session_flash(response, msg)
+ finally:
+ fixture.destroy_user_group(obj_name)
+
+ def test_delete_owner_of_user_group_deleting(self, request, user_util):
+ self.log_user()
+ obj_name = 'test_user_group'
+ usr = user_util.create_user(auto_cleanup=False)
+ username = usr.username
+ fixture.create_user_group(obj_name, cur_user=usr.username)
+
+ new_user = Session().query(User)\
+ .filter(User.username == username).one()
+ response = self.app.post(
+ route_path('user_delete', user_id=new_user.user_id),
+ params={'user_user_groups': 'delete', 'csrf_token': self.csrf_token})
+
+ msg = 'Deleted 1 user groups'
+ assert_session_flash(response, msg)
+
+ def test_edit(self, user_util):
+ self.log_user()
+ user = user_util.create_user()
+ self.app.get(route_path('user_edit', user_id=user.user_id))
+
+ def test_edit_default_user_redirect(self):
+ self.log_user()
+ user = User.get_default_user()
+ self.app.get(route_path('user_edit', user_id=user.user_id), status=302)
+
+ @pytest.mark.parametrize(
+ 'repo_create, repo_create_write, user_group_create, repo_group_create,'
+ 'fork_create, inherit_default_permissions, expect_error,'
+ 'expect_form_error', [
+ ('hg.create.none', 'hg.create.write_on_repogroup.false',
+ 'hg.usergroup.create.false', 'hg.repogroup.create.false',
+ 'hg.fork.none', 'hg.inherit_default_perms.false', False, False),
+ ('hg.create.repository', 'hg.create.write_on_repogroup.false',
+ 'hg.usergroup.create.false', 'hg.repogroup.create.false',
+ 'hg.fork.none', 'hg.inherit_default_perms.false', False, False),
+ ('hg.create.repository', 'hg.create.write_on_repogroup.true',
+ 'hg.usergroup.create.true', 'hg.repogroup.create.true',
+ 'hg.fork.repository', 'hg.inherit_default_perms.false', False,
+ False),
+ ('hg.create.XXX', 'hg.create.write_on_repogroup.true',
+ 'hg.usergroup.create.true', 'hg.repogroup.create.true',
+ 'hg.fork.repository', 'hg.inherit_default_perms.false', False,
+ True),
+ ('', '', '', '', '', '', True, False),
+ ])
+ def test_global_perms_on_user(
+ self, repo_create, repo_create_write, user_group_create,
+ repo_group_create, fork_create, expect_error, expect_form_error,
+ inherit_default_permissions, user_util):
+ self.log_user()
+ user = user_util.create_user()
+ uid = user.user_id
+
+ # ENABLE REPO CREATE ON A GROUP
+ perm_params = {
+ 'inherit_default_permissions': False,
+ 'default_repo_create': repo_create,
+ 'default_repo_create_on_write': repo_create_write,
+ 'default_user_group_create': user_group_create,
+ 'default_repo_group_create': repo_group_create,
+ 'default_fork_create': fork_create,
+ 'default_inherit_default_permissions': inherit_default_permissions,
+ 'csrf_token': self.csrf_token,
+ }
+ response = self.app.post(
+ route_path('user_edit_global_perms_update', user_id=uid),
+ params=perm_params)
+
+ if expect_form_error:
+ assert response.status_int == 200
+ response.mustcontain('Value must be one of')
+ else:
+ if expect_error:
+ msg = 'An error occurred during permissions saving'
+ else:
+ msg = 'User global permissions updated successfully'
+ ug = User.get(uid)
+ del perm_params['inherit_default_permissions']
+ del perm_params['csrf_token']
+ assert perm_params == ug.get_default_perms()
+ assert_session_flash(response, msg)
+
+ def test_global_permissions_initial_values(self, user_util):
+ self.log_user()
+ user = user_util.create_user()
+ uid = user.user_id
+ response = self.app.get(
+ route_path('user_edit_global_perms', user_id=uid))
+ default_user = User.get_default_user()
+ default_permissions = default_user.get_default_perms()
+ assert_response = response.assert_response()
+ expected_permissions = (
+ 'default_repo_create', 'default_repo_create_on_write',
+ 'default_fork_create', 'default_repo_group_create',
+ 'default_user_group_create', 'default_inherit_default_permissions')
+ for permission in expected_permissions:
+ css_selector = '[name={}][checked=checked]'.format(permission)
+ element = assert_response.get_element(css_selector)
+ assert element.value == default_permissions[permission]
+
+ def test_perms_summary_page(self):
+ user = self.log_user()
+ response = self.app.get(
+ route_path('edit_user_perms_summary', user_id=user['user_id']))
+ for repo in Repository.query().all():
+ response.mustcontain(repo.repo_name)
+
+ def test_perms_summary_page_json(self):
+ user = self.log_user()
+ response = self.app.get(
+ route_path('edit_user_perms_summary_json', user_id=user['user_id']))
+ for repo in Repository.query().all():
+ response.mustcontain(repo.repo_name)
+
+ def test_audit_log_page(self):
+ user = self.log_user()
+ self.app.get(
+ route_path('edit_user_audit_logs', user_id=user['user_id']))
diff --git a/rhodecode/apps/admin/tests/test_admin_users_ssh_keys.py b/rhodecode/apps/admin/tests/test_admin_users_ssh_keys.py
new file mode 100644
--- /dev/null
+++ b/rhodecode/apps/admin/tests/test_admin_users_ssh_keys.py
@@ -0,0 +1,173 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2010-2017 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see %s
""" % datetime.date.today())
+ response = self.app.get(route_path('journal'))
+ # response.mustcontain(
+ # """%s
""" % datetime.date.today())
+
+ @pytest.mark.parametrize("feed_type, content_type", [
+ ('rss', "application/rss+xml"),
+ ('atom', "application/atom+xml")
+ ])
+ def test_journal_feed(self, feed_type, content_type):
+ self.log_user()
+ response = self.app.get(
+ route_path(
+ 'journal_{}'.format(feed_type)),
+ status=200)
+
+ assert response.content_type == content_type
def test_toggle_following_repository(self, backend):
user = self.log_user()
repo = Repository.get_by_repo_name(backend.repo_name)
repo_id = repo.repo_id
- self.app.post(url('toggle_following'), {'follows_repo_id': repo_id,
- 'csrf_token': self.csrf_token})
+ self.app.post(
+ route_path('toggle_following'), {'follows_repo_id': repo_id,
+ 'csrf_token': self.csrf_token})
followings = UserFollowing.query()\
.filter(UserFollowing.user_id == user['user_id'])\
@@ -44,8 +82,9 @@ class TestJournalController(TestControll
assert len(followings) == 0
- self.app.post(url('toggle_following'), {'follows_repo_id': repo_id,
- 'csrf_token': self.csrf_token})
+ self.app.post(
+ route_path('toggle_following'), {'follows_repo_id': repo_id,
+ 'csrf_token': self.csrf_token})
followings = UserFollowing.query()\
.filter(UserFollowing.user_id == user['user_id'])\
@@ -53,12 +92,15 @@ class TestJournalController(TestControll
assert len(followings) == 1
- def test_public_journal_atom(self):
+ @pytest.mark.parametrize("feed_type, content_type", [
+ ('rss', "application/rss+xml"),
+ ('atom', "application/atom+xml")
+ ])
+ def test_public_journal_feed(self, feed_type, content_type):
self.log_user()
- response = self.app.get(url(controller='journal',
- action='public_journal_atom'),)
+ response = self.app.get(
+ route_path(
+ 'journal_public_{}'.format(feed_type)),
+ status=200)
- def test_public_journal_rss(self):
- self.log_user()
- response = self.app.get(url(controller='journal',
- action='public_journal_rss'),)
+ assert response.content_type == content_type
diff --git a/rhodecode/apps/journal/views.py b/rhodecode/apps/journal/views.py
new file mode 100644
--- /dev/null
+++ b/rhodecode/apps/journal/views.py
@@ -0,0 +1,384 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2010-2017 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see No notifications here yet
')
- cur_user = self._get_logged_user()
- notif = NotificationModel().create(
- created_by=u1, notification_subject=u'test_notification_1',
- notification_body=u'notification_1', recipients=[cur_user])
+ notification = NotificationModel().create(
+ created_by=user_id, notification_subject=u'test_notification_1',
+ notification_body=u'notification_1', recipients=[user_id])
Session().commit()
- response = self.app.get(url('notifications'))
- response.mustcontain('id="notification_%s"' % notif.notification_id)
+ notification_id = notification.notification_id
+
+ response = self.app.get(route_path('notifications_show_all',
+ params={'type': 'all'}))
+ response.mustcontain('id="notification_%s"' % notification_id)
+
+ def test_show_unread(self, user_util):
+ user = user_util.create_user(password='qweqwe')
+ user_id = user.user_id
+ self.log_user(user.username, 'qweqwe')
+
+ response = self.app.get(route_path('notifications_show_all'))
+ response.mustcontain(
+ 'No notifications here yet
')
+
+ notification = NotificationModel().create(
+ created_by=user_id, notification_subject=u'test_notification_1',
+ notification_body=u'notification_1', recipients=[user_id])
+
+ # mark the USER notification as unread
+ user_notification = NotificationModel().get_user_notification(
+ user_id, notification)
+ user_notification.read = False
+
+ Session().commit()
+ notification_id = notification.notification_id
+
+ response = self.app.get(route_path('notifications_show_all'))
+ response.mustcontain('id="notification_%s"' % notification_id)
+ response.mustcontain('
.
+#
+# This program is dual-licensed. If you wish to learn more about the
+# RhodeCode Enterprise Edition, including its added features, Support services,
+# and proprietary license terms, please see https://rhodecode.com/licenses/
diff --git a/rhodecode/apps/my_account/views.py b/rhodecode/apps/my_account/views/my_account.py
rename from rhodecode/apps/my_account/views.py
rename to rhodecode/apps/my_account/views/my_account.py
--- a/rhodecode/apps/my_account/views.py
+++ b/rhodecode/apps/my_account/views/my_account.py
@@ -22,22 +22,29 @@ import logging
import datetime
import formencode
+import formencode.htmlfill
from pyramid.httpexceptions import HTTPFound
from pyramid.view import view_config
+from pyramid.renderers import render
+from pyramid.response import Response
-from rhodecode.apps._base import BaseAppView
+from rhodecode.apps._base import BaseAppView, DataGridAppView
from rhodecode import forms
from rhodecode.lib import helpers as h
from rhodecode.lib import audit_logger
from rhodecode.lib.ext_json import json
from rhodecode.lib.auth import LoginRequired, NotAnonymous, CSRFRequired
-from rhodecode.lib.channelstream import channelstream_request, \
- ChannelstreamException
-from rhodecode.lib.utils2 import safe_int, md5
+from rhodecode.lib.channelstream import (
+ channelstream_request, ChannelstreamException)
+from rhodecode.lib.utils2 import safe_int, md5, str2bool
from rhodecode.model.auth_token import AuthTokenModel
+from rhodecode.model.comment import CommentsModel
from rhodecode.model.db import (
- Repository, UserEmailMap, UserApiKeys, UserFollowing, joinedload)
+ Repository, UserEmailMap, UserApiKeys, UserFollowing, joinedload,
+ PullRequest)
+from rhodecode.model.forms import UserForm
from rhodecode.model.meta import Session
+from rhodecode.model.pull_request import PullRequestModel
from rhodecode.model.scm import RepoList
from rhodecode.model.user import UserModel
from rhodecode.model.repo import RepoModel
@@ -46,7 +53,7 @@ from rhodecode.model.validation_schema.s
log = logging.getLogger(__name__)
-class MyAccountView(BaseAppView):
+class MyAccountView(BaseAppView, DataGridAppView):
ALLOW_SCOPED_TOKENS = False
"""
This view has alternative version inside EE, if modified please take a look
@@ -84,7 +91,9 @@ class MyAccountView(BaseAppView):
username=c.user.username)
form = forms.Form(
- schema, buttons=(forms.buttons.save, forms.buttons.reset))
+ schema,
+ action=h.route_path('my_account_password_update'),
+ buttons=(forms.buttons.save, forms.buttons.reset))
c.form = form
return self._get_template_context(c)
@@ -93,7 +102,7 @@ class MyAccountView(BaseAppView):
@NotAnonymous()
@CSRFRequired()
@view_config(
- route_name='my_account_password', request_method='POST',
+ route_name='my_account_password_update', request_method='POST',
renderer='rhodecode:templates/admin/my_account/my_account.mako')
def my_account_password_update(self):
_ = self.request.translate
@@ -143,21 +152,14 @@ class MyAccountView(BaseAppView):
c = self.load_default_context()
c.active = 'auth_tokens'
-
- c.lifetime_values = [
- (str(-1), _('forever')),
- (str(5), _('5 minutes')),
- (str(60), _('1 hour')),
- (str(60 * 24), _('1 day')),
- (str(60 * 24 * 30), _('1 month')),
- ]
- c.lifetime_options = [(c.lifetime_values, _("Lifetime"))]
+ c.lifetime_values = AuthTokenModel.get_lifetime_values(translator=_)
c.role_values = [
(x, AuthTokenModel.cls._get_role_name(x))
for x in AuthTokenModel.cls.ROLES]
c.role_options = [(c.role_values, _("Role"))]
c.user_auth_tokens = AuthTokenModel().get_auth_tokens(
c.user.user_id, show_expired=True)
+ c.role_vcs = AuthTokenModel.cls.ROLE_VCS
return self._get_template_context(c)
def maybe_attach_token_scope(self, token):
@@ -203,7 +205,7 @@ class MyAccountView(BaseAppView):
del_auth_token = self.request.POST.get('del_auth_token')
if del_auth_token:
- token = UserApiKeys.get_or_404(del_auth_token, pyramid_exc=True)
+ token = UserApiKeys.get_or_404(del_auth_token)
token_data = token.get_api_data()
AuthTokenModel().delete(del_auth_token, c.user.user_id)
@@ -271,7 +273,7 @@ class MyAccountView(BaseAppView):
del_email_id = self.request.POST.get('del_email_id')
if del_email_id:
- email = UserEmailMap.get_or_404(del_email_id, pyramid_exc=True).email
+ email = UserEmailMap.get_or_404(del_email_id).email
UserModel().delete_extra_email(c.user.user_id, del_email_id)
audit_logger.store_web(
'user.edit.email.delete', action_data={
@@ -396,4 +398,183 @@ class MyAccountView(BaseAppView):
new_status = not user.user_data.get('notification_status', True)
user.update_userdata(notification_status=new_status)
Session().commit()
- return user.user_data['notification_status']
\ No newline at end of file
+ return user.user_data['notification_status']
+
+ @LoginRequired()
+ @NotAnonymous()
+ @view_config(
+ route_name='my_account_edit',
+ request_method='GET',
+ renderer='rhodecode:templates/admin/my_account/my_account.mako')
+ def my_account_edit(self):
+ c = self.load_default_context()
+ c.active = 'profile_edit'
+
+ c.perm_user = c.auth_user
+ c.extern_type = c.user.extern_type
+ c.extern_name = c.user.extern_name
+
+ defaults = c.user.get_dict()
+
+ data = render('rhodecode:templates/admin/my_account/my_account.mako',
+ self._get_template_context(c), self.request)
+ html = formencode.htmlfill.render(
+ data,
+ defaults=defaults,
+ encoding="UTF-8",
+ force_defaults=False
+ )
+ return Response(html)
+
+ @LoginRequired()
+ @NotAnonymous()
+ @CSRFRequired()
+ @view_config(
+ route_name='my_account_update',
+ request_method='POST',
+ renderer='rhodecode:templates/admin/my_account/my_account.mako')
+ def my_account_update(self):
+ _ = self.request.translate
+ c = self.load_default_context()
+ c.active = 'profile_edit'
+
+ c.perm_user = c.auth_user
+ c.extern_type = c.user.extern_type
+ c.extern_name = c.user.extern_name
+
+ _form = UserForm(edit=True,
+ old_data={'user_id': self._rhodecode_user.user_id,
+ 'email': self._rhodecode_user.email})()
+ form_result = {}
+ try:
+ post_data = dict(self.request.POST)
+ post_data['new_password'] = ''
+ post_data['password_confirmation'] = ''
+ form_result = _form.to_python(post_data)
+ # skip updating those attrs for my account
+ skip_attrs = ['admin', 'active', 'extern_type', 'extern_name',
+ 'new_password', 'password_confirmation']
+ # TODO: plugin should define if username can be updated
+ if c.extern_type != "rhodecode":
+ # forbid updating username for external accounts
+ skip_attrs.append('username')
+
+ UserModel().update_user(
+ self._rhodecode_user.user_id, skip_attrs=skip_attrs,
+ **form_result)
+ h.flash(_('Your account was updated successfully'),
+ category='success')
+ Session().commit()
+
+ except formencode.Invalid as errors:
+ data = render(
+ 'rhodecode:templates/admin/my_account/my_account.mako',
+ self._get_template_context(c), self.request)
+
+ html = formencode.htmlfill.render(
+ data,
+ defaults=errors.value,
+ errors=errors.error_dict or {},
+ prefix_error=False,
+ encoding="UTF-8",
+ force_defaults=False)
+ return Response(html)
+
+ except Exception:
+ log.exception("Exception updating user")
+ h.flash(_('Error occurred during update of user %s')
+ % form_result.get('username'), category='error')
+ raise HTTPFound(h.route_path('my_account_profile'))
+
+ raise HTTPFound(h.route_path('my_account_profile'))
+
+ def _get_pull_requests_list(self, statuses):
+ draw, start, limit = self._extract_chunk(self.request)
+ search_q, order_by, order_dir = self._extract_ordering(self.request)
+ _render = self.request.get_partial_renderer(
+ 'data_table/_dt_elements.mako')
+
+ pull_requests = PullRequestModel().get_im_participating_in(
+ user_id=self._rhodecode_user.user_id,
+ statuses=statuses,
+ offset=start, length=limit, order_by=order_by,
+ order_dir=order_dir)
+
+ pull_requests_total_count = PullRequestModel().count_im_participating_in(
+ user_id=self._rhodecode_user.user_id, statuses=statuses)
+
+ data = []
+ comments_model = CommentsModel()
+ for pr in pull_requests:
+ repo_id = pr.target_repo_id
+ comments = comments_model.get_all_comments(
+ repo_id, pull_request=pr)
+ owned = pr.user_id == self._rhodecode_user.user_id
+
+ data.append({
+ 'target_repo': _render('pullrequest_target_repo',
+ pr.target_repo.repo_name),
+ 'name': _render('pullrequest_name',
+ pr.pull_request_id, pr.target_repo.repo_name,
+ short=True),
+ 'name_raw': pr.pull_request_id,
+ 'status': _render('pullrequest_status',
+ pr.calculated_review_status()),
+ 'title': _render(
+ 'pullrequest_title', pr.title, pr.description),
+ 'description': h.escape(pr.description),
+ 'updated_on': _render('pullrequest_updated_on',
+ h.datetime_to_time(pr.updated_on)),
+ 'updated_on_raw': h.datetime_to_time(pr.updated_on),
+ 'created_on': _render('pullrequest_updated_on',
+ h.datetime_to_time(pr.created_on)),
+ 'created_on_raw': h.datetime_to_time(pr.created_on),
+ 'author': _render('pullrequest_author',
+ pr.author.full_contact, ),
+ 'author_raw': pr.author.full_name,
+ 'comments': _render('pullrequest_comments', len(comments)),
+ 'comments_raw': len(comments),
+ 'closed': pr.is_closed(),
+ 'owned': owned
+ })
+
+ # json used to render the grid
+ data = ({
+ 'draw': draw,
+ 'data': data,
+ 'recordsTotal': pull_requests_total_count,
+ 'recordsFiltered': pull_requests_total_count,
+ })
+ return data
+
+ @LoginRequired()
+ @NotAnonymous()
+ @view_config(
+ route_name='my_account_pullrequests',
+ request_method='GET',
+ renderer='rhodecode:templates/admin/my_account/my_account.mako')
+ def my_account_pullrequests(self):
+ c = self.load_default_context()
+ c.active = 'pullrequests'
+ req_get = self.request.GET
+
+ c.closed = str2bool(req_get.get('pr_show_closed'))
+
+ return self._get_template_context(c)
+
+ @LoginRequired()
+ @NotAnonymous()
+ @view_config(
+ route_name='my_account_pullrequests_data',
+ request_method='GET', renderer='json_ext')
+ def my_account_pullrequests_data(self):
+ req_get = self.request.GET
+ closed = str2bool(req_get.get('closed'))
+
+ statuses = [PullRequest.STATUS_NEW, PullRequest.STATUS_OPEN]
+ if closed:
+ statuses += [PullRequest.STATUS_CLOSED]
+
+ data = self._get_pull_requests_list(statuses=statuses)
+ return data
+
diff --git a/rhodecode/apps/my_account/views/my_account_notifications.py b/rhodecode/apps/my_account/views/my_account_notifications.py
new file mode 100644
--- /dev/null
+++ b/rhodecode/apps/my_account/views/my_account_notifications.py
@@ -0,0 +1,198 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2010-2017 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see "{}" is not one of -1'.format(
+ group.group_id))
diff --git a/rhodecode/apps/repo_group/views/__init__.py b/rhodecode/apps/repo_group/views/__init__.py
new file mode 100644
--- /dev/null
+++ b/rhodecode/apps/repo_group/views/__init__.py
@@ -0,0 +1,19 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2016-2017 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see r(\d+):[\da-f]+ ')
+def route_path(name, params=None, **kwargs):
+ import urllib
+
+ base_url = {
+ 'repo_changelog':'/{repo_name}/changelog',
+ 'repo_changelog_file':'/{repo_name}/changelog/{commit_id}/{f_path}',
+ 'repo_changelog_elements':'/{repo_name}/changelog_elements',
+ }[name].format(**kwargs)
+
+ if params:
+ base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
+ return base_url
+
+
class TestChangelogController(TestController):
- def test_index(self, backend):
+ def test_changelog(self, backend):
self.log_user()
- response = self.app.get(url(controller='changelog', action='index',
- repo_name=backend.repo_name))
+ response = self.app.get(
+ route_path('repo_changelog', repo_name=backend.repo_name))
first_idx = -1
last_idx = -DEFAULT_CHANGELOG_SIZE
@@ -43,39 +55,30 @@ class TestChangelogController(TestContro
response, first_idx, last_idx, backend)
@pytest.mark.backends("hg", "git")
- def test_index_filtered_by_branch(self, backend):
+ def test_changelog_filtered_by_branch(self, backend):
self.log_user()
self.app.get(
- url(
- controller='changelog',
- action='index',
- repo_name=backend.repo_name,
- branch=backend.default_branch_name),
+ route_path('repo_changelog', repo_name=backend.repo_name,
+ params=dict(branch=backend.default_branch_name)),
status=200)
@pytest.mark.backends("svn")
- def test_index_filtered_by_branch_svn(self, autologin_user, backend):
+ def test_changelog_filtered_by_branch_svn(self, autologin_user, backend):
repo = backend['svn-simple-layout']
response = self.app.get(
- url(
- controller='changelog',
- action='index',
- repo_name=repo.repo_name,
- branch='trunk'),
+ route_path('repo_changelog', repo_name=repo.repo_name,
+ params=dict(branch='trunk')),
status=200)
self.assert_commits_on_page(
response, indexes=[15, 12, 7, 3, 2, 1])
- def test_index_filtered_by_wrong_branch(self, backend):
+ def test_changelog_filtered_by_wrong_branch(self, backend):
self.log_user()
branch = 'wrong-branch-name'
response = self.app.get(
- url(
- controller='changelog',
- action='index',
- repo_name=backend.repo_name,
- branch=branch),
+ route_path('repo_changelog', repo_name=backend.repo_name,
+ params=dict(branch=branch)),
status=302)
expected_url = '/{repo}/changelog/{branch}'.format(
repo=backend.repo_name, branch=branch)
@@ -89,7 +92,7 @@ class TestChangelogController(TestContro
assert found_indexes == indexes
@pytest.mark.xfail_backends("svn", reason="Depends on branch support")
- def test_index_filtered_by_branch_with_merges(
+ def test_changelog_filtered_by_branch_with_merges(
self, autologin_user, backend):
# Note: The changelog of branch "b" does not contain the commit "a1"
@@ -104,33 +107,27 @@ class TestChangelogController(TestContro
backend.create_repo(commits)
self.app.get(
- url('changelog_home',
- controller='changelog',
- action='index',
- repo_name=backend.repo_name,
- branch='b'),
+ route_path('repo_changelog', repo_name=backend.repo_name,
+ params=dict(branch='b')),
status=200)
@pytest.mark.backends("hg")
- def test_index_closed_branches(self, autologin_user, backend):
+ def test_changelog_closed_branches(self, autologin_user, backend):
repo = backend['closed_branch']
response = self.app.get(
- url(
- controller='changelog',
- action='index',
- repo_name=repo.repo_name,
- branch='experimental'),
+ route_path('repo_changelog', repo_name=repo.repo_name,
+ params=dict(branch='experimental')),
status=200)
self.assert_commits_on_page(
response, indexes=[3, 1])
- def test_index_pagination(self, backend):
+ def test_changelog_pagination(self, backend):
self.log_user()
# pagination, walk up to page 6
- changelog_url = url(
- controller='changelog', action='index',
- repo_name=backend.repo_name)
+ changelog_url = route_path(
+ 'repo_changelog', repo_name=backend.repo_name)
+
for page in range(1, 7):
response = self.app.get(changelog_url, {'page': page})
@@ -166,27 +163,33 @@ class TestChangelogController(TestContro
first_commit_of_next_page.idx, first_commit_of_next_page.short_id)
assert first_span_of_next_page not in response
- def test_index_with_filenode(self, backend):
+ @pytest.mark.parametrize('test_path', [
+ 'vcs/exceptions.py',
+ '/vcs/exceptions.py',
+ '//vcs/exceptions.py'
+ ])
+ def test_changelog_with_filenode(self, backend, test_path):
self.log_user()
- response = self.app.get(url(
- controller='changelog', action='index', revision='tip',
- f_path='/vcs/exceptions.py', repo_name=backend.repo_name))
+ response = self.app.get(
+ route_path('repo_changelog_file', repo_name=backend.repo_name,
+ commit_id='tip', f_path=test_path),
+ )
# history commits messages
response.mustcontain('Added exceptions module, this time for real')
response.mustcontain('Added not implemented hg backend test case')
response.mustcontain('Added BaseChangeset class')
- def test_index_with_filenode_that_is_dirnode(self, backend):
+ def test_changelog_with_filenode_that_is_dirnode(self, backend):
self.log_user()
- response = self.app.get(url(controller='changelog', action='index',
- revision='tip', f_path='/tests',
- repo_name=backend.repo_name))
- assert response.status == '302 Found'
+ self.app.get(
+ route_path('repo_changelog_file', repo_name=backend.repo_name,
+ commit_id='tip', f_path='/tests'),
+ status=302)
- def test_index_with_filenode_not_existing(self, backend):
+ def test_changelog_with_filenode_not_existing(self, backend):
self.log_user()
- response = self.app.get(url(controller='changelog', action='index',
- revision='tip', f_path='/wrong_path',
- repo_name=backend.repo_name))
- assert response.status == '302 Found'
+ self.app.get(
+ route_path('repo_changelog_file', repo_name=backend.repo_name,
+ commit_id='tip', f_path='wrong_path'),
+ status=302)
diff --git a/rhodecode/tests/functional/test_commit_comments.py b/rhodecode/apps/repository/tests/test_repo_commit_comments.py
rename from rhodecode/tests/functional/test_commit_comments.py
rename to rhodecode/apps/repository/tests/test_repo_commit_comments.py
--- a/rhodecode/tests/functional/test_commit_comments.py
+++ b/rhodecode/apps/repository/tests/test_repo_commit_comments.py
@@ -18,18 +18,33 @@
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
-from pylons.i18n import ungettext
import pytest
-from rhodecode.tests import *
+from rhodecode.tests import TestController
+
from rhodecode.model.db import (
ChangesetComment, Notification, UserNotification)
from rhodecode.model.meta import Session
from rhodecode.lib import helpers as h
+def route_path(name, params=None, **kwargs):
+ import urllib
+
+ base_url = {
+ 'repo_commit': '/{repo_name}/changeset/{commit_id}',
+ 'repo_commit_comment_create': '/{repo_name}/changeset/{commit_id}/comment/create',
+ 'repo_commit_comment_preview': '/{repo_name}/changeset/{commit_id}/comment/preview',
+ 'repo_commit_comment_delete': '/{repo_name}/changeset/{commit_id}/comment/{comment_id}/delete',
+ }[name].format(**kwargs)
+
+ if params:
+ base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
+ return base_url
+
+
@pytest.mark.backends("git", "hg", "svn")
-class TestCommitCommentsController(TestController):
+class TestRepoCommitCommentsView(TestController):
@pytest.fixture(autouse=True)
def prepare(self, request, pylonsapp):
@@ -62,12 +77,13 @@ class TestCommitCommentsController(TestC
params = {'text': text, 'csrf_token': self.csrf_token,
'comment_type': comment_type}
self.app.post(
- url(controller='changeset', action='comment',
- repo_name=backend.repo_name, revision=commit_id), params=params)
+ route_path('repo_commit_comment_create',
+ repo_name=backend.repo_name, commit_id=commit_id),
+ params=params)
response = self.app.get(
- url(controller='changeset', action='index',
- repo_name=backend.repo_name, revision=commit_id))
+ route_path('repo_commit',
+ repo_name=backend.repo_name, commit_id=commit_id))
# test DB
assert ChangesetComment.query().count() == 1
@@ -103,12 +119,13 @@ class TestCommitCommentsController(TestC
'csrf_token': self.csrf_token}
self.app.post(
- url(controller='changeset', action='comment',
- repo_name=backend.repo_name, revision=commit_id), params=params)
+ route_path('repo_commit_comment_create',
+ repo_name=backend.repo_name, commit_id=commit_id),
+ params=params)
response = self.app.get(
- url(controller='changeset', action='index',
- repo_name=backend.repo_name, revision=commit_id))
+ route_path('repo_commit',
+ repo_name=backend.repo_name, commit_id=commit_id))
# test DB
assert ChangesetComment.query().count() == 1
@@ -153,12 +170,13 @@ class TestCommitCommentsController(TestC
params = {'text': text, 'csrf_token': self.csrf_token}
self.app.post(
- url(controller='changeset', action='comment',
- repo_name=backend.repo_name, revision=commit_id), params=params)
+ route_path('repo_commit_comment_create',
+ repo_name=backend.repo_name, commit_id=commit_id),
+ params=params)
response = self.app.get(
- url(controller='changeset', action='index',
- repo_name=backend.repo_name, revision=commit_id))
+ route_path('repo_commit',
+ repo_name=backend.repo_name, commit_id=commit_id))
# test DB
assert ChangesetComment.query().count() == 1
assert_comment_links(response, ChangesetComment.query().count(), 0)
@@ -183,12 +201,14 @@ class TestCommitCommentsController(TestC
'csrf_token': self.csrf_token}
self.app.post(
- url(controller='changeset', action='comment',
- repo_name=backend.repo_name, revision=commit_id), params=params)
+ route_path(
+ 'repo_commit_comment_create',
+ repo_name=backend.repo_name, commit_id=commit_id),
+ params=params)
response = self.app.get(
- url(controller='changeset', action='index',
- repo_name=backend.repo_name, revision=commit_id))
+ route_path('repo_commit',
+ repo_name=backend.repo_name, commit_id=commit_id))
# test DB
assert ChangesetComment.query().count() == 1
@@ -218,9 +238,9 @@ class TestCommitCommentsController(TestC
params = {'text': text, 'csrf_token': self.csrf_token}
self.app.post(
- url(
- controller='changeset', action='comment',
- repo_name=backend.repo_name, revision=commit_id),
+ route_path(
+ 'repo_commit_comment_create',
+ repo_name=backend.repo_name, commit_id=commit_id),
params=params)
comments = ChangesetComment.query().all()
@@ -228,16 +248,18 @@ class TestCommitCommentsController(TestC
comment_id = comments[0].comment_id
self.app.post(
- url(controller='changeset', action='delete_comment',
- repo_name=backend.repo_name, comment_id=comment_id),
- params={'_method': 'delete', 'csrf_token': self.csrf_token})
+ route_path('repo_commit_comment_delete',
+ repo_name=backend.repo_name,
+ commit_id=commit_id,
+ comment_id=comment_id),
+ params={'csrf_token': self.csrf_token})
comments = ChangesetComment.query().all()
assert len(comments) == 0
response = self.app.get(
- url(controller='changeset', action='index',
- repo_name=backend.repo_name, revision=commit_id))
+ route_path('repo_commit',
+ repo_name=backend.repo_name, commit_id=commit_id))
assert_comment_links(response, 0, 0)
@pytest.mark.parametrize('renderer, input, output', [
@@ -251,36 +273,39 @@ class TestCommitCommentsController(TestC
('markdown', '**bold**', 'bold '),
], ids=['rst-plain', 'rst-header', 'rst-italics', 'rst-bold', 'md-plain',
'md-header', 'md-italics', 'md-bold', ])
- def test_preview(self, renderer, input, output, backend):
+ def test_preview(self, renderer, input, output, backend, xhr_header):
self.log_user()
params = {
'renderer': renderer,
'text': input,
'csrf_token': self.csrf_token
}
- environ = {
- 'HTTP_X_PARTIAL_XHR': 'true'
- }
+ commit_id = '0' * 16 # fake this for tests
response = self.app.post(
- url(controller='changeset',
- action='preview_comment',
- repo_name=backend.repo_name),
+ route_path('repo_commit_comment_preview',
+ repo_name=backend.repo_name, commit_id=commit_id,),
params=params,
- extra_environ=environ)
+ extra_environ=xhr_header)
response.mustcontain(output)
def assert_comment_links(response, comments, inline_comments):
- comments_text = ungettext("%d Commit comment",
- "%d Commit comments", comments) % comments
+ if comments == 1:
+ comments_text = "%d Commit comment" % comments
+ else:
+ comments_text = "%d Commit comments" % comments
+
+ if inline_comments == 1:
+ inline_comments_text = "%d Inline Comment" % inline_comments
+ else:
+ inline_comments_text = "%d Inline Comments" % inline_comments
+
if comments:
response.mustcontain('%s ,' % comments_text)
else:
response.mustcontain(comments_text)
- inline_comments_text = ungettext("%d Inline Comment", "%d Inline Comments",
- inline_comments) % inline_comments
if inline_comments:
response.mustcontain(
'id="inline-comments-counter">%s'],
+ "application/rss+xml"),
+ ('atom', ['xmlns="http://www.w3.org/2005/Atom"', 'xml:lang="en-us"'],
+ "application/atom+xml"),
+ ])
+ def test_feed(self, backend, feed_type, response_types, content_type):
self.log_user()
- response = self.app.get(url(controller='feed', action='rss',
- repo_name=backend.repo_name))
+ response = self.app.get(
+ route_path('{}_feed_home'.format(feed_type),
+ repo_name=backend.repo_name))
+
+ for content in response_types:
+ response.mustcontain(content)
- assert response.content_type == "application/rss+xml"
- assert """""" in response
+ assert response.content_type == content_type
- def test_rss_with_auth_token(self, backend, user_admin):
+ @pytest.mark.parametrize("feed_type, content_type", [
+ ('rss', "application/rss+xml"),
+ ('atom', "application/atom+xml")
+ ])
+ def test_feed_with_auth_token(
+ self, backend, user_admin, feed_type, content_type):
auth_token = user_admin.feed_token
assert auth_token != ''
- response = self.app.get(
- url(controller='feed', action='rss',
- repo_name=backend.repo_name, auth_token=auth_token,
- status=200))
- assert response.content_type == "application/rss+xml"
- assert """""" in response
+ response = self.app.get(
+ route_path(
+ '{}_feed_home'.format(feed_type),
+ repo_name=backend.repo_name,
+ params=dict(auth_token=auth_token)),
+ status=200)
- def test_rss_with_auth_token_of_wrong_type(self, backend, user_util):
+ assert response.content_type == content_type
+
+ @pytest.mark.parametrize("feed_type", ['rss', 'atom'])
+ def test_feed_with_auth_token_of_wrong_type(
+ self, backend, user_util, feed_type):
user = user_util.create_user()
auth_token = AuthTokenModel().create(
user.user_id, 'test-token', -1, AuthTokenModel.cls.ROLE_API)
auth_token = auth_token.api_key
self.app.get(
- url(controller='feed', action='rss',
- repo_name=backend.repo_name, auth_token=auth_token),
+ route_path(
+ '{}_feed_home'.format(feed_type),
+ repo_name=backend.repo_name,
+ params=dict(auth_token=auth_token)),
status=302)
auth_token = AuthTokenModel().create(
user.user_id, 'test-token', -1, AuthTokenModel.cls.ROLE_FEED)
auth_token = auth_token.api_key
self.app.get(
- url(controller='feed', action='rss',
- repo_name=backend.repo_name, auth_token=auth_token),
+ route_path(
+ '{}_feed_home'.format(feed_type),
+ repo_name=backend.repo_name,
+ params=dict(auth_token=auth_token)),
status=200)
-
- def test_atom(self, backend):
- self.log_user()
- response = self.app.get(url(controller='feed', action='atom',
- repo_name=backend.repo_name))
-
- assert response.content_type == """application/atom+xml"""
- assert """""" in response
-
- tag1 = ''
- tag2 = ''
- assert tag1 in response or tag2 in response
diff --git a/rhodecode/tests/functional/test_files.py b/rhodecode/apps/repository/tests/test_repo_files.py
rename from rhodecode/tests/functional/test_files.py
rename to rhodecode/apps/repository/tests/test_repo_files.py
--- a/rhodecode/tests/functional/test_files.py
+++ b/rhodecode/apps/repository/tests/test_repo_files.py
@@ -23,34 +23,92 @@ import os
import mock
import pytest
-from rhodecode.controllers.files import FilesController
+from rhodecode.apps.repository.views.repo_files import RepoFilesView
from rhodecode.lib import helpers as h
from rhodecode.lib.compat import OrderedDict
from rhodecode.lib.ext_json import json
from rhodecode.lib.vcs import nodes
from rhodecode.lib.vcs.conf import settings
-from rhodecode.tests import (
- url, assert_session_flash, assert_not_in_session_flash)
+from rhodecode.tests import assert_session_flash
from rhodecode.tests.fixture import Fixture
fixture = Fixture()
-NODE_HISTORY = {
- 'hg': json.loads(fixture.load_resource('hg_node_history_response.json')),
- 'git': json.loads(fixture.load_resource('git_node_history_response.json')),
- 'svn': json.loads(fixture.load_resource('svn_node_history_response.json')),
-}
+
+def get_node_history(backend_type):
+ return {
+ 'hg': json.loads(fixture.load_resource('hg_node_history_response.json')),
+ 'git': json.loads(fixture.load_resource('git_node_history_response.json')),
+ 'svn': json.loads(fixture.load_resource('svn_node_history_response.json')),
+ }[backend_type]
+
+
+def route_path(name, params=None, **kwargs):
+ import urllib
+ base_url = {
+ 'repo_archivefile': '/{repo_name}/archive/{fname}',
+ 'repo_files_diff': '/{repo_name}/diff/{f_path}',
+ 'repo_files_diff_2way_redirect': '/{repo_name}/diff-2way/{f_path}',
+ 'repo_files': '/{repo_name}/files/{commit_id}/{f_path}',
+ 'repo_files:default_path': '/{repo_name}/files/{commit_id}/',
+ 'repo_files:default_commit': '/{repo_name}/files',
+ 'repo_files:rendered': '/{repo_name}/render/{commit_id}/{f_path}',
+ 'repo_files:annotated': '/{repo_name}/annotate/{commit_id}/{f_path}',
+ 'repo_files:annotated_previous': '/{repo_name}/annotate-previous/{commit_id}/{f_path}',
+ 'repo_files_nodelist': '/{repo_name}/nodelist/{commit_id}/{f_path}',
+ 'repo_file_raw': '/{repo_name}/raw/{commit_id}/{f_path}',
+ 'repo_file_download': '/{repo_name}/download/{commit_id}/{f_path}',
+ 'repo_file_history': '/{repo_name}/history/{commit_id}/{f_path}',
+ 'repo_file_authors': '/{repo_name}/authors/{commit_id}/{f_path}',
+ 'repo_files_remove_file': '/{repo_name}/remove_file/{commit_id}/{f_path}',
+ 'repo_files_delete_file': '/{repo_name}/delete_file/{commit_id}/{f_path}',
+ 'repo_files_edit_file': '/{repo_name}/edit_file/{commit_id}/{f_path}',
+ 'repo_files_update_file': '/{repo_name}/update_file/{commit_id}/{f_path}',
+ 'repo_files_add_file': '/{repo_name}/add_file/{commit_id}/{f_path}',
+ 'repo_files_create_file': '/{repo_name}/create_file/{commit_id}/{f_path}',
+ 'repo_nodetree_full': '/{repo_name}/nodetree_full/{commit_id}/{f_path}',
+ 'repo_nodetree_full:default_path': '/{repo_name}/nodetree_full/{commit_id}/',
+ }[name].format(**kwargs)
+
+ if params:
+ base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
+ return base_url
+
+
+def assert_files_in_response(response, files, params):
+ template = (
+ 'href="/%(repo_name)s/files/%(commit_id)s/%(name)s"')
+ _assert_items_in_response(response, files, template, params)
+
+
+def assert_dirs_in_response(response, dirs, params):
+ template = (
+ 'href="/%(repo_name)s/files/%(commit_id)s/%(name)s"')
+ _assert_items_in_response(response, dirs, template, params)
+
+
+def _assert_items_in_response(response, items, template, params):
+ for item in items:
+ item_params = {'name': item}
+ item_params.update(params)
+ response.mustcontain(template % item_params)
+
+
+def assert_timeago_in_response(response, items, params):
+ for item in items:
+ response.mustcontain(h.age_component(params['date']))
@pytest.mark.usefixtures("app")
-class TestFilesController:
+class TestFilesViews(object):
- def test_index(self, backend):
- response = self.app.get(url(
- controller='files', action='index',
- repo_name=backend.repo_name, revision='tip', f_path='/'))
+ def test_show_files(self, backend):
+ response = self.app.get(
+ route_path('repo_files',
+ repo_name=backend.repo_name,
+ commit_id='tip', f_path='/'))
commit = backend.repo.get_commit()
params = {
@@ -77,21 +135,23 @@ class TestFilesController:
assert_files_in_response(response, files, params)
assert_timeago_in_response(response, files, params)
- def test_index_links_submodules_with_absolute_url(self, backend_hg):
+ def test_show_files_links_submodules_with_absolute_url(self, backend_hg):
repo = backend_hg['subrepos']
- response = self.app.get(url(
- controller='files', action='index',
- repo_name=repo.repo_name, revision='tip', f_path='/'))
+ response = self.app.get(
+ route_path('repo_files',
+ repo_name=repo.repo_name,
+ commit_id='tip', f_path='/'))
assert_response = response.assert_response()
assert_response.contains_one_link(
'absolute-path @ 000000000000', 'http://example.com/absolute-path')
- def test_index_links_submodules_with_absolute_url_subpaths(
+ def test_show_files_links_submodules_with_absolute_url_subpaths(
self, backend_hg):
repo = backend_hg['subrepos']
- response = self.app.get(url(
- controller='files', action='index',
- repo_name=repo.repo_name, revision='tip', f_path='/'))
+ response = self.app.get(
+ route_path('repo_files',
+ repo_name=repo.repo_name,
+ commit_id='tip', f_path='/'))
assert_response = response.assert_response()
assert_response.contains_one_link(
'subpaths-path @ 000000000000',
@@ -108,29 +168,29 @@ class TestFilesController:
backend.repo.landing_rev = "branch:%s" % new_branch
- # get response based on tip and not new revision
- response = self.app.get(url(
- controller='files', action='index',
- repo_name=backend.repo_name, revision='tip', f_path='/'),
- status=200)
+ # get response based on tip and not new commit
+ response = self.app.get(
+ route_path('repo_files',
+ repo_name=backend.repo_name,
+ commit_id='tip', f_path='/'))
- # make sure Files menu url is not tip but new revision
+ # make sure Files menu url is not tip but new commit
landing_rev = backend.repo.landing_rev[1]
- files_url = url('files_home', repo_name=backend.repo_name,
- revision=landing_rev)
+ files_url = route_path('repo_files:default_path',
+ repo_name=backend.repo_name,
+ commit_id=landing_rev)
assert landing_rev != 'tip'
- response.mustcontain('