kallithea Changeset - 974d6470cbec

Changeset - 974d6470cbec

Parent rev.

Child rev.

[Not reviewed]

default

0 5 0

Søren Løvborg - 9 years ago 2016-08-03 16:51:34
sorenl@unity3d.com

model: inline BaseModel.get_all

This is just needless indirection (and doesn't actually add any
abstraction on top of the database object), so inline all calls to it.

Don't touch PullRequestModel.get_all, though, since it is an entirely
unrelated method that just shadows the inherited get_all.

5 files changed with 9 insertions and 16 deletions:

kallithea/controllers/api/api.py

kallithea/model/__init__.py

kallithea/tests/api/api_base.py

kallithea/tests/fixture.py

kallithea/tests/models/test_user_groups.py

0 comments (0 inline, 0 general)

kallithea/controllers/api/api.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.controllers.api.api
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 API controller 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: Aug 20, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import time
 import traceback
 import logging
 from sqlalchemy import or_
 from kallithea import EXTERN_TYPE_INTERNAL
 from kallithea.controllers.api import JSONRPCController, JSONRPCError
 from kallithea.lib.auth import (
     PasswordGenerator, AuthUser, HasPermissionAnyDecorator,
     HasPermissionAnyDecorator, HasPermissionAnyApi, HasRepoPermissionAnyApi,
     HasRepoGroupPermissionAnyApi, HasUserGroupPermissionAny)
 from kallithea.lib.utils import map_groups, repo2db_mapper
 from kallithea.lib.utils2 import (
     str2bool, time_to_datetime, safe_int, Optional, OAttr)
 from kallithea.model.meta import Session
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.scm import ScmModel, UserGroupList
 from kallithea.model.repo import RepoModel
 from kallithea.model.user import UserModel
 from kallithea.model.user_group import UserGroupModel
 from kallithea.model.gist import GistModel
 from kallithea.model.db import (
     Repository, Setting, UserIpMap, Permission, User, Gist,
     RepoGroup)
+    RepoGroup, UserGroup)
 from kallithea.lib.compat import json
 from kallithea.lib.exceptions import (
     DefaultUserException, UserGroupsAssignedException)
 log = logging.getLogger(__name__)
 def store_update(updates, attr, name):
     """
     Stores param in updates dict if it's not instance of Optional
     allows easy updates of passed in params
     """
     if not isinstance(attr, Optional):
         updates[name] = attr
 def get_user_or_error(userid):
     """
     Get user by id or name or return JsonRPCError if not found
     :param userid:
     """
     user = UserModel().get_user(userid)
     if user is None:
         raise JSONRPCError("user `%s` does not exist" % (userid,))
     return user
 def get_repo_or_error(repoid):
     """
     Get repo by id or name or return JsonRPCError if not found
     :param repoid:
     """
     repo = RepoModel().get_repo(repoid)
     if repo is None:
         raise JSONRPCError('repository `%s` does not exist' % (repoid,))
     return repo
 def get_repo_group_or_error(repogroupid):
     """
     Get repo group by id or name or return JsonRPCError if not found
     :param repogroupid:
     """
     repo_group = RepoGroupModel()._get_repo_group(repogroupid)
     if repo_group is None:
         raise JSONRPCError(
             'repository group `%s` does not exist' % (repogroupid,))
     return repo_group
 def get_user_group_or_error(usergroupid):
     """
     Get user group by id or name or return JsonRPCError if not found
     :param usergroupid:
     """
     user_group = UserGroupModel().get_group(usergroupid)
     if user_group is None:
         raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
     return user_group
 def get_perm_or_error(permid, prefix=None):
     """
     Get permission by id or name or return JsonRPCError if not found
     :param permid:
     """
     perm = Permission.get_by_key(permid)
     if perm is None:
         raise JSONRPCError('permission `%s` does not exist' % (permid,))
     if prefix:
         if not perm.permission_name.startswith(prefix):
             raise JSONRPCError('permission `%s` is invalid, '
                                'should start with %s' % (permid, prefix))
     return perm
 def get_gist_or_error(gistid):
     """
     Get gist by id or gist_access_id or return JsonRPCError if not found
     :param gistid:
     """
     gist = GistModel().get_gist(gistid)
     if gist is None:
         raise JSONRPCError('gist `%s` does not exist' % (gistid,))
     return gist
 class ApiController(JSONRPCController):
     """
     API Controller
     Each method takes USER as first argument. This is then, based on given
     API_KEY propagated as instance of user object who's making the call.
     example function::
         def func(apiuser,arg1, arg2,...):
             pass
     Each function should also **raise** JSONRPCError for any
     errors that happens.
     """
     @HasPermissionAnyDecorator('hg.admin')
     def test(self, apiuser, args):
         return args
     @HasPermissionAnyDecorator('hg.admin')
     def pull(self, apiuser, repoid):
         """
         Triggers a pull from remote location on given repo. Can be used to
         automatically keep remote repos up to date. This command can be executed
         only using api_key belonging to user with admin rights
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param repoid: repository name or repository id
         :type repoid: str or int
         OUTPUT::
           id : <id_given_in_input>
           result : {
             "msg": "Pulled from `<repository name>`"
             "repository": "<repository name>"
+          }
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "Unable to pull changes from `<reponame>`"
+          }
         """
         repo = get_repo_or_error(repoid)
         try:
             ScmModel().pull_changes(repo.repo_name,
                                     self.authuser.username)
             return dict(
                 msg='Pulled from `%s`' % repo.repo_name,
                 repository=repo.repo_name
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Unable to pull changes from `%s`' % repo.repo_name
+            )
     @HasPermissionAnyDecorator('hg.admin')
     def rescan_repos(self, apiuser, remove_obsolete=Optional(False)):
         """
         Triggers rescan repositories action. If remove_obsolete is set
         than also delete repos that are in database but not in the filesystem.
         aka "clean zombies". This command can be executed only using api_key
         belonging to user with admin rights.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param remove_obsolete: deletes repositories from
             database that are not found on the filesystem
         :type remove_obsolete: Optional(bool)
         OUTPUT::
           id : <id_given_in_input>
           result : {
             'added': [<added repository name>,...]
             'removed': [<removed repository name>,...]
+          }
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             'Error occurred during rescan repositories action'
+          }
         """
@@ @@ -696,385 +696,385 @@ class ApiController(JSONRPCController): @@
             Session().commit()
             return dict(
                 msg='created new user `%s`' % username,
                 user=user.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create user `%s`' % (username,))
     @HasPermissionAnyDecorator('hg.admin')
     def update_user(self, apiuser, userid, username=Optional(None),
                     email=Optional(None), password=Optional(None),
                     firstname=Optional(None), lastname=Optional(None),
                     active=Optional(None), admin=Optional(None),
                     extern_type=Optional(None), extern_name=Optional(None)):
         """
         updates given user if such user exists. This command can
         be executed only using api_key belonging to user with admin rights.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param userid: userid to update
         :type userid: str or int
         :param username: new username
         :type username: str or int
         :param email: email
         :type email: str
         :param password: password
         :type password: Optional(str)
         :param firstname: firstname
         :type firstname: Optional(str)
         :param lastname: lastname
         :type lastname: Optional(str)
         :param active: active
         :type active: Optional(bool)
         :param admin: admin
         :type admin: Optional(bool)
         :param extern_name:
         :type extern_name: Optional(str)
         :param extern_type:
         :type extern_type: Optional(str)
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg" : "updated user ID:<userid> <username>",
                       "user": <user_object>,
+                    }
             error:  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to update user `<username>`"
+          }
         """
         user = get_user_or_error(userid)
         # only non optional arguments will be stored in updates
         updates = {}
         try:
             store_update(updates, username, 'username')
             store_update(updates, password, 'password')
             store_update(updates, email, 'email')
             store_update(updates, firstname, 'name')
             store_update(updates, lastname, 'lastname')
             store_update(updates, active, 'active')
             store_update(updates, admin, 'admin')
             store_update(updates, extern_name, 'extern_name')
             store_update(updates, extern_type, 'extern_type')
             user = UserModel().update_user(user, **updates)
             Session().commit()
             return dict(
                 msg='updated user ID:%s %s' % (user.user_id, user.username),
                 user=user.get_api_data()
+            )
         except DefaultUserException:
             log.error(traceback.format_exc())
             raise JSONRPCError('editing default user is forbidden')
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update user `%s`' % (userid,))
     @HasPermissionAnyDecorator('hg.admin')
     def delete_user(self, apiuser, userid):
         """
         deletes given user if such user exists. This command can
         be executed only using api_key belonging to user with admin rights.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param userid: user to delete
         :type userid: str or int
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg" : "deleted user ID:<userid> <username>",
                       "user": null
+                    }
             error:  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to delete user ID:<userid> <username>"
+          }
         """
         user = get_user_or_error(userid)
         try:
             UserModel().delete(userid)
             Session().commit()
             return dict(
                 msg='deleted user ID:%s %s' % (user.user_id, user.username),
                 user=None
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to delete user ID:%s %s'
                                % (user.user_id, user.username))
     # permission check inside
     def get_user_group(self, apiuser, usergroupid):
         """
         Gets an existing user group. This command can be executed only using api_key
         belonging to user with admin rights or user who has at least
         read access to user group.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param usergroupid: id of user_group to edit
         :type usergroupid: str or int
         OUTPUT::
             id : <id_given_in_input>
             result : None if group not exist
+                     {
                        "users_group_id" : "<id>",
                        "group_name" :     "<groupname>",
                        "active":          "<bool>",
                        "members" :  [<user_obj>,...]
+                     }
             error : null
         """
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAnyApi('hg.admin')(user=apiuser):
             # check if we have at least read permission for this user group !
             _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
             if not HasUserGroupPermissionAny(*_perms)(
                     user=apiuser, user_group_name=user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
         data = user_group.get_api_data()
         return data
     # permission check inside
     def get_user_groups(self, apiuser):
         """
         Lists all existing user groups. This command can be executed only using
         api_key belonging to user with admin rights or user who has at least
         read access to user group.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         OUTPUT::
             id : <id_given_in_input>
             result : [<user_group_obj>,...]
             error : null
         """
         result = []
         _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
         extras = {'user': apiuser}
-        for user_group in UserGroupList(UserGroupModel().get_all(),
+        for user_group in UserGroupList(UserGroup.get_all(),
                                         perm_set=_perms, extra_kwargs=extras):
             result.append(user_group.get_api_data())
         return result
     @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')
     def create_user_group(self, apiuser, group_name, description=Optional(''),
                           owner=Optional(OAttr('apiuser')), active=Optional(True)):
         """
         Creates new user group. This command can be executed only using api_key
         belonging to user with admin rights or an user who has create user group
         permission
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param group_name: name of new user group
         :type group_name: str
         :param description: group description
         :type description: str
         :param owner: owner of group. If not passed apiuser is the owner
         :type owner: Optional(str or int)
         :param active: group is active
         :type active: Optional(bool)
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg": "created new user group `<groupname>`",
                       "user_group": <user_group_object>
+                    }
             error:  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           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 isinstance(owner, Optional):
                 owner = apiuser.user_id
             owner = get_user_or_error(owner)
             active = Optional.extract(active)
             description = Optional.extract(description)
             ug = UserGroupModel().create(name=group_name, description=description,
                                          owner=owner, active=active)
             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, apiuser, usergroupid, group_name=Optional(''),
                           description=Optional(''), owner=Optional(None),
                           active=Optional(True)):
         """
         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 apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :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 HasPermissionAnyApi('hg.admin')(user=apiuser):
             # check if we have admin permission for this user group !
             _perms = ('usergroup.admin',)
             if not HasUserGroupPermissionAny(*_perms)(
                     user=apiuser, user_group_name=user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
         if not isinstance(owner, Optional):
             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, 'user')
         store_update(updates, active, 'users_group_active')
         try:
             UserGroupModel().update(user_group, updates)
             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, apiuser, 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 apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :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 HasPermissionAnyApi('hg.admin')(user=apiuser):
             # check if we have admin permission for this user group !
             _perms = ('usergroup.admin',)
             if not HasUserGroupPermissionAny(*_perms)(
                     user=apiuser, user_group_name=user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
         try:
             UserGroupModel().delete(user_group)
             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, apiuser, usergroupid, userid):
         """
@@ @@ -1134,385 +1134,385 @@ class ApiController(JSONRPCController): @@
         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, apiuser, 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 apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :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 HasPermissionAnyApi('hg.admin')(user=apiuser):
             # check if we have admin permission for this user group !
             _perms = ('usergroup.admin',)
             if not HasUserGroupPermissionAny(*_perms)(
                     user=apiuser, user_group_name=user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
         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"
             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, apiuser, repoid):
         """
         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 apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :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_locking":    "<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>, ...]
+                 ]
+            }
+          }
           error :  null
         """
         repo = get_repo_or_error(repoid)
         if not HasPermissionAnyApi('hg.admin')(user=apiuser):
             # check if we have admin permission for this repo !
             perms = ('repository.admin', 'repository.write', 'repository.read')
             if not HasRepoPermissionAnyApi(*perms)(user=apiuser, repo_name=repo.repo_name):
                 raise JSONRPCError('repository `%s` does not exist' % (repoid,))
         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())
         data = repo.get_api_data()
         data['members'] = members
         data['followers'] = followers
         return data
     # permission check inside
     def get_repos(self, apiuser):
         """
         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.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         OUTPUT::
             id : <id_given_in_input>
             result: [
+                      {
                         "repo_id" :          "<repo_id>",
                         "repo_name" :        "<reponame>"
                         "repo_type" :        "<repo_type>",
                         "clone_uri" :        "<clone_uri>",
                         "private": :         "<bool>",
                         "created_on" :       "<datetimecreated>",
                         "description" :      "<description>",
                         "landing_rev":       "<landing_rev>",
                         "owner":             "<repo_owner>",
                         "fork_of":           "<name_of_fork_parent>",
                         "enable_downloads":  "<bool>",
                         "enable_locking":    "<bool>",
                         "enable_statistics": "<bool>",
                       },
                       …
+                    ]
             error:  null
         """
         result = []
         if not HasPermissionAnyApi('hg.admin')(user=apiuser):
             repos = RepoModel().get_all_user_repos(user=apiuser)
         else:
-            repos = RepoModel().get_all()
+            repos = Repository.get_all()
         for repo in repos:
             result.append(repo.get_api_data())
         return result
     # permission check inside
     def get_repo_nodes(self, apiuser, repoid, revision, root_path,
                        ret_type=Optional('all')):
         """
         returns a list of nodes and it's children in a flat list for a given path
         at given revision. It's possible to specify ret_type to show only `files` or
         `dirs`.  This command can be executed only using api_key belonging to
         user with admin rights or regular user that have at least read access to repository.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :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 HasPermissionAnyApi('hg.admin')(user=apiuser):
             # check if we have admin permission for this repo !
             perms = ('repository.admin', 'repository.write', 'repository.read')
             if not HasRepoPermissionAnyApi(*perms)(user=apiuser, repo_name=repo.repo_name):
                 raise JSONRPCError('repository `%s` does not exist' % (repoid,))
         ret_type = Optional.extract(ret_type)
         _map = {}
         try:
             _d, _f = ScmModel().get_nodes(repo, revision, root_path,
                                           flat=False)
             _map = {
                 'all': _d + _f,
                 'files': _f,
                 'dirs': _d,
+            }
             return _map[ret_type]
         except KeyError:
             raise JSONRPCError('ret_type must be one of %s'
                                % (','.join(_map.keys())))
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to get repo: `%s` nodes' % repo.repo_name
+            )
     @HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
     def create_repo(self, apiuser, repo_name, owner=Optional(OAttr('apiuser')),
                     repo_type=Optional('hg'), description=Optional(''),
                     private=Optional(False), clone_uri=Optional(None),
                     landing_rev=Optional('rev:tip'),
                     enable_statistics=Optional(False),
                     enable_locking=Optional(False),
                     enable_downloads=Optional(False),
                     copy_permissions=Optional(False)):
         """
         Creates a repository. If repository name contains "/", all needed repository
         groups will be created. For example "foo/bar/baz" will create groups
         "foo", "bar" (with "foo" as parent), and create "baz" repository with
         "bar" as group. This command can be executed only using api_key
         belonging to user with admin rights or regular user that have create
         repository permission. Regular users cannot specify owner parameter
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param repo_name: repository name
         :type repo_name: str
         :param owner: user_id or username
         :type owner: Optional(str)
         :param repo_type: 'hg' or 'git'
         :type repo_type: Optional(str)
         :param description: repository description
         :type description: Optional(str)
         :param private:
         :type private: bool
         :param clone_uri:
         :type clone_uri: str
         :param landing_rev: <rev_type>:<rev>
         :type landing_rev: str
         :param enable_locking:
         :type enable_locking: bool
         :param enable_downloads:
         :type enable_downloads: bool
         :param enable_statistics:
         :type enable_statistics: bool
         :param copy_permissions: Copy permission from group that repository is
             being created.
         :type copy_permissions: bool
         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 HasPermissionAnyApi('hg.admin')(user=apiuser):
             if not isinstance(owner, Optional):
                 # forbid setting owner for non-admins
                 raise JSONRPCError(
                     'Only Kallithea admin can specify `owner` param'
+                )
         if isinstance(owner, Optional):
             owner = apiuser.user_id
         owner = get_user_or_error(owner)
         if RepoModel().get_by_repo_name(repo_name):
             raise JSONRPCError("repo `%s` already exist" % repo_name)
         defs = Setting.get_default_repo_settings(strip_prefix=True)
         if isinstance(private, Optional):
             private = defs.get('repo_private') or Optional.extract(private)
         if isinstance(repo_type, Optional):
             repo_type = defs.get('repo_type')
         if isinstance(enable_statistics, Optional):
             enable_statistics = defs.get('repo_enable_statistics')
         if isinstance(enable_locking, Optional):
             enable_locking = defs.get('repo_enable_locking')
         if isinstance(enable_downloads, Optional):
             enable_downloads = defs.get('repo_enable_downloads')
         clone_uri = Optional.extract(clone_uri)
         description = Optional.extract(description)
         landing_rev = Optional.extract(landing_rev)
         copy_permissions = Optional.extract(copy_permissions)
         try:
             repo_name_cleaned = repo_name.split('/')[-1]
             # create structure of groups and return the last group
             repo_group = map_groups(repo_name)
             data = dict(
                 repo_name=repo_name_cleaned,
                 repo_name_full=repo_name,
                 repo_type=repo_type,
                 repo_description=description,
                 owner=owner,
                 repo_private=private,
                 clone_uri=clone_uri,
                 repo_group=repo_group,
                 repo_landing_rev=landing_rev,
                 enable_statistics=enable_statistics,
                 enable_locking=enable_locking,
                 enable_downloads=enable_downloads,
                 repo_copy_permissions=copy_permissions,
+            )
             task = RepoModel().create(form_data=data, cur_user=owner)
             from celery.result import BaseAsyncResult
             task_id = None
             if isinstance(task, BaseAsyncResult):
                 task_id = task.task_id
             # no commit, it's done in RepoModel, or async via celery
             return dict(
                 msg="Created new repository `%s`" % (repo_name,),
                 success=True,  # cannot return the repo data here since fork
                                # can be done async
                 task=task_id
+            )
         except Exception:

kallithea/model/__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/>.
 """
 kallithea.model
 ~~~~~~~~~~~~~~~
 The application's model objects
 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: Nov 25, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 :example:
     .. code-block:: python
        from paste.deploy import appconfig
        from pylons import config
        from sqlalchemy import engine_from_config
        from kallithea.config.environment import load_environment
        conf = appconfig('config:development.ini', relative_to = './../../')
        load_environment(conf.global_conf, conf.local_conf)
        engine = engine_from_config(config, 'sqlalchemy.')
        init_model(engine)
        # RUN YOUR CODE HERE
 """
 import logging
 from kallithea.model import meta
 from kallithea.lib.utils2 import safe_str, obfuscate_url_pw
 log = logging.getLogger(__name__)
 def init_model(engine):
     """
     Initializes db session, bind the engine with the metadata,
     Call this before using any of the tables or classes in the model,
     preferably once in application start
     :param engine: engine to bind to
     """
     engine_str = obfuscate_url_pw(str(engine.url))
     log.info("initializing db for %s", engine_str)
     meta.Base.metadata.bind = engine
 class BaseModel(object):
     """
     Base Model for all Kallithea models, it adds sql alchemy session
     into instance of model
     :param sa: If passed it reuses this session instead of creating a new one
     """
     cls = None  # override in child class
     def __init__(self, sa=None):
         if sa is not None:
             self.sa = sa
         else:
             self.sa = meta.Session()
     def _get_instance(self, cls, instance, callback=None):
         """
         Gets instance of given cls using some simple lookup mechanism.
         :param cls: class to fetch
         :param instance: int or Instance
         :param callback: callback to call if all lookups failed
         """
         if isinstance(instance, cls):
             return instance
         elif isinstance(instance, (int, long)) or safe_str(instance).isdigit():
             return cls.get(instance)
         else:
             if instance is not None:
                 if callback is None:
                     raise Exception(
                         'given object must be int, long or Instance of %s '
                         'got %s, no callback provided' % (cls, type(instance))
+                    )
                 else:
                     return callback(instance)
     def _get_user(self, user):
         """
         Helper method to get user by ID, or username fallback
         :param user: UserID, username, or User instance
         """
         from kallithea.model.db import User
         return self._get_instance(User, user,
                                   callback=User.get_by_username)
     def _get_repo(self, repository):
         """
         Helper method to get repository by ID, or repository name
         :param repository: RepoID, repository name or Repository Instance
         """
         from kallithea.model.db import Repository
         return self._get_instance(Repository, repository,
                                   callback=Repository.get_by_repo_name)
     def _get_perm(self, permission):
         """
         Helper method to get permission by ID, or permission name
         :param permission: PermissionID, permission_name or Permission instance
         """
         from kallithea.model.db import Permission
         return self._get_instance(Permission, permission,
                                   callback=Permission.get_by_key)
     @classmethod
     def get_all(cls):
         """
         Returns all instances of what is defined in `cls` class variable
         """
         return cls.cls.get_all()

kallithea/tests/api/api_base.py

➞

Show inline comments

@@ @@ -691,385 +691,385 @@ class _BaseTestApi(object): @@
         ('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 = 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(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 = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         ret = {
             'msg': 'updated user ID:%s %s' % (
                 usr.user_id, 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 = User.get_by_username(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, 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 = 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', crash)
     def test_api_update_user_when_exception_happens(self):
         usr = User.get_by_username(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 = u'some_new_group'
         make_user_group(new_group)
         RepoModel().grant_user_group_permission(repo=self.REPO,
                                                 group_name=new_group,
                                                 perm='repository.read')
         Session().commit()
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
         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)
     @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)
         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)
         ret = repo.get_api_data()
         members = []
         followers = []
         assert 2 == len(repo.repo_to_perm)
         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)
         for user in repo.followers:
             followers.append(user.user.get_api_data())
         ret['members'] = members
         ret['followers'] = followers
         expected = ret
         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=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)
     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)
         result = []
-        for repo in RepoModel().get_all():
+        for repo in Repository.get_all():
             result.append(repo.get_api_data())
         ret = jsonify(result)
         expected = ret
         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)
         result = []
         for repo in RepoModel().get_all_user_repos(self.TEST_USER_LOGIN):
             result.append(repo.get_api_data())
         ret = jsonify(result)
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     @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(['files', 'dirs', 'all'])))
         self._compare_error(id_, expected, given=response.body)
     @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)
         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 = u'api-repo'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo != 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_and_repo_group(self):
         repo_name = u'my_gr/api-repo'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,)
         response = api_call(self, params)
         print params
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo != 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)
         fixture.destroy_repo_group(u'my_gr')
     def test_api_create_repo_in_repo_group_without_permission(self):
         repo_group_name = u'%s/api-repo-repo' % TEST_REPO_GROUP
         repo_name = u'%s/api-repo' % repo_group_name
         rg = fixture.create_repo_group(repo_group_name)
         Session().commit()
         RepoGroupModel().grant_user_permission(repo_group_name,
                                                self.TEST_USER_LOGIN,
                                                'group.none')
         Session().commit()
         id_, params = _build_data(self.apikey_regular, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         # Current result when API access control is different from Web:
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
             'task': None,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         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 = u'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 = u'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
+        )

kallithea/tests/fixture.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/>.
 """
 Helpers for fixture generation
 """
 import os
 import time
 from kallithea.tests import *
 from kallithea.model.db import Repository, User, RepoGroup, UserGroup
+from kallithea.model.db import Repository, User, RepoGroup, UserGroup, Gist
 from kallithea.model.meta import Session
 from kallithea.model.repo import RepoModel
 from kallithea.model.user import UserModel
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.user_group import UserGroupModel
 from kallithea.model.gist import GistModel
 from kallithea.model.scm import ScmModel
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from os.path import dirname
 FIXTURES = os.path.join(dirname(dirname(os.path.abspath(__file__))), 'tests', 'fixtures')
 def error_function(*args, **kwargs):
     raise Exception('Total Crash !')
 class Fixture(object):
     def __init__(self):
         pass
     def anon_access(self, status):
         """
         Context manager for controlling anonymous access.
         Anon access will be set and committed, but restored again when exiting the block.
         Usage:
         fixture = Fixture()
         with fixture.anon_access(False):
             stuff
         """
         class context(object):
             def __enter__(self):
                 anon = User.get_default_user()
                 self._before = anon.active
                 anon.active = status
                 Session().add(anon)
                 Session().commit()
                 time.sleep(1.5)  # hack: wait for beaker sql_cache_short to expire
             def __exit__(self, exc_type, exc_val, exc_tb):
                 anon = User.get_default_user()
                 anon.active = self._before
                 Session().add(anon)
                 Session().commit()
         return context()
     def _get_repo_create_params(self, **custom):
         defs = dict(
             repo_name=None,
             repo_type='hg',
             clone_uri='',
             repo_group=u'-1',
             repo_description=u'DESC',
             repo_private=False,
             repo_landing_rev='rev:tip',
             repo_copy_permissions=False,
             repo_state=Repository.STATE_CREATED,
+        )
         defs.update(custom)
         if 'repo_name_full' not in custom:
             defs.update({'repo_name_full': defs['repo_name']})
         # fix the repo name if passed as repo_name_full
         if defs['repo_name']:
             defs['repo_name'] = defs['repo_name'].split('/')[-1]
         return defs
     def _get_group_create_params(self, **custom):
         defs = dict(
             group_name=None,
             group_description=u'DESC',
             group_parent_id=None,
             perms_updates=[],
             perms_new=[],
             enable_locking=False,
             recursive=False
+        )
         defs.update(custom)
         return defs
     def _get_user_create_params(self, name, **custom):
         defs = dict(
             username=name,
             password='qweqwe',
             email='%s+test@example.com' % name,
             firstname=u'TestUser',
             lastname=u'Test',
             active=True,
             admin=False,
             extern_type='internal',
             extern_name=None
+        )
         defs.update(custom)
         return defs
     def _get_user_group_create_params(self, name, **custom):
         defs = dict(
             users_group_name=name,
             user_group_description=u'DESC',
             users_group_active=True,
             user_group_data={},
+        )
         defs.update(custom)
         return defs
     def create_repo(self, name, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             r = Repository.get_by_repo_name(name)
             if r:
                 return r
         if isinstance(kwargs.get('repo_group'), RepoGroup):
             kwargs['repo_group'] = kwargs['repo_group'].group_id
         form_data = self._get_repo_create_params(repo_name=name, **kwargs)
         cur_user = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
         RepoModel().create(form_data, cur_user)
         Session().commit()
         ScmModel().mark_for_invalidation(name)
         return Repository.get_by_repo_name(name)
     def create_fork(self, repo_to_fork, fork_name, **kwargs):
         repo_to_fork = Repository.get_by_repo_name(repo_to_fork)
         form_data = self._get_repo_create_params(repo_name=fork_name,
                                             fork_parent_id=repo_to_fork,
                                             repo_type=repo_to_fork.repo_type,
                                             **kwargs)
         form_data['update_after_clone'] = False
         #TODO: fix it !!
         form_data['description'] = form_data['repo_description']
         form_data['private'] = form_data['repo_private']
         form_data['landing_rev'] = form_data['repo_landing_rev']
         owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
         RepoModel().create_fork(form_data, cur_user=owner)
         Session().commit()
         ScmModel().mark_for_invalidation(fork_name)
         r = Repository.get_by_repo_name(fork_name)
         assert r
         return r
     def destroy_repo(self, repo_name, **kwargs):
         RepoModel().delete(repo_name, **kwargs)
         Session().commit()
     def create_repo_group(self, name, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             gr = RepoGroup.get_by_group_name(group_name=name)
             if gr:
                 return gr
         form_data = self._get_group_create_params(group_name=name, **kwargs)
         owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
         gr = RepoGroupModel().create(
             group_name=form_data['group_name'],
             group_description=form_data['group_name'],
             owner=owner, parent=form_data['group_parent_id'])
         Session().commit()
         gr = RepoGroup.get_by_group_name(gr.group_name)
         return gr
     def destroy_repo_group(self, repogroupid):
         RepoGroupModel().delete(repogroupid)
         Session().commit()
     def create_user(self, name, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             user = User.get_by_username(name)
             if user:
                 return user
         form_data = self._get_user_create_params(name, **kwargs)
         user = UserModel().create(form_data)
         Session().commit()
         user = User.get_by_username(user.username)
         return user
     def destroy_user(self, userid):
         UserModel().delete(userid)
         Session().commit()
     def create_user_group(self, name, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             gr = UserGroup.get_by_group_name(group_name=name)
             if gr:
                 return gr
         form_data = self._get_user_group_create_params(name, **kwargs)
         owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
         user_group = UserGroupModel().create(
             name=form_data['users_group_name'],
             description=form_data['user_group_description'],
             owner=owner, active=form_data['users_group_active'],
             group_data=form_data['user_group_data'])
         Session().commit()
         user_group = UserGroup.get_by_group_name(user_group.users_group_name)
         return user_group
     def destroy_user_group(self, usergroupid):
         UserGroupModel().delete(user_group=usergroupid, force=True)
         Session().commit()
     def create_gist(self, **kwargs):
         form_data = {
             'description': u'new-gist',
             'owner': TEST_USER_ADMIN_LOGIN,
             'gist_type': GistModel.cls.GIST_PUBLIC,
             'lifetime': -1,
             'gist_mapping': {'filename1.txt':{'content':'hello world'},}
+        }
         form_data.update(kwargs)
         gist = GistModel().create(
             description=form_data['description'],owner=form_data['owner'],
             gist_mapping=form_data['gist_mapping'], gist_type=form_data['gist_type'],
             lifetime=form_data['lifetime']
+        )
         Session().commit()
         return gist
     def destroy_gists(self, gistid=None):
-        for g in GistModel.cls.get_all():
+        for g in Gist.get_all():
             if gistid:
                 if gistid == g.gist_access_id:
                     GistModel().delete(g)
             else:
                 GistModel().delete(g)
         Session().commit()
     def load_resource(self, resource_name, strip=True):
         with open(os.path.join(FIXTURES, resource_name), 'rb') as f:
             source = f.read()
             if strip:
                 source = source.strip()
         return source
     def commit_change(self, repo, filename, content, message, vcs_type, parent=None, newfile=False):
         repo = Repository.get_by_repo_name(repo)
         _cs = parent
         if not parent:
             _cs = EmptyChangeset(alias=vcs_type)
         if newfile:
             nodes = {
                 filename: {
                     'content': content
+                }
+            }
             cs = ScmModel().create_nodes(
                 user=TEST_USER_ADMIN_LOGIN, repo=repo,
                 message=message,
                 nodes=nodes,
                 parent_cs=_cs,
                 author=TEST_USER_ADMIN_LOGIN,
+            )
         else:
             cs = ScmModel().commit_change(
                 repo=repo.scm_instance, repo_name=repo.repo_name,
                 cs=parent, user=TEST_USER_ADMIN_LOGIN,
                 author=TEST_USER_ADMIN_LOGIN,
                 message=message,
                 content=content,
                 f_path=filename
+            )
         return cs

kallithea/tests/models/test_user_groups.py

➞

Show inline comments

 from kallithea.model.db import User
+from kallithea.model.db import User, UserGroup
 from kallithea.tests import *
 from kallithea.tests.fixture import Fixture
 from kallithea.model.user_group import UserGroupModel
 from kallithea.model.meta import Session
 fixture = Fixture()
 class TestUserGroups(TestController):
     def teardown_method(self, method):
         # delete all groups
-        for gr in UserGroupModel.get_all():
+        for gr in UserGroup.get_all():
             fixture.destroy_user_group(gr)
         Session().commit()
     @parametrize('pre_existing,regular_should_be,external_should_be,groups,expected', [
         ([], [], [], [], []),
         ([], [u'regular'], [], [], [u'regular']),  # no changes of regular
         ([u'some_other'], [], [], [u'some_other'], []),   # not added to regular group
         ([], [u'regular'], [u'container'], [u'container'], [u'regular', u'container']),
         ([], [u'regular'], [], [u'container', u'container2'], [u'regular', u'container', u'container2']),
         ([], [u'regular'], [u'other'], [], [u'regular']),  # remove not used
         ([u'some_other'], [u'regular'], [u'other', u'container'], [u'container', u'container2'], [u'regular', u'container', u'container2']),
     ])
     def test_enforce_groups(self, pre_existing, regular_should_be,
                             external_should_be, groups, expected):
         # delete all groups
-        for gr in UserGroupModel.get_all():
+        for gr in UserGroup.get_all():
             fixture.destroy_user_group(gr)
         Session().commit()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         for gr in pre_existing:
             gr = fixture.create_user_group(gr)
         Session().commit()
         # make sure use is just in those groups
         for gr in regular_should_be:
             gr = fixture.create_user_group(gr)
             Session().commit()
             UserGroupModel().add_user_to_group(gr, user)
             Session().commit()
         # now special external groups created by auth plugins
         for gr in external_should_be:
             gr = fixture.create_user_group(gr, user_group_data={'extern_type': 'container'})
             Session().commit()
             UserGroupModel().add_user_to_group(gr, user)
             Session().commit()
         UserGroupModel().enforce_groups(user, groups, 'container')
         Session().commit()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         in_groups = user.group_member
         assert expected == [x.users_group.users_group_name for x in in_groups]

0 comments (0 inline, 0 general)