added API call for locking/unlocking repositories

docs/api/api.rst

@@ -1,850 +1,876 b''
 .. _api:
 
 ===
 API
 ===
 
 
 Starting from RhodeCode version 1.2 a simple API was implemented.
 There's a single schema for calling all api methods. API is implemented
 with JSON protocol both ways. An url to send API request to RhodeCode is
 <your_server>/_admin/api
 
 API ACCESS FOR WEB VIEWS
 ++++++++++++++++++++++++
 
 API access can also be turned on for each web view in RhodeCode that is 
 decorated with `@LoginRequired` decorator. To enable API access simple change 
 the standard login decorator to `@LoginRequired(api_access=True)`. 
 After this change, a rhodecode view can be accessed without login by adding a 
 GET parameter `?api_key=<api_key>` to url. By default this is only
 enabled on RSS/ATOM feed views.
 
 
 API ACCESS
 ++++++++++
 
 All clients are required to send JSON-RPC spec JSON data::
 
     {   
         "id:"<id>",
         "api_key":"<api_key>",
         "method":"<method_name>",
         "args":{"<arg_key>":"<arg_val>"}
     }
 
 Example call for autopulling remotes repos using curl::
     curl https://server.com/_admin/api -X POST -H 'content-type:text/plain' --data-binary '{"id":1,"api_key":"xe7cdb2v278e4evbdf5vs04v832v0efvcbcve4a3","method":"pull","args":{"repo":"CPython"}}'
 
 Simply provide
  - *id* A value of any type, which is used to match the response with the request that it is replying to.
  - *api_key* for access and permission validation.
  - *method* is name of method to call
  - *args* is an key:value list of arguments to pass to method
 
 .. note::
 
     api_key can be found in your user account page
 
 
 RhodeCode API will return always a JSON-RPC response::
 
     {   
         "id":<id>, # matching id sent by request
         "result": "<result>"|null, # JSON formatted result, null if any errors
         "error": "null"|<error_message> # JSON formatted error (if any)
     }
 
 All responses from API will be `HTTP/1.0 200 OK`, if there's an error while
 calling api *error* key from response will contain failure description
 and result will be null.
 
 
 API CLIENT
 ++++++++++
 
 From version 1.4 RhodeCode adds a script that allows to easily
 communicate with API. After installing RhodeCode a `rhodecode-api` script
 will be available.
 
 To get started quickly simply run::
 
   rhodecode-api _create_config --apikey=<youapikey> --apihost=<rhodecode host>
  
 This will create a file named .config in the directory you executed it storing
 json config file with credentials. You can skip this step and always provide
 both of the arguments to be able to communicate with server
 
 
 after that simply run any api command for example get_repo::
  
  rhodecode-api get_repo
 
  calling {"api_key": "<apikey>", "id": 75, "args": {}, "method": "get_repo"} to http://127.0.0.1:5000
  rhodecode said:
  {'error': 'Missing non optional `repoid` arg in JSON DATA',
   'id': 75,
   'result': None}
 
 Ups looks like we forgot to add an argument
 
 Let's try again now giving the repoid as parameters::
 
     rhodecode-api get_repo repoid:rhodecode   
  
     calling {"api_key": "<apikey>", "id": 39, "args": {"repoid": "rhodecode"}, "method": "get_repo"} to http://127.0.0.1:5000
     rhodecode said:
     {'error': None,
      'id': 39,
      'result': <json data...>}
 
 
 
 API METHODS
 +++++++++++
 
 
 pull
 ----
 
 Pulls given repo from remote location. Can be used to automatically keep
 remote repos up to date. This command can be executed only using api_key
 belonging to user with admin rights
 
 INPUT::
 
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "pull"
     args :    {
                 "repoid" : "<reponame or repo_id>"
               }
 
 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.
+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 :  "lock"
+    args :    {
+                "repoid" : "<reponame or repo_id>"
+                "userid" : "<user_id or username>",
+                "locked" : "<bool true|false>"
+                
+              }
+
+OUTPUT::
+
+    id : <id_given_in_input>
+    result : "User `<username>` set lock state for repo `<reponame>` to `true|false`"
+    error :  null
+
+
 get_user
 --------
 
 Get's an user by username or user_id, Returns empty result if user is not 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 :  "get_user"
     args :    { 
                 "userid" : "<username or user_id>"
               }
 
 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>",
                 "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>",
                 "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)"
               }
 
 OUTPUT::
 
     id : <id_given_in_input>
     result: {
               "msg" : "created new user `<username>`",
               "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>",
               },
             }
     error:  null
 
 
 update_user
 -----------
 
 updates given user if such user exists. 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 :  "update_user"
     args :    {
                 "userid" : "<user_id or username>",
                 "username" :  "<username> = Optional",
                 "email" :     "<useremail> = Optional",
                 "password" :  "<password> = Optional",
                 "firstname" : "<firstname> = Optional",
                 "lastname" :  "<lastname> = Optional",
                 "active" :    "<bool> = Optional",
                 "admin" :     "<bool> = Optional",
                 "ldap_dn" :   "<ldap_dn> = Optional"
               }
 
 OUTPUT::
 
     id : <id_given_in_input>
     result: {
               "msg" : "updated user ID:<userid> <username>",
               "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>",
               },              
             }
     error:  null
 
 
 delete_user
 -----------
 
 
 deletes givenuser if such user exists. 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 :  "delete_user"
     args :    {
                 "userid" : "<user_id or username>",
               }
 
 OUTPUT::
 
     id : <id_given_in_input>
     result: {
               "msg" : "deleted user ID:<userid> <username>",
               "user": null
             }
     error:  null
 
 
 get_users_group
 ---------------
 
 Gets an existing 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 :  "get_users_group"
     args :    {
                 "usersgroupid" : "<users group id or name>"
               }
 
 OUTPUT::
 
     id : <id_given_in_input>
     result : None if group not exist
              {
                "users_group_id" : "<id>",
                "group_name" :     "<groupname>",
                "active":          "<bool>",
                "members" :  [
                               { 
                                 "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>",
                               },
                               …
                             ]
              }
     error : null
 
 
 get_users_groups
 ----------------
 
 Lists all existing users groups. 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_groups"
     args :    { }
 
 OUTPUT::
 
     id : <id_given_in_input>
     result : [
                {
                "users_group_id" : "<id>",
                "group_name" :     "<groupname>",
                "active":          "<bool>",
                "members" :  [
                               { 
                                 "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>",
                               },
                               …
                             ]
                },
                …
               ]
     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>",
                      "members" :  [
                                   { 
                                     "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>",
                                   },
                                   …
                      ]
                },
             }
     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.
 
 
 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>",
                 "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
 
 
 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>",
               },
               …
             ]
     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.
 
 
 INPUT::
 
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "create_repo"
     args:     {
                 "repo_name" :   "<reponame>",
                 "owner" :       "<onwer_name_or_id>",
                 "repo_type" :   "<repo_type>",
                 "description" : "<description> = Optional('')",
                 "private" :     "<bool> = Optional(False)",
                 "clone_uri" :   "<clone_uri> = Optional(None)",
                 "landing_rev" : "<landing_rev> = Optional('tip')",
               }
 
 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":        "<repo_owner>",
                 "fork_of":  "<name_of_fork_parent>",
               },
             }
     error:  null
 
 
 delete_repo
 -----------
 
 Deletes a 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 :  "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))",
               }
 
 OUTPUT::
 
     id : <id_given_in_input>
     result: {
               "msg" : "Granted perm: `<perm>` for group: `<usersgroupname>` in repo: `<reponame>`",
               "success": true
             }
     error:  null
     
     
 revoke_users_group_permission
 -----------------------------
 
 Revoke permission for users group 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_users_group_permission"
     args:     {
                 "repoid" : "<reponame or repo_id>"
                 "usersgroupid" : "<users group id or name>"
               }
 
 OUTPUT::
 
     id : <id_given_in_input>
     result: {
               "msg" : "Revoked perm for group: `<usersgroupname>` in repo: `<reponame>`",
               "success": true
             }
     error:  null
  No newline at end of file

rhodecode/controllers/api/api.py

@@ -1,784 +1,812 b''
 # -*- coding: utf-8 -*-
 """
     rhodecode.controllers.api
     ~~~~~~~~~~~~~~~~~~~~~~~~~
 
     API controller for RhodeCode
 
     :created_on: Aug 20, 2011
     :author: marcink
     :copyright: (C) 2011-2012 Marcin Kuzminski <[e-mail]>
     :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 traceback
 import logging
 
 from rhodecode.controllers.api import JSONRPCController, JSONRPCError
 from rhodecode.lib.auth import HasPermissionAllDecorator, \
     HasPermissionAnyDecorator, PasswordGenerator, AuthUser
 from rhodecode.lib.utils import map_groups, repo2db_mapper
 from rhodecode.model.meta import Session
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 from rhodecode.model.users_group import UsersGroupModel
 from rhodecode.model.permission import PermissionModel
+from rhodecode.model.db import Repository
 
 log = logging.getLogger(__name__)
 
 
 class Optional(object):
     """
     Defines an optional parameter::
 
         param = param.getval() if isinstance(param, Optional) else param
         param = param() if isinstance(param, Optional) else param
 
     is equivalent of::
 
         param = Optional.extract(param)
 
     """
     def __init__(self, type_):
         self.type_ = type_
 
     def __repr__(self):
         return '<Optional:%s>' % self.type_.__repr__()
 
     def __call__(self):
         return self.getval()
 
     def getval(self):
         """
         returns value from this Optional instance
         """
         return self.type_
 
     @classmethod
     def extract(cls, val):
         if isinstance(val, cls):
             return val.getval()
         return val
 
 
 def get_user_or_error(userid):
     """
     Get user by id or name or return JsonRPCError if not found
 
     :param userid:
     """
     user = UserModel().get_user(userid)
     if user is None:
         raise JSONRPCError("user `%s` does not exist" % userid)
     return user
 
 
 def get_repo_or_error(repoid):
     """
     Get repo by id or name or return JsonRPCError if not found
 
     :param userid:
     """
     repo = RepoModel().get_repo(repoid)
     if repo is None:
         raise JSONRPCError('repository `%s` does not exist' % (repoid))
     return repo
 
 
 def get_users_group_or_error(usersgroupid):
     """
     Get users group by id or name or return JsonRPCError if not found
 
     :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(
-                'Unable to rescan repositories'
+                'Error occurred during rescan repositories action'
+            )
+
+    @HasPermissionAllDecorator('hg.admin')
+    def lock(self, apiuser, repoid, userid, locked):
+        """
+        Set locking state on particular repository by given user
+
+        :param apiuser:
+        :param repoid:
+        :param userid:
+        :param locked:
+        """
+        repo = get_repo_or_error(repoid)
+        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 get_user(self, apiuser, userid):
         """"
         Get a user by username
 
         :param apiuser:
         :param userid:
         """
 
         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 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):
             if not isinstance(attr, Optional):
                 updates[name] = attr
 
         try:
 
             store_update(username, 'username')
             store_update(password, 'password')
             store_update(email, 'email')
             store_update(firstname, 'name')
             store_update(lastname, 'lastname')
             store_update(active, 'active')
             store_update(admin, 'admin')
             store_update(ldap_dn, 'ldap_dn')
 
             user = UserModel().update_user(user, **updates)
             Session().commit()
             return dict(
                 msg='updated user ID:%s %s' % (user.user_id, user.username),
                 user=user.get_api_data()
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update user `%s`' % userid)
 
     @HasPermissionAllDecorator('hg.admin')
     def delete_user(self, apiuser, userid):
         """"
         Deletes an user
 
         :param apiuser:
         :param userid:
         """
         user = get_user_or_error(userid)
 
         try:
             UserModel().delete(userid)
             Session().commit()
             return dict(
                 msg='deleted user ID:%s %s' % (user.user_id, user.username),
                 user=None
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to delete ID:%s %s' % (user.user_id,
                                                               user.username))
 
     @HasPermissionAllDecorator('hg.admin')
     def get_users_group(self, apiuser, usersgroupid):
         """"
         Get users group by name or id
 
         :param apiuser:
         :param usersgroupid:
         """
         users_group = get_users_group_or_error(usersgroupid)
 
         data = users_group.get_api_data()
 
         members = []
         for user in users_group.members:
             user = user.user
             members.append(user.get_api_data())
         data['members'] = members
         return data
 
     @HasPermissionAllDecorator('hg.admin')
     def get_users_groups(self, apiuser):
         """"
         Get all users groups
 
         :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
                     )
             )
 
     @HasPermissionAnyDecorator('hg.admin')
     def get_repo(self, apiuser, repoid):
         """"
         Get repository by name
 
         :param apiuser:
         :param repoid:
         """
         repo = get_repo_or_error(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
 
     @HasPermissionAnyDecorator('hg.admin')
     def get_repos(self, apiuser):
         """"
         Get all repositories
 
         :param apiuser:
         """
 
         result = []
         for repo in RepoModel().get_all():
             result.append(repo.get_api_data())
         return result
 
     @HasPermissionAnyDecorator('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,
                     description=Optional(''), private=Optional(False),
                     clone_uri=Optional(None), landing_rev=Optional('tip')):
         """
         Create repository, if clone_url is given it makes a remote clone
         if repo_name is withina  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:
         """
         owner = get_user_or_error(owner)
 
         if RepoModel().get_by_repo_name(repo_name):
             raise JSONRPCError("repo `%s` already exist" % repo_name)
 
         private = Optional.extract(private)
         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,
             )
 
             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)
 
     @HasPermissionAnyDecorator('hg.admin')
     def fork_repo(self, apiuser, repoid, fork_name, owner,
                   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))
 
         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)
             )
 
     @HasPermissionAnyDecorator('hg.admin')
     def delete_repo(self, apiuser, repoid):
         """
         Deletes a given repository
 
         :param apiuser:
         :param repoid:
         """
         repo = get_repo_or_error(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
             )
 
     @HasPermissionAnyDecorator('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
                 )
             )
 
     @HasPermissionAnyDecorator('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
                 )
             )
 
     @HasPermissionAnyDecorator('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,
                                                      perm=perm)
 
             Session().commit()
             return dict(
                 msg='Granted perm: `%s` for users group: `%s` in '
                     'repo: `%s`' % (
                     perm.permission_name, users_group.users_group_name,
                     repo.repo_name
                 ),
                 success=True
             )
         except Exception:
             print traceback.format_exc()
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for users group: `%s` in '
                 'repo: `%s`' % (
                     usersgroupid, repo.repo_name
                 )
             )
 
     @HasPermissionAnyDecorator('hg.admin')
     def revoke_users_group_permission(self, apiuser, repoid, usersgroupid):
         """
         Revoke permission for users group on given repository
 
         :param apiuser:
         :param repoid:
         :param usersgroupid:
         """
         repo = get_repo_or_error(repoid)
         users_group = get_users_group_or_error(usersgroupid)
 
         try:
             RepoModel().revoke_users_group_permission(repo=repo,
                                                       group_name=users_group)
 
             Session().commit()
             return dict(
                 msg='Revoked perm for users group: `%s` in repo: `%s`' % (
                     users_group.users_group_name, repo.repo_name
                 ),
                 success=True
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for users group: `%s` in '
                 'repo: `%s`' % (
                     users_group.users_group_name, repo.repo_name
                 )
             )

rhodecode/tests/api/api_base.py

@@ -1,1008 +1,1043 b''
 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 !')
 
 
 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):
     # create new repo
     form_data = dict(repo_name=repo_name,
                      repo_name_full=repo_name,
                      fork_name=None,
                      description='description %s' % repo_name,
                      repo_group=None,
                      private=False,
                      repo_type=repo_type,
                      clone_uri=None,
                      landing_rev='tip')
     cur_user = UserModel().get_by_username(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(
             username='test-api',
             password='test',
             email='[e-mail]',
             firstname='first',
             lastname='last'
         )
         Session().commit()
         self.TEST_USER_LOGIN = self.TEST_USER.username
 
     @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 = self.app.post(API_URL, content_type='application/json',
                                  params=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')
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = 'Missing non optional `userid` 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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = "user `%s` does not exist" % 'trololo'
         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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
-        expected = 'Unable to rescan repositories'
+        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 = self.app.post(API_URL, content_type='application/json',
+                                 params=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_release(self):
+        id_, params = _build_data(self.apikey, 'lock',
+                                  userid=TEST_USER_ADMIN_LOGIN,
+                                  repoid=self.REPO,
+                                  locked=False)
+        response = self.app.post(API_URL, content_type='application/json',
+                                 params=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)
+
+    @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 = self.app.post(API_URL, content_type='application/json',
+                                 params=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='[e-mail]',
                                   password='trololo')
         response = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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)
         self.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 = self.app.post(API_URL, content_type='application/json',
                                  params=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'[e-mail]',
                                            firstname=u'u1', lastname=u'u1')
         self.Session().commit()
         username = usr.username
         email = usr.email
         usr_id = usr.user_id
         ## DELETE THIS USER NOW
 
         id_, params = _build_data(self.apikey, 'delete_user',
                                   userid=username,)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         ret = {'msg': 'deleted user ID:%s %s' % (usr_id, username),
                'user': None}
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
 
     @mock.patch.object(UserModel, 'delete', crash)
     def test_api_delete_user_when_exception_happened(self):
         usr = UserModel().create_or_update(username=u'test_user',
                                            password=u'qweqwe',
                                            email=u'[e-mail]',
                                            firstname=u'u1', lastname=u'u1')
         self.Session().commit()
         username = usr.username
 
         id_, params = _build_data(self.apikey, 'delete_user',
                                   userid=username,)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
         ret = 'failed to delete ID:%s %s' % (usr.user_id,
                                              usr.username)
         expected = ret
         self._compare_error(id_, expected, given=response.body)
 
     @parameterized.expand([('firstname', 'new_username'),
                            ('lastname', 'new_username'),
                            ('email', 'new_username'),
                            ('admin', True),
                            ('admin', False),
                            ('ldap_dn', 'test'),
                            ('ldap_dn', None),
                            ('active', False),
                            ('active', True),
                            ('password', 'newpass')
                            ])
     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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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')
         self.Session().commit()
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid=self.REPO)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=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_that_doesn_not_exist(self):
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid='no-such-repo')
         response = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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)
 
     @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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=params)
         expected = 'user `%s` does not exist' % owner
         self._compare_error(id_, expected, given=response.body)
 
     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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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_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 = self.app.post(API_URL, content_type='application/json',
                                          params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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_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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                      params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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 = self.app.post(API_URL, content_type='application/json',
                                  params=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')
         self.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 = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         ret = {
             'msg': 'created new users group `%s`' % group_name,
             'users_group': jsonify(UsersGroupModel()\
                                    .get_by_name(group_name)\
                                    .get_api_data())
         }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
 
         destroy_users_group(group_name)
 
     def test_api_get_users_group_that_exist(self):
         id_, params = _build_data(self.apikey, 'create_users_group',
                                   group_name=TEST_USERS_GROUP)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = "users group `%s` already exist" % TEST_USERS_GROUP
         self._compare_error(id_, expected, given=response.body)
 
     @mock.patch.object(UsersGroupModel, 'create', crash)
     def test_api_get_users_group_exception_occurred(self):
         group_name = 'exception_happens'
         id_, params = _build_data(self.apikey, 'create_users_group',
                                   group_name=group_name)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = 'failed to create group `%s`' % group_name
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_add_user_to_users_group(self):
         gr_name = 'test_group'
         UsersGroupModel().create(gr_name)
         self.Session().commit()
         id_, params = _build_data(self.apikey, 'add_user_to_users_group',
                                   usersgroupid=gr_name,
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = {
                     'msg': 'added member `%s` to users group `%s`' % (
                                 TEST_USER_ADMIN_LOGIN, gr_name
                             ),
                     'success': True}
         self._compare_ok(id_, expected, given=response.body)
 
         UsersGroupModel().delete(users_group=gr_name)
         self.Session().commit()
 
     def test_api_add_user_to_users_group_that_doesnt_exist(self):
         id_, params = _build_data(self.apikey, 'add_user_to_users_group',
                                   usersgroupid='false-group',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = 'users group `%s` does not exist' % 'false-group'
         self._compare_error(id_, expected, given=response.body)
 
     @mock.patch.object(UsersGroupModel, 'add_user_to_group', crash)
     def test_api_add_user_to_users_group_exception_occurred(self):
         gr_name = 'test_group'
         UsersGroupModel().create(gr_name)
         self.Session().commit()
         id_, params = _build_data(self.apikey, 'add_user_to_users_group',
                                   usersgroupid=gr_name,
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = 'failed to add member to users group `%s`' % gr_name
         self._compare_error(id_, expected, given=response.body)
 
         UsersGroupModel().delete(users_group=gr_name)
         self.Session().commit()
 
     def test_api_remove_user_from_users_group(self):
         gr_name = 'test_group_3'
         gr = UsersGroupModel().create(gr_name)
         UsersGroupModel().add_user_to_group(gr, user=TEST_USER_ADMIN_LOGIN)
         self.Session().commit()
         id_, params = _build_data(self.apikey, 'remove_user_from_users_group',
                                   usersgroupid=gr_name,
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = {
                     'msg': 'removed member `%s` from users group `%s`' % (
                                 TEST_USER_ADMIN_LOGIN, gr_name
                             ),
                     'success': True}
         self._compare_ok(id_, expected, given=response.body)
 
         UsersGroupModel().delete(users_group=gr_name)
         self.Session().commit()
 
     @mock.patch.object(UsersGroupModel, 'remove_user_from_group', crash)
     def test_api_remove_user_from_users_group_exception_occurred(self):
         gr_name = 'test_group_3'
         gr = UsersGroupModel().create(gr_name)
         UsersGroupModel().add_user_to_group(gr, user=TEST_USER_ADMIN_LOGIN)
         self.Session().commit()
         id_, params = _build_data(self.apikey, 'remove_user_from_users_group',
                                   usersgroupid=gr_name,
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = 'failed to remove member from users group `%s`' % gr_name
         self._compare_error(id_, expected, given=response.body)
 
         UsersGroupModel().delete(users_group=gr_name)
         self.Session().commit()
 
     @parameterized.expand([('none', 'repository.none'),
                            ('read', 'repository.read'),
                            ('write', 'repository.write'),
                            ('admin', 'repository.admin')])
     def test_api_grant_user_permission(self, name, perm):
         id_, params = _build_data(self.apikey, 'grant_user_permission',
                                   repoid=self.REPO,
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   perm=perm)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         ret = {
                 'msg': 'Granted perm: `%s` for user: `%s` in repo: `%s`' % (
                     perm, TEST_USER_ADMIN_LOGIN, self.REPO
                 ),
                 'success': True
             }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_grant_user_permission_wrong_permission(self):
         perm = 'haha.no.permission'
         id_, params = _build_data(self.apikey, 'grant_user_permission',
                                   repoid=self.REPO,
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   perm=perm)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = 'permission `%s` does not exist' % perm
         self._compare_error(id_, expected, given=response.body)
 
     @mock.patch.object(RepoModel, 'grant_user_permission', crash)
     def test_api_grant_user_permission_exception_when_adding(self):
         perm = 'repository.read'
         id_, params = _build_data(self.apikey, 'grant_user_permission',
                                   repoid=self.REPO,
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   perm=perm)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
                     TEST_USER_ADMIN_LOGIN, self.REPO
                 )
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_revoke_user_permission(self):
         id_, params = _build_data(self.apikey, 'revoke_user_permission',
                                   repoid=self.REPO,
                                   userid=TEST_USER_ADMIN_LOGIN,)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = {
             'msg': 'Revoked perm for user: `%s` in repo: `%s`' % (
                 TEST_USER_ADMIN_LOGIN, self.REPO
             ),
             'success': True
         }
         self._compare_ok(id_, expected, given=response.body)
 
     @mock.patch.object(RepoModel, 'revoke_user_permission', crash)
     def test_api_revoke_user_permission_exception_when_adding(self):
         id_, params = _build_data(self.apikey, 'revoke_user_permission',
                                   repoid=self.REPO,
                                   userid=TEST_USER_ADMIN_LOGIN,)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
                     TEST_USER_ADMIN_LOGIN, self.REPO
                 )
         self._compare_error(id_, expected, given=response.body)
 
     @parameterized.expand([('none', 'repository.none'),
                            ('read', 'repository.read'),
                            ('write', 'repository.write'),
                            ('admin', 'repository.admin')])
     def test_api_grant_users_group_permission(self, name, perm):
         id_, params = _build_data(self.apikey, 'grant_users_group_permission',
                                   repoid=self.REPO,
                                   usersgroupid=TEST_USERS_GROUP,
                                   perm=perm)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         ret = {
             'msg': 'Granted perm: `%s` for users group: `%s` in repo: `%s`' % (
                 perm, TEST_USERS_GROUP, self.REPO
             ),
             'success': True
         }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_grant_users_group_permission_wrong_permission(self):
         perm = 'haha.no.permission'
         id_, params = _build_data(self.apikey, 'grant_users_group_permission',
                                   repoid=self.REPO,
                                   usersgroupid=TEST_USERS_GROUP,
                                   perm=perm)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = 'permission `%s` does not exist' % perm
         self._compare_error(id_, expected, given=response.body)
 
     @mock.patch.object(RepoModel, 'grant_users_group_permission', crash)
     def test_api_grant_users_group_permission_exception_when_adding(self):
         perm = 'repository.read'
         id_, params = _build_data(self.apikey, 'grant_users_group_permission',
                                   repoid=self.REPO,
                                   usersgroupid=TEST_USERS_GROUP,
                                   perm=perm)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = 'failed to edit permission for users group: `%s` in repo: `%s`' % (
                     TEST_USERS_GROUP, self.REPO
                 )
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_revoke_users_group_permission(self):
         RepoModel().grant_users_group_permission(repo=self.REPO,
                                                  group_name=TEST_USERS_GROUP,
                                                  perm='repository.read')
         self.Session().commit()
         id_, params = _build_data(self.apikey, 'revoke_users_group_permission',
                                   repoid=self.REPO,
                                   usersgroupid=TEST_USERS_GROUP,)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = {
             'msg': 'Revoked perm for users group: `%s` in repo: `%s`' % (
                 TEST_USERS_GROUP, self.REPO
             ),
             'success': True
         }
         self._compare_ok(id_, expected, given=response.body)
 
     @mock.patch.object(RepoModel, 'revoke_users_group_permission', crash)
     def test_api_revoke_users_group_permission_exception_when_adding(self):
 
         id_, params = _build_data(self.apikey, 'revoke_users_group_permission',
                                   repoid=self.REPO,
                                   usersgroupid=TEST_USERS_GROUP,)
         response = self.app.post(API_URL, content_type='application/json',
                                  params=params)
 
         expected = 'failed to edit permission for users group: `%s` in repo: `%s`' % (
                     TEST_USERS_GROUP, self.REPO
                 )
         self._compare_error(id_, expected, given=response.body)