kallithea Changeset - 7bffccee3a49

Changeset - 7bffccee3a49

Parent rev.

Child rev.

[Not reviewed]

default

0 11 0

Søren Løvborg - 9 years ago 2016-09-23 14:05:42
sorenl@unity3d.com

db: inline calls to get_all

This method saves basically no typing, compared to "query().all()".
Additionally, "all()" returns a list, forcing all records to be loaded
into a memory at the same time, but some callers just need to iterate
over the objects one at a time, in which case "query()" alone is more
efficient. In one case, the caller can even use "count()" and avoid
loading any objects from the database at all.

11 files changed with 16 insertions and 20 deletions:

kallithea/controllers/admin/settings.py

kallithea/controllers/api/api.py

kallithea/lib/paster_commands/update_repoinfo.py

kallithea/model/db.py

kallithea/model/repo.py

kallithea/tests/api/api_base.py

kallithea/tests/fixture.py

kallithea/tests/functional/test_admin_gists.py

kallithea/tests/functional/test_home.py

kallithea/tests/models/test_user_groups.py

kallithea/tests/other/manual_test_vcs_operations.py

0 comments (0 inline, 0 general)

kallithea/controllers/admin/settings.py

➞

Show inline comments

@@ @@ -74,193 +74,193 @@ class SettingsController(BaseController) @@
             k = k.replace('.', '_')
             if each.ui_section in ['hooks', 'extensions']:
                 v = each.ui_active
             settings[k] = v
         return settings
     @HasPermissionAnyDecorator('hg.admin')
     def settings_vcs(self):
         c.active = 'vcs'
         if request.POST:
             application_form = ApplicationUiSettingsForm()()
             try:
                 form_result = application_form.to_python(dict(request.POST))
             except formencode.Invalid as errors:
                 return htmlfill.render(
                      render('admin/settings/settings.html'),
                      defaults=errors.value,
                      errors=errors.error_dict or {},
                      prefix_error=False,
                      encoding="UTF-8",
                      force_defaults=False)
             try:
                 if c.visual.allow_repo_location_change:
                     sett = Ui.get_by_key('paths', '/')
                     sett.ui_value = form_result['paths_root_path']
                 #HOOKS
                 sett = Ui.get_by_key('hooks', Ui.HOOK_UPDATE)
                 sett.ui_active = form_result['hooks_changegroup_update']
                 sett = Ui.get_by_key('hooks', Ui.HOOK_REPO_SIZE)
                 sett.ui_active = form_result['hooks_changegroup_repo_size']
                 sett = Ui.get_by_key('hooks', Ui.HOOK_PUSH)
                 sett.ui_active = form_result['hooks_changegroup_push_logger']
                 sett = Ui.get_by_key('hooks', Ui.HOOK_PULL)
                 sett.ui_active = form_result['hooks_outgoing_pull_logger']
                 ## EXTENSIONS
                 sett = Ui.get_or_create('extensions', 'largefiles')
                 sett.ui_active = form_result['extensions_largefiles']
                 sett = Ui.get_or_create('extensions', 'hgsubversion')
                 sett.ui_active = form_result['extensions_hgsubversion']
                 if sett.ui_active:
                     try:
                         import hgsubversion  # pragma: no cover
                     except ImportError:
                         raise HgsubversionImportError
 #                sett = Ui.get_or_create('extensions', 'hggit')
 #                sett.ui_active = form_result['extensions_hggit']
                 Session().commit()
                 h.flash(_('Updated VCS settings'), category='success')
             except HgsubversionImportError:
                 log.error(traceback.format_exc())
                 h.flash(_('Unable to activate hgsubversion support. '
                           'The "hgsubversion" library is missing'),
                         category='error')
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred while updating '
                           'application settings'), category='error')
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_mapping(self):
         c.active = 'mapping'
         if request.POST:
             rm_obsolete = request.POST.get('destroy', False)
             install_git_hooks = request.POST.get('hooks', False)
             overwrite_git_hooks = request.POST.get('hooks_overwrite', False);
             invalidate_cache = request.POST.get('invalidate', False)
             log.debug('rescanning repo location with destroy obsolete=%s, '
                       'install git hooks=%s and '
                       'overwrite git hooks=%s' % (rm_obsolete, install_git_hooks, overwrite_git_hooks))
             if invalidate_cache:
                 log.debug('invalidating all repositories cache')
-                for repo in Repository.get_all():
+                for repo in Repository.query():
                     ScmModel().mark_for_invalidation(repo.repo_name)
             filesystem_repos = ScmModel().repo_scan()
             added, removed = repo2db_mapper(filesystem_repos, rm_obsolete,
                                             install_git_hooks=install_git_hooks,
                                             user=c.authuser.username,
                                             overwrite_git_hooks=overwrite_git_hooks)
             h.flash(h.literal(_('Repositories successfully rescanned. Added: %s. Removed: %s.') %
                 (', '.join(h.link_to(safe_unicode(repo_name), h.url('summary_home', repo_name=repo_name))
                  for repo_name in added) or '-',
                  ', '.join(h.escape(safe_unicode(repo_name)) for repo_name in removed) or '-')),
                 category='success')
             raise HTTPFound(location=url('admin_settings_mapping'))
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_global(self):
         c.active = 'global'
         if request.POST:
             application_form = ApplicationSettingsForm()()
             try:
                 form_result = application_form.to_python(dict(request.POST))
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('admin/settings/settings.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             try:
                 sett1 = Setting.create_or_update('title',
                                             form_result['title'])
                 Session().add(sett1)
                 sett2 = Setting.create_or_update('realm',
                                             form_result['realm'])
                 Session().add(sett2)
                 sett3 = Setting.create_or_update('ga_code',
                                             form_result['ga_code'])
                 Session().add(sett3)
                 sett4 = Setting.create_or_update('captcha_public_key',
                                     form_result['captcha_public_key'])
                 Session().add(sett4)
                 sett5 = Setting.create_or_update('captcha_private_key',
                                     form_result['captcha_private_key'])
                 Session().add(sett5)
                 Session().commit()
                 set_app_settings(config)
                 h.flash(_('Updated application settings'), category='success')
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred while updating '
                           'application settings'),
                           category='error')
             raise HTTPFound(location=url('admin_settings_global'))
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_visual(self):
         c.active = 'visual'
         if request.POST:
             application_form = ApplicationVisualisationForm()()
             try:
                 form_result = application_form.to_python(dict(request.POST))
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('admin/settings/settings.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)

kallithea/controllers/api/api.py

➞

Show inline comments

@@ @@ -350,193 +350,193 @@ class ApiController(JSONRPCController): @@
+                )
         else:
             raise JSONRPCError('repository `%s` does not exist' % (repoid,))
         if isinstance(userid, Optional):
             userid = self.authuser.user_id
         user = get_user_or_error(userid)
         if isinstance(locked, Optional):
             lockobj = Repository.getlock(repo)
             if lockobj[0] is None:
                 _d = {
                     'repo': repo.repo_name,
                     'locked': False,
                     'locked_since': None,
                     'locked_by': None,
                     'lock_state_changed': False,
                     'msg': 'Repo `%s` not locked.' % repo.repo_name
+                }
                 return _d
             else:
                 userid, time_ = lockobj
                 lock_user = get_user_or_error(userid)
                 _d = {
                     'repo': repo.repo_name,
                     'locked': True,
                     'locked_since': time_,
                     'locked_by': lock_user.username,
                     'lock_state_changed': False,
                     'msg': ('Repo `%s` locked by `%s` on `%s`.'
                             % (repo.repo_name, lock_user.username,
                                json.dumps(time_to_datetime(time_))))
+                }
                 return _d
         # force locked state through a flag
         else:
             locked = str2bool(locked)
             try:
                 if locked:
                     lock_time = time.time()
                     Repository.lock(repo, user.user_id, lock_time)
                 else:
                     lock_time = None
                     Repository.unlock(repo)
                 _d = {
                     'repo': repo.repo_name,
                     'locked': locked,
                     'locked_since': lock_time,
                     'locked_by': user.username,
                     'lock_state_changed': True,
                     'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                             % (user.username, repo.repo_name, locked))
+                }
                 return _d
             except Exception:
                 log.error(traceback.format_exc())
                 raise JSONRPCError(
                     'Error occurred locking repository `%s`' % repo.repo_name
+                )
     def get_locks(self, userid=Optional(OAttr('apiuser'))):
         """
         Get all repositories with locks for given userid, if
         this command is run by non-admin account userid is set to user
         who is calling this method, thus returning locks for himself.
         :param userid: User to get locks for
         :type userid: Optional(str or int)
         OUTPUT::
           id : <id_given_in_input>
           result : {
             [repo_object, repo_object,...]
+          }
           error :  null
         """
         if not HasPermissionAny('hg.admin')():
             # make sure normal user does not pass someone else userid,
             # he is not allowed to do that
             if not isinstance(userid, Optional) and userid != self.authuser.user_id:
                 raise JSONRPCError(
                     'userid is not the same as your user'
+                )
         ret = []
         if isinstance(userid, Optional):
             user = None
         else:
             user = get_user_or_error(userid)
         # show all locks
-        for r in Repository.get_all():
+        for r in Repository.query():
             userid, time_ = r.locked
             if time_:
                 _api_data = r.get_api_data()
                 # if we use userfilter just show the locks for this user
                 if user is not None:
                     if safe_int(userid) == user.user_id:
                         ret.append(_api_data)
                 else:
                     ret.append(_api_data)
         return ret
     @HasPermissionAnyDecorator('hg.admin')
     def get_ip(self, userid=Optional(OAttr('apiuser'))):
         """
         Shows IP address as seen from Kallithea server, together with all
         defined IP addresses for given user. If userid is not passed data is
         returned for user who's calling this function.
         This command can be executed only using api_key belonging to user with
         admin rights.
         :param userid: username to show ips for
         :type userid: Optional(str or int)
         OUTPUT::
             id : <id_given_in_input>
             result : {
                          "server_ip_addr": "<ip_from_clien>",
                          "user_ips": [
+                                        {
                                            "ip_addr": "<ip_with_mask>",
                                            "ip_range": ["<start_ip>", "<end_ip>"],
                                         },
                                         ...
+                                     ]
+            }
         """
         if isinstance(userid, Optional):
             userid = self.authuser.user_id
         user = get_user_or_error(userid)
         ips = UserIpMap.query().filter(UserIpMap.user == user).all()
         return dict(
             server_ip_addr=self.ip_addr,
             user_ips=ips
+        )
     # alias for old
     show_ip = get_ip
     @HasPermissionAnyDecorator('hg.admin')
     def get_server_info(self):
         """
         return server info, including Kallithea version and installed packages
         OUTPUT::
           id : <id_given_in_input>
           result : {
             'modules': [<module name>,...]
             'py_version': <python version>,
             'platform': <platform type>,
             'kallithea_version': <kallithea version>
+          }
           error :  null
         """
         return Setting.get_server_info()
     def get_user(self, userid=Optional(OAttr('apiuser'))):
         """
         Gets a user by username or user_id, Returns empty result if user is
         not found. If userid param is skipped it is set to id of user who is
         calling this method. This command can be executed only using api_key
         belonging to user with admin rights, or regular users that cannot
         specify different userid than theirs
         :param userid: user to get data for
         :type userid: Optional(str or int)
         OUTPUT::
             id : <id_given_in_input>
             result: None if user does not exist or
+                    {
                         "user_id" :     "<user_id>",
                         "api_key" :     "<api_key>",
                         "api_keys":     "[<list of all API keys including additional ones>]"
                         "username" :    "<username>",
                         "firstname":    "<firstname>",
                         "lastname" :    "<lastname>",
                         "email" :       "<email>",
                         "emails":       "[<list of all emails including additional ones>]",
                         "ip_addresses": "[<ip_address_for_user>,...]",
                         "active" :      "<bool: user active>",
@@ @@ -755,193 +755,193 @@ class ApiController(JSONRPCController): @@
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update user `%s`' % (userid,))
     @HasPermissionAnyDecorator('hg.admin')
     def delete_user(self, userid):
         """
         deletes given user if such user exists. This command can
         be executed only using api_key belonging to user with admin rights.
         :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, 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 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 HasPermissionAny('hg.admin')():
             # check if we have at least read permission for this user group !
             _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
             if not HasUserGroupPermissionAny(*_perms)(
                     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):
         """
         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.
         OUTPUT::
             id : <id_given_in_input>
             result : [<user_group_obj>,...]
             error : null
         """
         result = []
         _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
-        for user_group in UserGroupList(UserGroup.get_all(),
+        for user_group in UserGroupList(UserGroup.query().all(),
                                         perm_set=_perms):
             result.append(user_group.get_api_data())
         return result
     @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')
     def create_user_group(self, 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 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 = self.authuser.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, 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 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 :  {
@@ @@ -1179,193 +1179,193 @@ class ApiController(JSONRPCController): @@
                 "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 HasPermissionAny('hg.admin')():
             # check if we have admin permission for this repo !
             perms = ('repository.admin', 'repository.write', 'repository.read')
             if not HasRepoPermissionAny(*perms)(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):
         """
         Lists all existing repositories. This command can be executed only using
         api_key belonging to user with admin rights or regular user that have
         admin, write or read access to repository.
         OUTPUT::
             id : <id_given_in_input>
             result: [
+                      {
                         "repo_id" :          "<repo_id>",
                         "repo_name" :        "<reponame>"
                         "repo_type" :        "<repo_type>",
                         "clone_uri" :        "<clone_uri>",
                         "private": :         "<bool>",
                         "created_on" :       "<datetimecreated>",
                         "description" :      "<description>",
                         "landing_rev":       "<landing_rev>",
                         "owner":             "<repo_owner>",
                         "fork_of":           "<name_of_fork_parent>",
                         "enable_downloads":  "<bool>",
                         "enable_locking":    "<bool>",
                         "enable_statistics": "<bool>",
                       },
                       …
+                    ]
             error:  null
         """
         result = []
         if not HasPermissionAny('hg.admin')():
             repos = RepoModel().get_all_user_repos(user=self.authuser.user_id)
         else:
-            repos = Repository.get_all()
+            repos = Repository.query()
         for repo in repos:
             result.append(repo.get_api_data())
         return result
     # permission check inside
     def get_repo_nodes(self, repoid, revision, root_path,
                        ret_type=Optional('all')):
         """
         returns a list of nodes and it's children in a flat list for a given path
         at given revision. It's possible to specify ret_type to show only `files` or
         `dirs`.  This command can be executed only using api_key belonging to
         user with admin rights or regular user that have at least read access to repository.
         :param repoid: repository name or repository id
         :type repoid: str or int
         :param revision: revision for which listing should be done
         :type revision: str
         :param root_path: path from which start displaying
         :type root_path: str
         :param ret_type: return type 'all|files|dirs' nodes
         :type ret_type: Optional(str)
         OUTPUT::
             id : <id_given_in_input>
             result: [
+                      {
                         "name" :        "<name>"
                         "type" :        "<type>",
                       },
                       …
+                    ]
             error:  null
         """
         repo = get_repo_or_error(repoid)
         if not HasPermissionAny('hg.admin')():
             # check if we have admin permission for this repo !
             perms = ('repository.admin', 'repository.write', 'repository.read')
             if not HasRepoPermissionAny(*perms)(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, repo_name, owner=Optional(OAttr('apiuser')),
                     repo_type=Optional('hg'), description=Optional(''),
                     private=Optional(False), clone_uri=Optional(None),
                     landing_rev=Optional('rev:tip'),
                     enable_statistics=Optional(False),
                     enable_locking=Optional(False),
                     enable_downloads=Optional(False),
                     copy_permissions=Optional(False)):
         """
         Creates a repository. If repository name contains "/", all needed repository
         groups will be created. For example "foo/bar/baz" will create groups
         "foo", "bar" (with "foo" as parent), and create "baz" repository with
         "bar" as group. This command can be executed only using api_key
         belonging to user with admin rights or regular user that have create
         repository permission. Regular users cannot specify owner parameter
         :param repo_name: repository name
         :type repo_name: str
         :param owner: user_id or username
         :type owner: Optional(str)
         :param repo_type: 'hg' or 'git'
         :type repo_type: Optional(str)
         :param description: repository description
         :type description: Optional(str)
         :param private:
         :type private: bool
         :param clone_uri:
         :type clone_uri: str
         :param landing_rev: <rev_type>:<rev>
         :type landing_rev: str
         :param enable_locking:
@@ @@ -1857,193 +1857,193 @@ class ApiController(JSONRPCController): @@
     def revoke_user_group_permission(self, repoid, usergroupid):
         """
         Revoke permission for user group on given repository. This command can be
         executed only using api_key belonging to user with admin rights.
         :param repoid: repository name or repository id
         :type repoid: str or int
         :param usergroupid:
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg" : "Revoked perm for group: `<usersgroupname>` in repo: `<reponame>`",
                       "success": true
+                    }
             error:  null
         """
         repo = get_repo_or_error(repoid)
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAny('hg.admin')():
             # check if we have admin permission for this repo !
             _perms = ('repository.admin',)
             if not HasRepoPermissionAny(*_perms)(
                     repo_name=repo.repo_name):
                 raise JSONRPCError('repository `%s` does not exist' % (repoid,))
             # check if we have at least read permission for this user group !
             _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
             if not HasUserGroupPermissionAny(*_perms)(
                     user_group_name=user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
         try:
             RepoModel().revoke_user_group_permission(
                 repo=repo, group_name=user_group)
             Session().commit()
             return dict(
                 msg='Revoked perm for user group: `%s` in repo: `%s`' % (
                     user_group.users_group_name, repo.repo_name
                 ),
                 success=True
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for user group: `%s` in '
                 'repo: `%s`' % (
                     user_group.users_group_name, repo.repo_name
+                )
+            )
     @HasPermissionAnyDecorator('hg.admin')
     def get_repo_group(self, repogroupid):
         """
         Returns given repo group together with permissions, and repositories
         inside the group
         :param repogroupid: id/name of repository group
         :type repogroupid: str or int
         """
         repo_group = get_repo_group_or_error(repogroupid)
         members = []
         for user in repo_group.repo_group_to_perm:
             perm = user.permission.permission_name
             user = user.user
             user_data = {
                 'name': user.username,
                 'type': "user",
                 'permission': perm
+            }
             members.append(user_data)
         for user_group in repo_group.users_group_to_perm:
             perm = user_group.permission.permission_name
             user_group = user_group.users_group
             user_group_data = {
                 'name': user_group.users_group_name,
                 'type': "user_group",
                 'permission': perm
+            }
             members.append(user_group_data)
         data = repo_group.get_api_data()
         data["members"] = members
         return data
     @HasPermissionAnyDecorator('hg.admin')
     def get_repo_groups(self):
         """
         Returns all repository groups
         """
         result = []
-        for repo_group in RepoGroup.get_all():
+        for repo_group in RepoGroup.query():
             result.append(repo_group.get_api_data())
         return result
     @HasPermissionAnyDecorator('hg.admin')
     def create_repo_group(self, group_name, description=Optional(''),
                           owner=Optional(OAttr('apiuser')),
                           parent=Optional(None),
                           copy_permissions=Optional(False)):
         """
         Creates a repository group. This command can be executed only using
         api_key belonging to user with admin rights.
         :param group_name:
         :type group_name:
         :param description:
         :type description:
         :param owner:
         :type owner:
         :param parent:
         :type parent:
         :param copy_permissions:
         :type copy_permissions:
         OUTPUT::
           id : <id_given_in_input>
           result : {
               "msg": "created new repo group `<repo_group_name>`"
               "repo_group": <repogroup_object>
+          }
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             failed to create repo group `<repogroupid>`
+          }
         """
         if RepoGroup.get_by_group_name(group_name):
             raise JSONRPCError("repo group `%s` already exist" % (group_name,))
         if isinstance(owner, Optional):
             owner = self.authuser.user_id
         group_description = Optional.extract(description)
         parent_group = Optional.extract(parent)
         if not isinstance(parent, Optional):
             parent_group = get_repo_group_or_error(parent_group)
         copy_permissions = Optional.extract(copy_permissions)
         try:
             repo_group = RepoGroupModel().create(
                 group_name=group_name,
                 group_description=group_description,
                 owner=owner,
                 parent=parent_group,
                 copy_permissions=copy_permissions
+            )
             Session().commit()
             return dict(
                 msg='created new repo group `%s`' % group_name,
                 repo_group=repo_group.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create repo group `%s`' % (group_name,))
     @HasPermissionAnyDecorator('hg.admin')
     def update_repo_group(self, repogroupid, group_name=Optional(''),
                           description=Optional(''),
                           owner=Optional(OAttr('apiuser')),
                           parent=Optional(None), enable_locking=Optional(False)):
         repo_group = get_repo_group_or_error(repogroupid)
         updates = {}
         try:
             store_update(updates, group_name, 'group_name')
             store_update(updates, description, 'group_description')
             store_update(updates, owner, 'owner')
             store_update(updates, parent, 'parent_group')
             store_update(updates, enable_locking, 'enable_locking')
             repo_group = RepoGroupModel().update(repo_group, updates)
             Session().commit()
             return dict(
                 msg='updated repository group ID:%s %s' % (repo_group.group_id,
                                                            repo_group.group_name),
                 repo_group=repo_group.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update repository group `%s`'
                                % (repogroupid,))

kallithea/lib/paster_commands/update_repoinfo.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.lib.paster_commands.update_repoinfo
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 update-repoinfo paster command 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: Jul 14, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import sys
 import string
 from kallithea.lib.paster_commands.common import BasePasterCommand
 from kallithea.lib.utils2 import safe_unicode
 from kallithea.model.db import Repository
 from kallithea.model.repo import RepoModel
 from kallithea.model.meta import Session
 class Command(BasePasterCommand):
     max_args = 1
     min_args = 1
     usage = "CONFIG_FILE"
     group_name = "Kallithea"
     takes_config_file = -1
     parser = BasePasterCommand.standard_parser(verbose=True)
     summary = "Updates repositories caches for last changeset"
     def command(self):
         #get SqlAlchemy session
         self._init_session()
         if self.options.repo_update_list is None:
-            repo_list = Repository.get_all()
+            repo_list = Repository.query().all()
         else:
             repo_names = [safe_unicode(n.strip())
                           for n in self.options.repo_update_list.split(',')]
             repo_list = list(Repository.query()
                 .filter(Repository.repo_name.in_(repo_names)))
         RepoModel.update_repoinfo(repositories=repo_list)
         Session().commit()
         if self.options.invalidate_cache:
             for r in repo_list:
                 r.set_invalidate()
             print 'Updated repo info and invalidated cache for %s repositories' % (len(repo_list))
         else:
             print 'Updated repo info for %s repositories' % (len(repo_list))
     def update_parser(self):
         self.parser.add_option('--update-only',
                            action='store',
                            dest='repo_update_list',
                            help="Specifies a comma separated list of repositories "
                                 "to update last commit info for. OPTIONAL")
         self.parser.add_option('--invalidate-cache',
                            action='store_true',
                            dest='invalidate_cache',
                            help="Trigger cache invalidation event for repos. "
                                 "OPTIONAL")

kallithea/model/db.py

➞

Show inline comments

@@ @@ -67,196 +67,192 @@ log = logging.getLogger(__name__) @@
 _hash_key = lambda k: hashlib.md5(safe_str(k)).hexdigest()
 class BaseModel(object):
     """
     Base Model for all classes
     """
     @classmethod
     def _get_keys(cls):
         """return column names for this model """
         return class_mapper(cls).c.keys()
     def get_dict(self):
         """
         return dict with keys and values corresponding
         to this model data """
         d = {}
         for k in self._get_keys():
             d[k] = getattr(self, k)
         # also use __json__() if present to get additional fields
         _json_attr = getattr(self, '__json__', None)
         if _json_attr:
             # update with attributes from __json__
             if callable(_json_attr):
                 _json_attr = _json_attr()
             for k, val in _json_attr.iteritems():
                 d[k] = val
         return d
     def get_appstruct(self):
         """return list with keys and values tuples corresponding
         to this model data """
         l = []
         for k in self._get_keys():
             l.append((k, getattr(self, k),))
         return l
     def populate_obj(self, populate_dict):
         """populate model with data from given populate_dict"""
         for k in self._get_keys():
             if k in populate_dict:
                 setattr(self, k, populate_dict[k])
     @classmethod
     def query(cls):
         return Session().query(cls)
     @classmethod
     def get(cls, id_):
         if id_:
             return cls.query().get(id_)
     @classmethod
     def guess_instance(cls, value, callback=None):
         """Haphazardly attempt to convert `value` to a `cls` instance.
         If `value` is None or already a `cls` instance, return it. If `value`
         is a number (or looks like one if you squint just right), assume it's
         a database primary key and let SQLAlchemy sort things out. Otherwise,
         fall back to resolving it using `callback` (if specified); this could
         e.g. be a function that looks up instances by name (though that won't
         work if the name begins with a digit). Otherwise, raise Exception.
         """
         if value is None:
             return None
         if isinstance(value, cls):
             return value
         if isinstance(value, (int, long)) or safe_str(value).isdigit():
             return cls.get(value)
         if callback is not None:
             return callback(value)
         raise Exception(
             'given object must be int, long or Instance of %s '
             'got %s, no callback provided' % (cls, type(value))
+        )
     @classmethod
     def get_or_404(cls, id_):
         try:
             id_ = int(id_)
         except (TypeError, ValueError):
             raise HTTPNotFound
         res = cls.query().get(id_)
         if res is None:
             raise HTTPNotFound
         return res
     @classmethod
     def get_all(cls):
         return cls.query().all()
     @classmethod
     def delete(cls, id_):
         obj = cls.query().get(id_)
         Session().delete(obj)
     def __repr__(self):
         if hasattr(self, '__unicode__'):
             # python repr needs to return str
             try:
                 return safe_str(self.__unicode__())
             except UnicodeDecodeError:
                 pass
         return '<DB:%s>' % (self.__class__.__name__)
 _table_args_default_dict = {'extend_existing': True,
                             'mysql_engine': 'InnoDB',
                             'mysql_charset': 'utf8',
                             'sqlite_autoincrement': True,
+                           }
 class Setting(Base, BaseModel):
     __tablename__ = 'settings'
     __table_args__ = (
         _table_args_default_dict,
+    )
     SETTINGS_TYPES = {
         'str': safe_str,
         'int': safe_int,
         'unicode': safe_unicode,
         'bool': str2bool,
         'list': functools.partial(aslist, sep=',')
+    }
     DEFAULT_UPDATE_URL = ''
     app_settings_id = Column(Integer(), primary_key=True)
     app_settings_name = Column(String(255), nullable=False, unique=True)
     _app_settings_value = Column("app_settings_value", Unicode(4096), nullable=False)
     _app_settings_type = Column("app_settings_type", String(255), nullable=True) # FIXME: not nullable?
     def __init__(self, key='', val='', type='unicode'):
         self.app_settings_name = key
         self.app_settings_value = val
         self.app_settings_type = type
     @validates('_app_settings_value')
     def validate_settings_value(self, key, val):
         assert type(val) == unicode
         return val
     @hybrid_property
     def app_settings_value(self):
         v = self._app_settings_value
         _type = self.app_settings_type
         converter = self.SETTINGS_TYPES.get(_type) or self.SETTINGS_TYPES['unicode']
         return converter(v)
     @app_settings_value.setter
     def app_settings_value(self, val):
         """
         Setter that will always make sure we use unicode in app_settings_value
         :param val:
         """
         self._app_settings_value = safe_unicode(val)
     @hybrid_property
     def app_settings_type(self):
         return self._app_settings_type
     @app_settings_type.setter
     def app_settings_type(self, val):
         if val not in self.SETTINGS_TYPES:
             raise Exception('type must be one of %s got %s'
                             % (self.SETTINGS_TYPES.keys(), val))
         self._app_settings_type = val
     def __unicode__(self):
         return u"<%s('%s:%s[%s]')>" % (
             self.__class__.__name__,
             self.app_settings_name, self.app_settings_value, self.app_settings_type
+        )
     @classmethod
     def get_by_name(cls, key):
         return cls.query() \
             .filter(cls.app_settings_name == key).scalar()
     @classmethod
     def get_by_name_or_create(cls, key, val='', type='unicode'):
         res = cls.get_by_name(key)
         if res is None:
             res = cls(key, val, type)
         return res
     @classmethod

kallithea/model/repo.py

➞

Show inline comments

@@ @@ -86,193 +86,193 @@ class RepoModel(BaseModel): @@
         return repo_to_perm
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = self.sa.query(Ui).filter(Ui.ui_key == '/').one()
         return q.ui_value
     def get(self, repo_id, cache=False):
         repo = self.sa.query(Repository) \
             .filter(Repository.repo_id == repo_id)
         if cache:
             repo = repo.options(FromCache("sql_cache_short",
                                           "get_repo_%s" % repo_id))
         return repo.scalar()
     def get_repo(self, repository):
         return self._get_repo(repository)
     def get_by_repo_name(self, repo_name, cache=False):
         repo = self.sa.query(Repository) \
             .filter(Repository.repo_name == repo_name)
         if cache:
             repo = repo.options(FromCache("sql_cache_short",
                                           "get_repo_%s" % repo_name))
         return repo.scalar()
     def get_all_user_repos(self, user):
         """
         Gets all repositories that user have at least read access
         :param user:
         """
         from kallithea.lib.auth import AuthUser
         user = self._get_user(user)
         repos = AuthUser(user_id=user.user_id).permissions['repositories']
         access_check = lambda r: r[1] in ['repository.read',
                                           'repository.write',
                                           'repository.admin']
         repos = [x[0] for x in filter(access_check, repos.items())]
         return Repository.query().filter(Repository.repo_name.in_(repos))
     def get_users_js(self):
         users = self.sa.query(User) \
             .filter(User.active == True) \
             .order_by(User.name, User.lastname) \
             .all()
         return json.dumps([
+            {
                 'id': u.user_id,
                 'fname': h.escape(u.name),
                 'lname': h.escape(u.lastname),
                 'nname': u.username,
                 'gravatar_lnk': h.gravatar_url(u.email, size=28, default='default'),
                 'gravatar_size': 14,
             } for u in users]
+        )
     def get_user_groups_js(self):
         user_groups = self.sa.query(UserGroup) \
             .filter(UserGroup.users_group_active == True) \
             .order_by(UserGroup.users_group_name) \
             .options(subqueryload(UserGroup.members)) \
             .all()
         user_groups = UserGroupList(user_groups, perm_set=['usergroup.read',
                                                            'usergroup.write',
                                                            'usergroup.admin'])
         return json.dumps([
+            {
                 'id': gr.users_group_id,
                 'grname': gr.users_group_name,
                 'grmembers': len(gr.members),
             } for gr in user_groups]
+        )
     @classmethod
     def _render_datatable(cls, tmpl, *args, **kwargs):
         import kallithea
         from pylons import tmpl_context as c
         from pylons.i18n.translation import _
         _tmpl_lookup = kallithea.CONFIG['pylons.app_globals'].mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         tmpl = template.get_def(tmpl)
         kwargs.update(dict(_=_, h=h, c=c))
         return tmpl.render(*args, **kwargs)
     @classmethod
     def update_repoinfo(cls, repositories=None):
         if repositories is None:
-            repositories = Repository.get_all()
+            repositories = Repository.query()
         for repo in repositories:
             repo.update_changeset_cache()
     def get_repos_as_dict(self, repos_list=None, admin=False, perm_check=True,
                           super_user_actions=False, short_name=False):
         _render = self._render_datatable
         from pylons import tmpl_context as c
         def quick_menu(repo_name):
             return _render('quick_menu', repo_name)
         def repo_lnk(name, rtype, rstate, private, fork_of):
             return _render('repo_name', name, rtype, rstate, private, fork_of,
                            short_name=short_name, admin=False)
         def last_change(last_change):
             return _render("last_change", last_change)
         def rss_lnk(repo_name):
             return _render("rss", repo_name)
         def atom_lnk(repo_name):
             return _render("atom", repo_name)
         def last_rev(repo_name, cs_cache):
             return _render('revision', repo_name, cs_cache.get('revision'),
                            cs_cache.get('raw_id'), cs_cache.get('author'),
                            cs_cache.get('message'))
         def desc(desc):
             return h.urlify_text(desc, truncate=80, stylize=c.visual.stylify_metatags)
         def state(repo_state):
             return _render("repo_state", repo_state)
         def repo_actions(repo_name):
             return _render('repo_actions', repo_name, super_user_actions)
         def owner_actions(user_id, username):
             return _render('user_name', user_id, username)
         repos_data = []
         for repo in repos_list:
             if perm_check:
                 # check permission at this level
                 if not HasRepoPermissionAny(
                         'repository.read', 'repository.write',
                         'repository.admin'
                 )(repo.repo_name, 'get_repos_as_dict check'):
                     continue
             cs_cache = repo.changeset_cache
             row = {
                 "menu": quick_menu(repo.repo_name),
                 "raw_name": repo.repo_name,
                 "name": repo_lnk(repo.repo_name, repo.repo_type,
                                  repo.repo_state, repo.private, repo.fork),
                 "last_change_iso": repo.last_db_change.isoformat(),
                 "last_change": last_change(repo.last_db_change),
                 "last_changeset": last_rev(repo.repo_name, cs_cache),
                 "last_rev_raw": cs_cache.get('revision'),
                 "desc": desc(repo.description),
                 "owner": h.person(repo.user),
                 "state": state(repo.repo_state),
                 "rss": rss_lnk(repo.repo_name),
                 "atom": atom_lnk(repo.repo_name),
+            }
             if admin:
                 row.update({
                     "action": repo_actions(repo.repo_name),
                     "owner": owner_actions(repo.user.user_id,
                                            h.person(repo.user))
                 })
             repos_data.append(row)
         return {
             "totalRecords": len(repos_list),
             "startIndex": 0,
             "sort": "name",
             "dir": "asc",
             "records": repos_data
+        }
     def _get_defaults(self, repo_name):
         """
         Gets information about repository, and returns a dict for
         usage in forms
         :param repo_name:
         """
         repo_info = Repository.get_by_repo_name(repo_name)
         if repo_info is None:
             return None

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 Repository.get_all():
+        for repo in Repository.query():
             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

@@ @@ -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': Gist.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 Gist.get_all():
+        for g in Gist.query():
             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/functional/test_admin_gists.py

➞

Show inline comments

 from kallithea.tests.base import *
 from kallithea.model.gist import GistModel
 from kallithea.model.meta import Session
 from kallithea.model.db import User, Gist
 def _create_gist(f_name, content='some gist', lifetime=-1,
                  description=u'gist-desc', gist_type='public',
                  owner=TEST_USER_ADMIN_LOGIN):
     gist_mapping = {
         f_name: {'content': content}
+    }
     user = User.get_by_username(owner)
     gist = GistModel().create(description, owner=user,
                        gist_mapping=gist_mapping, gist_type=gist_type,
                        lifetime=lifetime)
     Session().commit()
     return gist
 class TestGistsController(TestController):
     def teardown_method(self, method):
-        for g in Gist.get_all():
+        for g in Gist.query():
             GistModel().delete(g)
         Session().commit()
     def test_index(self):
         self.log_user()
         response = self.app.get(url('gists'))
         # Test response...
         response.mustcontain('There are no gists yet')
         g1 = _create_gist('gist1').gist_access_id
         g2 = _create_gist('gist2', lifetime=1400).gist_access_id
         g3 = _create_gist('gist3', description=u'gist3-desc').gist_access_id
         g4 = _create_gist('gist4', gist_type='private').gist_access_id
         response = self.app.get(url('gists'))
         # Test response...
         response.mustcontain('gist: %s' % g1)
         response.mustcontain('gist: %s' % g2)
         response.mustcontain('Expires: in 23 hours')  # we don't care about the end
         response.mustcontain('gist: %s' % g3)
         response.mustcontain('gist3-desc')
         response.mustcontain(no=['gist: %s' % g4])
     def test_index_private_gists(self):
         self.log_user()
         gist = _create_gist('gist5', gist_type='private')
         response = self.app.get(url('gists', private=1))
         # Test response...
         #and privates
         response.mustcontain('gist: %s' % gist.gist_access_id)
     def test_create_missing_description(self):
         self.log_user()
         response = self.app.post(url('gists'),
                                  params={'lifetime': -1, '_authentication_token': self.authentication_token()},
                                  status=200)
         response.mustcontain('Missing value')
     def test_create(self):
         self.log_user()
         response = self.app.post(url('gists'),
                                  params={'lifetime': -1,
                                          'content': 'gist test',
                                          'filename': 'foo',
                                          'public': 'public',
                                          '_authentication_token': self.authentication_token()},
                                  status=302)
         response = response.follow()
         response.mustcontain('added file: foo')
         response.mustcontain('gist test')
         response.mustcontain('<div class="btn btn-mini btn-success disabled">Public Gist</div>')
     def test_create_with_path_with_dirs(self):
         self.log_user()
         response = self.app.post(url('gists'),
                                  params={'lifetime': -1,
                                          'content': 'gist test',
                                          'filename': '/home/foo',
                                          'public': 'public',
                                          '_authentication_token': self.authentication_token()},
                                  status=200)
         response.mustcontain('Filename cannot be inside a directory')
     def test_access_expired_gist(self):
         self.log_user()
         gist = _create_gist('never-see-me')
         gist.gist_expires = 0  # 1970
         Session().add(gist)
         Session().commit()
         response = self.app.get(url('gist', gist_id=gist.gist_access_id), status=404)
     def test_create_private(self):
         self.log_user()
         response = self.app.post(url('gists'),
                                  params={'lifetime': -1,
                                          'content': 'private gist test',
                                          'filename': 'private-foo',
                                          'private': 'private',
                                          '_authentication_token': self.authentication_token()},
                                  status=302)
         response = response.follow()
         response.mustcontain('added file: private-foo<')
         response.mustcontain('private gist test')
         response.mustcontain('<div class="btn btn-mini btn-warning disabled">Private Gist</div>')
     def test_create_with_description(self):
         self.log_user()
         response = self.app.post(url('gists'),
                                  params={'lifetime': -1,
                                          'content': 'gist test',
                                          'filename': 'foo-desc',
                                          'description': 'gist-desc',
                                          'public': 'public',
                                          '_authentication_token': self.authentication_token()},

kallithea/tests/functional/test_home.py

➞

Show inline comments

 from kallithea.tests.base import *
 from kallithea.tests.fixture import Fixture
 from kallithea.model.meta import Session
 from kallithea.model.db import Repository
 from kallithea.model.repo import RepoModel
 from kallithea.model.repo_group import RepoGroupModel
 fixture = Fixture()
 class TestHomeController(TestController):
     def test_index(self):
         self.log_user()
         response = self.app.get(url(controller='home', action='index'))
         #if global permission is set
         response.mustcontain('Add Repository')
         # html in javascript variable:
-        response.mustcontain('var data = {"totalRecords": %s' % len(Repository.get_all()))
+        response.mustcontain('var data = {"totalRecords": %s' % Repository.query().count())
         response.mustcontain(r'href=\"/%s\"' % HG_REPO)
         response.mustcontain(r'<span class="repotag">git')
         response.mustcontain(r'<i class=\"icon-globe\"')
         response.mustcontain("""fixes issue with having custom format for git-log""")
         response.mustcontain("""/%s/changeset/5f2c6ee195929b0be80749243c18121c9864a3b3""" % GIT_REPO)
         response.mustcontain("""disable security checks on hg clone for travis""")
         response.mustcontain("""/%s/changeset/96507bd11ecc815ebc6270fdf6db110928c09c1e""" % HG_REPO)
     def test_repo_summary_with_anonymous_access_disabled(self):
         with fixture.anon_access(False):
             response = self.app.get(url(controller='summary',
                                         action='index', repo_name=HG_REPO),
                                         status=302)
             assert 'login' in response.location
     def test_index_with_anonymous_access_disabled(self):
         with fixture.anon_access(False):
             response = self.app.get(url(controller='home', action='index'),
                                     status=302)
             assert 'login' in response.location
     def test_index_page_on_groups(self):
         self.log_user()
         gr = fixture.create_repo_group(u'gr1')
         fixture.create_repo(name=u'gr1/repo_in_group', repo_group=gr)
         response = self.app.get(url('repos_group_home', group_name=u'gr1'))
         try:
             response.mustcontain(u"gr1/repo_in_group")
         finally:
             RepoModel().delete(u'gr1/repo_in_group')
             RepoGroupModel().delete(repo_group=u'gr1', force_delete=True)
             Session().commit()

kallithea/tests/models/test_user_groups.py

➞

Show inline comments

 from kallithea.model.db import User, UserGroup
 from kallithea.tests.base 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 UserGroup.get_all():
+        for gr in UserGroup.query():
             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 UserGroup.get_all():
+        for gr in UserGroup.query():
             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]

kallithea/tests/other/manual_test_vcs_operations.py

➞

Show inline comments

@@ @@ -179,193 +179,193 @@ def _check_proper_git_push(stdout, stder @@
     assert 'Pushing to' in stderr
     assert 'master -> master' in stderr
 class TestVCSOperations(TestController):
     @classmethod
     def setup_class(cls):
         #DISABLE ANONYMOUS ACCESS
         set_anonymous_access(False)
     def setup_method(self, method):
         r = Repository.get_by_repo_name(GIT_REPO)
         Repository.unlock(r)
         r.enable_locking = False
         Session().add(r)
         Session().commit()
         r = Repository.get_by_repo_name(HG_REPO)
         Repository.unlock(r)
         r.enable_locking = False
         Session().add(r)
         Session().commit()
     def test_clone_hg_repo_by_admin(self):
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         assert 'requesting all changes' in stdout
         assert 'adding changesets' in stdout
         assert 'adding manifests' in stdout
         assert 'adding file changes' in stdout
         assert stderr == ''
     def test_clone_git_repo_by_admin(self):
         clone_url = _construct_url(GIT_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         assert 'Cloning into' in stdout + stderr
         assert stderr == '' or stdout == ''
     def test_clone_wrong_credentials_hg(self):
         clone_url = _construct_url(HG_REPO, passwd='bad!')
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         assert 'abort: authorization failed' in stderr
     def test_clone_wrong_credentials_git(self):
         clone_url = _construct_url(GIT_REPO, passwd='bad!')
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         assert 'fatal: Authentication failed' in stderr
     def test_clone_git_dir_as_hg(self):
         clone_url = _construct_url(GIT_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         assert 'HTTP Error 404: Not Found' in stderr
     def test_clone_hg_repo_as_git(self):
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         assert 'not found' in stderr
     def test_clone_non_existing_path_hg(self):
         clone_url = _construct_url('trololo')
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         assert 'HTTP Error 404: Not Found' in stderr
     def test_clone_non_existing_path_git(self):
         clone_url = _construct_url('trololo')
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         assert 'not found' in stderr
     def test_push_new_file_hg(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())
         fixture.create_fork(HG_REPO, fork_name)
         clone_url = _construct_url(fork_name).split()[0]
         stdout, stderr = _add_files_and_push('hg', DEST, clone_url=clone_url)
         assert 'pushing to' in stdout
         assert 'Repository size' in stdout
         assert 'Last revision is now' in stdout
     def test_push_new_file_git(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         # commit some stuff into this repo
         fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())
         fixture.create_fork(GIT_REPO, fork_name)
         clone_url = _construct_url(fork_name).split()[0]
         stdout, stderr = _add_files_and_push('git', DEST, clone_url=clone_url)
-        print [(x.repo_full_path,x.repo_path) for x in Repository.get_all()] # TODO: what is this for
+        print [(x.repo_full_path,x.repo_path) for x in Repository.query()] # TODO: what is this for
         _check_proper_git_push(stdout, stderr)
     def test_push_invalidates_cache_hg(self):
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==HG_REPO).scalar()
         if not key:
             key = CacheInvalidation(HG_REPO, HG_REPO)
         key.cache_active = True
         Session().add(key)
         Session().commit()
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())
         fixture.create_fork(HG_REPO, fork_name)
         clone_url = _construct_url(fork_name).split()[0]
         stdout, stderr = _add_files_and_push('hg', DEST, files_no=1, clone_url=clone_url)
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==fork_name).all()
         assert key == []
     def test_push_invalidates_cache_git(self):
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==GIT_REPO).scalar()
         if not key:
             key = CacheInvalidation(GIT_REPO, GIT_REPO)
         key.cache_active = True
         Session().add(key)
         Session().commit()
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         # commit some stuff into this repo
         fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())
         fixture.create_fork(GIT_REPO, fork_name)
         clone_url = _construct_url(fork_name).split()[0]
         stdout, stderr = _add_files_and_push('git', DEST, files_no=1, clone_url=clone_url)
         _check_proper_git_push(stdout, stderr)
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                ==fork_name).all()
         assert key == []
     def test_push_wrong_credentials_hg(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         stdout, stderr = _add_files_and_push('hg', DEST, user='bad',
                                              passwd='name')
         assert 'abort: authorization failed' in stderr
     def test_push_wrong_credentials_git(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         stdout, stderr = _add_files_and_push('git', DEST, user='bad',
                                              passwd='name')
         assert 'fatal: Authentication failed' in stderr
     def test_push_back_to_wrong_url_hg(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(HG_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         stdout, stderr = _add_files_and_push('hg', DEST,
                                     clone_url='http://%s/tmp' % HOST)
         assert 'HTTP Error 404: Not Found' in stderr
     def test_push_back_to_wrong_url_git(self):
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         stdout, stderr = _add_files_and_push('git', DEST,
                                     clone_url='http://%s/tmp' % HOST)
         assert 'not found' in stderr
     def test_clone_and_create_lock_hg(self):
         # enable locking
         r = Repository.get_by_repo_name(HG_REPO)
         r.enable_locking = True
         Session().add(r)
         Session().commit()
@@ @@ -434,136 +434,136 @@ class TestVCSOperations(TestController): @@
                 % (HG_REPO, TEST_USER_ADMIN_LOGIN))
         assert msg in stderr
     def test_push_on_locked_repo_by_other_user_git(self):
         #clone some temp
         DEST = _get_tmp_dir()
         clone_url = _construct_url(GIT_REPO, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         #lock repo
         r = Repository.get_by_repo_name(GIT_REPO)
         # let this user actually push !
         RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,
                                           perm='repository.write')
         Session().commit()
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         #push fails repo is locked by other user !
         stdout, stderr = _add_files_and_push('git', DEST,
                                              user=TEST_USER_REGULAR_LOGIN,
                                              passwd=TEST_USER_REGULAR_PASS)
         err = 'Repository `%s` locked by user `%s`' % (GIT_REPO, TEST_USER_ADMIN_LOGIN)
         assert err in stderr
         #TODO: fix this somehow later on Git, Git is stupid and even if we throw
         #back 423 to it, it makes ANOTHER request and we fail there with 405 :/
         msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""
                 % (GIT_REPO, TEST_USER_ADMIN_LOGIN))
         #msg = "405 Method Not Allowed"
         #assert msg in stderr
     def test_push_unlocks_repository_hg(self):
         # enable locking
         fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())
         fixture.create_fork(HG_REPO, fork_name)
         r = Repository.get_by_repo_name(fork_name)
         r.enable_locking = True
         Session().add(r)
         Session().commit()
         #clone some temp
         DEST = _get_tmp_dir()
         clone_url = _construct_url(fork_name, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         #check for lock repo after clone
         r = Repository.get_by_repo_name(fork_name)
         uid = User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
         assert r.locked[0] == uid
         #push is ok and repo is now unlocked
         stdout, stderr = _add_files_and_push('hg', DEST, clone_url=clone_url.split()[0])
         assert str('remote: Released lock on repo `%s`' % fork_name) in stdout
         #we need to cleanup the Session Here !
         Session.remove()
         r = Repository.get_by_repo_name(fork_name)
         assert r.locked == [None, None]
     #TODO: fix me ! somehow during tests hooks don't get called on Git
     def test_push_unlocks_repository_git(self):
         # enable locking
         fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())
         fixture.create_fork(GIT_REPO, fork_name)
         r = Repository.get_by_repo_name(fork_name)
         r.enable_locking = True
         Session().add(r)
         Session().commit()
         #clone some temp
         DEST = _get_tmp_dir()
         clone_url = _construct_url(fork_name, dest=DEST)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         #check for lock repo after clone
         r = Repository.get_by_repo_name(fork_name)
         assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
         #push is ok and repo is now unlocked
         stdout, stderr = _add_files_and_push('git', DEST, clone_url=clone_url.split()[0])
         _check_proper_git_push(stdout, stderr)
         assert ('remote: Released lock on repo `%s`' % fork_name) in stderr
         #we need to cleanup the Session Here !
         Session.remove()
         r = Repository.get_by_repo_name(fork_name)
         assert r.locked == [None, None]
     def test_ip_restriction_hg(self):
         user_model = UserModel()
         try:
             user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             clone_url = _construct_url(HG_REPO)
             stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
             assert 'abort: HTTP Error 403: Forbidden' in stderr
         finally:
             #release IP restrictions
-            for ip in UserIpMap.get_all():
+            for ip in UserIpMap.query():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
         # IP permissions are cached, need to invalidate this cache explicitly
         invalidate_all_caches()
         clone_url = _construct_url(HG_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
         assert 'requesting all changes' in stdout
         assert 'adding changesets' in stdout
         assert 'adding manifests' in stdout
         assert 'adding file changes' in stdout
         assert stderr == ''
     def test_ip_restriction_git(self):
         user_model = UserModel()
         try:
             user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             clone_url = _construct_url(GIT_REPO)
             stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
             # The message apparently changed in Git 1.8.3, so match it loosely.
             assert re.search(r'\b403\b', stderr)
         finally:
             #release IP restrictions
-            for ip in UserIpMap.get_all():
+            for ip in UserIpMap.query():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
         # IP permissions are cached, need to invalidate this cache explicitly
         invalidate_all_caches()
         clone_url = _construct_url(GIT_REPO)
         stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
         assert 'Cloning into' in stdout + stderr
         assert stderr == '' or stdout == ''

0 comments (0 inline, 0 general)