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
@@ @@ -792,193 +792,193 @@ class ApiController(JSONRPCController): @@
         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::
@@ @@ -1230,193 +1230,193 @@ class ApiController(JSONRPCController): @@
                 "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:

kallithea/model/__init__.py

➞

Show inline comments

@@ @@ -39,103 +39,96 @@ Original author and date, and relevant c @@
        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

@@ @@ -787,193 +787,193 @@ class _BaseTestApi(object): @@
         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:

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,
@@ @@ -159,141 +159,141 @@ class Fixture(object): @@
                                             **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)