@@ -1294,771 +1294,772 @@ class ApiController(JSONRPCController):
api_key belonging to user with admin rights or regular user that have
admin, write or read access to repository.
:param apiuser: filled automatically from apikey
:type apiuser: AuthUser
OUTPUT::
id : <id_given_in_input>
result: [
{
"repo_id" : "<repo_id>",
"repo_name" : "<reponame>"
"repo_type" : "<repo_type>",
"clone_uri" : "<clone_uri>",
"private": : "<bool>",
"created_on" : "<datetimecreated>",
"description" : "<description>",
"landing_rev": "<landing_rev>",
"owner": "<repo_owner>",
"fork_of": "<name_of_fork_parent>",
"enable_downloads": "<bool>",
"enable_locking": "<bool>",
"enable_statistics": "<bool>",
},
…
]
error: null
"""
result = []
if not HasPermissionAnyApi('hg.admin')(user=apiuser):
repos = RepoModel().get_all_user_repos(user=apiuser)
else:
repos = RepoModel().get_all()
for repo in repos:
result.append(repo.get_api_data())
return result
# permission check inside
def get_repo_nodes(self, apiuser, repoid, revision, root_path,
ret_type=Optional('all')):
returns a list of nodes and it's children in a flat list for a given path
at given revision. It's possible to specify ret_type to show only `files` or
`dirs`. This command can be executed only using api_key belonging to
user with admin rights or regular user that have at least read access to repository.
:param repoid: repository name or repository id
:type repoid: str or int
:param revision: revision for which listing should be done
:type revision: str
:param root_path: path from which start displaying
:type root_path: str
:param ret_type: return type 'all|files|dirs' nodes
:type ret_type: Optional(str)
"name" : "<name>"
"type" : "<type>",
repo = get_repo_or_error(repoid)
# check if we have admin permission for this repo !
perms = ('repository.admin', 'repository.write', 'repository.read')
if not HasRepoPermissionAnyApi(*perms)(user=apiuser, repo_name=repo.repo_name):
raise JSONRPCError('repository `%s` does not exist' % (repoid,))
ret_type = Optional.extract(ret_type)
_map = {}
try:
_d, _f = ScmModel().get_nodes(repo, revision, root_path,
flat=False)
_map = {
'all': _d + _f,
'files': _f,
'dirs': _d,
}
return _map[ret_type]
except KeyError:
raise JSONRPCError('ret_type must be one of %s'
% (','.join(_map.keys())))
except Exception:
log.error(traceback.format_exc())
raise JSONRPCError(
'failed to get repo: `%s` nodes' % repo.repo_name
)
@HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
def create_repo(self, apiuser, repo_name, owner=Optional(OAttr('apiuser')),
repo_type=Optional('hg'), description=Optional(''),
private=Optional(False), clone_uri=Optional(None),
landing_rev=Optional('rev:tip'),
enable_statistics=Optional(False),
enable_locking=Optional(False),
enable_downloads=Optional(False),
copy_permissions=Optional(False)):
Creates a repository. If repository name contains "/", all needed repository
groups will be created. For example "foo/bar/baz" will create groups
"foo", "bar" (with "foo" as parent), and create "baz" repository with
"bar" as group. This command can be executed only using api_key
belonging to user with admin rights or regular user that have create
repository permission. Regular users cannot specify owner parameter
:param repo_name: repository name
:type repo_name: str
:param owner: user_id or username
:type owner: Optional(str)
:param repo_type: 'hg' or 'git'
:type repo_type: Optional(str)
:param description: repository description
:type description: Optional(str)
:param private:
:type private: bool
:param clone_uri:
:type clone_uri: str
:param landing_rev: <rev_type>:<rev>
:type landing_rev: str
:param enable_locking:
:type enable_locking: bool
:param enable_downloads:
:type enable_downloads: bool
:param enable_statistics:
:type enable_statistics: bool
:param copy_permissions: Copy permission from group that repository is
being created.
:type copy_permissions: bool
result: {
"msg": "Created new repository `<reponame>`",
"success": true,
"task": "<celery task id or None if done sync>"
ERROR OUTPUT::
result : null
error : {
'failed to create repository `<repo_name>`
if not isinstance(owner, Optional):
#forbid setting owner for non-admins
'Only Kallithea admin can specify `owner` param'
if isinstance(owner, Optional):
owner = apiuser.user_id
owner = get_user_or_error(owner)
if RepoModel().get_by_repo_name(repo_name):
raise JSONRPCError("repo `%s` already exist" % repo_name)
defs = Setting.get_default_repo_settings(strip_prefix=True)
if isinstance(private, Optional):
private = defs.get('repo_private') or Optional.extract(private)
if isinstance(repo_type, Optional):
repo_type = defs.get('repo_type')
if isinstance(enable_statistics, Optional):
enable_statistics = defs.get('repo_enable_statistics')
if isinstance(enable_locking, Optional):
enable_locking = defs.get('repo_enable_locking')
if isinstance(enable_downloads, Optional):
enable_downloads = defs.get('repo_enable_downloads')
clone_uri = Optional.extract(clone_uri)
description = Optional.extract(description)
landing_rev = Optional.extract(landing_rev)
copy_permissions = Optional.extract(copy_permissions)
repo_name_cleaned = repo_name.split('/')[-1]
# create structure of groups and return the last group
repo_group = map_groups(repo_name)
data = dict(
repo_name=repo_name_cleaned,
repo_name_full=repo_name,
repo_type=repo_type,
repo_description=description,
owner=owner,
repo_private=private,
clone_uri=clone_uri,
repo_group=repo_group,
repo_landing_rev=landing_rev,
enable_statistics=enable_statistics,
enable_locking=enable_locking,
enable_downloads=enable_downloads,
repo_copy_permissions=copy_permissions,
task = RepoModel().create(form_data=data, cur_user=owner)
from celery.result import BaseAsyncResult
task_id = None
if isinstance(task, BaseAsyncResult):
task_id = task.task_id
# no commit, it's done in RepoModel, or async via celery
return dict(
msg="Created new repository `%s`" % (repo_name,),
success=True, # cannot return the repo data here since fork
# can be done async
task=task_id
'failed to create repository `%s`' % (repo_name,))
def update_repo(self, apiuser, repoid, name=Optional(None),
owner=Optional(OAttr('apiuser')),
group=Optional(None),
description=Optional(''), private=Optional(False),
clone_uri=Optional(None), landing_rev=Optional('rev:tip'),
enable_downloads=Optional(False)):
Updates repo
:param name:
:param owner:
:param group:
:param description:
:param landing_rev:
if not HasRepoPermissionAnyApi('repository.admin')(user=apiuser,
repo_name=repo.repo_name):
if (name != repo.repo_name and
not HasPermissionAnyApi('hg.create.repository')(user=apiuser)
):
raise JSONRPCError('no permission to create (or move) repositories')
updates = {}
repo_group = group
if not isinstance(repo_group, Optional):
repo_group = get_repo_group_or_error(repo_group)
repo_group = repo_group.group_id
store_update(updates, name, 'repo_name')
store_update(updates, repo_group, 'repo_group')
store_update(updates, owner, 'user')
store_update(updates, description, 'repo_description')
store_update(updates, private, 'repo_private')
store_update(updates, clone_uri, 'clone_uri')
store_update(updates, landing_rev, 'repo_landing_rev')
store_update(updates, enable_statistics, 'repo_enable_statistics')
store_update(updates, enable_locking, 'repo_enable_locking')
store_update(updates, enable_downloads, 'repo_enable_downloads')
RepoModel().update(repo, **updates)
Session().commit()
msg='updated repo ID:%s %s' % (repo.repo_id, repo.repo_name),
repository=repo.get_api_data()
raise JSONRPCError('failed to update repo `%s`' % repoid)
@HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
def fork_repo(self, apiuser, repoid, fork_name,
description=Optional(''), copy_permissions=Optional(False),
private=Optional(False), landing_rev=Optional('rev:tip')):
Creates a fork of given repo. In case of using celery this will
immediately return success message, while fork is going to be created
asynchronous. This command can be executed only using api_key belonging to
user with admin rights or regular user that have fork permission, and at least
read access to forking repository. Regular users cannot specify owner parameter.
:param fork_name:
:param copy_permissions:
INPUT::
id : <id_for_response>
api_key : "<api_key>"
args: {
"repoid" : "<reponame or repo_id>",
"fork_name": "<forkname>",
"owner": "<username or user_id = Optional(=apiuser)>",
"description": "<description>",
"copy_permissions": "<bool>",
"private": "<bool>",
"landing_rev": "<landing_rev>"
"msg": "Created fork of `<reponame>` as `<forkname>`",
repo_name = repo.repo_name
_repo = RepoModel().get_by_repo_name(fork_name)
if _repo:
type_ = 'fork' if _repo.fork else 'repo'
raise JSONRPCError("%s `%s` already exist" % (type_, fork_name))
if HasPermissionAnyApi('hg.admin')(user=apiuser):
pass
elif HasRepoPermissionAnyApi('repository.admin',
'repository.write',
'repository.read')(user=apiuser,
if not HasPermissionAnyApi('hg.create.repository')(user=apiuser):
raise JSONRPCError('no permission to create repositories')
group = map_groups(fork_name)
fork_base_name = fork_name.rsplit('/', 1)[-1]
form_data = dict(
repo_name=fork_name,
repo_name=fork_base_name,
repo_name_full=fork_name,
repo_group=group,
repo_type=repo.repo_type,
description=Optional.extract(description),
private=Optional.extract(private),
copy_permissions=Optional.extract(copy_permissions),
landing_rev=Optional.extract(landing_rev),
update_after_clone=False,
fork_parent_id=repo.repo_id,
task = RepoModel().create_fork(form_data, cur_user=owner)
msg='Created fork of `%s` as `%s`' % (repo.repo_name,
fork_name),
'failed to fork repository `%s` as `%s`' % (repo_name,
fork_name)
def delete_repo(self, apiuser, repoid, forks=Optional('')):
Deletes a repository. This command can be executed only using api_key belonging
to user with admin rights or regular user that have admin access to repository.
When `forks` param is set it's possible to detach or delete forks of deleting
repository
:param forks: `detach` or `delete`, what do do with attached forks for repo
:type forks: Optional(str)
"msg": "Deleted repository `<reponame>`",
"success": true
handle_forks = Optional.extract(forks)
_forks_msg = ''
_forks = [f for f in repo.forks]
if handle_forks == 'detach':
_forks_msg = ' ' + 'Detached %s forks' % len(_forks)
elif handle_forks == 'delete':
_forks_msg = ' ' + 'Deleted %s forks' % len(_forks)
elif _forks:
'Cannot delete `%s` it still contains attached forks' %
(repo.repo_name,)
RepoModel().delete(repo, forks=forks)
msg='Deleted repository `%s`%s' % (repo.repo_name, _forks_msg),
success=True
'failed to delete repository `%s`' % (repo.repo_name,)
@HasPermissionAllDecorator('hg.admin')
def grant_user_permission(self, apiuser, repoid, userid, perm):
Grant permission for user on given repository, or update existing one
if found. This command can be executed only using api_key belonging to user
with admin rights.
:param userid:
:param perm: (repository.(none|read|write|admin))
:type perm: str
"msg" : "Granted perm: `<perm>` for user: `<username>` in repo: `<reponame>`",
user = get_user_or_error(userid)
perm = get_perm_or_error(perm)
RepoModel().grant_user_permission(repo=repo, user=user, perm=perm)
msg='Granted perm: `%s` for user: `%s` in repo: `%s`' % (
perm.permission_name, user.username, repo.repo_name
),
'failed to edit permission for user: `%s` in repo: `%s`' % (
userid, repoid
def revoke_user_permission(self, apiuser, repoid, userid):
Revoke permission for user on given repository. This command can be executed
only using api_key belonging to user with admin rights.
"msg" : "Revoked perm for user: `<username>` in repo: `<reponame>`",
RepoModel().revoke_user_permission(repo=repo, user=user)
msg='Revoked perm for user: `%s` in repo: `%s`' % (
user.username, repo.repo_name
def grant_user_group_permission(self, apiuser, repoid, usergroupid, perm):
Grant permission for user group on given repository, or update
existing one if found. This command can be executed only using
api_key belonging to user with admin rights.
:param usergroupid: id of usergroup
:type usergroupid: str or int
result : {
"msg" : "Granted perm: `<perm>` for group: `<usersgroupname>` in repo: `<reponame>`",
error : null
"failed to edit permission for user group: `<usergroup>` in repo `<repo>`'
user_group = get_user_group_or_error(usergroupid)
_perms = ('repository.admin',)
if not HasRepoPermissionAnyApi(*_perms)(
user=apiuser, repo_name=repo.repo_name):
# check if we have at least read permission for this user group !
_perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
if not HasUserGroupPermissionAny(*_perms)(
user=apiuser, user_group_name=user_group.users_group_name):
raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
RepoModel().grant_user_group_permission(
repo=repo, group_name=user_group, perm=perm)
msg='Granted perm: `%s` for user group: `%s` in '
'repo: `%s`' % (
perm.permission_name, user_group.users_group_name,
repo.repo_name
'failed to edit permission for user group: `%s` in '
usergroupid, repo.repo_name
def revoke_user_group_permission(self, apiuser, repoid, usergroupid):
Revoke permission for user group on given repository. This command can be
executed only using api_key belonging to user with admin rights.
:param usergroupid:
"msg" : "Revoked perm for group: `<usersgroupname>` in repo: `<reponame>`",
RepoModel().revoke_user_group_permission(
repo=repo, group_name=user_group)
msg='Revoked perm for user group: `%s` in repo: `%s`' % (
user_group.users_group_name, repo.repo_name
def get_repo_group(self, apiuser, repogroupid):
Returns given repo group together with permissions, and repositories
inside the group
:param repogroupid: id/name of repository group
:type repogroupid: str or int
repo_group = get_repo_group_or_error(repogroupid)
members = []
for user in repo_group.repo_group_to_perm:
perm = user.permission.permission_name
user = user.user
user_data = {
'name': user.username,
'type': "user",
'permission': perm
members.append(user_data)
for user_group in repo_group.users_group_to_perm:
perm = user_group.permission.permission_name
user_group = user_group.users_group
user_group_data = {
'name': user_group.users_group_name,
'type': "user_group",
members.append(user_group_data)
data = repo_group.get_api_data()
data["members"] = members
return data
def get_repo_groups(self, apiuser):
Returns all repository groups
for repo_group in RepoGroupModel().get_all():
result.append(repo_group.get_api_data())
def create_repo_group(self, apiuser, group_name, description=Optional(''),
parent=Optional(None),
Creates a repository group. This command can be executed only using
:param group_name:
:type group_name:
:type description:
:type owner:
:param parent:
:type parent:
:type copy_permissions:
@@ -1010,770 +1010,773 @@ class _BaseTestApi(object):
'task': None,
expected = ret
self._compare_ok(id_, expected, given=response.body)
fixture.destroy_repo(repo_name)
fixture.destroy_repo_group('my_gr')
def test_api_create_repo_in_repo_group_without_permission(self):
repo_group_name = '%s/api-repo-repo' % TEST_REPO_GROUP
repo_name = '%s/api-repo' % repo_group_name
rg = fixture.create_repo_group(repo_group_name)
RepoGroupModel().grant_user_permission(repo_group_name,
self.TEST_USER_LOGIN,
'group.none')
id_, params = _build_data(self.apikey_regular, 'create_repo',
repo_name=repo_name,
repo_type=self.REPO_TYPE,
response = api_call(self, params)
# Current result when API access control is different from Web:
ret = {
'msg': 'Created new repository `%s`' % repo_name,
'success': True,
# Expected and arguably more correct result:
#expected = 'failed to create repository `%s`' % repo_name
#self._compare_error(id_, expected, given=response.body)
fixture.destroy_repo_group(repo_group_name)
def test_api_create_repo_unknown_owner(self):
repo_name = 'api-repo'
owner = 'i-dont-exist'
id_, params = _build_data(self.apikey, 'create_repo',
expected = 'user `%s` does not exist' % owner
self._compare_error(id_, expected, given=response.body)
def test_api_create_repo_dont_specify_owner(self):
repo = RepoModel().get_by_repo_name(repo_name)
self.assertNotEqual(repo, None)
def test_api_create_repo_by_non_admin(self):
def test_api_create_repo_by_non_admin_specify_owner(self):
owner=owner)
expected = 'Only Kallithea admin can specify `owner` param'
def test_api_create_repo_exists(self):
repo_name = self.REPO
owner=TEST_USER_ADMIN_LOGIN,
repo_type=self.REPO_TYPE,)
expected = "repo `%s` already exist" % repo_name
@mock.patch.object(RepoModel, 'create', crash)
def test_api_create_repo_exception_occurred(self):
expected = 'failed to create repository `%s`' % repo_name
@parameterized.expand([
('owner', {'owner': TEST_USER_REGULAR_LOGIN}),
('description', {'description': 'new description'}),
('active', {'active': True}),
('active', {'active': False}),
('clone_uri', {'clone_uri': 'http://example.com/repo'}),
('clone_uri', {'clone_uri': None}),
('landing_rev', {'landing_rev': 'branch:master'}),
('enable_statistics', {'enable_statistics': True}),
('enable_locking', {'enable_locking': True}),
('enable_downloads', {'enable_downloads': True}),
('name', {'name': 'new_repo_name'}),
('repo_group', {'group': 'test_group_for_update'}),
])
def test_api_update_repo(self, changing_attr, updates):
repo_name = 'api_update_me'
repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
if changing_attr == 'repo_group':
fixture.create_repo_group(updates['group'])
id_, params = _build_data(self.apikey, 'update_repo',
repoid=repo_name, **updates)
if changing_attr == 'name':
repo_name = updates['name']
repo_name = '/'.join([updates['group'], repo_name])
expected = {
'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo_name),
'repository': repo.get_api_data()
finally:
fixture.destroy_repo_group(updates['group'])
('description', {'description': u'new description'}),
('name', {'name': u'new_repo_name'}),
('repo_group', {'group': u'test_group_for_update'}),
def test_api_update_group_repo(self, changing_attr, updates):
group_name = u'lololo'
fixture.create_repo_group(group_name)
repo_name = u'%s/api_update_me' % group_name
repo = fixture.create_repo(repo_name, repo_group=group_name, repo_type=self.REPO_TYPE)
repo_name = u'%s/%s' % (group_name, updates['name'])
repo_name = u'/'.join([updates['group'], repo_name.rsplit('/', 1)[-1]])
fixture.destroy_repo_group(group_name)
def test_api_update_repo_repo_group_does_not_exist(self):
repo_name = 'admin_owned'
fixture.create_repo(repo_name)
updates = {'group': 'test_group_for_update'}
expected = 'repository group `%s` does not exist' % updates['group']
def test_api_update_repo_regular_user_not_allowed(self):
updates = {'active': False}
id_, params = _build_data(self.apikey_regular, 'update_repo',
expected = 'repository `%s` does not exist' % repo_name
@mock.patch.object(RepoModel, 'update', crash)
def test_api_update_repo_exception_occurred(self):
fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
repoid=repo_name, owner=TEST_USER_ADMIN_LOGIN,)
expected = 'failed to update repo `%s`' % repo_name
def test_api_update_repo_regular_user_change_repo_name(self):
new_repo_name = 'new_repo_name'
RepoModel().grant_user_permission(repo=repo_name,
user=self.TEST_USER_LOGIN,
perm='repository.admin')
UserModel().revoke_perm('default', 'hg.create.repository')
UserModel().grant_perm('default', 'hg.create.none')
updates = {'name': new_repo_name}
expected = 'no permission to create (or move) repositories'
fixture.destroy_repo(new_repo_name)
def test_api_update_repo_regular_user_change_repo_name_allowed(self):
UserModel().revoke_perm('default', 'hg.create.none')
UserModel().grant_perm('default', 'hg.create.repository')
'msg': 'updated repo ID:%s %s' % (repo.repo_id, new_repo_name),
def test_api_update_repo_regular_user_change_owner(self):
updates = {'owner': TEST_USER_ADMIN_LOGIN}
def test_api_delete_repo(self):
repo_name = 'api_delete_me'
id_, params = _build_data(self.apikey, 'delete_repo',
repoid=repo_name, )
'msg': 'Deleted repository `%s`' % repo_name,
'success': True
def test_api_delete_repo_by_non_admin(self):
fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
cur_user=self.TEST_USER_LOGIN)
id_, params = _build_data(self.apikey_regular, 'delete_repo',
def test_api_delete_repo_by_non_admin_no_permission(self):
expected = 'repository `%s` does not exist' % (repo_name)
def test_api_delete_repo_exception_occurred(self):
with mock.patch.object(RepoModel, 'delete', crash):
expected = 'failed to delete repository `%s`' % repo_name
def test_api_fork_repo(self):
fork_name = 'api-repo-fork'
id_, params = _build_data(self.apikey, 'fork_repo',
repoid=self.REPO,
fork_name=fork_name,
'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
fixture.destroy_repo(fork_name)
def test_api_fork_repo_non_admin(self):
(u'api-repo-fork',),
(u'%s/api-repo-fork' % TEST_REPO_GROUP,),
def test_api_fork_repo_non_admin(self, fork_name):
id_, params = _build_data(self.apikey_regular, 'fork_repo',
def test_api_fork_repo_non_admin_specify_owner(self):
def test_api_fork_repo_non_admin_no_permission_to_fork(self):
RepoModel().grant_user_permission(repo=self.REPO,
perm='repository.none')
expected = 'repository `%s` does not exist' % (self.REPO)
@parameterized.expand([('read', 'repository.read'),
('write', 'repository.write'),
('admin', 'repository.admin')])
def test_api_fork_repo_non_admin_no_create_repo_permission(self, name, perm):
# regardless of base repository permission, forking is disallowed
# when repository creation is disabled
perm=perm)
expected = 'no permission to create repositories'
def test_api_fork_repo_unknown_owner(self):
def test_api_fork_repo_fork_exists(self):
fixture.create_fork(self.REPO, fork_name)
expected = "fork `%s` already exist" % fork_name
def test_api_fork_repo_repo_exists(self):
fork_name = self.REPO
expected = "repo `%s` already exist" % fork_name
@mock.patch.object(RepoModel, 'create_fork', crash)
def test_api_fork_repo_exception_occurred(self):
expected = 'failed to fork repository `%s` as `%s`' % (self.REPO,
def test_api_get_user_group(self):
id_, params = _build_data(self.apikey, 'get_user_group',
usergroupid=TEST_USER_GROUP)
user_group = UserGroupModel().get_group(TEST_USER_GROUP)
for user in user_group.members:
members.append(user.get_api_data())
ret = user_group.get_api_data()
ret['members'] = members
def test_api_get_user_groups(self):
gr_name = 'test_user_group2'
make_user_group(gr_name)
id_, params = _build_data(self.apikey, 'get_user_groups', )
expected = []
for gr_name in [TEST_USER_GROUP, 'test_user_group2']:
user_group = UserGroupModel().get_group(gr_name)
expected.append(ret)
fixture.destroy_user_group(gr_name)
def test_api_create_user_group(self):
group_name = 'some_new_group'
id_, params = _build_data(self.apikey, 'create_user_group',
group_name=group_name)
'msg': 'created new user group `%s`' % group_name,
'user_group': jsonify(UserGroupModel() \
.get_by_name(group_name) \
.get_api_data())
fixture.destroy_user_group(group_name)
def test_api_get_user_group_that_exist(self):
group_name=TEST_USER_GROUP)
expected = "user group `%s` already exist" % TEST_USER_GROUP
@mock.patch.object(UserGroupModel, 'create', crash)
def test_api_get_user_group_exception_occurred(self):
group_name = 'exception_happens'
expected = 'failed to create group `%s`' % group_name
@parameterized.expand([('group_name', {'group_name': 'new_group_name'}),
('group_name', {'group_name': 'test_group_for_update'}),
('active', {'active': True})])
def test_api_update_user_group(self, changing_attr, updates):
gr_name = 'test_group_for_update'
user_group = fixture.create_user_group(gr_name)
id_, params = _build_data(self.apikey, 'update_user_group',
usergroupid=gr_name, **updates)
'msg': 'updated user group ID:%s %s' % (user_group.users_group_id,
user_group.users_group_name),
'user_group': user_group.get_api_data()
if changing_attr == 'group_name':
# switch to updated name for proper cleanup
gr_name = updates['group_name']
@mock.patch.object(UserGroupModel, 'update', crash)
def test_api_update_user_group_exception_occurred(self):
gr_name = 'test_group'
fixture.create_user_group(gr_name)
usergroupid=gr_name)
expected = 'failed to update user group `%s`' % gr_name
def test_api_add_user_to_user_group(self):
id_, params = _build_data(self.apikey, 'add_user_to_user_group',
usergroupid=gr_name,
userid=TEST_USER_ADMIN_LOGIN)
'msg': 'added member `%s` to user group `%s`' % (
TEST_USER_ADMIN_LOGIN, gr_name),
def test_api_add_user_to_user_group_that_doesnt_exist(self):
usergroupid='false-group',
expected = 'user group `%s` does not exist' % 'false-group'
@mock.patch.object(UserGroupModel, 'add_user_to_group', crash)
def test_api_add_user_to_user_group_exception_occurred(self):
expected = 'failed to add member to user group `%s`' % gr_name
def test_api_remove_user_from_user_group(self):
gr_name = 'test_group_3'
gr = fixture.create_user_group(gr_name)
UserGroupModel().add_user_to_group(gr, user=TEST_USER_ADMIN_LOGIN)
id_, params = _build_data(self.apikey, 'remove_user_from_user_group',
'msg': 'removed member `%s` from user group `%s`' % (
TEST_USER_ADMIN_LOGIN, gr_name
'success': True}
@mock.patch.object(UserGroupModel, 'remove_user_from_group', crash)
def test_api_remove_user_from_user_group_exception_occurred(self):
expected = 'failed to remove member from user group `%s`' % gr_name
def test_api_delete_user_group(self):
ugroup = fixture.create_user_group(gr_name)
gr_id = ugroup.users_group_id
id_, params = _build_data(self.apikey, 'delete_user_group',
'user_group': None,
'msg': 'deleted user group ID:%s %s' % (gr_id, gr_name)
if UserGroupModel().get_by_name(gr_name):
def test_api_delete_user_group_that_is_assigned(self):
ugr_to_perm = RepoModel().grant_user_group_permission(self.REPO, gr_name, 'repository.write')
msg = 'User Group assigned to %s' % ugr_to_perm.repository.repo_name
expected = msg
def test_api_delete_user_group_exception_occurred(self):
with mock.patch.object(UserGroupModel, 'delete', crash):
expected = 'failed to delete user group ID:%s %s' % (gr_id, gr_name)
@parameterized.expand([('none', 'repository.none'),
('read', 'repository.read'),
def test_api_grant_user_permission(self, name, perm):
id_, params = _build_data(self.apikey,
'grant_user_permission',
userid=TEST_USER_ADMIN_LOGIN,
'msg': 'Granted perm: `%s` for user: `%s` in repo: `%s`' % (
perm, TEST_USER_ADMIN_LOGIN, self.REPO
def test_api_grant_user_permission_wrong_permission(self):
perm = 'haha.no.permission'
Status change: