kallithea Changeset - 87c2cd07166a

Changeset - 87c2cd07166a

Parent rev.

Child rev.

[Not reviewed]

default

0 5 0

Mads Kiilerich (mads) - 5 years ago 2020-11-01 23:04:25
mads@kiilerich.com

Grafted from: 90dd5831e497

lib: move get_all_user_repos to AuthUser

5 files changed with 13 insertions and 17 deletions:

kallithea/controllers/api/api.py

kallithea/lib/auth.py

kallithea/lib/auth_modules/__init__.py

kallithea/model/repo.py

kallithea/tests/api/api_base.py

0 comments (0 inline, 0 general)

kallithea/controllers/api/api.py

➞

Show inline comments

@@ @@ -716,769 +716,769 @@ class ApiController(JSONRPCController): @@
           error :  {
             "user group `<group name>` already exist"
             or
             "failed to create group `<group name>`"
+          }
         """
         if UserGroupModel().get_by_name(group_name):
             raise JSONRPCError("user group `%s` already exist" % (group_name,))
         try:
             if owner is None:
                 owner = request.authuser.user_id
             owner = get_user_or_error(owner)
             ug = UserGroupModel().create(name=group_name, description=description,
                                          owner=owner, active=active)
             meta.Session().commit()
             return dict(
                 msg='created new user group `%s`' % group_name,
                 user_group=ug.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create group `%s`' % (group_name,))
     # permission check inside
     def update_user_group(self, usergroupid, group_name=None,
                           description=None, owner=None,
                           active=None):
         """
         Updates given usergroup.  This command can be executed only using api_key
         belonging to user with admin rights or an admin of given user group
         :param usergroupid: id of user group to update
         :type usergroupid: str or int
         :param group_name: name of new user group
         :type group_name: str
         :param description: group description
         :type description: str
         :param owner: owner of group.
         :type owner: Optional(str or int)
         :param active: group is active
         :type active: Optional(bool)
         OUTPUT::
           id : <id_given_in_input>
           result : {
             "msg": 'updated user group ID:<user group id> <user group name>',
             "user_group": <user_group_object>
+          }
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to update user group `<user group name>`"
+          }
         """
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAny('hg.admin')():
             if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
         if owner is not None:
             owner = get_user_or_error(owner)
         updates = {}
         store_update(updates, group_name, 'users_group_name')
         store_update(updates, description, 'user_group_description')
         store_update(updates, owner, 'owner')
         store_update(updates, active, 'users_group_active')
         try:
             UserGroupModel().update(user_group, updates)
             meta.Session().commit()
             return dict(
                 msg='updated user group ID:%s %s' % (user_group.users_group_id,
                                                      user_group.users_group_name),
                 user_group=user_group.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update user group `%s`' % (usergroupid,))
     # permission check inside
     def delete_user_group(self, usergroupid):
         """
         Delete given user group by user group id or name.
         This command can be executed only using api_key
         belonging to user with admin rights or an admin of given user group
         :param usergroupid:
         :type usergroupid: int
         OUTPUT::
           id : <id_given_in_input>
           result : {
             "msg": "deleted user group ID:<user_group_id> <user_group_name>"
+          }
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to delete user group ID:<user_group_id> <user_group_name>"
             or
             "RepoGroup assigned to <repo_groups_list>"
+          }
         """
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAny('hg.admin')():
             if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
         try:
             UserGroupModel().delete(user_group)
             meta.Session().commit()
             return dict(
                 msg='deleted user group ID:%s %s' %
                     (user_group.users_group_id, user_group.users_group_name),
                 user_group=None
+            )
         except UserGroupsAssignedException as e:
             log.error(traceback.format_exc())
             raise JSONRPCError(str(e))
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to delete user group ID:%s %s' %
                                (user_group.users_group_id,
                                 user_group.users_group_name)
+                               )
     # permission check inside
     def add_user_to_user_group(self, usergroupid, userid):
         """
         Adds a user to a user group. If user exists in that group success will be
         `false`. This command can be executed only using api_key
         belonging to user with admin rights  or an admin of given user group
         :param usergroupid:
         :type usergroupid: int
         :param userid:
         :type userid: int
         OUTPUT::
           id : <id_given_in_input>
           result : {
               "success": True|False # depends on if member is in group
               "msg": "added member `<username>` to user group `<groupname>` |
                       User is already in that group"
+          }
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to add member to user group `<user_group_name>`"
+          }
         """
         user = get_user_or_error(userid)
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAny('hg.admin')():
             if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
         try:
             ugm = UserGroupModel().add_user_to_group(user_group, user)
             success = True if ugm is not True else False
             msg = 'added member `%s` to user group `%s`' % (
                 user.username, user_group.users_group_name
+            )
             msg = msg if success else 'User is already in that group'
             meta.Session().commit()
             return dict(
                 success=success,
                 msg=msg
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to add member to user group `%s`' % (
                     user_group.users_group_name,
+                )
+            )
     # permission check inside
     def remove_user_from_user_group(self, usergroupid, userid):
         """
         Removes a user from a user group. If user is not in given group success will
         be `false`. This command can be executed only
         using api_key belonging to user with admin rights or an admin of given user group
         :param usergroupid:
         :param userid:
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "success":  True|False,  # depends on if member is in group
                       "msg": "removed member <username> from user group <groupname> |
                               User wasn't in group"
+                    }
             error:  null
         """
         user = get_user_or_error(userid)
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAny('hg.admin')():
             if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
         try:
             success = UserGroupModel().remove_user_from_group(user_group, user)
             msg = 'removed member `%s` from user group `%s`' % (
                 user.username, user_group.users_group_name
+            )
             msg = msg if success else "User wasn't in group"
             meta.Session().commit()
             return dict(success=success, msg=msg)
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to remove member from user group `%s`' % (
                     user_group.users_group_name,
+                )
+            )
     # permission check inside
     def get_repo(self, repoid,
                  with_revision_names=False,
                  with_pullrequests=False):
         """
         Gets an existing repository by it's name or repository_id. Members will return
         either users_group or user associated to that repository. 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
         OUTPUT::
           id : <id_given_in_input>
           result : {
+            {
                 "repo_id" :          "<repo_id>",
                 "repo_name" :        "<reponame>"
                 "repo_type" :        "<repo_type>",
                 "clone_uri" :        "<clone_uri>",
                 "enable_downloads":  "<bool>",
                 "enable_statistics": "<bool>",
                 "private":           "<bool>",
                 "created_on" :       "<date_time_created>",
                 "description" :      "<description>",
                 "landing_rev":       "<landing_rev>",
                 "last_changeset":    {
                                        "author":   "<full_author>",
                                        "date":     "<date_time_of_commit>",
                                        "message":  "<commit_message>",
                                        "raw_id":   "<raw_id>",
                                        "revision": "<numeric_revision>",
                                        "short_id": "<short_id>"
+                                     }
                 "owner":             "<repo_owner>",
                 "fork_of":           "<name_of_fork_parent>",
                 "members" :     [
+                                  {
                                     "name":     "<username>",
                                     "type" :    "user",
                                     "permission" : "repository.(read|write|admin)"
                                   },
                                   …
+                                  {
                                     "name":     "<usergroup name>",
                                     "type" :    "user_group",
                                     "permission" : "usergroup.(read|write|admin)"
                                   },
                                   …
+                                ]
                  "followers":   [<user_obj>, ...],
                  <if with_revision_names == True>
                  "tags": {
                             "<tagname>": "<raw_id>",
                             ...
                          },
                  "branches": {
                             "<branchname>": "<raw_id>",
                             ...
                          },
                  "bookmarks": {
                             "<bookmarkname>": "<raw_id>",
                             ...
                          },
+            }
+          }
           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,))
         members = []
         for user in repo.repo_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.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",
                 'permission': perm
+            }
             members.append(user_group_data)
         followers = [
             uf.user.get_api_data()
             for uf in repo.followers
+        ]
         data = repo.get_api_data(with_revision_names=with_revision_names,
                                  with_pullrequests=with_pullrequests)
         data['members'] = members
         data['followers'] = followers
         return data
     # permission check inside
     def get_repos(self):
         """
         Lists all existing repositories. This command can be executed only using
         api_key belonging to user with admin rights or regular user that have
         admin, write or read access to repository.
         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_statistics": "<bool>",
                       },
                       …
+                    ]
             error:  null
         """
         if not HasPermissionAny('hg.admin')():
-            repos = RepoModel().get_all_user_repos(user=request.authuser.user_id)
+            repos = request.authuser.get_all_user_repos()
         else:
             repos = db.Repository.query()
         return [
             repo.get_api_data()
             for repo in repos
+        ]
     # permission check inside
     def get_repo_nodes(self, repoid, revision, root_path,
                        ret_type='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)
         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 = 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')
     def create_repo(self, repo_name, owner=None,
                     repo_type=None, description='',
                     private=False, clone_uri=None,
                     landing_rev='rev:tip',
                     enable_statistics=None,
                     enable_downloads=None,
                     copy_permissions=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>`
+          }
         """
         if not HasPermissionAny('hg.admin')():
             if owner is not None:
                 # forbid setting owner for non-admins
                 raise JSONRPCError(
                     'Only Kallithea admin can specify `owner` param'
+                )
         if owner is None:
             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 = db.Setting.get_default_repo_settings(strip_prefix=True)
         if private is None:
             private = defs.get('repo_private') or False
         if repo_type is None:
             repo_type = defs.get('repo_type')
         if enable_statistics is None:
             enable_statistics = defs.get('repo_enable_statistics')
         if enable_downloads is None:
             enable_downloads = defs.get('repo_enable_downloads')
         try:
             repo_name_parts = repo_name.split('/')
             repo_group = None
             if len(repo_name_parts) > 1:
                 group_name = '/'.join(repo_name_parts[:-1])
                 repo_group = db.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=repo_group,
                 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)
             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=None,
                     owner=None,
                     group=None,
                     description=None, private=None,
                     clone_uri=None, landing_rev=None,
                     enable_statistics=None,
                     enable_downloads=None):
         """
         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 owner is not None:
                 # forbid setting owner for non-admins
                 raise JSONRPCError(
                     'Only Kallithea admin can specify `owner` param'
+                )
         updates = {}
         repo_group = group
         if repo_group is not None:
             repo_group = get_repo_group_or_error(repo_group)
             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)
             meta.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)
     @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
     def fork_repo(self, repoid, fork_name,
                   owner=None,
                   description='', copy_permissions=False,
                   private=False, landing_rev='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))
         if HasPermissionAny('hg.admin')():
             pass
         elif HasRepoPermissionLevel('read')(repo.repo_name):
             if owner is not None:
                 # 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 owner is None:
             owner = request.authuser.user_id
         owner = get_user_or_error(owner)
         try:
             fork_name_parts = fork_name.split('/')
             repo_group = None
             if len(fork_name_parts) > 1:
                 group_name = '/'.join(fork_name_parts[:-1])
                 repo_group = db.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=repo_group,
                 repo_type=repo.repo_type,
                 description=description,
                 private=private,
                 copy_permissions=copy_permissions,
                 landing_rev=landing_rev,
                 update_after_clone=False,
                 fork_parent_id=repo.repo_id,
+            )
             task = RepoModel().create_fork(form_data, cur_user=owner)
             # 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=''):
         """
         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
+                    }
             error:  null
         """
         repo = get_repo_or_error(repoid)

kallithea/lib/auth.py

➞

Show inline comments

@@ @@ -86,628 +86,638 @@ class AuthUser(object): @@
     "no user". On the other hand, if the default user is disabled (and
     there is no login information), we instead get "no user"; this should
     only happen on the login page (as all other requests are redirected).
     `is_default_user` specifically checks if the AuthUser is the user named
     "default". Use `is_anonymous` to check for both "default" and "no user".
     """
     @classmethod
     def make(cls, dbuser=None, is_external_auth=False, ip_addr=None):
         """Create an AuthUser to be authenticated ... or return None if user for some reason can't be authenticated.
         Checks that a non-None dbuser is provided, is active, and that the IP address is ok.
         """
         assert ip_addr is not None
         if dbuser is None:
             log.info('No db user for authentication')
             return None
         if not dbuser.active:
             log.info('Db user %s not active', dbuser.username)
             return None
         allowed_ips = AuthUser.get_allowed_ips(dbuser.user_id)
         if not check_ip_access(source_ip=ip_addr, allowed_ips=allowed_ips):
             log.info('Access for %s from %s forbidden - not in %s', dbuser.username, ip_addr, allowed_ips)
             return None
         return cls(dbuser=dbuser, is_external_auth=is_external_auth)
     def __init__(self, user_id=None, dbuser=None, is_external_auth=False):
         self.is_external_auth = is_external_auth # container auth - don't show logout option
         # These attributes will be overridden below if the requested user is
         # found or anonymous access (using the default user) is enabled.
         self.user_id = None
         self.username = None
         self.api_key = None
         self.name = ''
         self.lastname = ''
         self.email = ''
         self.admin = False
         # Look up database user, if necessary.
         if user_id is not None:
             assert dbuser is None
             log.debug('Auth User lookup by USER ID %s', user_id)
             dbuser = UserModel().get(user_id)
             assert dbuser is not None
         else:
             assert dbuser is not None
             log.debug('Auth User lookup by database user %s', dbuser)
         log.debug('filling %s data', dbuser)
         self.is_anonymous = dbuser.is_default_user
         if dbuser.is_default_user and not dbuser.active:
             self.username = 'None'
             self.is_default_user = False
         else:
             # copy non-confidential database fields from a `db.User` to this `AuthUser`.
             for k, v in dbuser.get_dict().items():
                 assert k not in ['api_keys', 'permissions']
                 setattr(self, k, v)
             self.is_default_user = dbuser.is_default_user
         log.debug('Auth User is now %s', self)
     @LazyProperty
     def global_permissions(self):
         log.debug('Getting global permissions for %s', self)
         if self.is_admin:
             return set(['hg.admin'])
         global_permissions = set()
         # default global permissions from the default user
         default_global_perms = db.UserToPerm.query() \
             .filter(db.UserToPerm.user_id == kallithea.DEFAULT_USER_ID) \
             .options(joinedload(db.UserToPerm.permission))
         for perm in default_global_perms:
             global_permissions.add(perm.permission.permission_name)
         # user group global permissions
         user_perms_from_users_groups = meta.Session().query(db.UserGroupToPerm) \
             .options(joinedload(db.UserGroupToPerm.permission)) \
             .join((db.UserGroupMember, db.UserGroupToPerm.users_group_id ==
                    db.UserGroupMember.users_group_id)) \
             .filter(db.UserGroupMember.user_id == self.user_id) \
             .join((db.UserGroup, db.UserGroupMember.users_group_id ==
                    db.UserGroup.users_group_id)) \
             .filter(db.UserGroup.users_group_active == True) \
             .order_by(db.UserGroupToPerm.users_group_id) \
             .all()
         # need to group here by groups since user can be in more than
         # one group
         _grouped = [[x, list(y)] for x, y in
                     itertools.groupby(user_perms_from_users_groups,
                                       lambda x:x.users_group)]
         for gr, perms in _grouped:
             for perm in perms:
                 global_permissions.add(perm.permission.permission_name)
         # user specific global permissions
         user_perms = meta.Session().query(db.UserToPerm) \
                 .options(joinedload(db.UserToPerm.permission)) \
                 .filter(db.UserToPerm.user_id == self.user_id).all()
         for perm in user_perms:
             global_permissions.add(perm.permission.permission_name)
         # for each kind of global permissions, only keep the one with heighest weight
         kind_max_perm = {}
         for perm in sorted(global_permissions, key=lambda n: PERM_WEIGHTS.get(n, -1)):
             kind = perm.rsplit('.', 1)[0]
             kind_max_perm[kind] = perm
         return set(kind_max_perm.values())
     @LazyProperty
     def repository_permissions(self):
         log.debug('Getting repository permissions for %s', self)
         repository_permissions = {}
         default_repo_perms = db.Permission.get_default_perms(kallithea.DEFAULT_USER_ID)
         if self.is_admin:
             for perm in default_repo_perms:
                 r_k = perm.repository.repo_name
                 p = 'repository.admin'
                 repository_permissions[r_k] = p
         else:
             # defaults for repositories from default user
             for perm in default_repo_perms:
                 r_k = perm.repository.repo_name
                 if perm.repository.owner_id == self.user_id:
                     p = 'repository.admin'
                 elif perm.repository.private:
                     p = 'repository.none'
                 else:
                     p = perm.permission.permission_name
                 repository_permissions[r_k] = p
             # user group repository permissions
             user_repo_perms_from_users_groups = \
              meta.Session().query(db.UserGroupRepoToPerm) \
                 .join((db.UserGroup, db.UserGroupRepoToPerm.users_group_id ==
                        db.UserGroup.users_group_id)) \
                 .filter(db.UserGroup.users_group_active == True) \
                 .join((db.UserGroupMember, db.UserGroupRepoToPerm.users_group_id ==
                        db.UserGroupMember.users_group_id)) \
                 .filter(db.UserGroupMember.user_id == self.user_id) \
                 .options(joinedload(db.UserGroupRepoToPerm.repository)) \
                 .options(joinedload(db.UserGroupRepoToPerm.permission)) \
                 .all()
             for perm in user_repo_perms_from_users_groups:
                 bump_permission(repository_permissions,
                     perm.repository.repo_name,
                     perm.permission.permission_name)
             # user permissions for repositories
             user_repo_perms = db.Permission.get_default_perms(self.user_id)
             for perm in user_repo_perms:
                 bump_permission(repository_permissions,
                     perm.repository.repo_name,
                     perm.permission.permission_name)
         return repository_permissions
     @LazyProperty
     def repository_group_permissions(self):
         log.debug('Getting repository group permissions for %s', self)
         repository_group_permissions = {}
         default_repo_groups_perms = db.Permission.get_default_group_perms(kallithea.DEFAULT_USER_ID)
         if self.is_admin:
             for perm in default_repo_groups_perms:
                 rg_k = perm.group.group_name
                 p = 'group.admin'
                 repository_group_permissions[rg_k] = p
         else:
             # defaults for repository groups taken from default user permission
             # on given group
             for perm in default_repo_groups_perms:
                 rg_k = perm.group.group_name
                 p = perm.permission.permission_name
                 repository_group_permissions[rg_k] = p
             # user group for repo groups permissions
             user_repo_group_perms_from_users_groups = \
                 meta.Session().query(db.UserGroupRepoGroupToPerm) \
                 .join((db.UserGroup, db.UserGroupRepoGroupToPerm.users_group_id ==
                        db.UserGroup.users_group_id)) \
                 .filter(db.UserGroup.users_group_active == True) \
                 .join((db.UserGroupMember, db.UserGroupRepoGroupToPerm.users_group_id
                        == db.UserGroupMember.users_group_id)) \
                 .filter(db.UserGroupMember.user_id == self.user_id) \
                 .options(joinedload(db.UserGroupRepoGroupToPerm.permission)) \
                 .all()
             for perm in user_repo_group_perms_from_users_groups:
                 bump_permission(repository_group_permissions,
                     perm.group.group_name,
                     perm.permission.permission_name)
             # user explicit permissions for repository groups
             user_repo_groups_perms = db.Permission.get_default_group_perms(self.user_id)
             for perm in user_repo_groups_perms:
                 bump_permission(repository_group_permissions,
                     perm.group.group_name,
                     perm.permission.permission_name)
         return repository_group_permissions
     @LazyProperty
     def user_group_permissions(self):
         log.debug('Getting user group permissions for %s', self)
         user_group_permissions = {}
         default_user_group_perms = db.Permission.get_default_user_group_perms(kallithea.DEFAULT_USER_ID)
         if self.is_admin:
             for perm in default_user_group_perms:
                 u_k = perm.user_group.users_group_name
                 p = 'usergroup.admin'
                 user_group_permissions[u_k] = p
         else:
             # defaults for user groups taken from default user permission
             # on given user group
             for perm in default_user_group_perms:
                 u_k = perm.user_group.users_group_name
                 p = perm.permission.permission_name
                 user_group_permissions[u_k] = p
             # user group for user group permissions
             user_group_user_groups_perms = \
                 meta.Session().query(db.UserGroupUserGroupToPerm) \
                 .join((db.UserGroup, db.UserGroupUserGroupToPerm.target_user_group_id
                        == db.UserGroup.users_group_id)) \
                 .join((db.UserGroupMember, db.UserGroupUserGroupToPerm.user_group_id
                        == db.UserGroupMember.users_group_id)) \
                 .filter(db.UserGroupMember.user_id == self.user_id) \
                 .join((db.UserGroup, db.UserGroupMember.users_group_id ==
                        db.UserGroup.users_group_id), aliased=True, from_joinpoint=True) \
                 .filter(db.UserGroup.users_group_active == True) \
                 .options(joinedload(db.UserGroupUserGroupToPerm.permission)) \
                 .all()
             for perm in user_group_user_groups_perms:
                 bump_permission(user_group_permissions,
                     perm.target_user_group.users_group_name,
                     perm.permission.permission_name)
             # user explicit permission for user groups
             user_user_groups_perms = db.Permission.get_default_user_group_perms(self.user_id)
             for perm in user_user_groups_perms:
                 bump_permission(user_group_permissions,
                     perm.user_group.users_group_name,
                     perm.permission.permission_name)
         return user_group_permissions
     @LazyProperty
     def permissions(self):
         """dict with all 4 kind of permissions - mainly for backwards compatibility"""
         return {
             'global': self.global_permissions,
             'repositories': self.repository_permissions,
             'repositories_groups': self.repository_group_permissions,
             'user_groups': self.user_group_permissions,
+        }
     def has_repository_permission_level(self, repo_name, level, purpose=None):
         required_perms = {
             'read': ['repository.read', 'repository.write', 'repository.admin'],
             'write': ['repository.write', 'repository.admin'],
             'admin': ['repository.admin'],
         }[level]
         actual_perm = self.repository_permissions.get(repo_name)
         ok = actual_perm in required_perms
         log.debug('Checking if user %r can %r repo %r (%s): %s (has %r)',
             self.username, level, repo_name, purpose, ok, actual_perm)
         return ok
     def has_repository_group_permission_level(self, repo_group_name, level, purpose=None):
         required_perms = {
             'read': ['group.read', 'group.write', 'group.admin'],
             'write': ['group.write', 'group.admin'],
             'admin': ['group.admin'],
         }[level]
         actual_perm = self.repository_group_permissions.get(repo_group_name)
         ok = actual_perm in required_perms
         log.debug('Checking if user %r can %r repo group %r (%s): %s (has %r)',
             self.username, level, repo_group_name, purpose, ok, actual_perm)
         return ok
     def has_user_group_permission_level(self, user_group_name, level, purpose=None):
         required_perms = {
             'read': ['usergroup.read', 'usergroup.write', 'usergroup.admin'],
             'write': ['usergroup.write', 'usergroup.admin'],
             'admin': ['usergroup.admin'],
         }[level]
         actual_perm = self.user_group_permissions.get(user_group_name)
         ok = actual_perm in required_perms
         log.debug('Checking if user %r can %r user group %r (%s): %s (has %r)',
             self.username, level, user_group_name, purpose, ok, actual_perm)
         return ok
     @property
     def api_keys(self):
         return self._get_api_keys()
     def _get_api_keys(self):
         api_keys = [self.api_key]
         for api_key in db.UserApiKeys.query() \
                 .filter_by(user_id=self.user_id, is_expired=False):
             api_keys.append(api_key.api_key)
         return api_keys
     @property
     def is_admin(self):
         return self.admin
     @property
     def repositories_admin(self):
         """
         Returns list of repositories you're an admin of
         """
         return [x[0] for x in self.repository_permissions.items()
                 if x[1] == 'repository.admin']
     @property
     def repository_groups_admin(self):
         """
         Returns list of repository groups you're an admin of
         """
         return [x[0] for x in self.repository_group_permissions.items()
                 if x[1] == 'group.admin']
     @property
     def user_groups_admin(self):
         """
         Returns list of user groups you're an admin of
         """
         return [x[0] for x in self.user_group_permissions.items()
                 if x[1] == 'usergroup.admin']
     def __repr__(self):
         return "<%s %s: %r>" % (self.__class__.__name__, self.user_id, self.username)
     def to_cookie(self):
         """ Serializes this login session to a cookie `dict`. """
         return {
             'user_id': self.user_id,
             'is_external_auth': self.is_external_auth,
+        }
     @staticmethod
     def from_cookie(cookie, ip_addr):
         """
         Deserializes an `AuthUser` from a cookie `dict` ... or return None.
         """
         return AuthUser.make(
             dbuser=UserModel().get(cookie.get('user_id')),
             is_external_auth=cookie.get('is_external_auth', False),
             ip_addr=ip_addr,
+        )
     @classmethod
     def get_allowed_ips(cls, user_id):
         _set = set()
         default_ips = db.UserIpMap.query().filter(db.UserIpMap.user_id == kallithea.DEFAULT_USER_ID)
         for ip in default_ips:
             try:
                 _set.add(ip.ip_addr)
             except ObjectDeletedError:
                 # since we use heavy caching sometimes it happens that we get
                 # deleted objects here, we just skip them
                 pass
         user_ips = db.UserIpMap.query().filter(db.UserIpMap.user_id == user_id)
         for ip in user_ips:
             try:
                 _set.add(ip.ip_addr)
             except ObjectDeletedError:
                 # since we use heavy caching sometimes it happens that we get
                 # deleted objects here, we just skip them
                 pass
         return _set or set(['0.0.0.0/0', '::/0'])
     def get_all_user_repos(self):
         """
         Gets all repositories that user have at least read access
         """
         repos = [repo_name
             for repo_name, perm in self.repository_permissions.items()
             if perm in ['repository.read', 'repository.write', 'repository.admin']
+            ]
         return db.Repository.query().filter(db.Repository.repo_name.in_(repos))
 #==============================================================================
 # CHECK DECORATORS
 #==============================================================================
 def _redirect_to_login(message=None):
     """Return an exception that must be raised. It will redirect to the login
     page which will redirect back to the current URL after authentication.
     The optional message will be shown in a flash message."""
     if message:
         webutils.flash(message, category='warning')
     p = request.path_qs
     log.debug('Redirecting to login page, origin: %s', p)
     return HTTPFound(location=url('login_home', came_from=p))
 # Use as decorator
 class LoginRequired(object):
     """Client must be logged in as a valid User, or we'll redirect to the login
     page.
     If the "default" user is enabled and allow_default_user is true, that is
     considered valid too.
     Also checks that IP address is allowed.
     """
     def __init__(self, allow_default_user=False):
         self.allow_default_user = allow_default_user
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         controller = fargs[0]
         user = request.authuser
         loc = "%s:%s" % (controller.__class__.__name__, func.__name__)
         log.debug('Checking access for user %s @ %s', user, loc)
         # regular user authentication
         if user.is_default_user:
             if self.allow_default_user:
                 log.info('default user @ %s', loc)
                 return func(*fargs, **fkwargs)
             log.info('default user is not accepted here @ %s', loc)
         elif user.is_anonymous: # default user is disabled and no proper authentication
             log.warning('user is anonymous and NOT authenticated with regular auth @ %s', loc)
         else: # regular authentication
             log.info('user %s authenticated with regular auth @ %s', user, loc)
             return func(*fargs, **fkwargs)
         raise _redirect_to_login()
 # Use as decorator
 class NotAnonymous(object):
     """Ensures that client is not logged in as the "default" user, and
     redirects to the login page otherwise. Must be used together with
     LoginRequired."""
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         cls = fargs[0]
         user = request.authuser
         log.debug('Checking that user %s is not anonymous @%s', user.username, cls)
         if user.is_default_user:
             raise _redirect_to_login(_('You need to be a registered user to '
                                        'perform this action'))
         else:
             return func(*fargs, **fkwargs)
 class _PermsDecorator(object):
     """Base class for controller decorators with multiple permissions"""
     def __init__(self, *required_perms):
         self.required_perms = required_perms # usually very short - a list is thus fine
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         cls = fargs[0]
         user = request.authuser
         log.debug('checking %s permissions %s for %s %s',
           self.__class__.__name__, self.required_perms, cls, user)
         if self.check_permissions(user):
             log.debug('Permission granted for %s %s', cls, user)
             return func(*fargs, **fkwargs)
         else:
             log.info('Permission denied for %s %s', cls, user)
             if user.is_default_user:
                 raise _redirect_to_login(_('You need to be signed in to view this page'))
             else:
                 raise HTTPForbidden()
     def check_permissions(self, user):
         raise NotImplementedError()
 class HasPermissionAnyDecorator(_PermsDecorator):
     """
     Checks the user has any of the given global permissions.
     """
     def check_permissions(self, user):
         return any(p in user.global_permissions for p in self.required_perms)
 class _PermDecorator(_PermsDecorator):
     """Base class for controller decorators with a single permission"""
     def __init__(self, required_perm):
         _PermsDecorator.__init__(self, [required_perm])
         self.required_perm = required_perm
 class HasRepoPermissionLevelDecorator(_PermDecorator):
     """
     Checks the user has at least the specified permission level for the requested repository.
     """
     def check_permissions(self, user):
         repo_name = get_repo_slug(request)
         return user.has_repository_permission_level(repo_name, self.required_perm)
 class HasRepoGroupPermissionLevelDecorator(_PermDecorator):
     """
     Checks the user has any of given permissions for the requested repository group.
     """
     def check_permissions(self, user):
         repo_group_name = get_repo_group_slug(request)
         return user.has_repository_group_permission_level(repo_group_name, self.required_perm)
 class HasUserGroupPermissionLevelDecorator(_PermDecorator):
     """
     Checks for access permission for any of given predicates for specific
     user group. In order to fulfill the request any of predicates must be meet
     """
     def check_permissions(self, user):
         user_group_name = get_user_group_slug(request)
         return user.has_user_group_permission_level(user_group_name, self.required_perm)
 #==============================================================================
 # CHECK FUNCTIONS
 #==============================================================================
 class _PermsFunction(object):
     """Base function for other check functions with multiple permissions"""
     def __init__(self, *required_perms):
         self.required_perms = required_perms # usually very short - a list is thus fine
     def __bool__(self):
         """ Defend against accidentally forgetting to call the object
             and instead evaluating it directly in a boolean context,
             which could have security implications.
         """
         raise AssertionError(self.__class__.__name__ + ' is not a bool and must be called!')
     def __call__(self, *a, **b):
         raise NotImplementedError()
 class HasPermissionAny(_PermsFunction):
     def __call__(self, purpose=None):
         ok = any(p in request.authuser.global_permissions for p in self.required_perms)
         log.debug('Check %s for global %s (%s): %s',
             request.authuser.username, self.required_perms, purpose, ok)
         return ok
 class _PermFunction(_PermsFunction):
     """Base function for other check functions with a single permission"""
     def __init__(self, required_perm):
         _PermsFunction.__init__(self, [required_perm])
         self.required_perm = required_perm
 class HasRepoPermissionLevel(_PermFunction):
     def __call__(self, repo_name, purpose=None):
         return request.authuser.has_repository_permission_level(repo_name, self.required_perm, purpose)
 class HasRepoGroupPermissionLevel(_PermFunction):
     def __call__(self, group_name, purpose=None):
         return request.authuser.has_repository_group_permission_level(group_name, self.required_perm, purpose)
 class HasUserGroupPermissionLevel(_PermFunction):
     def __call__(self, user_group_name, purpose=None):
         return request.authuser.has_user_group_permission_level(user_group_name, self.required_perm, purpose)
 #==============================================================================
 # SPECIAL VERSION TO HANDLE MIDDLEWARE AUTH
 #==============================================================================
 class HasPermissionAnyMiddleware(object):
     def __init__(self, *perms):
         self.required_perms = set(perms)
     def __call__(self, authuser, repo_name, purpose=None):
         try:
             ok = authuser.repository_permissions[repo_name] in self.required_perms
         except KeyError:
             ok = False
         log.debug('Middleware check %s for %s for repo %s (%s): %s', authuser.username, self.required_perms, repo_name, purpose, ok)
         return ok
 def check_ip_access(source_ip, allowed_ips=None):
     """
     Checks if source_ip is a subnet of any of allowed_ips.
     :param source_ip:
     :param allowed_ips: list of allowed ips together with mask
     """
     source_ip = source_ip.split('%', 1)[0]
     log.debug('checking if ip:%s is subnet of %s', source_ip, allowed_ips)
     if isinstance(allowed_ips, (tuple, list, set)):
         for ip in allowed_ips:
             if ipaddr.IPAddress(source_ip) in ipaddr.IPNetwork(ip):
                 log.debug('IP %s is network %s',
                           ipaddr.IPAddress(source_ip), ipaddr.IPNetwork(ip))
                 return True
     return False

kallithea/lib/auth_modules/__init__.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 Authentication modules
 """
 import importlib
 import logging
 import traceback
 from inspect import isfunction
 from kallithea.lib.auth import AuthUser
 from kallithea.lib.compat import hybrid_property
-from kallithea.lib.utils2 import asbool, PasswordGenerator
+from kallithea.lib.utils2 import PasswordGenerator, asbool
 from kallithea.model import db, meta, validators
 from kallithea.model.user import UserModel
 from kallithea.model.user_group import UserGroupModel
 log = logging.getLogger(__name__)
 class LazyFormencode(object):
     def __init__(self, formencode_obj, *args, **kwargs):
         self.formencode_obj = formencode_obj
         self.args = args
         self.kwargs = kwargs
     def __call__(self, *args, **kwargs):
         formencode_obj = self.formencode_obj
         if isfunction(formencode_obj):
             # case we wrap validators into functions
             formencode_obj = self.formencode_obj(*args, **kwargs)
         return formencode_obj(*self.args, **self.kwargs)
 class KallitheaAuthPluginBase(object):
     auth_func_attrs = {
         "username": "unique username",
         "firstname": "first name",
         "lastname": "last name",
         "email": "email address",
         "groups": '["list", "of", "groups"]',
         "extern_name": "name in external source of record",
         "admin": 'True|False defines if user should be Kallithea admin',
+    }
     @property
     def validators(self):
         """
         Exposes Kallithea validators modules
         """
         # this is a hack to overcome issues with pylons threadlocals and
         # translator object _() not being registered properly.
         class LazyCaller(object):
             def __init__(self, name):
                 self.validator_name = name
             def __call__(self, *args, **kwargs):
                 obj = getattr(validators, self.validator_name)
                 #log.debug('Initializing lazy formencode object: %s', obj)
                 return LazyFormencode(obj, *args, **kwargs)
         class ProxyGet(object):
             def __getattribute__(self, name):
                 return LazyCaller(name)
         return ProxyGet()
     @hybrid_property
     def name(self):
         """
         Returns the name of this authentication plugin.
         :returns: string
         """
         raise NotImplementedError("Not implemented in base class")
     @hybrid_property
     def is_container_auth(self):
         """
         Returns bool if this module uses container auth.
         This property will trigger an automatic call to authenticate on
         a visit to the website or during a push/pull.
         :returns: bool
         """
         return False
     def accepts(self, user, accepts_empty=True):
         """
         Checks if this authentication module should accept a request for
         the current user.
         :param user: user object fetched using plugin's get_user() method.
         :param accepts_empty: if True accepts don't allow the user to be empty
         :returns: boolean
         """
         plugin_name = self.name
         if not user and not accepts_empty:
             log.debug('User is empty not allowed to authenticate')
             return False
         if user and user.extern_type and user.extern_type != plugin_name:
             log.debug('User %s should authenticate using %s this is %s, skipping',
                       user, user.extern_type, plugin_name)
             return False
         return True
     def get_user(self, username=None, **kwargs):
         """
         Helper method for user fetching in plugins, by default it's using
         simple fetch by username, but this method can be customized in plugins
         eg. container auth plugin to fetch user by environ params
         :param username: username if given to fetch from database
         :param kwargs: extra arguments needed for user fetching.
         """
         user = None
         log.debug('Trying to fetch user `%s` from Kallithea database',
                   username)
         if username:
             user = db.User.get_by_username_or_email(username)
         else:
             log.debug('provided username:`%s` is empty skipping...', username)
         return user
     def settings(self):
         """
         Return a list of the form:
+        [
+            {
                 "name": "OPTION_NAME",
                 "type": "[bool|password|string|int|select]",
                 ["values": ["opt1", "opt2", ...]]
                 "validator": "expr"
                 "description": "A short description of the option" [,
                 "default": Default Value],
                 ["formname": "Friendly Name for Forms"]
             } [, ...]
+        ]
         This is used to interrogate the authentication plugin as to what
         settings it expects to be present and configured.
         'type' is a shorthand notation for what kind of value this option is.
         This is primarily used by the auth web form to control how the option
         is configured.
                 bool : checkbox
                 password : password input box
                 string : input box
                 select : single select dropdown
         'validator' is an lazy instantiated form field validator object, ala
         formencode. You need to *call* this object to init the validators.
         All calls to Kallithea validators should be used through self.validators
         which is a lazy loading proxy of formencode module.
         """
         raise NotImplementedError("Not implemented in base class")
     def plugin_settings(self):
         """
         This method is called by the authentication framework, not the .settings()
         method. This method adds a few default settings (e.g., "enabled"), so that
         plugin authors don't have to maintain a bunch of boilerplate.
         OVERRIDING THIS METHOD WILL CAUSE YOUR PLUGIN TO FAIL.
         """
         settings = self.settings()
         settings.insert(0, {
             "name": "enabled",
             "validator": self.validators.StringBoolean(if_missing=False),
             "type": "bool",
             "description": "Enable or Disable this Authentication Plugin",
             "formname": "Enabled"
+            }
+        )
         return settings
     def auth(self, userobj, username, passwd, settings, **kwargs):
         """
         Given a user object (which may be None), username, a plaintext password,
         and a settings object (containing all the keys needed as listed in settings()),
         authenticate this user's login attempt.
         Return None on failure. On success, return a dictionary with keys from
         KallitheaAuthPluginBase.auth_func_attrs.
         This is later validated for correctness.
         """
         raise NotImplementedError("not implemented in base class")
     def _authenticate(self, userobj, username, passwd, settings, **kwargs):
         """
         Wrapper to call self.auth() that validates call on it
         """
         user_data = self.auth(userobj, username, passwd, settings, **kwargs)
         if user_data is not None:
             return self._validate_auth_return(user_data)
         return None
     def _validate_auth_return(self, user_data):
         if not isinstance(user_data, dict):
             raise Exception('returned value from auth must be a dict')
         for k in self.auth_func_attrs:
             if k not in user_data:
                 raise Exception('Missing %s attribute from returned data' % k)
         return user_data
 class KallitheaExternalAuthPlugin(KallitheaAuthPluginBase):
     def use_fake_password(self):
         """
         Return a boolean that indicates whether or not we should set the user's
         password to a random value when it is authenticated by this plugin.
         If your plugin provides authentication, then you will generally want this.
         :returns: boolean
         """
         raise NotImplementedError("Not implemented in base class")
     def _authenticate(self, userobj, username, passwd, settings, **kwargs):
         user_data = super(KallitheaExternalAuthPlugin, self)._authenticate(
             userobj, username, passwd, settings, **kwargs)
         if user_data is not None:
             if userobj is None: # external authentication of unknown user that will be created soon
                 def_user_perms = AuthUser(dbuser=db.User.get_default_user()).global_permissions
                 active = 'hg.extern_activate.auto' in def_user_perms
             else:
                 active = userobj.active
             if self.use_fake_password():
                 # Randomize the PW because we don't need it, but don't want
                 # them blank either
                 passwd = PasswordGenerator().gen_password(length=8)
             log.debug('Updating or creating user info from %s plugin',
                       self.name)
             user = UserModel().create_or_update(
                 username=user_data['username'],
                 password=passwd,
                 email=user_data["email"],
                 firstname=user_data["firstname"],
                 lastname=user_data["lastname"],
                 active=active,
                 admin=user_data["admin"],
                 extern_name=user_data["extern_name"],
                 extern_type=self.name,
+            )
             # enforce user is just in given groups, all of them has to be ones
             # created from plugins. We store this info in _group_data JSON field
             groups = user_data['groups'] or []
             UserGroupModel().enforce_groups(user, groups, self.name)
             meta.Session().commit()
         return user_data
 def loadplugin(plugin):
     """
     Imports, instantiates, and returns the authentication plugin in the module named by plugin
     (e.g., plugin='kallithea.lib.auth_modules.auth_internal'). Returns an instance of the
     KallitheaAuthPluginBase subclass on success, raises exceptions on failure.
     raises:
         AttributeError -- no KallitheaAuthPlugin class in the module
         TypeError -- if the KallitheaAuthPlugin is not a subclass of ours KallitheaAuthPluginBase
         ImportError -- if we couldn't import the plugin at all
     """
     log.debug("Importing %s", plugin)
     if not plugin.startswith('kallithea.lib.auth_modules.auth_'):
         parts = plugin.split('.lib.auth_modules.auth_', 1)
         if len(parts) == 2:
             _module, pn = parts
             plugin = 'kallithea.lib.auth_modules.auth_' + pn
     PLUGIN_CLASS_NAME = "KallitheaAuthPlugin"
     try:
         module = importlib.import_module(plugin)
     except (ImportError, TypeError):
         log.error(traceback.format_exc())
         # TODO: make this more error prone, if by some accident we screw up
         # the plugin name, the crash is pretty bad and hard to recover
         raise
     log.debug("Loaded auth plugin from %s (module:%s, file:%s)",
               plugin, module.__name__, module.__file__)
     pluginclass = getattr(module, PLUGIN_CLASS_NAME)
     if not issubclass(pluginclass, KallitheaAuthPluginBase):
         raise TypeError("Authentication class %s.KallitheaAuthPlugin is not "
                         "a subclass of %s" % (plugin, KallitheaAuthPluginBase))
     plugin = pluginclass()
     if plugin.plugin_settings.__func__ != KallitheaAuthPluginBase.plugin_settings:
         raise TypeError("Authentication class %s.KallitheaAuthPluginBase "
                         "has overridden the plugin_settings method, which is "
                         "forbidden." % plugin)
     return plugin
 def get_auth_plugins():
     """Return a list of instances of plugins that are available and enabled"""
     auth_plugins = []
     for plugin_name in db.Setting.get_by_name("auth_plugins").app_settings_value:
         try:
             plugin = loadplugin(plugin_name)
         except Exception:
             log.exception('Failed to load authentication module %s' % (plugin_name))
         else:
             auth_plugins.append(plugin)
     return auth_plugins
 def authenticate(username, password, environ=None):
     """
     Authentication function used for access control,
     It tries to authenticate based on enabled authentication modules.
     :param username: username can be empty for container auth
     :param password: password can be empty for container auth
     :param environ: environ headers passed for container auth
     :returns: None if auth failed, user_data dict if auth is correct
     """
     auth_plugins = get_auth_plugins()
     for plugin in auth_plugins:
         module = plugin.__class__.__module__
         log.debug('Trying authentication using %s', module)
         # load plugin settings from Kallithea database
         plugin_name = plugin.name
         plugin_settings = {}
         for v in plugin.plugin_settings():
             conf_key = "auth_%s_%s" % (plugin_name, v["name"])
             setting = db.Setting.get_by_name(conf_key)
             plugin_settings[v["name"]] = setting.app_settings_value if setting else None
         log.debug('Settings for auth plugin %s: %s', plugin_name, plugin_settings)
         if not asbool(plugin_settings["enabled"]):
             log.info("Authentication plugin %s is disabled, skipping for %s",
                      module, username)
             continue
         # use plugin's method of user extraction.
         user = plugin.get_user(username, environ=environ,
                                settings=plugin_settings)
         log.debug('Plugin %s extracted user `%s`', module, user)
         if user is not None and not user.active: # give up, way before creating AuthUser
             log.error("Rejecting authentication of in-active user %s", user)
             continue
         if not plugin.accepts(user):
             log.debug('Plugin %s does not accept user `%s` for authentication',
                       module, user)
             continue
         else:
             log.debug('Plugin %s accepted user `%s` for authentication',
                       module, user)
             # The user might have tried to authenticate using their email address,
             # then the username variable wouldn't contain a valid username.
             # But as the plugin has accepted the user, .username field should
             # have a valid username, so use it for authentication purposes.
             if user is not None:
                 username = user.username
         log.info('Authenticating user using %s plugin', module)
         # _authenticate is a wrapper for .auth() method of plugin.
         # it checks if .auth() sends proper data. For KallitheaExternalAuthPlugin
         # it also maps users to Database and maps the attributes returned
         # from .auth() to Kallithea database. If this function returns data
         # then auth is correct.
         user_data = plugin._authenticate(user, username, password,
                                            plugin_settings,
                                            environ=environ or {})
         log.debug('Plugin user data: %s', user_data)
         if user_data is not None:
             log.debug('Plugin returned proper authentication data')
             return user_data
         # we failed to Auth because .auth() method didn't return the user
         if username:
             log.warning("User `%s` failed to authenticate against %s",
                         username, module)
     return None
 def get_managed_fields(user):
     """return list of fields that are managed by the user's auth source, usually some of
     'username', 'firstname', 'lastname', 'email', 'password'
     """
     auth_plugins = get_auth_plugins()
     for plugin in auth_plugins:
         module = plugin.__class__.__module__
         log.debug('testing %s (%s) with auth plugin %s', user, user.extern_type, module)

kallithea/model/repo.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.repo
 ~~~~~~~~~~~~~~~~~~~~
 Repository model for kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jun 5, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import os
 import shutil
 import traceback
 from datetime import datetime
 import kallithea.lib.utils2
 from kallithea.lib.auth import HasRepoPermissionLevel, HasUserGroupPermissionLevel
 from kallithea.lib.exceptions import AttachedForksError
 from kallithea.lib.hooks import log_delete_repository
 from kallithea.lib.utils import is_valid_repo_uri, make_ui
 from kallithea.lib.utils2 import LazyProperty, get_current_authuser, obfuscate_url_pw, remove_prefix
 from kallithea.lib.vcs.backends import get_backend
 from kallithea.model import db, meta, scm
 log = logging.getLogger(__name__)
 class RepoModel(object):
     def _create_default_perms(self, repository, private):
         # create default permission
         default = 'repository.read'
         def_user = db.User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('repository.'):
                 default = p.permission.permission_name
                 break
         default_perm = 'repository.none' if private else default
         repo_to_perm = db.UserRepoToPerm()
         repo_to_perm.permission = db.Permission.get_by_key(default_perm)
         repo_to_perm.repository = repository
         repo_to_perm.user_id = def_user.user_id
         meta.Session().add(repo_to_perm)
         return repo_to_perm
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = db.Ui.query().filter(db.Ui.ui_key == '/').one()
         return q.ui_value
     def get(self, repo_id):
         repo = db.Repository.query() \
             .filter(db.Repository.repo_id == repo_id)
         return repo.scalar()
     def get_repo(self, repository):
         return db.Repository.guess_instance(repository)
     def get_by_repo_name(self, repo_name):
         repo = db.Repository.query() \
             .filter(db.Repository.repo_name == repo_name)
         return repo.scalar()
     def get_all_user_repos(self, user):
         """
         Gets all repositories that user have at least read access
         :param user:
         """
         from kallithea.lib.auth import AuthUser
         auth_user = AuthUser(dbuser=db.User.guess_instance(user))
         repos = [repo_name
             for repo_name, perm in auth_user.repository_permissions.items()
             if perm in ['repository.read', 'repository.write', 'repository.admin']
+            ]
         return db.Repository.query().filter(db.Repository.repo_name.in_(repos))
     @classmethod
     def _render_datatable(cls, tmpl, *args, **kwargs):
         from tg import app_globals, request
         from tg import tmpl_context as c
         from tg.i18n import ugettext as _
         import kallithea.lib.helpers as h
         _tmpl_lookup = app_globals.mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         tmpl = template.get_def(tmpl)
         kwargs.update(dict(_=_, h=h, c=c, request=request))
         return tmpl.render_unicode(*args, **kwargs)
     def get_repos_as_dict(self, repos_list, repo_groups_list=None,
                           admin=False,
                           short_name=False):
         """Return repository list for use by DataTable.
         repos_list: list of repositories - but will be filtered for read permission.
         repo_groups_list: added at top of list without permission check.
         admin: return data for action column.
         """
         _render = self._render_datatable
         from tg import request
         from tg import tmpl_context as c
         import kallithea.lib.helpers as h
         def repo_lnk(name, rtype, rstate, private, fork_of):
             return _render('repo_name', name, rtype, rstate, private, fork_of,
                            short_name=short_name)
         def following(repo_id, is_following):
             return _render('following', repo_id, is_following)
         def last_change(last_change):
             return _render("last_change", last_change)
         def rss_lnk(repo_name):
             return _render("rss", repo_name)
         def atom_lnk(repo_name):
             return _render("atom", repo_name)
         def last_rev(repo_name, cs_cache):
             return _render('revision', repo_name, cs_cache.get('revision'),
                            cs_cache.get('raw_id'), cs_cache.get('author'),
                            cs_cache.get('message'))
         def desc(desc):
             import kallithea.lib.helpers as h
             return h.urlify_text(desc, truncate=80, stylize=c.visual.stylify_metalabels)
         def state(repo_state):
             return _render("repo_state", repo_state)
         def repo_actions(repo_name):
             return _render('repo_actions', repo_name)
         def owner_actions(owner_id, username):
             return _render('user_name', owner_id, username)
         repos_data = []
         for gr in repo_groups_list or []:
             repos_data.append(dict(
                 raw_name='\0' + gr.name, # sort before repositories
                 just_name=gr.name,
                 name=_render('group_name_html', group_name=gr.group_name, name=gr.name),
                 desc=gr.group_description))
         for repo in repos_list:
             if not HasRepoPermissionLevel('read')(repo.repo_name, 'get_repos_as_dict check'):
                 continue
             cs_cache = repo.changeset_cache
             row = {
                 "raw_name": repo.repo_name,
                 "just_name": repo.just_name,
                 "name": repo_lnk(repo.repo_name, repo.repo_type,
                                  repo.repo_state, repo.private, repo.fork),
                 "following": following(
                     repo.repo_id,
                     scm.ScmModel().is_following_repo(repo.repo_name, request.authuser.user_id),
                 ),
                 "last_change_iso": repo.last_db_change.isoformat(),
                 "last_change": last_change(repo.last_db_change),
                 "last_changeset": last_rev(repo.repo_name, cs_cache),
                 "last_rev_raw": cs_cache.get('revision'),
                 "desc": desc(repo.description),
                 "owner": h.person(repo.owner),
                 "state": state(repo.repo_state),
                 "rss": rss_lnk(repo.repo_name),
                 "atom": atom_lnk(repo.repo_name),
+            }
             if admin:
                 row.update({
                     "action": repo_actions(repo.repo_name),
                     "owner": owner_actions(repo.owner_id,
                                            h.person(repo.owner))
                 })
             repos_data.append(row)
         return {
             "sort": "name",
             "dir": "asc",
             "records": repos_data
+        }
     def _get_defaults(self, repo_name):
         """
         Gets information about repository, and returns a dict for
         usage in forms
         :param repo_name:
         """
         repo_info = db.Repository.get_by_repo_name(repo_name)
         if repo_info is None:
             return None
         defaults = repo_info.get_dict()
         defaults['repo_name'] = repo_info.just_name
         defaults['repo_group'] = repo_info.group_id
         for strip, k in [(0, 'repo_type'), (1, 'repo_enable_downloads'),
                          (1, 'repo_description'),
                          (1, 'repo_landing_rev'), (0, 'clone_uri'),
                          (1, 'repo_private'), (1, 'repo_enable_statistics')]:
             attr = k
             if strip:
                 attr = remove_prefix(k, 'repo_')
             val = defaults[attr]
             if k == 'repo_landing_rev':
                 val = ':'.join(defaults[attr])
             defaults[k] = val
             if k == 'clone_uri':
                 defaults['clone_uri_hidden'] = repo_info.clone_uri_hidden
         # fill owner
         if repo_info.owner:
             defaults.update({'owner': repo_info.owner.username})
         else:
             replacement_user = db.User.query().filter(db.User.admin ==
                                                    True).first().username
             defaults.update({'owner': replacement_user})
         # fill repository users
         for p in repo_info.repo_to_perm:
             defaults.update({'u_perm_%s' % p.user.username:
                                  p.permission.permission_name})
         # fill repository groups
         for p in repo_info.users_group_to_perm:
             defaults.update({'g_perm_%s' % p.users_group.users_group_name:
                                  p.permission.permission_name})
         return defaults
     def update(self, repo, **kwargs):
         try:
             cur_repo = db.Repository.guess_instance(repo)
             org_repo_name = cur_repo.repo_name
             if 'owner' in kwargs:
                 cur_repo.owner = db.User.get_by_username(kwargs['owner'])
             if 'repo_group' in kwargs:
                 assert kwargs['repo_group'] != '-1', kwargs # RepoForm should have converted to None
                 cur_repo.group = db.RepoGroup.get(kwargs['repo_group'])
                 cur_repo.repo_name = cur_repo.get_new_name(cur_repo.just_name)
             log.debug('Updating repo %s with params:%s', cur_repo, kwargs)
             for k in ['repo_enable_downloads',
                       'repo_description',
                       'repo_landing_rev',
                       'repo_private',
                       'repo_enable_statistics',
                       ]:
                 if k in kwargs:
                     setattr(cur_repo, remove_prefix(k, 'repo_'), kwargs[k])
             clone_uri = kwargs.get('clone_uri')
             if clone_uri is not None and clone_uri != cur_repo.clone_uri_hidden:
                 # clone_uri is modified - if given a value, check it is valid
                 if clone_uri != '':
                     # will raise exception on error
                     is_valid_repo_uri(cur_repo.repo_type, clone_uri, make_ui())
                 cur_repo.clone_uri = clone_uri
             if 'repo_name' in kwargs:
                 repo_name = kwargs['repo_name']
                 if kallithea.lib.utils2.repo_name_slug(repo_name) != repo_name:
                     raise Exception('invalid repo name %s' % repo_name)
                 cur_repo.repo_name = cur_repo.get_new_name(repo_name)
             # if private flag is set, reset default permission to NONE
             if kwargs.get('repo_private'):
                 EMPTY_PERM = 'repository.none'
                 RepoModel().grant_user_permission(
                     repo=cur_repo, user='default', perm=EMPTY_PERM
+                )
                 # handle extra fields
             for field in [k for k in kwargs if k.startswith(db.RepositoryField.PREFIX)]:
                 k = db.RepositoryField.un_prefix_key(field)
                 ex_field = db.RepositoryField.get_by_key_name(key=k, repo=cur_repo)
                 if ex_field:
                     ex_field.field_value = kwargs[field]
             if org_repo_name != cur_repo.repo_name:
                 # rename repository
                 self._rename_filesystem_repo(old=org_repo_name, new=cur_repo.repo_name)
             return cur_repo
         except Exception:
             log.error(traceback.format_exc())
             raise
     def _create_repo(self, repo_name, repo_type, description, owner,
                      private=False, clone_uri=None, repo_group=None,
                      landing_rev='rev:tip', fork_of=None,
                      copy_fork_permissions=False, enable_statistics=False,
                      enable_downloads=False,
                      copy_group_permissions=False, state=db.Repository.STATE_PENDING):
         """
         Create repository inside database with PENDING state. This should only be
         executed by create() repo, with exception of importing existing repos.
         """
         owner = db.User.guess_instance(owner)
         fork_of = db.Repository.guess_instance(fork_of)
         repo_group = db.RepoGroup.guess_instance(repo_group)
         try:
             repo_name = repo_name
             description = description
             # repo name is just a name of repository
             # while repo_name_full is a full qualified name that is combined
             # with name and path of group
             repo_name_full = repo_name
             repo_name = repo_name.split(kallithea.URL_SEP)[-1]
             if kallithea.lib.utils2.repo_name_slug(repo_name) != repo_name:
                 raise Exception('invalid repo name %s' % repo_name)
             new_repo = db.Repository()
             new_repo.repo_state = state
             new_repo.enable_statistics = False
             new_repo.repo_name = repo_name_full
             new_repo.repo_type = repo_type
             new_repo.owner = owner
             new_repo.group = repo_group
             new_repo.description = description or repo_name
             new_repo.private = private
             if clone_uri:
                 # will raise exception on error
                 is_valid_repo_uri(repo_type, clone_uri, make_ui())
             new_repo.clone_uri = clone_uri
             new_repo.landing_rev = landing_rev
             new_repo.enable_statistics = enable_statistics
             new_repo.enable_downloads = enable_downloads
             if fork_of:
                 parent_repo = fork_of
                 new_repo.fork = parent_repo
             meta.Session().add(new_repo)
             if fork_of and copy_fork_permissions:
                 repo = fork_of
                 user_perms = db.UserRepoToPerm.query() \
                     .filter(db.UserRepoToPerm.repository == repo).all()
                 group_perms = db.UserGroupRepoToPerm.query() \
                     .filter(db.UserGroupRepoToPerm.repository == repo).all()
                 for perm in user_perms:
                     db.UserRepoToPerm.create(perm.user, new_repo, perm.permission)
                 for perm in group_perms:
                     db.UserGroupRepoToPerm.create(perm.users_group, new_repo,
                                                perm.permission)
             elif repo_group and copy_group_permissions:
                 user_perms = db.UserRepoGroupToPerm.query() \
                     .filter(db.UserRepoGroupToPerm.group == repo_group).all()
                 group_perms = db.UserGroupRepoGroupToPerm.query() \
                     .filter(db.UserGroupRepoGroupToPerm.group == repo_group).all()
                 for perm in user_perms:
                     perm_name = perm.permission.permission_name.replace('group.', 'repository.')
                     perm_obj = db.Permission.get_by_key(perm_name)
                     db.UserRepoToPerm.create(perm.user, new_repo, perm_obj)
                 for perm in group_perms:
                     perm_name = perm.permission.permission_name.replace('group.', 'repository.')
                     perm_obj = db.Permission.get_by_key(perm_name)
                     db.UserGroupRepoToPerm.create(perm.users_group, new_repo, perm_obj)
             else:
                 self._create_default_perms(new_repo, private)
             # now automatically start following this repository as owner
             scm.ScmModel().toggle_following_repo(new_repo.repo_id, owner.user_id)
             # we need to flush here, in order to check if database won't
             # throw any exceptions, create filesystem dirs at the very end
             meta.Session().flush()
             return new_repo
         except Exception:
             log.error(traceback.format_exc())
             raise
     def create(self, form_data, cur_user):
         """
         Create repository using celery tasks
         :param form_data:
         :param cur_user:
         """
         from kallithea.lib.celerylib import tasks
         return tasks.create_repo(form_data, cur_user)
     def _update_permissions(self, repo, perms_new=None, perms_updates=None,
                             check_perms=True):
         if not perms_new:
             perms_new = []
         if not perms_updates:
             perms_updates = []
         # update permissions
         for member, perm, member_type in perms_updates:
             if member_type == 'user':
                 # this updates existing one
                 self.grant_user_permission(
                     repo=repo, user=member, perm=perm
+                )
             else:
                 # check if we have permissions to alter this usergroup's access
                 if not check_perms or HasUserGroupPermissionLevel('read')(member):
                     self.grant_user_group_permission(
                         repo=repo, group_name=member, perm=perm
+                    )
             # set new permissions
         for member, perm, member_type in perms_new:
             if member_type == 'user':
                 self.grant_user_permission(
                     repo=repo, user=member, perm=perm
+                )
             else:
                 # check if we have permissions to alter this usergroup's access
                 if not check_perms or HasUserGroupPermissionLevel('read')(member):
                     self.grant_user_group_permission(
                         repo=repo, group_name=member, perm=perm
+                    )
     def create_fork(self, form_data, cur_user):
         """
         Simple wrapper into executing celery task for fork creation
         :param form_data:
         :param cur_user:
         """
         from kallithea.lib.celerylib import tasks
         return tasks.create_repo_fork(form_data, cur_user)
     def delete(self, repo, forks=None, fs_remove=True, cur_user=None):
         """
         Delete given repository, forks parameter defines what do do with
         attached forks. Throws AttachedForksError if deleted repo has attached
         forks
         :param repo:
         :param forks: str 'delete' or 'detach'
         :param fs_remove: remove(archive) repo from filesystem
         """
         if not cur_user:
             cur_user = getattr(get_current_authuser(), 'username', None)
         repo = db.Repository.guess_instance(repo)
         if repo is not None:
             if forks == 'detach':
                 for r in repo.forks:
                     r.fork = None
             elif forks == 'delete':
                 for r in repo.forks:
                     self.delete(r, forks='delete')

kallithea/tests/api/api_base.py

➞

Show inline comments

@@ @@ -307,769 +307,769 @@ class _BaseTestApi(object): @@
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=repo_name,
                                   clone_uri=custom_remote_path)
         response = api_call(self, params)
         expected = {'msg': 'Pulled from `%s`' % repo_name,
                     'repository': repo_name}
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_rescan_repos(self):
         id_, params = _build_data(self.apikey, 'rescan_repos')
         response = api_call(self, params)
         expected = {'added': [], 'removed': []}
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(ScmModel, 'repo_scan', raise_exception)
     def test_api_rescann_error(self):
         id_, params = _build_data(self.apikey, 'rescan_repos', )
         response = api_call(self, params)
         expected = 'Error occurred during rescan repositories action'
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_existing_user(self):
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=base.TEST_USER_ADMIN_LOGIN,
                                   email='test@example.com',
                                   password='trololo')
         response = api_call(self, params)
         expected = "user `%s` already exist" % base.TEST_USER_ADMIN_LOGIN
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_user_with_existing_email(self):
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=base.TEST_USER_ADMIN_LOGIN + 'new',
                                   email=base.TEST_USER_REGULAR_EMAIL,
                                   password='trololo')
         response = api_call(self, params)
         expected = "email `%s` already exist" % base.TEST_USER_REGULAR_EMAIL
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_user(self):
         username = 'test_new_api_user'
         email = username + "@example.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email,
                                   password='trololo')
         response = api_call(self, params)
         usr = db.User.get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
+        )
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
     def test_api_create_user_without_password(self):
         username = 'test_new_api_user_passwordless'
         email = username + "@example.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email)
         response = api_call(self, params)
         usr = db.User.get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
+        )
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
     def test_api_create_user_with_extern_name(self):
         username = 'test_new_api_user_passwordless'
         email = username + "@example.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email, extern_name='internal')
         response = api_call(self, params)
         usr = db.User.get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
+        )
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
     @mock.patch.object(UserModel, 'create_or_update', raise_exception)
     def test_api_create_user_when_exception_happened(self):
         username = 'test_new_api_user'
         email = username + "@example.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email,
                                   password='trololo')
         response = api_call(self, params)
         expected = 'failed to create user `%s`' % username
         self._compare_error(id_, expected, given=response.body)
     def test_api_delete_user(self):
         usr = UserModel().create_or_update(username='test_user',
                                            password='qweqwe',
                                            email='u232@example.com',
                                            firstname='u1', lastname='u1')
         meta.Session().commit()
         username = usr.username
         email = usr.email
         usr_id = usr.user_id
         ## DELETE THIS USER NOW
         id_, params = _build_data(self.apikey, 'delete_user',
                                   userid=username, )
         response = api_call(self, params)
         ret = {'msg': 'deleted user ID:%s %s' % (usr_id, username),
                'user': None}
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(UserModel, 'delete', raise_exception)
     def test_api_delete_user_when_exception_happened(self):
         usr = UserModel().create_or_update(username='test_user',
                                            password='qweqwe',
                                            email='u232@example.com',
                                            firstname='u1', lastname='u1')
         meta.Session().commit()
         username = usr.username
         id_, params = _build_data(self.apikey, 'delete_user',
                                   userid=username, )
         response = api_call(self, params)
         ret = 'failed to delete user ID:%s %s' % (usr.user_id,
                                                   usr.username)
         expected = ret
         self._compare_error(id_, expected, given=response.body)
     @base.parametrize('name,expected', [
         ('firstname', 'new_username'),
         ('lastname', 'new_username'),
         ('email', 'new_username'),
         ('admin', True),
         ('admin', False),
         ('extern_type', 'ldap'),
         ('extern_type', None),
         ('extern_name', 'test'),
         ('extern_name', None),
         ('active', False),
         ('active', True),
         ('password', 'newpass'),
     ])
     def test_api_update_user(self, name, expected):
         usr = db.User.get_by_username(self.TEST_USER_LOGIN)
         kw = {name: expected,
               'userid': usr.user_id}
         id_, params = _build_data(self.apikey, 'update_user', **kw)
         response = api_call(self, params)
         ret = {
             'msg': 'updated user ID:%s %s' % (
                 usr.user_id, self.TEST_USER_LOGIN),
             'user': jsonify(db.User \
                 .get_by_username(self.TEST_USER_LOGIN) \
                 .get_api_data())
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_update_user_no_changed_params(self):
         usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=base.TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         ret = {
             'msg': 'updated user ID:%s %s' % (
                 usr.user_id, base.TEST_USER_ADMIN_LOGIN),
             'user': ret
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_update_user_by_user_id(self):
         usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
         response = api_call(self, params)
         ret = {
             'msg': 'updated user ID:%s %s' % (
                 usr.user_id, base.TEST_USER_ADMIN_LOGIN),
             'user': ret
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_update_user_default_user(self):
         usr = db.User.get_default_user()
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
         response = api_call(self, params)
         expected = 'editing default user is forbidden'
         self._compare_error(id_, expected, given=response.body)
     @mock.patch.object(UserModel, 'update_user', raise_exception)
     def test_api_update_user_when_exception_happens(self):
         usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
         response = api_call(self, params)
         ret = 'failed to update user `%s`' % usr.user_id
         expected = ret
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo(self):
         new_group = 'some_new_group'
         make_user_group(new_group)
         RepoModel().grant_user_group_permission(repo=self.REPO,
                                                 group_name=new_group,
                                                 perm='repository.read')
         meta.Session().commit()
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
         assert "tags" not in response.json['result']
         assert 'pull_requests' not in response.json['result']
         repo = RepoModel().get_by_repo_name(self.REPO)
         ret = repo.get_api_data()
         members = []
         followers = []
         for user in repo.repo_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.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", 'permission': perm}
             members.append(user_group_data)
         for user in repo.followers:
             followers.append(user.user.get_api_data())
         ret['members'] = members
         ret['followers'] = followers
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_user_group(new_group)
         id_, params = _build_data(self.apikey, 'get_repo', repoid=self.REPO,
                                   with_revision_names=True,
                                   with_pullrequests=True)
         response = api_call(self, params)
         assert "v0.2.0" in response.json['result']['tags']
         assert 'pull_requests' in response.json['result']
     @base.parametrize('grant_perm', [
         ('repository.admin'),
         ('repository.write'),
         ('repository.read'),
     ])
     def test_api_get_repo_by_non_admin(self, grant_perm):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm=grant_perm)
         meta.Session().commit()
         id_, params = _build_data(self.apikey_regular, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(self.REPO)
         assert len(repo.repo_to_perm) >= 2  # make sure we actually are testing something - probably the default 2 permissions, possibly more
         expected = repo.get_api_data()
         members = []
         for user in repo.repo_to_perm:
             perm = user.permission.permission_name
             user_obj = user.user
             user_data = {'name': user_obj.username, 'type': "user",
                          'permission': perm}
             members.append(user_data)
         for user_group in repo.users_group_to_perm:
             perm = user_group.permission.permission_name
             user_group_obj = user_group.users_group
             user_group_data = {'name': user_group_obj.users_group_name,
                                'type': "user_group", 'permission': perm}
             members.append(user_group_data)
         expected['members'] = members
         followers = []
         for user in repo.followers:
             followers.append(user.user.get_api_data())
         expected['followers'] = followers
         try:
             self._compare_ok(id_, expected, given=response.body)
         finally:
             RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)
     def test_api_get_repo_by_non_admin_no_permission_to_repo(self):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=db.User.DEFAULT_USER_NAME,
                                           perm='repository.none')
         try:
             RepoModel().grant_user_permission(repo=self.REPO,
                                               user=self.TEST_USER_LOGIN,
                                               perm='repository.none')
             id_, params = _build_data(self.apikey_regular, 'get_repo',
                                       repoid=self.REPO)
             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=db.User.DEFAULT_USER_NAME,
                                               perm='repository.read')
     def test_api_get_repo_that_doesn_not_exist(self):
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid='no-such-repo')
         response = api_call(self, params)
         ret = 'repository `%s` does not exist' % 'no-such-repo'
         expected = ret
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repos(self):
         id_, params = _build_data(self.apikey, 'get_repos')
         response = api_call(self, params)
         expected = jsonify([
             repo.get_api_data()
             for repo in db.Repository.query()
         ])
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_repos_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_repos')
         response = api_call(self, params)
         expected = jsonify([
             repo.get_api_data()
-            for repo in RepoModel().get_all_user_repos(self.TEST_USER_LOGIN)
+            for repo in AuthUser(dbuser=db.User.get_by_username(self.TEST_USER_LOGIN)).get_all_user_repos()
         ])
         self._compare_ok(id_, expected, given=response.body)
     @base.parametrize('name,ret_type', [
         ('all', 'all'),
         ('dirs', 'dirs'),
         ('files', 'files'),
     ])
     def test_api_get_repo_nodes(self, name, ret_type):
         rev = 'tip'
         path = '/'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
         # we don't the actual return types here since it's tested somewhere
         # else
         expected = response.json['result']
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_repo_nodes_bad_revisions(self):
         rev = 'i-dont-exist'
         path = '/'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path, )
         response = api_call(self, params)
         expected = 'failed to get repo: `%s` nodes' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo_nodes_bad_path(self):
         rev = 'tip'
         path = '/idontexits'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path, )
         response = api_call(self, params)
         expected = 'failed to get repo: `%s` nodes' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo_nodes_bad_ret_type(self):
         rev = 'tip'
         path = '/'
         ret_type = 'error'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
         expected = ('ret_type must be one of %s'
                     % (','.join(sorted(['files', 'dirs', 'all']))))
         self._compare_error(id_, expected, given=response.body)
     @base.parametrize('name,ret_type,grant_perm', [
         ('all', 'all', 'repository.write'),
         ('dirs', 'dirs', 'repository.admin'),
         ('files', 'files', 'repository.read'),
     ])
     def test_api_get_repo_nodes_by_regular_user(self, name, ret_type, grant_perm):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm=grant_perm)
         meta.Session().commit()
         rev = 'tip'
         path = '/'
         id_, params = _build_data(self.apikey_regular, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
         # we don't the actual return types here since it's tested somewhere
         # else
         expected = response.json['result']
         try:
             self._compare_ok(id_, expected, given=response.body)
         finally:
             RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)
     def test_api_create_repo(self):
         repo_name = 'api-repo'
         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)
         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)
     @base.parametrize('repo_name', [
         '',
         '.',
         '..',
         ':',
         '/',
         '<test>',
     ])
     def test_api_create_repo_bad_names(self, repo_name):
         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)
         if repo_name == '/':
             expected = "repo group `` not found"
             self._compare_error(id_, expected, given=response.body)
         else:
             expected = "failed to create repository `%s`" % repo_name
             self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_create_repo_clone_uri_local(self):
         # cloning from local repos was a mis-feature - it would bypass access control
         # TODO: introduce other test coverage of actual remote cloning
         clone_uri = os.path.join(base.TESTS_TMP_PATH, self.REPO)
         repo_name = 'api-repo'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,
                                   clone_uri=clone_uri,
+        )
         response = api_call(self, params)
         expected = "failed to create repository `%s`" % repo_name
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_create_repo_and_repo_group(self):
         repo_group_name = 'my_gr'
         repo_name = '%s/api-repo' % repo_group_name
         # repo creation can no longer also create repo group
         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 = '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)
         meta.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 = db.RepoGroup.get_by_group_name(TEST_REPO_GROUP)
         assert top_group
         rg = fixture.create_repo_group(repo_group_basename, parent_group_id=top_group)
         meta.Session().commit()
         RepoGroupModel().grant_user_permission(repo_group_name,
                                                self.TEST_USER_LOGIN,
                                                'group.none')
         meta.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)
         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,
             '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_specify_owner(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,
                                   owner=owner)
         response = api_call(self, params)
         expected = 'Only Kallithea admin can specify `owner` param'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_create_repo_exists(self):
         repo_name = self.REPO
         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 = "repo `%s` already exist" % repo_name
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_repo_dot_dot(self):
         # it is only possible to create repositories in existing repo groups - and '..' can't be used
         group_name = '%s/..' % TEST_REPO_GROUP
         repo_name = '%s/%s' % (group_name, 'could-be-outside')
         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 = 'repo group `%s` not found' % group_name
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     @mock.patch.object(RepoModel, 'create', raise_exception)
     def test_api_create_repo_exception_occurred(self):
         repo_name = 'api-repo'
         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 = 'failed to create repository `%s`' % repo_name
         self._compare_error(id_, expected, given=response.body)
     @base.parametrize('changing_attr,updates', [
         ('owner', {'owner': base.TEST_USER_REGULAR_LOGIN}),
         ('description', {'description': 'new description'}),
         ('clone_uri', {'clone_uri': 'http://example.com/repo'}), # will fail - pulling from non-existing repo should fail
         ('clone_uri', {'clone_uri': '/repo'}), # will fail - pulling from local repo was a mis-feature - it would bypass access control
         ('clone_uri', {'clone_uri': None}),
         ('landing_rev', {'landing_rev': 'branch:master'}),
         ('enable_statistics', {'enable_statistics': 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)
         response = api_call(self, params)
         if changing_attr == 'name':
             repo_name = updates['name']
         if changing_attr == 'repo_group':
             repo_name = '/'.join([updates['group'], repo_name])
         try:
             if changing_attr == 'clone_uri' and updates['clone_uri']:
                 expected = 'failed to update repo `%s`' % repo_name
                 self._compare_error(id_, expected, given=response.body)
             else:
                 expected = {
                     'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo_name),
                     'repository': repo.get_api_data()
+                }
                 self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             if changing_attr == 'repo_group':
                 fixture.destroy_repo_group(updates['group'])
     @base.parametrize('changing_attr,updates', [
         ('owner', {'owner': base.TEST_USER_REGULAR_LOGIN}),
         ('description', {'description': 'new description'}),
         ('clone_uri', {'clone_uri': 'http://example.com/repo'}), # will fail - pulling from non-existing repo should fail
         ('clone_uri', {'clone_uri': '/repo'}), # will fail - pulling from local repo was a mis-feature - it would bypass access control
         ('clone_uri', {'clone_uri': None}),
         ('landing_rev', {'landing_rev': 'branch:master'}),
         ('enable_statistics', {'enable_statistics': True}),
         ('enable_downloads', {'enable_downloads': True}),
         ('name', {'name': 'new_repo_name'}),
         ('repo_group', {'group': 'test_group_for_update'}),
     ])
     def test_api_update_group_repo(self, changing_attr, updates):
         group_name = 'lololo'
         fixture.create_repo_group(group_name)
         repo_name = '%s/api_update_me' % group_name
         repo = fixture.create_repo(repo_name, repo_group=group_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)
         response = api_call(self, params)
         if changing_attr == 'name':
             repo_name = '%s/%s' % (group_name, updates['name'])
         if changing_attr == 'repo_group':
             repo_name = '/'.join([updates['group'], repo_name.rsplit('/', 1)[-1]])
         try:
             if changing_attr == 'clone_uri' and updates['clone_uri']:

0 comments (0 inline, 0 general)