kallithea Changeset - 28571535dd61

Changeset - 28571535dd61

Parent rev.

Child rev.

[Not reviewed]

beta

0 5 0

Marcin Kuzminski - 13 years ago 2013-01-14 00:38:24
marcin@python-works.com

API methods create_repo, fork_repo, delete_repo, get_repo, get_repos
can be executed by non-admin users ref #539

5 files changed with 352 insertions and 48 deletions:

docs/api/api.rst

rhodecode/controllers/api/__init__.py

rhodecode/controllers/api/api.py

rhodecode/model/repo.py

rhodecode/tests/api/api_base.py

249

0 comments (0 inline, 0 general)

docs/api/api.rst

➞

Show inline comments

@@ @@ -125,193 +125,193 @@ OUTPUT:: @@
     id : <id_given_in_input>
     result : "Pulled from `<reponame>`"
     error :  null
 rescan_repos
 ------------
 Dispatch rescan repositories action. If remove_obsolete is set
 RhodeCode will delete repos that are in database but not in the filesystem.
 This command can be executed only using api_key belonging to user with admin
 rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "rescan_repos"
     args :    {
                 "remove_obsolete" : "<boolean = Optional(False)>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result : "{'added': [<list of names of added repos>],
                'removed': [<list of names of removed repos>]}"
     error :  null
 lock
 ----
 Set locking state on given repository by given user. If userid param is skipped
 , then it is set to id of user whos calling this method.
 This command can be executed only using api_key belonging to user with admin
 rights or regular user that have admin or write access to repository.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "lock"
     args :    {
                 "repoid" : "<reponame or repo_id>"
                 "userid" : "<user_id or username = Optional(=apiuser)>",
                 "locked" : "<bool true|false>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result : "User `<username>` set lock state for repo `<reponame>` to `true|false`"
     error :  null
 show_ip
 -------
 Shows IP address as seen from RhodeCode server, together with all
 defined IP addresses for given user.
 This command can be executed only using api_key belonging to user with admin
 rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "show_ip"
     args :    {
                 "userid" : "<user_id or username>",
+              }
 OUTPUT::
     id : <id_given_in_input>
     result : {
                  "ip_addr_server": <ip_from_clien>",
                  "user_ips": [
+                                {
                                    "ip_addr": "<ip_with_mask>",
                                    "ip_range": ["<start_ip>", "<end_ip>"],
                                 },
                                 ...
+                             ]
+             }
     error :  null
 get_user
 --------
 Get's an 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 which cannot specify userid parameter.
+rights, or regular users that cannot specify different userid than theirs
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "get_user"
     args :    {
                 "userid" : "<username or user_id Optional(=apiuser)>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: None if user does not exist or
+            {
                 "user_id" :     "<user_id>",
                 "username" :    "<username>",
                 "firstname":    "<firstname>",
                 "lastname" :    "<lastname>",
                 "email" :       "<email>",
                 "emails":       "<list_of_all_additional_emails>",
                 "ip_addresses": "<list_of_ip_addresses_for_user>",
                 "active" :      "<bool>",
                 "admin" :       "<bool>",
                 "ldap_dn" :     "<ldap_dn>",
                 "last_login":   "<last_login>",
                 "permissions": {
                     "global": ["hg.create.repository",
                                "repository.read",
                                "hg.register.manual_activate"],
                     "repositories": {"repo1": "repository.none"},
                     "repositories_groups": {"Group1": "group.read"}
                  },
+            }
     error:  null
 get_users
 ---------
 Lists all existing users. This command can be executed only using api_key
 belonging to user with admin rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "get_users"
     args :    { }
 OUTPUT::
     id : <id_given_in_input>
     result: [
+              {
                 "user_id" :     "<user_id>",
                 "username" :    "<username>",
                 "firstname":    "<firstname>",
                 "lastname" :    "<lastname>",
                 "email" :       "<email>",
                 "emails":       "<list_of_all_additional_emails>",
                 "ip_addresses": "<list_of_ip_addresses_for_user>",
                 "active" :      "<bool>",
                 "admin" :       "<bool>",
                 "ldap_dn" :     "<ldap_dn>",
                 "last_login":   "<last_login>",
               },
     	      …
+            ]
     error:  null
 create_user
 -----------
 Creates new user. This command can
 be executed only using api_key belonging to user with admin rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "create_user"
     args :    {
                 "username" :  "<username>",
                 "email" :     "<useremail>",
                 "password" :  "<password>",
                 "firstname" : "<firstname> = Optional(None)",
                 "lastname" :  "<lastname> = Optional(None)",
                 "active" :    "<bool> = Optional(True)",
                 "admin" :     "<bool> = Optional(False)",
                 "ldap_dn" :   "<ldap_dn> = Optional(None)"
@@ @@ -481,416 +481,420 @@ OUTPUT:: @@
     error : null
 create_users_group
 ------------------
 Creates new users group. This command can be executed only using api_key
 belonging to user with admin rights
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "create_users_group"
     args:     {
                 "group_name":  "<groupname>",
                 "active":"<bool> = Optional(True)"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: {
               "msg": "created new users group `<groupname>`",
               "users_group": {
                      "users_group_id" : "<id>",
                      "group_name" :     "<groupname>",
                      "active":          "<bool>",
                },
+            }
     error:  null
 add_user_to_users_group
 -----------------------
 Adds a user to a users group. If user exists in that group success will be
 `false`. This command can be executed only using api_key
 belonging to user with admin rights
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "add_user_users_group"
     args:     {
                 "usersgroupid" : "<users group id or name>",
                 "userid" : "<user_id or username>",
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: {
               "success": True|False # depends on if member is in group
               "msg": "added member `<username>` to users group `<groupname>` |
                       User is already in that group"
+            }
     error:  null
 remove_user_from_users_group
 ----------------------------
 Removes a user from a users group. If user is not in given group success will
 be `false`. This command can be executed only
 using api_key belonging to user with admin rights
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "remove_user_from_users_group"
     args:     {
                 "usersgroupid" : "<users group id or name>",
                 "userid" : "<user_id or username>",
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: {
               "success":  True|False,  # depends on if member is in group
               "msg": "removed member <username> from users group <groupname> |
                       User wasn't in group"
+            }
     error:  null
 get_repo
 --------
 Gets an existing repository by it's name or repository_id. Members will return
 either users_group or user associated to that repository. This command can
 be executed only using api_key belonging to user with admin rights.
 either users_group or user associated to that repository. This command can be
 executed only using api_key belonging to user with admin
 rights or regular user that have at least read access to repository.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "get_repo"
     args:     {
                 "repoid" : "<reponame or repo_id>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: None if repository does not exist or
+            {
                 "repo_id" :          "<repo_id>",
                 "repo_name" :        "<reponame>"
                 "repo_type" :        "<repo_type>",
                 "clone_uri" :        "<clone_uri>",
                 "enable_downloads":  "<bool>",
                 "enable_locking":    "<bool>",
                 "enable_statistics": "<bool>",
                 "private":           "<bool>",
                 "created_on" :       "<datetimecreated>",
                 "description" :      "<description>",
                 "landing_rev":       "<landing_rev>",
                 "owner":             "<repo_owner>",
                 "fork_of":           "<name_of_fork_parent>",
                 "members" :     [
+                                  {
                                     "type": "user",
                                     "user_id" :  "<user_id>",
                                     "username" : "<username>",
                                     "firstname": "<firstname>",
                                     "lastname" : "<lastname>",
                                     "email" :    "<email>",
                                     "emails":    "<list_of_all_additional_emails>",
                                     "active" :   "<bool>",
                                     "admin" :    "<bool>",
                                     "ldap_dn" :  "<ldap_dn>",
                                     "last_login": "<last_login>",
                                     "permission" : "repository.(read|write|admin)"
                                   },
                                   …
+                                  {
                                     "type": "users_group",
                                     "id" :       "<usersgroupid>",
                                     "name" :     "<usersgroupname>",
                                     "active":    "<bool>",
                                     "permission" : "repository.(read|write|admin)"
                                   },
                                   …
+                                ]
+            }
     error:  null
 get_repos
 ---------
 Lists all existing repositories. This command can be executed only using api_key
 belonging to user with admin rights
 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.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "get_repos"
     args:     { }
 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
 get_repo_nodes
 --------------
 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
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "get_repo_nodes"
     args:     {
                 "repoid" : "<reponame or repo_id>"
                 "revision"  : "<revision>",
                 "root_path" : "<root_path>",
                 "ret_type"  : "<ret_type> = Optional('all')"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: [
+              {
                 "name" :        "<name>"
                 "type" :        "<type>",
               },
               …
+            ]
     error:  null
 create_repo
 -----------
 Creates a repository. This command can be executed only using api_key
 belonging to user with admin rights.
 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.
 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
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "create_repo"
     args:     {
                 "repo_name" :        "<reponame>",
                 "owner" :            "<onwer_name_or_id>",
+                "owner" :            "<onwer_name_or_id = Optional(=apiuser)>",
                 "repo_type" :        "<repo_type> = Optional('hg')",
                 "description" :      "<description> = Optional('')",
                 "private" :          "<bool> = Optional(False)",
                 "clone_uri" :        "<clone_uri> = Optional(None)",
                 "landing_rev" :      "<landing_rev> = Optional('tip')",
                 "enable_downloads":  "<bool> = Optional(False)",
                 "enable_locking":    "<bool> = Optional(False)",
                 "enable_statistics": "<bool> = Optional(False)",
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: {
               "msg": "Created new repository `<reponame>`",
               "repo": {
                 "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":             "<username or user_id>",
                 "fork_of":           "<name_of_fork_parent>",
                 "enable_downloads":  "<bool>",
                 "enable_locking":    "<bool>",
                 "enable_statistics": "<bool>",
               },
+            }
     error:  null
 fork_repo
 ---------
 Creates a fork of given repo. This command can be executed only using api_key
 belonging to user with admin rights. In case of using celery this will
 Creates a fork of given repo. In case of using celery this will
 immidiatelly return success message, while fork is going to be created
 asynchronous
 asynchronous. This command can be executed only using api_key belonging to
 user with admin rights or regular user that have fork permission, and at least
 read access to forking repository. Regular users cannot specify owner parameter.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "fork_repo"
     args:     {
                 "repoid" :          "<reponame or repo_id>",
                 "fork_name":        "<forkname>",
                 "owner":            "<username or user_id>",
+                "owner":            "<username or user_id = Optional(=apiuser)>",
                 "description":      "<description>",
                 "copy_permissions": "<bool>",
                 "private":          "<bool>",
                 "landing_rev":      "<landing_rev>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: {
               "msg": "Created fork of `<reponame>` as `<forkname>`",
               "success": true
+            }
     error:  null
 delete_repo
 -----------
 Deletes a repository. This command can be executed only using api_key
 belonging to user with admin rights.
 Deletes a repository. This command can be executed only using api_key belonging to user with admin
 rights or regular user that have admin access to repository.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "delete_repo"
     args:     {
                 "repoid" : "<reponame or repo_id>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: {
               "msg": "Deleted repository `<reponame>`",
               "success": true
+            }
     error:  null
 grant_user_permission
 ---------------------
 Grant permission for user on given repository, or update existing one
 if found. This command can be executed only using api_key belonging to user
 with admin rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "grant_user_permission"
     args:     {
                 "repoid" : "<reponame or repo_id>"
                 "userid" : "<username or user_id>"
                 "perm" :       "(repository.(none|read|write|admin))",
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: {
               "msg" : "Granted perm: `<perm>` for user: `<username>` in repo: `<reponame>`",
               "success": true
+            }
     error:  null
 revoke_user_permission
 ----------------------
 Revoke permission for user on given repository. This command can be executed
 only using api_key belonging to user with admin rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method  : "revoke_user_permission"
     args:     {
                 "repoid" : "<reponame or repo_id>"
                 "userid" : "<username or user_id>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: {
               "msg" : "Revoked perm for user: `<username>` in repo: `<reponame>`",
               "success": true
+            }
     error:  null
 grant_users_group_permission
 ----------------------------
 Grant permission for users group on given repository, or update
 existing one if found. This command can be executed only using
 api_key belonging to user with admin rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "grant_users_group_permission"
     args:     {
                 "repoid" : "<reponame or repo_id>"
                 "usersgroupid" : "<users group id or name>"
                 "perm" : "(repository.(none|read|write|admin))",
+              }

rhodecode/controllers/api/__init__.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.controllers.api
     ~~~~~~~~~~~~~~~~~~~~~~~~~
     JSON RPC controller
     :created_on: Aug 20, 2011
     :author: marcink
     :copyright: (C) 2011-2012 Marcin Kuzminski <marcin@python-works.com>
     :license: GPLv3, see COPYING for more details.
 """
 # 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; version 2
 # of the License or (at your opinion) any later version of the license.
+#
 # 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, write to the Free Software
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
 # MA  02110-1301, USA.
 import inspect
 import logging
 import types
 import urllib
 import traceback
 import time
 from rhodecode.lib.compat import izip_longest, json
 from paste.response import replace_header
 from pylons.controllers import WSGIController
 from webob.exc import HTTPNotFound, HTTPForbidden, HTTPInternalServerError, \
 HTTPBadRequest, HTTPError
 from rhodecode.model.db import User
 from rhodecode.model import meta
 from rhodecode.lib.compat import izip_longest, json
 from rhodecode.lib.auth import AuthUser
 from rhodecode.lib.base import _get_ip_addr, _get_access_path
 from rhodecode.lib.utils2 import safe_unicode
 log = logging.getLogger('JSONRPC')
 class JSONRPCError(BaseException):
     def __init__(self, message):
         self.message = message
         super(JSONRPCError, self).__init__()
     def __str__(self):
         return str(self.message)
 def jsonrpc_error(message, retid=None, code=None):
     """
     Generate a Response object with a JSON-RPC error body
     """
     from pylons.controllers.util import Response
     return Response(
             body=json.dumps(dict(id=retid, result=None, error=message)),
             status=code,
             content_type='application/json'
+    )
 class JSONRPCController(WSGIController):
     """
      A WSGI-speaking JSON-RPC controller class
      See the specification:
      <http://json-rpc.org/wiki/specification>`.
      Valid controller return values should be json-serializable objects.
      Sub-classes should catch their exceptions and raise JSONRPCError
      if they want to pass meaningful errors to the client.
      """
     def _get_ip_addr(self, environ):
         return _get_ip_addr(environ)
     def _get_method_args(self):
         """
         Return `self._rpc_args` to dispatched controller method
         chosen by __call__
         """
         return self._rpc_args
     def __call__(self, environ, start_response):
         """
         Parse the request body as JSON, look up the method on the
         controller and if it exists, dispatch to it.
         """
         start = time.time()
         ip_addr = self.ip_addr = self._get_ip_addr(environ)
         self._req_id = None
         if 'CONTENT_LENGTH' not in environ:
             log.debug("No Content-Length")
             return jsonrpc_error(retid=self._req_id,
                                  message="No Content-Length in request")
         else:
             length = environ['CONTENT_LENGTH'] or 0
             length = int(environ['CONTENT_LENGTH'])
             log.debug('Content-Length: %s' % length)
         if length == 0:
             log.debug("Content-Length is 0")
             return jsonrpc_error(retid=self._req_id,
                                  message="Content-Length is 0")
         raw_body = environ['wsgi.input'].read(length)
         try:
             json_body = json.loads(urllib.unquote_plus(raw_body))
         except ValueError, e:
             # catch JSON errors Here
             return jsonrpc_error(retid=self._req_id,
                                  message="JSON parse error ERR:%s RAW:%r" \
                                  % (e, urllib.unquote_plus(raw_body)))
         # check AUTH based on API KEY
         try:
             self._req_api_key = json_body['api_key']
             self._req_id = json_body['id']
             self._req_method = json_body['method']
             self._request_params = json_body['args']
             log.debug(
                 'method: %s, params: %s' % (self._req_method,
                                             self._request_params)
+            )
         except KeyError, e:

rhodecode/controllers/api/api.py

➞

Show inline comments

@@ @@ -126,247 +126,251 @@ def get_users_group_or_error(usersgroupi @@
     :param userid:
     """
     users_group = UsersGroupModel().get_group(usersgroupid)
     if users_group is None:
         raise JSONRPCError('users group `%s` does not exist' % usersgroupid)
     return users_group
 def get_perm_or_error(permid):
     """
     Get permission by id or name or return JsonRPCError if not found
     :param userid:
     """
     perm = PermissionModel().get_permission_by_name(permid)
     if perm is None:
         raise JSONRPCError('permission `%s` does not exist' % (permid))
     return perm
 class ApiController(JSONRPCController):
     """
     API Controller
     Each method needs to have USER as argument this is then based on given
     API_KEY propagated as instance of user object
     Preferably this should be first argument also
     Each function should also **raise** JSONRPCError for any
     errors that happens
     """
     @HasPermissionAllDecorator('hg.admin')
     def pull(self, apiuser, repoid):
         """
         Dispatch pull action on given repo
         :param apiuser:
         :param repoid:
         """
         repo = get_repo_or_error(repoid)
         try:
             ScmModel().pull_changes(repo.repo_name,
                                     self.rhodecode_user.username)
             return 'Pulled from `%s`' % repo.repo_name
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Unable to pull changes from `%s`' % repo.repo_name
+            )
     @HasPermissionAllDecorator('hg.admin')
     def rescan_repos(self, apiuser, remove_obsolete=Optional(False)):
         """
         Dispatch rescan repositories action. If remove_obsolete is set
         than also delete repos that are in database but not in the filesystem.
         aka "clean zombies"
         :param apiuser:
         :param remove_obsolete:
         """
         try:
             rm_obsolete = Optional.extract(remove_obsolete)
             added, removed = repo2db_mapper(ScmModel().repo_scan(),
                                             remove_obsolete=rm_obsolete)
             return {'added': added, 'removed': removed}
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Error occurred during rescan repositories action'
+            )
     def lock(self, apiuser, repoid, locked, userid=Optional(OAttr('apiuser'))):
         """
         Set locking state on particular repository by given user, if
         this command is runned by non-admin account userid is set to user
         who is calling this method
         :param apiuser:
         :param repoid:
         :param userid:
         :param locked:
         """
         repo = get_repo_or_error(repoid)
         if HasPermissionAnyApi('hg.admin')(user=apiuser):
             pass
         elif HasRepoPermissionAnyApi('repository.admin',
                                      'repository.write')(user=apiuser,
                                                          repo_name=repo.repo_name):
             #make sure normal user does not pass userid, he is not allowed to do that
             if not isinstance(userid, Optional):
             #make sure normal user does not pass someone else userid,
             #he is not allowed to do that
             if not isinstance(userid, Optional) and userid != apiuser.user_id:
                 raise JSONRPCError(
-                    'Only RhodeCode admin can specify `userid` param'
+                    'userid is not the same as your user'
+                )
         else:
             return abort(403)
             raise JSONRPCError('repository `%s` does not exist' % (repoid))
         if isinstance(userid, Optional):
             userid = apiuser.user_id
         user = get_user_or_error(userid)
         locked = bool(locked)
         try:
             if locked:
                 Repository.lock(repo, user.user_id)
             else:
                 Repository.unlock(repo)
             return ('User `%s` set lock state for repo `%s` to `%s`'
                     % (user.username, repo.repo_name, locked))
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Error occurred locking repository `%s`' % repo.repo_name
+            )
     @HasPermissionAllDecorator('hg.admin')
     def show_ip(self, apiuser, userid):
         """
         Shows IP address as seen from RhodeCode server, together with all
         defined IP addresses for given user
         :param apiuser:
         :param userid:
         """
         user = get_user_or_error(userid)
         ips = UserIpMap.query().filter(UserIpMap.user == user).all()
         return dict(
             ip_addr_server=self.ip_addr,
             user_ips=ips
+        )
     def get_user(self, apiuser, userid=Optional(OAttr('apiuser'))):
         """"
         Get a user by username, or userid, if userid is given
         :param apiuser:
         :param userid:
         """
         if HasPermissionAnyApi('hg.admin')(user=apiuser):
             pass
         else:
             if not isinstance(userid, Optional):
         if HasPermissionAnyApi('hg.admin')(user=apiuser) is False:
             #make sure normal user does not pass someone else userid,
             #he is not allowed to do that
             if not isinstance(userid, Optional) and userid != apiuser.user_id:
                 raise JSONRPCError(
-                    'Only RhodeCode admin can specify `userid` params'
+                    'userid is not the same as your user'
+                )
         if isinstance(userid, Optional):
             userid = apiuser.user_id
         user = get_user_or_error(userid)
         data = user.get_api_data()
         data['permissions'] = AuthUser(user_id=user.user_id).permissions
         return data
     @HasPermissionAllDecorator('hg.admin')
     def get_users(self, apiuser):
         """"
         Get all users
         :param apiuser:
         """
         result = []
         for user in UserModel().get_all():
             result.append(user.get_api_data())
         return result
     @HasPermissionAllDecorator('hg.admin')
     def create_user(self, apiuser, username, email, password,
                     firstname=Optional(None), lastname=Optional(None),
                     active=Optional(True), admin=Optional(False),
                     ldap_dn=Optional(None)):
         """
         Create new user
         :param apiuser:
         :param username:
         :param email:
         :param password:
         :param firstname:
         :param lastname:
         :param active:
         :param admin:
         :param ldap_dn:
         """
         if UserModel().get_by_username(username):
             raise JSONRPCError("user `%s` already exist" % username)
         if UserModel().get_by_email(email, case_insensitive=True):
             raise JSONRPCError("email `%s` already exist" % email)
         if Optional.extract(ldap_dn):
             # generate temporary password if ldap_dn
             password = PasswordGenerator().gen_password(length=8)
         try:
             user = UserModel().create_or_update(
                 username=Optional.extract(username),
                 password=Optional.extract(password),
                 email=Optional.extract(email),
                 firstname=Optional.extract(firstname),
                 lastname=Optional.extract(lastname),
                 active=Optional.extract(active),
                 admin=Optional.extract(admin),
                 ldap_dn=Optional.extract(ldap_dn)
+            )
             Session().commit()
             return dict(
                 msg='created new user `%s`' % username,
                 user=user.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create user `%s`' % username)
     @HasPermissionAllDecorator('hg.admin')
     def update_user(self, apiuser, userid, username=Optional(None),
                     email=Optional(None), firstname=Optional(None),
                     lastname=Optional(None), active=Optional(None),
                     admin=Optional(None), ldap_dn=Optional(None),
                     password=Optional(None)):
         """
         Updates given user
         :param apiuser:
         :param userid:
         :param username:
         :param email:
         :param firstname:
         :param lastname:
         :param active:
         :param admin:
         :param ldap_dn:
         :param password:
         """
         user = get_user_or_error(userid)
         # call function and store only updated arguments
         updates = {}
         def store_update(attr, name):
@@ @@ -442,392 +446,431 @@ class ApiController(JSONRPCController): @@
         :param apiuser:
         """
         result = []
         for users_group in UsersGroupModel().get_all():
             result.append(users_group.get_api_data())
         return result
     @HasPermissionAllDecorator('hg.admin')
     def create_users_group(self, apiuser, group_name, active=Optional(True)):
         """
         Creates an new usergroup
         :param apiuser:
         :param group_name:
         :param active:
         """
         if UsersGroupModel().get_by_name(group_name):
             raise JSONRPCError("users group `%s` already exist" % group_name)
         try:
             active = Optional.extract(active)
             ug = UsersGroupModel().create(name=group_name, active=active)
             Session().commit()
             return dict(
                 msg='created new users group `%s`' % group_name,
                 users_group=ug.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create group `%s`' % group_name)
     @HasPermissionAllDecorator('hg.admin')
     def add_user_to_users_group(self, apiuser, usersgroupid, userid):
         """"
         Add a user to a users group
         :param apiuser:
         :param usersgroupid:
         :param userid:
         """
         user = get_user_or_error(userid)
         users_group = get_users_group_or_error(usersgroupid)
         try:
             ugm = UsersGroupModel().add_user_to_group(users_group, user)
             success = True if ugm != True else False
             msg = 'added member `%s` to users group `%s`' % (
                         user.username, users_group.users_group_name
+                    )
             msg = msg if success else 'User is already in that group'
             Session().commit()
             return dict(
                 success=success,
                 msg=msg
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to add member to users group `%s`' % (
                     users_group.users_group_name
+                )
+            )
     @HasPermissionAllDecorator('hg.admin')
     def remove_user_from_users_group(self, apiuser, usersgroupid, userid):
         """
         Remove user from a group
         :param apiuser:
         :param usersgroupid:
         :param userid:
         """
         user = get_user_or_error(userid)
         users_group = get_users_group_or_error(usersgroupid)
         try:
             success = UsersGroupModel().remove_user_from_group(users_group,
                                                                user)
             msg = 'removed member `%s` from users group `%s`' % (
                         user.username, users_group.users_group_name
+                    )
             msg = msg if success else "User wasn't in group"
             Session().commit()
             return dict(success=success, msg=msg)
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to remove member from users group `%s`' % (
                         users_group.users_group_name
+                    )
+            )
     @HasPermissionAllDecorator('hg.admin')
     def get_repo(self, apiuser, repoid):
         """"
         Get repository by name
         :param apiuser:
         :param repoid:
         """
         repo = get_repo_or_error(repoid)
         if HasPermissionAnyApi('hg.admin')(user=apiuser) is False:
             # check if we have admin permission for this repo !
             if HasRepoPermissionAnyApi('repository.admin')(user=apiuser,
                                             repo_name=repo.repo_name) is False:
                 raise JSONRPCError('repository `%s` does not exist' % (repoid))
         members = []
         for user in repo.repo_to_perm:
             perm = user.permission.permission_name
             user = user.user
             user_data = user.get_api_data()
             user_data['type'] = "user"
             user_data['permission'] = perm
             members.append(user_data)
         for users_group in repo.users_group_to_perm:
             perm = users_group.permission.permission_name
             users_group = users_group.users_group
             users_group_data = users_group.get_api_data()
             users_group_data['type'] = "users_group"
             users_group_data['permission'] = perm
             members.append(users_group_data)
         data = repo.get_api_data()
         data['members'] = members
         return data
     @HasPermissionAllDecorator('hg.admin')
     def get_repos(self, apiuser):
         """"
         Get all repositories
         :param apiuser:
         """
         result = []
         if HasPermissionAnyApi('hg.admin')(user=apiuser) is False:
             repos = RepoModel().get_all_user_repos(user=apiuser)
         else:
             repos = RepoModel().get_all()
         result = []
         for repo in RepoModel().get_all():
         for repo in repos:
             result.append(repo.get_api_data())
         return result
     @HasPermissionAllDecorator('hg.admin')
     def get_repo_nodes(self, apiuser, repoid, revision, root_path,
                        ret_type='all'):
         """
         returns a list of nodes and it's children
         for a given path at given revision. It's possible to specify ret_type
         to show only files or dirs
         :param apiuser:
         :param repoid: name or id of repository
         :param revision: revision for which listing should be done
         :param root_path: path from which start displaying
         :param ret_type: return type 'all|files|dirs' nodes
         """
         repo = get_repo_or_error(repoid)
         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' % _map.keys())
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to get repo: `%s` nodes' % repo.repo_name
+            )
     @HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
     def create_repo(self, apiuser, repo_name, owner, repo_type=Optional('hg'),
     def create_repo(self, apiuser, repo_name, owner=Optional(OAttr('apiuser')),
                     repo_type=Optional('hg'),
                     description=Optional(''), private=Optional(False),
                     clone_uri=Optional(None), landing_rev=Optional('tip'),
                     enable_statistics=Optional(False),
                     enable_locking=Optional(False),
                     enable_downloads=Optional(False)):
         """
         Create repository, if clone_url is given it makes a remote clone
-        if repo_name is withina  group name the groups will be created
+        if repo_name is within a group name the groups will be created
         automatically if they aren't present
         :param apiuser:
         :param repo_name:
         :param onwer:
         :param repo_type:
         :param description:
         :param private:
         :param clone_uri:
         :param landing_rev:
         """
         if HasPermissionAnyApi('hg.admin')(user=apiuser) is False:
             if not isinstance(owner, Optional):
                 #forbid setting owner for non-admins
                 raise JSONRPCError(
                     'Only RhodeCode admin can specify `owner` param'
+                )
         if isinstance(owner, Optional):
             owner = apiuser.user_id
         owner = get_user_or_error(owner)
         if RepoModel().get_by_repo_name(repo_name):
             raise JSONRPCError("repo `%s` already exist" % repo_name)
         defs = RhodeCodeSetting.get_default_repo_settings(strip_prefix=True)
         if isinstance(private, Optional):
             private = defs.get('repo_private') or Optional.extract(private)
         if isinstance(repo_type, Optional):
             repo_type = defs.get('repo_type')
         if isinstance(enable_statistics, Optional):
             enable_statistics = defs.get('repo_enable_statistics')
         if isinstance(enable_locking, Optional):
             enable_locking = defs.get('repo_enable_locking')
         if isinstance(enable_downloads, Optional):
             enable_downloads = defs.get('repo_enable_downloads')
         clone_uri = Optional.extract(clone_uri)
         description = Optional.extract(description)
         landing_rev = Optional.extract(landing_rev)
         try:
             # create structure of groups and return the last group
             group = map_groups(repo_name)
             repo = RepoModel().create_repo(
                 repo_name=repo_name,
                 repo_type=repo_type,
                 description=description,
                 owner=owner,
                 private=private,
                 clone_uri=clone_uri,
                 repos_group=group,
                 landing_rev=landing_rev,
                 enable_statistics=enable_statistics,
                 enable_downloads=enable_downloads,
                 enable_locking=enable_locking
+            )
             Session().commit()
             return dict(
                 msg="Created new repository `%s`" % (repo.repo_name),
                 repo=repo.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create repository `%s`' % repo_name)
     @HasPermissionAllDecorator('hg.admin')
     def fork_repo(self, apiuser, repoid, fork_name, owner,
     @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
     def fork_repo(self, apiuser, repoid, fork_name, owner=Optional(OAttr('apiuser')),
                   description=Optional(''), copy_permissions=Optional(False),
                   private=Optional(False), landing_rev=Optional('tip')):
         repo = get_repo_or_error(repoid)
         repo_name = repo.repo_name
         owner = get_user_or_error(owner)
         _repo = RepoModel().get_by_repo_name(fork_name)
         if _repo:
             type_ = 'fork' if _repo.fork else 'repo'
             raise JSONRPCError("%s `%s` already exist" % (type_, fork_name))
         if HasPermissionAnyApi('hg.admin')(user=apiuser):
             pass
         elif HasRepoPermissionAnyApi('repository.admin',
                                      'repository.write',
                                      'repository.read')(user=apiuser,
                                                         repo_name=repo.repo_name):
             if not isinstance(owner, Optional):
                 #forbid setting owner for non-admins
                 raise JSONRPCError(
                     'Only RhodeCode admin can specify `owner` param'
+                )
         else:
             raise JSONRPCError('repository `%s` does not exist' % (repoid))
         if isinstance(owner, Optional):
             owner = apiuser.user_id
         owner = get_user_or_error(owner)
         try:
             # create structure of groups and return the last group
             group = map_groups(fork_name)
             form_data = dict(
                 repo_name=fork_name,
                 repo_name_full=fork_name,
                 repo_group=group,
                 repo_type=repo.repo_type,
                 description=Optional.extract(description),
                 private=Optional.extract(private),
                 copy_permissions=Optional.extract(copy_permissions),
                 landing_rev=Optional.extract(landing_rev),
                 update_after_clone=False,
                 fork_parent_id=repo.repo_id,
+            )
             RepoModel().create_fork(form_data, cur_user=owner)
             return dict(
                 msg='Created fork of `%s` as `%s`' % (repo.repo_name,
                                                       fork_name),
                 success=True  # cannot return the repo data here since fork
                               # cann be done async
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to fork repository `%s` as `%s`' % (repo_name,
                                                             fork_name)
+            )
     @HasPermissionAllDecorator('hg.admin')
     def delete_repo(self, apiuser, repoid):
         """
         Deletes a given repository
         :param apiuser:
         :param repoid:
         """
         repo = get_repo_or_error(repoid)
         if HasPermissionAnyApi('hg.admin')(user=apiuser) is False:
             # check if we have admin permission for this repo !
             if HasRepoPermissionAnyApi('repository.admin')(user=apiuser,
                                             repo_name=repo.repo_name) is False:
                  raise JSONRPCError('repository `%s` does not exist' % (repoid))
         try:
             RepoModel().delete(repo)
             Session().commit()
             return dict(
                 msg='Deleted repository `%s`' % repo.repo_name,
                 success=True
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to delete repository `%s`' % repo.repo_name
+            )
     @HasPermissionAllDecorator('hg.admin')
     def grant_user_permission(self, apiuser, repoid, userid, perm):
         """
         Grant permission for user on given repository, or update existing one
         if found
         :param repoid:
         :param userid:
         :param perm:
         """
         repo = get_repo_or_error(repoid)
         user = get_user_or_error(userid)
         perm = get_perm_or_error(perm)
         try:
             RepoModel().grant_user_permission(repo=repo, user=user, perm=perm)
             Session().commit()
             return dict(
                 msg='Granted perm: `%s` for user: `%s` in repo: `%s`' % (
                     perm.permission_name, user.username, repo.repo_name
                 ),
                 success=True
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for user: `%s` in repo: `%s`' % (
                     userid, repoid
+                )
+            )
     @HasPermissionAllDecorator('hg.admin')
     def revoke_user_permission(self, apiuser, repoid, userid):
         """
         Revoke permission for user on given repository
         :param apiuser:
         :param repoid:
         :param userid:
         """
         repo = get_repo_or_error(repoid)
         user = get_user_or_error(userid)
         try:
             RepoModel().revoke_user_permission(repo=repo, user=user)
             Session().commit()
             return dict(
                 msg='Revoked perm for user: `%s` in repo: `%s`' % (
                     user.username, repo.repo_name
                 ),
                 success=True
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for user: `%s` in repo: `%s`' % (
                     userid, repoid
+                )
+            )
     @HasPermissionAllDecorator('hg.admin')
     def grant_users_group_permission(self, apiuser, repoid, usersgroupid,
                                      perm):
         """
         Grant permission for users group on given repository, or update
         existing one if found
         :param apiuser:
         :param repoid:
         :param usersgroupid:
         :param perm:
         """
         repo = get_repo_or_error(repoid)
         perm = get_perm_or_error(perm)
         users_group = get_users_group_or_error(usersgroupid)
         try:
             RepoModel().grant_users_group_permission(repo=repo,
                                                      group_name=users_group,

rhodecode/model/repo.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.model.repo
     ~~~~~~~~~~~~~~~~~~~~
     Repository model for rhodecode
     :created_on: Jun 5, 2010
     :author: marcink
     :copyright: (C) 2010-2012 Marcin Kuzminski <marcin@python-works.com>
     :license: GPLv3, see COPYING for more details.
 """
 # 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/>.
 from __future__ import with_statement
 import os
 import shutil
 import logging
 import traceback
 from datetime import datetime
 from rhodecode.lib.vcs.backends import get_backend
 from rhodecode.lib.compat import json
 from rhodecode.lib.utils2 import LazyProperty, safe_str, safe_unicode,\
     remove_prefix
 from rhodecode.lib.caching_query import FromCache
 from rhodecode.lib.hooks import log_create_repository, log_delete_repository
 from rhodecode.model import BaseModel
 from rhodecode.model.db import Repository, UserRepoToPerm, User, Permission, \
     Statistics, UsersGroup, UsersGroupRepoToPerm, RhodeCodeUi, RepoGroup,\
     RhodeCodeSetting
 from rhodecode.lib import helpers as h
 from rhodecode.lib.auth import HasRepoPermissionAny
 log = logging.getLogger(__name__)
 class RepoModel(BaseModel):
     cls = Repository
     URL_SEPARATOR = Repository.url_sep()
     def __get_users_group(self, users_group):
         return self._get_instance(UsersGroup, users_group,
                                   callback=UsersGroup.get_by_group_name)
     def _get_repos_group(self, repos_group):
         return self._get_instance(RepoGroup, repos_group,
                                   callback=RepoGroup.get_by_group_name)
     @LazyProperty
     def repos_path(self):
         """
         Get's the repositories root path from database
         """
         q = self.sa.query(RhodeCodeUi).filter(RhodeCodeUi.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):
         """
         Get's all repositories that user have at least read access
         :param user:
         :type user:
         """
         from rhodecode.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).all()
         return json.dumps([
+            {
              'id': u.user_id,
              'fname': u.name,
              'lname': u.lastname,
              'nname': u.username,
              'gravatar_lnk': h.gravatar_url(u.email, 14)
             } for u in users]
+        )
     def get_users_groups_js(self):
         users_groups = self.sa.query(UsersGroup)\
             .filter(UsersGroup.users_group_active == True).all()
         return json.dumps([
+            {
              'id': gr.users_group_id,
              'grname': gr.users_group_name,
              'grmembers': len(gr.members),
             } for gr in users_groups]
+        )
     @classmethod
     def _render_datatable(cls, tmpl, *args, **kwargs):
         import rhodecode
         from pylons import tmpl_context as c
         from pylons.i18n.translation import _
         _tmpl_lookup = rhodecode.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)
     def get_repos_as_dict(self, repos_list=None, admin=False, perm_check=True):
         _render = self._render_datatable
         def quick_menu(repo_name):
             return _render('quick_menu', repo_name)
         def repo_lnk(name, rtype, private, fork_of):
             return _render('repo_name', name, rtype, private, fork_of,
                            short_name=not admin, 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):
             from pylons import tmpl_context as c
             if c.visual.stylify_metatags:
                 return h.urlify_text(h.desc_stylize(h.truncate(desc, 60)))
             else:
                 return h.urlify_text(h.truncate(desc, 60))
         def repo_actions(repo_name):
             return _render('repo_actions', repo_name)
         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.lower(),
                 "name": repo_lnk(repo.repo_name, repo.repo_type,
                                  repo.private, repo.fork),
                 "last_change": last_change(repo.last_db_change),
                 "last_changeset": last_rev(repo.repo_name, cs_cache),
                 "raw_tip": cs_cache.get('revision'),
                 "desc": desc(repo.description),
                 "owner": h.person(repo.user.username),
                 "rss": rss_lnk(repo.repo_name),
                 "atom": atom_lnk(repo.repo_name),

rhodecode/tests/api/api_base.py

➞

Show inline comments

 from __future__ import with_statement
 import random
 import mock
 from rhodecode.tests import *
 from rhodecode.lib.compat import json
 from rhodecode.lib.auth import AuthUser
 from rhodecode.model.user import UserModel
 from rhodecode.model.users_group import UsersGroupModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.meta import Session
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.db import Repository
 API_URL = '/_admin/api'
 def _build_data(apikey, method, **kw):
     """
     Builds API data with given random ID
     :param random_id:
     :type random_id:
     """
     random_id = random.randrange(1, 9999)
     return random_id, json.dumps({
         "id": random_id,
         "api_key": apikey,
         "method": method,
         "args": kw
     })
 jsonify = lambda obj: json.loads(json.dumps(obj))
 def crash(*args, **kwargs):
     raise Exception('Total Crash !')
 def api_call(test_obj, params):
     response = test_obj.app.post(API_URL, content_type='application/json',
                                  params=params)
     return response
 TEST_USERS_GROUP = 'test_users_group'
 def make_users_group(name=TEST_USERS_GROUP):
     gr = UsersGroupModel().create(name=name)
     UsersGroupModel().add_user_to_group(users_group=gr,
                                         user=TEST_USER_ADMIN_LOGIN)
     Session().commit()
     return gr
 def destroy_users_group(name=TEST_USERS_GROUP):
     UsersGroupModel().delete(users_group=name, force=True)
     Session().commit()
 def create_repo(repo_name, repo_type):
+def create_repo(repo_name, repo_type, owner=None):
     # create new repo
     form_data = _get_repo_create_params(
                     repo_name_full=repo_name,
                     repo_description='description %s' % repo_name,
+                )
     cur_user = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
+    cur_user = UserModel().get_by_username(owner or TEST_USER_ADMIN_LOGIN)
     r = RepoModel().create(form_data, cur_user)
     Session().commit()
     return r
 def create_fork(fork_name, fork_type, fork_of):
     fork = RepoModel(Session())._get_repo(fork_of)
     r = create_repo(fork_name, fork_type)
     r.fork = fork
     Session().add(r)
     Session().commit()
     return r
 def destroy_repo(repo_name):
     RepoModel().delete(repo_name)
     Session().commit()
 class BaseTestApi(object):
     REPO = None
     REPO_TYPE = None
     @classmethod
     def setUpClass(self):
         self.usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         self.apikey = self.usr.api_key
-        self.TEST_USER = UserModel().create_or_update(
+        self.test_user = UserModel().create_or_update(
             username='test-api',
             password='test',
             email='test@api.rhodecode.org',
             firstname='first',
             lastname='last'
+        )
         Session().commit()
         self.TEST_USER_LOGIN = self.TEST_USER.username
         self.TEST_USER_LOGIN = self.test_user.username
         self.apikey_regular = self.test_user.api_key
     @classmethod
     def teardownClass(self):
         pass
     def setUp(self):
         self.maxDiff = None
         make_users_group()
     def tearDown(self):
         destroy_users_group()
     def _compare_ok(self, id_, expected, given):
         expected = jsonify({
             'id': id_,
             'error': None,
             'result': expected
         })
         given = json.loads(given)
         self.assertEqual(expected, given)
     def _compare_error(self, id_, expected, given):
         expected = jsonify({
             'id': id_,
             'error': expected,
             'result': None
         })
         given = json.loads(given)
         self.assertEqual(expected, given)
 #    def test_Optional(self):
 #        from rhodecode.controllers.api.api import Optional
 #        option1 = Optional(None)
 #        self.assertEqual('<Optional:%s>' % None, repr(option1))
+#
 #        self.assertEqual(1, Optional.extract(Optional(1)))
 #        self.assertEqual('trololo', Optional.extract('trololo'))
     def test_api_wrong_key(self):
         id_, params = _build_data('trololo', 'get_user')
         response = api_call(self, params)
         expected = 'Invalid API KEY'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param(self):
-        id_, params = _build_data(self.apikey, 'get_user')
+        id_, params = _build_data(self.apikey, 'get_repo')
         response = api_call(self, params)
-        expected = 'Missing non optional `userid` arg in JSON DATA'
+        expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_users(self):
         id_, params = _build_data(self.apikey, 'get_users',)
         response = api_call(self, params)
         ret_all = []
         for usr in UserModel().get_all():
             ret = usr.get_api_data()
             ret_all.append(jsonify(ret))
         expected = ret_all
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user(self):
         id_, params = _build_data(self.apikey, 'get_user',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(usr.user_id).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_that_does_not_exist(self):
         id_, params = _build_data(self.apikey, 'get_user',
                                   userid='trololo')
         response = api_call(self, params)
         expected = "user `%s` does not exist" % 'trololo'
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_user_without_giving_userid(self):
         id_, params = _build_data(self.apikey, 'get_user')
         response = api_call(self, params)
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(usr.user_id).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_without_giving_userid_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_user')
         response = api_call(self, params)
         usr = UserModel().get_by_username(self.TEST_USER_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(usr.user_id).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_with_giving_userid_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_user',
                                   userid=self.TEST_USER_LOGIN)
         response = api_call(self, params)
         expected = 'userid is not the same as your user'
         self._compare_error(id_, expected, given=response.body)
     def test_api_pull(self):
         #TODO: issues with rhodecode_extras here.. not sure why !
         pass
 #        repo_name = 'test_pull'
 #        r = create_repo(repo_name, self.REPO_TYPE)
 #        r.clone_uri = TEST_self.REPO
 #        Session.add(r)
 #        Session.commit()
+#
 #        id_, params = _build_data(self.apikey, 'pull',
 #                                  repoid=repo_name,)
 #        response = self.app.post(API_URL, content_type='application/json',
 #                                 params=params)
+#
 #        expected = 'Pulled from `%s`' % repo_name
 #        self._compare_ok(id_, expected, given=response.body)
+#
 #        destroy_repo(repo_name)
     def test_api_pull_error(self):
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=self.REPO,)
         response = api_call(self, params)
         expected = 'Unable to pull changes from `%s`' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_rescan_repos(self):
         id_, params = _build_data(self.apikey, 'rescan_repos')
         response = api_call(self, params)
         expected = {'added': [], 'removed': []}
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(ScmModel, 'repo_scan', crash)
     def test_api_rescann_error(self):
         id_, params = _build_data(self.apikey, 'rescan_repos',)
         response = api_call(self, params)
         expected = 'Error occurred during rescan repositories action'
         self._compare_error(id_, expected, given=response.body)
     def test_api_lock_repo_lock_aquire(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = ('User `%s` set lock state for repo `%s` to `%s`'
                    % (TEST_USER_ADMIN_LOGIN, self.REPO, True))
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_aquire_by_non_admin(self):
         repo_name = 'api_delete_me'
         create_repo(repo_name, self.REPO_TYPE, owner=self.TEST_USER_LOGIN)
         try:
             id_, params = _build_data(self.apikey_regular, 'lock',
                                       repoid=repo_name,
                                       locked=True)
             response = api_call(self, params)
             expected = ('User `%s` set lock state for repo `%s` to `%s`'
                        % (self.TEST_USER_LOGIN, repo_name, True))
             self._compare_ok(id_, expected, given=response.body)
         finally:
             destroy_repo(repo_name)
     def test_api_lock_repo_lock_aquire_non_admin_with_userid(self):
         repo_name = 'api_delete_me'
         create_repo(repo_name, self.REPO_TYPE, owner=self.TEST_USER_LOGIN)
         try:
             id_, params = _build_data(self.apikey_regular, 'lock',
                                       userid=TEST_USER_ADMIN_LOGIN,
                                       repoid=repo_name,
                                       locked=True)
             response = api_call(self, params)
             expected = 'userid is not the same as your user'
             self._compare_error(id_, expected, given=response.body)
         finally:
             destroy_repo(repo_name)
     def test_api_lock_repo_lock_aquire_non_admin_not_his_repo(self):
         id_, params = _build_data(self.apikey_regular, 'lock',
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = 'repository `%s` does not exist' % (self.REPO)
         self._compare_error(id_, expected, given=response.body)
     def test_api_lock_repo_lock_release(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=False)
         response = api_call(self, params)
         expected = ('User `%s` set lock state for repo `%s` to `%s`'
                    % (TEST_USER_ADMIN_LOGIN, self.REPO, False))
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_aquire_optional_userid(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = ('User `%s` set lock state for repo `%s` to `%s`'
                    % (TEST_USER_ADMIN_LOGIN, self.REPO, True))
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(Repository, 'lock', crash)
     def test_api_lock_error(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = 'Error occurred locking repository `%s`' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_existing_user(self):
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=TEST_USER_ADMIN_LOGIN,
                                   email='test@foo.com',
                                   password='trololo')
         response = api_call(self, params)
         expected = "user `%s` already exist" % TEST_USER_ADMIN_LOGIN
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_user_with_existing_email(self):
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=TEST_USER_ADMIN_LOGIN + 'new',
                                   email=TEST_USER_REGULAR_EMAIL,
                                   password='trololo')
         response = api_call(self, params)
         expected = "email `%s` already exist" % TEST_USER_REGULAR_EMAIL
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_user(self):
         username = 'test_new_api_user'
         email = username + "@foo.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email,
                                   password='trololo')
         response = api_call(self, params)
         usr = UserModel().get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
+        )
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         UserModel().delete(usr.user_id)
         Session().commit()
     @mock.patch.object(UserModel, 'create_or_update', crash)
     def test_api_create_user_when_exception_happened(self):
         username = 'test_new_api_user'
         email = username + "@foo.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email,
                                   password='trololo')
         response = api_call(self, params)
         expected = 'failed to create user `%s`' % username
         self._compare_error(id_, expected, given=response.body)
     def test_api_delete_user(self):
         usr = UserModel().create_or_update(username=u'test_user',
                                            password=u'qweqwe',
                                            email=u'u232@rhodecode.org',
                                            firstname=u'u1', lastname=u'u1')
         Session().commit()
         username = usr.username
         email = usr.email
         usr_id = usr.user_id
         ## DELETE THIS USER NOW
@@ @@ -373,365 +440,541 @@ class BaseTestApi(object): @@
                            ])
     def test_api_update_user(self, name, expected):
         usr = UserModel().get_by_username(self.TEST_USER_LOGIN)
         kw = {name: expected,
               'userid': usr.user_id}
         id_, params = _build_data(self.apikey, 'update_user', **kw)
         response = api_call(self, params)
         ret = {
         'msg': 'updated user ID:%s %s' % (usr.user_id, self.TEST_USER_LOGIN),
         'user': jsonify(UserModel()\
                             .get_by_username(self.TEST_USER_LOGIN)\
                             .get_api_data())
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_update_user_no_changed_params(self):
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         ret = {
         'msg': 'updated user ID:%s %s' % (usr.user_id, TEST_USER_ADMIN_LOGIN),
         'user': ret
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_update_user_by_user_id(self):
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
         response = api_call(self, params)
         ret = {
         'msg': 'updated user ID:%s %s' % (usr.user_id, TEST_USER_ADMIN_LOGIN),
         'user': ret
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(UserModel, 'update_user', crash)
     def test_api_update_user_when_exception_happens(self):
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
         response = api_call(self, params)
         ret = 'failed to update user `%s`' % usr.user_id
         expected = ret
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo(self):
         new_group = 'some_new_group'
         make_users_group(new_group)
         RepoModel().grant_users_group_permission(repo=self.REPO,
                                                  group_name=new_group,
                                                  perm='repository.read')
         Session().commit()
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(self.REPO)
         ret = repo.get_api_data()
         members = []
         for user in repo.repo_to_perm:
             perm = user.permission.permission_name
             user = user.user
             user_data = user.get_api_data()
             user_data['type'] = "user"
             user_data['permission'] = perm
             members.append(user_data)
         for users_group in repo.users_group_to_perm:
             perm = users_group.permission.permission_name
             users_group = users_group.users_group
             users_group_data = users_group.get_api_data()
             users_group_data['type'] = "users_group"
             users_group_data['permission'] = perm
             members.append(users_group_data)
         ret['members'] = members
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         destroy_users_group(new_group)
     def test_api_get_repo_by_non_admin(self):
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(self.REPO)
         ret = repo.get_api_data()
         members = []
         for user in repo.repo_to_perm:
             perm = user.permission.permission_name
             user = user.user
             user_data = user.get_api_data()
             user_data['type'] = "user"
             user_data['permission'] = perm
             members.append(user_data)
         for users_group in repo.users_group_to_perm:
             perm = users_group.permission.permission_name
             users_group = users_group.users_group
             users_group_data = users_group.get_api_data()
             users_group_data['type'] = "users_group"
             users_group_data['permission'] = perm
             members.append(users_group_data)
         ret['members'] = members
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_repo_by_non_admin_no_permission_to_repo(self):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm='repository.none')
         id_, params = _build_data(self.apikey_regular, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
         expected = 'repository `%s` does not exist' % (self.REPO)
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo_that_doesn_not_exist(self):
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid='no-such-repo')
         response = api_call(self, params)
         ret = 'repository `%s` does not exist' % 'no-such-repo'
         expected = ret
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repos(self):
         id_, params = _build_data(self.apikey, 'get_repos')
         response = api_call(self, params)
         result = []
         for repo in RepoModel().get_all():
             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)
     @parameterized.expand([('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 = json.loads(response.body)['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' % (['files', 'dirs', 'all'])
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_repo(self):
         repo_name = 'api-repo'
         id_, params = _build_data(self.apikey, 'create_repo',
                                     repo_name=repo_name,
                                     owner=TEST_USER_ADMIN_LOGIN,
                                     repo_type='hg',
+                                  )
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(repo_name)
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'repo': jsonify(repo.get_api_data())
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         destroy_repo(repo_name)
     def test_api_create_repo_unknown_owner(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'create_repo',
                                     repo_name=repo_name,
                                     owner=owner,
                                     repo_type='hg',
+                                  )
         response = api_call(self, params)
         expected = 'user `%s` does not exist' % owner
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_repo_dont_specify_owner(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'create_repo',
                                     repo_name=repo_name,
                                     repo_type='hg',
+                                  )
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(repo_name)
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'repo': jsonify(repo.get_api_data())
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         destroy_repo(repo_name)
     def test_api_create_repo_by_non_admin(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey_regular, 'create_repo',
                                     repo_name=repo_name,
                                     repo_type='hg',
+                                  )
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(repo_name)
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'repo': jsonify(repo.get_api_data())
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         destroy_repo(repo_name)
     def test_api_create_repo_by_non_admin_specify_owner(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey_regular, 'create_repo',
                                     repo_name=repo_name,
                                     repo_type='hg',
                                     owner=owner
+                                  )
         response = api_call(self, params)
         expected = 'Only RhodeCode admin can specify `owner` param'
         self._compare_error(id_, expected, given=response.body)
         destroy_repo(repo_name)
     def test_api_create_repo_exists(self):
         repo_name = self.REPO
         id_, params = _build_data(self.apikey, 'create_repo',
                                     repo_name=repo_name,
                                     owner=TEST_USER_ADMIN_LOGIN,
                                     repo_type='hg',
+                                  )
         response = api_call(self, params)
         expected = "repo `%s` already exist" % repo_name
         self._compare_error(id_, expected, given=response.body)
     @mock.patch.object(RepoModel, 'create_repo', crash)
     def test_api_create_repo_exception_occurred(self):
         repo_name = 'api-repo'
         id_, params = _build_data(self.apikey, 'create_repo',
                                     repo_name=repo_name,
                                     owner=TEST_USER_ADMIN_LOGIN,
                                     repo_type='hg',
+                                  )
         response = api_call(self, params)
         expected = 'failed to create repository `%s`' % repo_name
         self._compare_error(id_, expected, given=response.body)
     def test_api_delete_repo(self):
         repo_name = 'api_delete_me'
         create_repo(repo_name, self.REPO_TYPE)
         id_, params = _build_data(self.apikey, 'delete_repo',
                                   repoid=repo_name,)
         response = api_call(self, params)
         ret = {
             'msg': 'Deleted repository `%s`' % repo_name,
             'success': True
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_delete_repo_by_non_admin(self):
         repo_name = 'api_delete_me'
         create_repo(repo_name, self.REPO_TYPE, owner=self.TEST_USER_LOGIN)
         try:
             id_, params = _build_data(self.apikey_regular, 'delete_repo',
                                       repoid=repo_name,)
             response = api_call(self, params)
             ret = {
                 'msg': 'Deleted repository `%s`' % repo_name,
                 'success': True
+            }
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             destroy_repo(repo_name)
     def test_api_delete_repo_by_non_admin_no_permission(self):
         repo_name = 'api_delete_me'
         create_repo(repo_name, self.REPO_TYPE)
         try:
             id_, params = _build_data(self.apikey_regular, 'delete_repo',
                                       repoid=repo_name,)
             response = api_call(self, params)
             expected = 'repository `%s` does not exist' % (repo_name)
             self._compare_error(id_, expected, given=response.body)
         finally:
             destroy_repo(repo_name)
     def test_api_delete_repo_exception_occurred(self):
         repo_name = 'api_delete_me'
         create_repo(repo_name, self.REPO_TYPE)
         try:
             with mock.patch.object(RepoModel, 'delete', crash):
                 id_, params = _build_data(self.apikey, 'delete_repo',
                                           repoid=repo_name,)
                 response = api_call(self, params)
                 expected = 'failed to delete repository `%s`' % repo_name
                 self._compare_error(id_, expected, given=response.body)
         finally:
             destroy_repo(repo_name)
     def test_api_fork_repo(self):
         fork_name = 'api-repo-fork'
         id_, params = _build_data(self.apikey, 'fork_repo',
                                     repoid=self.REPO,
                                     fork_name=fork_name,
                                     owner=TEST_USER_ADMIN_LOGIN,
+                                  )
         response = api_call(self, params)
         ret = {
             'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
                                                      fork_name),
             'success': True
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         destroy_repo(fork_name)
     def test_api_fork_repo_non_admin(self):
         fork_name = 'api-repo-fork'
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                     repoid=self.REPO,
                                     fork_name=fork_name,
+                                  )
         response = api_call(self, params)
         ret = {
             'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
                                                      fork_name),
             'success': True
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         destroy_repo(fork_name)
     def test_api_fork_repo_non_admin_specify_owner(self):
         fork_name = 'api-repo-fork'
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                     repoid=self.REPO,
                                     fork_name=fork_name,
                                     owner=TEST_USER_ADMIN_LOGIN,
+                                  )
         response = api_call(self, params)
         expected = 'Only RhodeCode admin can specify `owner` param'
         self._compare_error(id_, expected, given=response.body)
         destroy_repo(fork_name)
     def test_api_fork_repo_non_admin_no_permission_to_fork(self):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm='repository.none')
         fork_name = 'api-repo-fork'
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                     repoid=self.REPO,
                                     fork_name=fork_name,
+                                  )
         response = api_call(self, params)
         expected = 'repository `%s` does not exist' % (self.REPO)
         self._compare_error(id_, expected, given=response.body)
         destroy_repo(fork_name)
     def test_api_fork_repo_unknown_owner(self):
         fork_name = 'api-repo-fork'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'fork_repo',
                                     repoid=self.REPO,
                                     fork_name=fork_name,
                                     owner=owner,
+                                  )
         response = api_call(self, params)
         expected = 'user `%s` does not exist' % owner
         self._compare_error(id_, expected, given=response.body)
     def test_api_fork_repo_fork_exists(self):
         fork_name = 'api-repo-fork'
         create_fork(fork_name, self.REPO_TYPE, self.REPO)
         try:
             fork_name = 'api-repo-fork'
             id_, params = _build_data(self.apikey, 'fork_repo',
                                         repoid=self.REPO,
                                         fork_name=fork_name,
                                         owner=TEST_USER_ADMIN_LOGIN,
+                                      )
             response = api_call(self, params)
             expected = "fork `%s` already exist" % fork_name
             self._compare_error(id_, expected, given=response.body)
         finally:
             destroy_repo(fork_name)
     def test_api_fork_repo_repo_exists(self):
         fork_name = self.REPO
         id_, params = _build_data(self.apikey, 'fork_repo',
                                     repoid=self.REPO,
                                     fork_name=fork_name,
                                     owner=TEST_USER_ADMIN_LOGIN,
+                                  )
         response = api_call(self, params)
         expected = "repo `%s` already exist" % fork_name
         self._compare_error(id_, expected, given=response.body)
     @mock.patch.object(RepoModel, 'create_fork', crash)
     def test_api_fork_repo_exception_occurred(self):
         fork_name = 'api-repo-fork'
         id_, params = _build_data(self.apikey, 'fork_repo',
                                     repoid=self.REPO,
                                     fork_name=fork_name,
                                     owner=TEST_USER_ADMIN_LOGIN,
+                                  )
         response = api_call(self, params)
         expected = 'failed to fork repository `%s` as `%s`' % (self.REPO,
                                                                fork_name)
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_users_group(self):
         id_, params = _build_data(self.apikey, 'get_users_group',
                                   usersgroupid=TEST_USERS_GROUP)
         response = api_call(self, params)
         users_group = UsersGroupModel().get_group(TEST_USERS_GROUP)
         members = []
         for user in users_group.members:
             user = user.user
             members.append(user.get_api_data())
         ret = users_group.get_api_data()
         ret['members'] = members
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_users_groups(self):
         make_users_group('test_users_group2')
         id_, params = _build_data(self.apikey, 'get_users_groups',)
         response = api_call(self, params)
         expected = []
         for gr_name in [TEST_USERS_GROUP, 'test_users_group2']:
             users_group = UsersGroupModel().get_group(gr_name)
             ret = users_group.get_api_data()
             expected.append(ret)
         self._compare_ok(id_, expected, given=response.body)
         UsersGroupModel().delete(users_group='test_users_group2')
         Session().commit()
     def test_api_create_users_group(self):
         group_name = 'some_new_group'
         id_, params = _build_data(self.apikey, 'create_users_group',
                                   group_name=group_name)
         response = api_call(self, params)

0 comments (0 inline, 0 general)