kallithea Changeset - 7643d8ecbb20

Changeset - 7643d8ecbb20

Parent rev.

Child rev.

[Not reviewed]

stable

0 3 0

Mads Kiilerich (mads) - 5 years ago 2021-01-02 23:41:37
mads@kiilerich.com

Grafted from: 217062ecb161

api: fix repo group permission check for repo creation

hg.create.repository is only controlling whether all users have write access at
top level. For other repo locations, check for write permission to the repo
group.

Note: This also covers "repo creation" by forking or by moving another repo.

3 files changed with 39 insertions and 34 deletions:

kallithea/controllers/api/api.py

kallithea/templates/admin/permissions/permissions_globals.html

kallithea/tests/api/api_base.py

0 comments (0 inline, 0 general)

kallithea/controllers/api/api.py

➞

Show inline comments

@@ @@ -1122,367 +1122,379 @@ class ApiController(JSONRPCController): @@
         :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)
         OUTPUT::
             id : <id_given_in_input>
             result: [
+                      {
                         "name" :        "<name>"
                         "type" :        "<type>",
                       },
                       …
+                    ]
             error:  null
         """
         repo = get_repo_or_error(repoid)
         if not HasPermissionAny('hg.admin')():
             if not HasRepoPermissionLevel('read')(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(sorted(_map))))
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to get repo: `%s` nodes' % repo.repo_name
+            )
-    @HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
+    # permission check inside
     def create_repo(self, 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_downloads=Optional(False),
                     copy_permissions=Optional(False)):
         """
         Creates a repository. The repository name contains the full path, but the
         parent repository group must exist. For example "foo/bar/baz" require the groups
         "foo" and "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_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
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg": "Created new repository `<reponame>`",
                       "success": true,
                       "task": "<celery task id or None if done sync>"
+                    }
             error:  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
              'failed to create repository `<repo_name>`
+          }
         """
         group_name = None
         repo_name_parts = repo_name.split('/')
         if len(repo_name_parts) > 1:
             group_name = '/'.join(repo_name_parts[:-1])
             repo_group = RepoGroup.get_by_group_name(group_name)
             if repo_group is None:
                 raise JSONRPCError("repo group `%s` not found" % group_name)
             if not(HasPermissionAny('hg.admin')() or HasRepoGroupPermissionLevel('write')(group_name)):
                 raise JSONRPCError("no permission to create repo in %s" % group_name)
         else:
             if not HasPermissionAny('hg.admin', 'hg.create.repository')():
                 raise JSONRPCError("no permission to create top level repo")
         if not HasPermissionAny('hg.admin')():
             if not isinstance(owner, Optional):
                 # forbid setting owner for non-admins
                 raise JSONRPCError(
                     'Only Kallithea admin can specify `owner` param'
+                )
         if isinstance(owner, Optional):
             owner = request.authuser.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_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)
         try:
             repo_name_parts = repo_name.split('/')
             group_name = None
             if len(repo_name_parts) > 1:
                 group_name = '/'.join(repo_name_parts[:-1])
                 repo_group = RepoGroup.get_by_group_name(group_name)
                 if repo_group is None:
                     raise JSONRPCError("repo group `%s` not found" % group_name)
             data = dict(
                 repo_name=repo_name_parts[-1],
                 repo_name_full=repo_name,
                 repo_type=repo_type,
                 repo_description=description,
                 owner=owner,
                 repo_private=private,
                 clone_uri=clone_uri,
                 repo_group=group_name,
                 repo_landing_rev=landing_rev,
                 enable_statistics=enable_statistics,
                 enable_downloads=enable_downloads,
                 repo_copy_permissions=copy_permissions,
+            )
             task = RepoModel().create(form_data=data, cur_user=owner.username)
             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
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to create repository `%s`' % (repo_name,))
     # permission check inside
     def update_repo(self, 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_statistics=Optional(False),
                     enable_downloads=Optional(False)):
         """
         Updates repo
         :param repoid: repository name or repository id
         :type repoid: str or int
         :param name:
         :param owner:
         :param group:
         :param description:
         :param private:
         :param clone_uri:
         :param landing_rev:
         :param enable_statistics:
         :param enable_downloads:
         """
         repo = get_repo_or_error(repoid)
         if not HasPermissionAny('hg.admin')():
             if not HasRepoPermissionLevel('admin')(repo.repo_name):
                 raise JSONRPCError('repository `%s` does not exist' % (repoid,))
             if (name != repo.repo_name and
                 not HasPermissionAny('hg.create.repository')()
             ):
                 raise JSONRPCError('no permission to create (or move) repositories')
             if not isinstance(owner, Optional):
                 # forbid setting owner for non-admins
                 raise JSONRPCError(
                     'Only Kallithea admin can specify `owner` param'
+                )
         updates = {}
         repo_group = group
         if not isinstance(repo_group, Optional):
             repo_group = get_repo_group_or_error(repo_group)
             if repo_group.group_id != repo.group_id:
                 if not(HasPermissionAny('hg.admin')() or HasRepoGroupPermissionLevel('write')(repo_group.group_name)):
                     raise JSONRPCError("no permission to create (or move) repo in %s" % repo_group.group_name)
             repo_group = repo_group.group_id
         try:
             store_update(updates, name, 'repo_name')
             store_update(updates, repo_group, 'repo_group')
             store_update(updates, owner, 'owner')
             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_downloads, 'repo_enable_downloads')
             RepoModel().update(repo, **updates)
             Session().commit()
             return dict(
                 msg='updated repo ID:%s %s' % (repo.repo_id, repo.repo_name),
                 repository=repo.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update repo `%s`' % repoid)
     # permission check inside
     @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
     def fork_repo(self, repoid, fork_name,
                   owner=Optional(OAttr('apiuser')),
                   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 repoid: repository name or repository id
         :type repoid: str or int
         :param fork_name:
         :param owner:
         :param description:
         :param copy_permissions:
         :param private:
         :param landing_rev:
         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>"
+                      }
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg": "Created fork of `<reponame>` as `<forkname>`",
                       "success": true,
                       "task": "<celery task id or None if done sync>"
+                    }
             error:  null
         """
         repo = get_repo_or_error(repoid)
         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))
         group_name = None
         fork_name_parts = fork_name.split('/')
         if len(fork_name_parts) > 1:
             group_name = '/'.join(fork_name_parts[:-1])
             repo_group = RepoGroup.get_by_group_name(group_name)
             if repo_group is None:
                 raise JSONRPCError("repo group `%s` not found" % group_name)
             if not(HasPermissionAny('hg.admin')() or HasRepoGroupPermissionLevel('write')(group_name)):
                 raise JSONRPCError("no permission to create repo in %s" % group_name)
         else:
             if not HasPermissionAny('hg.admin', 'hg.create.repository')():
                 raise JSONRPCError("no permission to create top level repo")
         if HasPermissionAny('hg.admin')():
             pass
         elif HasRepoPermissionLevel('read')(repo.repo_name):
             if not isinstance(owner, Optional):
                 # forbid setting owner for non-admins
                 raise JSONRPCError(
                     'Only Kallithea admin can specify `owner` param'
+                )
             if not HasPermissionAny('hg.create.repository')():
                 raise JSONRPCError('no permission to create repositories')
         else:
             raise JSONRPCError('repository `%s` does not exist' % (repoid,))
         if isinstance(owner, Optional):
             owner = request.authuser.user_id
         owner = get_user_or_error(owner)
         try:
             fork_name_parts = fork_name.split('/')
             group_name = None
             if len(fork_name_parts) > 1:
                 group_name = '/'.join(fork_name_parts[:-1])
                 repo_group = RepoGroup.get_by_group_name(group_name)
                 if repo_group is None:
                     raise JSONRPCError("repo group `%s` not found" % group_name)
             form_data = dict(
                 repo_name=fork_name_parts[-1],
                 repo_name_full=fork_name,
                 repo_group=group_name,
                 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.username)
             # no commit, it's done in RepoModel, or async via celery
             task_id = task.task_id
             return dict(
                 msg='Created fork of `%s` as `%s`' % (repo.repo_name,
                                                       fork_name),
                 success=True,  # cannot return the repo data here since fork
                                # can be done async
                 task=task_id
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to fork repository `%s` as `%s`' % (repo_name,
                                                             fork_name)
+            )
     # permission check inside
     def delete_repo(self, 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 repoid: repository name or repository id
         :type repoid: str or int
         :param forks: `detach` or `delete`, what do do with attached forks for repo
         :type forks: Optional(str)
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg": "Deleted repository `<reponame>`",
                       "success": true

kallithea/templates/admin/permissions/permissions_globals.html

➞

Show inline comments

@@ @@ -9,92 +9,91 @@ ${h.form(url('admin_permissions'), metho @@
                     </label>
                     <span class="help-block">${h.HTML(_('Allow access to Kallithea without needing to log in. Anonymous users use %s user permissions.')) % (h.link_to('*default*',h.url('admin_permissions_perms')))}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_repo_perm">${_('Repository')}:</label>
                 <div class="form-inline">
                     ${h.select('default_repo_perm','',c.repo_perms_choices,class_='form-control')}
                     <label>
                         ${h.checkbox('overwrite_default_repo','true')}
                         <span data-toggle="tooltip" title="${_('All default permissions on each repository will be reset to chosen permission, note that all custom default permission on repositories will be lost')}">
                             ${_('Apply to all existing repositories')}
                         </span>
                     </label>
                     <span class="help-block">${_('Permissions for the Default user on new repositories.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_group_perm">${_('Repository group')}:</label>
                 <div class="form-inline">
                     ${h.select('default_group_perm','',c.group_perms_choices,class_='form-control')}
                     <label>
                         ${h.checkbox('overwrite_default_group','true')}
                         <span data-toggle="tooltip" title="${_('All default permissions on each repository group will be reset to chosen permission, note that all custom default permission on repository groups will be lost')}">
                             ${_('Apply to all existing repository groups')}
                         </span>
                     </label>
                     <span class="help-block">${_('Permissions for the Default user on new repository groups.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_user_group_perm">${_('User group')}:</label>
                 <div class="form-inline">
                     ${h.select('default_user_group_perm','',c.user_group_perms_choices,class_='form-control')}
                     <label>
                         ${h.checkbox('overwrite_default_user_group','true')}
                         <span data-toggle="tooltip" title="${_('All default permissions on each user group will be reset to chosen permission, note that all custom default permission on user groups will be lost')}">
                             ${_('Apply to all existing user groups')}
                         </span>
                     </label>
                     <span class="help-block">${_('Permissions for the Default user on new user groups.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_repo_create">${_('Top level repository creation')}:</label>
                 <div>
                     ${h.select('default_repo_create','',c.repo_create_choices,class_='form-control')}
                     <span class="help-block">${_('Enable this to allow non-admins to create repositories at the top level.')}</span>
                     <span class="help-block">${_('Note: This will also give all users API access to create repositories everywhere. That might change in future versions.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="create_on_write">${_('Repository creation with group write access')}:</label>
                 <div>
                     ${h.select('create_on_write','',c.repo_create_on_write_choices,class_='form-control')}
                     <span class="help-block">${_('With this, write permission to a repository group allows creating repositories inside that group. Without this, group write permissions mean nothing.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_user_group_create">${_('User group creation')}:</label>
                 <div>
                     ${h.select('default_user_group_create','',c.user_group_create_choices,class_='form-control')}
                     <span class="help-block">${_('Enable this to allow non-admins to create user groups.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_fork">${_('Repository forking')}:</label>
                 <div>
                     ${h.select('default_fork','',c.fork_choices,class_='form-control')}
                     <span class="help-block">${_('Enable this to allow non-admins to fork repositories.')}</span>
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_register">${_('Registration')}:</label>
                 <div>
                     ${h.select('default_register','',c.register_choices,class_='form-control')}
                 </div>
             </div>
             <div class="form-group">
                 <label class="control-label" for="default_extern_activate">${_('External auth account activation')}:</label>
                 <div>
                     ${h.select('default_extern_activate','',c.extern_activate_choices,class_='form-control')}
                 </div>
             </div>
             <div class="form-group">
                 <div class="buttons">
                     ${h.submit('save',_('Save'),class_="btn btn-default")}
                     ${h.reset('reset',_('Reset'),class_="btn btn-default")}
                 </div>
             </div>
     </div>
 ${h.end_form()}

kallithea/tests/api/api_base.py

➞

Show inline comments

@@ @@ -871,110 +871,101 @@ class _BaseTestApi(object): @@
                                   owner=base.TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,)
         response = api_call(self, params)
         expected = 'repo group `%s` not found' % repo_group_name
         self._compare_error(id_, expected, given=response.body)
         assert RepoModel().get_by_repo_name(repo_name) is None
         # create group before creating repo
         rg = fixture.create_repo_group(repo_group_name)
         Session().commit()
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,)
         response = api_call(self, params)
         expected = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
             'task': None,
+        }
         self._compare_ok(id_, expected, given=response.body)
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         fixture.destroy_repo(repo_name)
         fixture.destroy_repo_group(repo_group_name)
     def test_api_create_repo_in_repo_group_without_permission(self):
         repo_group_basename = 'api-repo-repo'
         repo_group_name = '%s/%s' % (TEST_REPO_GROUP, repo_group_basename)
         repo_name = '%s/api-repo' % repo_group_name
         top_group = RepoGroup.get_by_group_name(TEST_REPO_GROUP)
         assert top_group
         rg = fixture.create_repo_group(repo_group_basename, parent_group_id=top_group)
         Session().commit()
         RepoGroupModel().grant_user_permission(repo_group_name,
                                                self.TEST_USER_LOGIN,
                                                'group.none')
         Session().commit()
         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,
             'task': None,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         # API access control match Web access control:
         expected = 'no permission to create repo in test_repo_group/api-repo-repo'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
         # 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',
                                   repo_name=repo_name,
                                   owner=owner,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         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_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
             'task': None,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_create_repo_by_non_admin(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey_regular, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
@@ @@ -1253,163 +1244,166 @@ class _BaseTestApi(object): @@
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         try:
             id_, params = _build_data(self.apikey_regular, 'delete_repo',
                                       repoid=repo_name, )
             response = api_call(self, params)
             expected = 'repository `%s` does not exist' % (repo_name)
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_delete_repo_exception_occurred(self):
         repo_name = 'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         try:
             with mock.patch.object(RepoModel, 'delete', crash):
                 id_, params = _build_data(self.apikey, 'delete_repo',
                                           repoid=repo_name, )
                 response = api_call(self, params)
                 expected = 'failed to delete repository `%s`' % repo_name
                 self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(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,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
+        )
         response = api_call(self, params)
         ret = {
             'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
                                                      fork_name),
             'success': True,
             'task': None,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     @base.parametrize('fork_name', [
         'api-repo-fork',
         '%s/api-repo-fork' % TEST_REPO_GROUP,
     ])
     def test_api_fork_repo_non_admin(self, fork_name):
         RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
                                                self.TEST_USER_LOGIN,
                                                'group.write')
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
+        )
         response = api_call(self, params)
         ret = {
             'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
                                                      fork_name),
             'success': True,
             'task': None,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     def test_api_fork_repo_non_admin_specify_owner(self):
         fork_name = 'api-repo-fork'
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
+        )
         response = api_call(self, params)
         expected = 'Only Kallithea admin can specify `owner` param'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     def test_api_fork_repo_non_admin_no_permission_to_fork(self):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=User.DEFAULT_USER_NAME,
                                           perm='repository.none')
         try:
             fork_name = 'api-repo-fork'
             id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                       repoid=self.REPO,
                                       fork_name=fork_name,
+            )
             response = api_call(self, params)
             expected = 'repository `%s` does not exist' % (self.REPO)
             self._compare_error(id_, expected, given=response.body)
         finally:
             RepoModel().grant_user_permission(repo=self.REPO,
                                               user=User.DEFAULT_USER_NAME,
                                               perm='repository.read')
             fixture.destroy_repo(fork_name)
     @base.parametrize('name,perm', [
         ('read', 'repository.read'),
         ('write', 'repository.write'),
         ('admin', 'repository.admin'),
     ])
     def test_api_fork_repo_non_admin_no_create_repo_permission(self, name, perm):
         fork_name = 'api-repo-fork'
         # regardless of base repository permission, forking is disallowed
         # when repository creation is disabled
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm=perm)
         UserModel().revoke_perm('default', 'hg.create.repository')
         UserModel().grant_perm('default', 'hg.create.none')
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
+        )
         response = api_call(self, params)
-        expected = 'no permission to create repositories'
+        expected = 'no permission to create top level repo'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     def test_api_fork_repo_unknown_owner(self):
         fork_name = 'api-repo-fork'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=owner,
+        )
         response = api_call(self, params)
         expected = 'user `%s` does not exist' % owner
         self._compare_error(id_, expected, given=response.body)
     def test_api_fork_repo_fork_exists(self):
         fork_name = 'api-repo-fork'
         fixture.create_fork(self.REPO, fork_name)
         try:
             fork_name = 'api-repo-fork'
             id_, params = _build_data(self.apikey, 'fork_repo',
                                       repoid=self.REPO,
                                       fork_name=fork_name,
                                       owner=base.TEST_USER_ADMIN_LOGIN,
+            )
             response = api_call(self, params)
             expected = "fork `%s` already exist" % fork_name
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(fork_name)
     def test_api_fork_repo_repo_exists(self):
         fork_name = self.REPO
         id_, params = _build_data(self.apikey, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
+        )
         response = api_call(self, params)
         expected = "repo `%s` already exist" % fork_name
         self._compare_error(id_, expected, given=response.body)
     @mock.patch.object(RepoModel, 'create_fork', crash)

0 comments (0 inline, 0 general)