kallithea Changeset - 647fb653048e

Changeset - 647fb653048e

Parent rev.

Child rev.

[Not reviewed]

beta

0 4 0

Marcin Kuzminski - 13 years ago 2013-05-06 23:35:29
marcin@python-works.com

make the password optional in API calls

4 files changed with 24 insertions and 3 deletions:

docs/api/api.rst

rhodecode/controllers/api/api.py

rhodecode/model/user.py

rhodecode/tests/api/api_base.py

0 comments (0 inline, 0 general)

docs/api/api.rst

➞

Show inline comments

@@ @@ -165,385 +165,385 @@ OUTPUT:: @@
     result : "{'added': [<list of names of added repos>],
                'removed': [<list of names of removed repos>]}"
     error :  null
 invalidate_cache
 ----------------
 Invalidate cache for repository.
 This command can be executed only using api_key belonging to user with admin
 rights or regular user that have write or admin or write access to repository.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "invalidate_cache"
     args :    {
                 "repoid" : "<reponame or repo_id>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result : "Caches of repository `<reponame>`"
     error :  null
 lock
 ----
 Set locking state on given repository by given user. If userid param is skipped
 , then it is set to id of user whos calling this method. If locked param is skipped
 then function shows current lock state of given repo.
 This command can be executed only using api_key belonging to user with admin
 rights or regular user that have admin or write access to repository.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "lock"
     args :    {
                 "repoid" : "<reponame or repo_id>"
                 "userid" : "<user_id or username = Optional(=apiuser)>",
                 "locked" : "<bool true|false = Optional(=None)>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result : {
                  "repo": "<reponame>",
                  "locked": "<bool true|false>",
                  "locked_since": "<float lock_time>",
                  "locked_by": "<username>",
                  "msg": "User `<username>` set lock state for repo `<reponame>` to `<false|true>`"
+             }
     error :  null
 show_ip
 -------
 Shows IP address as seen from RhodeCode server, together with all
 defined IP addresses for given user.
 This command can be executed only using api_key belonging to user with admin
 rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "show_ip"
     args :    {
                 "userid" : "<user_id or username>",
+              }
 OUTPUT::
     id : <id_given_in_input>
     result : {
                  "ip_addr_server": <ip_from_clien>",
                  "user_ips": [
+                                {
                                    "ip_addr": "<ip_with_mask>",
                                    "ip_range": ["<start_ip>", "<end_ip>"],
                                 },
                                 ...
+                             ]
+             }
     error :  null
 get_user
 --------
 Get's an user by username or user_id, Returns empty result if user is not found.
 If userid param is skipped it is set to id of user who is calling this method.
 This command can be executed only using api_key belonging to user with admin
 rights, or regular users that cannot specify different userid than theirs
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "get_user"
     args :    {
                 "userid" : "<username or user_id Optional(=apiuser)>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: None if user does not exist or
+            {
                 "user_id" :     "<user_id>",
                 "api_key" :     "<api_key>",
                 "username" :    "<username>",
                 "firstname":    "<firstname>",
                 "lastname" :    "<lastname>",
                 "email" :       "<email>",
                 "emails":       "<list_of_all_additional_emails>",
                 "ip_addresses": "<list_of_ip_addresses_for_user>",
                 "active" :      "<bool>",
                 "admin" :       "<bool>",
                 "ldap_dn" :     "<ldap_dn>",
                 "last_login":   "<last_login>",
                 "permissions": {
                     "global": ["hg.create.repository",
                                "repository.read",
                                "hg.register.manual_activate"],
                     "repositories": {"repo1": "repository.none"},
                     "repositories_groups": {"Group1": "group.read"}
                  },
+            }
     error:  null
 get_users
 ---------
 Lists all existing users. This command can be executed only using api_key
 belonging to user with admin rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "get_users"
     args :    { }
 OUTPUT::
     id : <id_given_in_input>
     result: [
+              {
                 "user_id" :     "<user_id>",
                 "username" :    "<username>",
                 "firstname":    "<firstname>",
                 "lastname" :    "<lastname>",
                 "email" :       "<email>",
                 "emails":       "<list_of_all_additional_emails>",
                 "ip_addresses": "<list_of_ip_addresses_for_user>",
                 "active" :      "<bool>",
                 "admin" :       "<bool>",
                 "ldap_dn" :     "<ldap_dn>",
                 "last_login":   "<last_login>",
               },
               …
+            ]
     error:  null
 create_user
 -----------
 Creates new user. This command can
 be executed only using api_key belonging to user with admin rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "create_user"
     args :    {
                 "username" :  "<username>",
                 "email" :     "<useremail>",
                 "password" :  "<password>",
+                "password" :  "<password = Optional(None)>",
                 "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(None)",
                 "email" :     "<useremail> = Optional(None)",
                 "password" :  "<password> = Optional(None)",
                 "firstname" : "<firstname> = Optional(None)",
                 "lastname" :  "<lastname> = Optional(None)",
                 "active" :    "<bool> = Optional(None)",
                 "admin" :     "<bool> = Optional(None)",
                 "ldap_dn" :   "<ldap_dn> = Optional(None)"
+              }
 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 user 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" : "<user 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 user 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>",
                },
                …
+              ]
     error : null
 create_users_group
 ------------------
 Creates new user 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>",
                 "owner" :     "<onwer_name_or_id = Optional(=apiuser)>",
                 "active":     "<bool> = Optional(True)"
+              }
 OUTPUT::
     id : <id_given_in_input>

rhodecode/controllers/api/api.py

➞

Show inline comments

@@ @@ -218,385 +218,385 @@ class ApiController(JSONRPCController): @@
             # check if we have admin permission for this repo !
             if HasRepoPermissionAnyApi('repository.admin',
                                        'repository.write')(user=apiuser,
                                             repo_name=repo.repo_name) is False:
                 raise JSONRPCError('repository `%s` does not exist' % (repoid))
         try:
             ScmModel().mark_for_invalidation(repo.repo_name)
             return ('Caches of repository `%s` was invalidated' % repoid)
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Error occurred during cache invalidation action'
+            )
     def lock(self, apiuser, repoid, locked=Optional(None),
              userid=Optional(OAttr('apiuser'))):
         """
         Set locking state on particular repository by given user, if
         this command is runned by non-admin account userid is set to user
         who is calling this method
         :param apiuser:
         :param repoid:
         :param userid:
         :param locked:
         """
         repo = get_repo_or_error(repoid)
         if HasPermissionAnyApi('hg.admin')(user=apiuser):
             pass
         elif HasRepoPermissionAnyApi('repository.admin',
                                      'repository.write')(user=apiuser,
                                                          repo_name=repo.repo_name):
             #make sure normal user does not pass someone else userid,
             #he is not allowed to do that
             if not isinstance(userid, Optional) and userid != apiuser.user_id:
                 raise JSONRPCError(
                     'userid is not the same as your user'
+                )
         else:
             raise JSONRPCError('repository `%s` does not exist' % (repoid))
         if isinstance(userid, Optional):
             userid = apiuser.user_id
         user = get_user_or_error(userid)
         if isinstance(locked, Optional):
             lockobj = Repository.getlock(repo)
             if lockobj[0] is None:
                 _d = {
                     'repo': repo.repo_name,
                     'locked': False,
                     'locked_since': None,
                     'locked_by': None,
                     'msg': 'Repo `%s` not locked.' % repo.repo_name
+                }
                 return _d
             else:
                 userid, time_ = lockobj
                 lock_user = get_user_or_error(userid)
                 _d = {
                     'repo': repo.repo_name,
                     'locked': True,
                     'locked_since': time_,
                     'locked_by': lock_user.username,
                     'msg': ('Repo `%s` locked by `%s`. '
                             % (repo.repo_name,
                                json.dumps(time_to_datetime(time_))))
+                }
                 return _d
         # force locked state through a flag
         else:
             locked = str2bool(locked)
             try:
                 if locked:
                     lock_time = time.time()
                     Repository.lock(repo, user.user_id, lock_time)
                 else:
                     lock_time = None
                     Repository.unlock(repo)
                 _d = {
                     'repo': repo.repo_name,
                     'locked': locked,
                     'locked_since': lock_time,
                     'locked_by': user.username,
                     'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                             % (user.username, repo.repo_name, locked))
+                }
                 return _d
             except Exception:
                 log.error(traceback.format_exc())
                 raise JSONRPCError(
                     'Error occurred locking repository `%s`' % repo.repo_name
+                )
     def get_locks(self, apiuser, userid=Optional(OAttr('apiuser'))):
         """
         Get all locks for given userid, if
         this command is runned by non-admin account userid is set to user
         who is calling this method, thus returning locks for himself
         :param apiuser:
         :param userid:
         """
         if HasPermissionAnyApi('hg.admin')(user=apiuser):
             pass
         else:
             #make sure normal user does not pass someone else userid,
             #he is not allowed to do that
             if not isinstance(userid, Optional) and userid != apiuser.user_id:
                 raise JSONRPCError(
                     'userid is not the same as your user'
+                )
         ret = []
         if isinstance(userid, Optional):
             user = None
         else:
             user = get_user_or_error(userid)
         #show all locks
         for r in Repository.getAll():
             userid, time_ = r.locked
             if time_:
                 _api_data = r.get_api_data()
                 # if we use userfilter just show the locks for this user
                 if user:
                     if safe_int(userid) == user.user_id:
                         ret.append(_api_data)
                 else:
                     ret.append(_api_data)
         return ret
     @HasPermissionAllDecorator('hg.admin')
     def show_ip(self, apiuser, userid):
         """
         Shows IP address as seen from RhodeCode server, together with all
         defined IP addresses for given user
         :param apiuser:
         :param userid:
         """
         user = get_user_or_error(userid)
         ips = UserIpMap.query().filter(UserIpMap.user == user).all()
         return dict(
             ip_addr_server=self.ip_addr,
             user_ips=ips
+        )
     def get_user(self, apiuser, userid=Optional(OAttr('apiuser'))):
         """"
         Get a user by username, or userid, if userid is given
         :param apiuser:
         :param userid:
         """
         if HasPermissionAnyApi('hg.admin')(user=apiuser) is False:
             #make sure normal user does not pass someone else userid,
             #he is not allowed to do that
             if not isinstance(userid, Optional) and userid != apiuser.user_id:
                 raise JSONRPCError(
                     'userid is not the same as your user'
+                )
         if isinstance(userid, Optional):
             userid = apiuser.user_id
         user = get_user_or_error(userid)
         data = user.get_api_data()
         data['permissions'] = AuthUser(user_id=user.user_id).permissions
         return data
     @HasPermissionAllDecorator('hg.admin')
     def get_users(self, apiuser):
         """"
         Get all users
         :param apiuser:
         """
         result = []
         users_list = User.query().order_by(User.username)\
                         .filter(User.username != User.DEFAULT_USER)\
                         .all()
         for user in users_list:
             result.append(user.get_api_data())
         return result
     @HasPermissionAllDecorator('hg.admin')
     def create_user(self, apiuser, username, email, password,
+    def create_user(self, apiuser, username, email, password=Optional(None),
                     firstname=Optional(None), lastname=Optional(None),
                     active=Optional(True), admin=Optional(False),
                     ldap_dn=Optional(None)):
         """
         Create new user
         :param apiuser:
         :param username:
         :param email:
         :param password:
         :param firstname:
         :param lastname:
         :param active:
         :param admin:
         :param ldap_dn:
         """
         if UserModel().get_by_username(username):
             raise JSONRPCError("user `%s` already exist" % username)
         if UserModel().get_by_email(email, case_insensitive=True):
             raise JSONRPCError("email `%s` already exist" % email)
         if Optional.extract(ldap_dn):
             # generate temporary password if ldap_dn
             password = PasswordGenerator().gen_password(length=8)
         try:
             user = UserModel().create_or_update(
                 username=Optional.extract(username),
                 password=Optional.extract(password),
                 email=Optional.extract(email),
                 firstname=Optional.extract(firstname),
                 lastname=Optional.extract(lastname),
                 active=Optional.extract(active),
                 admin=Optional.extract(admin),
                 ldap_dn=Optional.extract(ldap_dn)
+            )
             Session().commit()
             return dict(
                 msg='created new user `%s`' % username,
                 user=user.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create user `%s`' % username)
     @HasPermissionAllDecorator('hg.admin')
     def update_user(self, apiuser, userid, username=Optional(None),
                     email=Optional(None), firstname=Optional(None),
                     lastname=Optional(None), active=Optional(None),
                     admin=Optional(None), ldap_dn=Optional(None),
                     password=Optional(None)):
         """
         Updates given user
         :param apiuser:
         :param userid:
         :param username:
         :param email:
         :param firstname:
         :param lastname:
         :param active:
         :param admin:
         :param ldap_dn:
         :param password:
         """
         user = get_user_or_error(userid)
         # call function and store only updated arguments
         updates = {}
         def store_update(attr, name):
             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 DefaultUserException:
             log.error(traceback.format_exc())
             raise JSONRPCError('editing default user is forbidden')
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update user `%s`' % userid)
     @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 user 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 user groups
         :param apiuser:
         """
         result = []
         for users_group in UserGroupModel().get_all():
             result.append(users_group.get_api_data())
         return result
     @HasPermissionAllDecorator('hg.admin')
     def create_users_group(self, apiuser, group_name,
                            owner=Optional(OAttr('apiuser')),
                            active=Optional(True)):
         """
         Creates an new usergroup
         :param apiuser:
         :param group_name:
         :param owner:
         :param active:
         """
         if UserGroupModel().get_by_name(group_name):
             raise JSONRPCError("user group `%s` already exist" % group_name)
         try:
             if isinstance(owner, Optional):
                 owner = apiuser.user_id
             owner = get_user_or_error(owner)
             active = Optional.extract(active)
             ug = UserGroupModel().create(name=group_name,
                                          owner=owner,
                                          active=active)
             Session().commit()
             return dict(
                 msg='created new user 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):
         """"

rhodecode/model/user.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.model.user
     ~~~~~~~~~~~~~~~~~~~~
     users model for RhodeCode
     :created_on: Apr 9, 2010
     :author: marcink
     :copyright: (C) 2010-2012 Marcin Kuzminski <marcin@python-works.com>
     :license: GPLv3, see COPYING for more details.
 """
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 import logging
 import traceback
 import itertools
 import collections
 from pylons import url
 from pylons.i18n.translation import _
 from sqlalchemy.exc import DatabaseError
 from sqlalchemy.orm import joinedload
 from rhodecode.lib.utils2 import safe_unicode, generate_api_key
 from rhodecode.lib.caching_query import FromCache
 from rhodecode.model import BaseModel
 from rhodecode.model.db import User, UserRepoToPerm, Repository, Permission, \
     UserToPerm, UserGroupRepoToPerm, UserGroupToPerm, UserGroupMember, \
     Notification, RepoGroup, UserRepoGroupToPerm, UserGroupRepoGroupToPerm, \
     UserEmailMap, UserIpMap, UserGroupUserGroupToPerm, UserGroup
 from rhodecode.lib.exceptions import DefaultUserException, \
     UserOwnsReposException
 from rhodecode.model.meta import Session
 log = logging.getLogger(__name__)
 PERM_WEIGHTS = Permission.PERM_WEIGHTS
 class UserModel(BaseModel):
     cls = User
     def get(self, user_id, cache=False):
         user = self.sa.query(User)
         if cache:
             user = user.options(FromCache("sql_cache_short",
                                           "get_user_%s" % user_id))
         return user.get(user_id)
     def get_user(self, user):
         return self._get_user(user)
     def get_by_username(self, username, cache=False, case_insensitive=False):
         if case_insensitive:
             user = self.sa.query(User).filter(User.username.ilike(username))
         else:
             user = self.sa.query(User)\
                 .filter(User.username == username)
         if cache:
             user = user.options(FromCache("sql_cache_short",
                                           "get_user_%s" % username))
         return user.scalar()
     def get_by_email(self, email, cache=False, case_insensitive=False):
         return User.get_by_email(email, case_insensitive, cache)
     def get_by_api_key(self, api_key, cache=False):
         return User.get_by_api_key(api_key, cache)
     def create(self, form_data):
         from rhodecode.lib.auth import get_crypt_password
         try:
             new_user = User()
             for k, v in form_data.items():
                 if k == 'password':
                     v = get_crypt_password(v)
                 if k == 'firstname':
                     k = 'name'
                 setattr(new_user, k, v)
             new_user.api_key = generate_api_key(form_data['username'])
             self.sa.add(new_user)
             return new_user
         except Exception:
             log.error(traceback.format_exc())
             raise
     def create_or_update(self, username, password, email, firstname='',
                          lastname='', active=True, admin=False, ldap_dn=None):
         """
         Creates a new instance if not found, or updates current one
         :param username:
         :param password:
         :param email:
         :param active:
         :param firstname:
         :param lastname:
         :param active:
         :param admin:
         :param ldap_dn:
         """
         from rhodecode.lib.auth import get_crypt_password
         log.debug('Checking for %s account in RhodeCode database' % username)
         user = User.get_by_username(username, case_insensitive=True)
         if user is None:
             log.debug('creating new user %s' % username)
             new_user = User()
             edit = False
         else:
             log.debug('updating user %s' % username)
             new_user = user
             edit = True
         try:
             new_user.username = username
             new_user.admin = admin
             # set password only if creating an user or password is changed
             if not edit or user.password != password:
                 new_user.password = get_crypt_password(password)
+                new_user.password = get_crypt_password(password) if password else None
                 new_user.api_key = generate_api_key(username)
             new_user.email = email
             new_user.active = active
             new_user.ldap_dn = safe_unicode(ldap_dn) if ldap_dn else None
             new_user.name = firstname
             new_user.lastname = lastname
             self.sa.add(new_user)
             return new_user
         except (DatabaseError,):
             log.error(traceback.format_exc())
             raise
     def create_for_container_auth(self, username, attrs):
         """
         Creates the given user if it's not already in the database
         :param username:
         :param attrs:
         """
         if self.get_by_username(username, case_insensitive=True) is None:
             # autogenerate email for container account without one
             generate_email = lambda usr: '%s@container_auth.account' % usr
             try:
                 new_user = User()
                 new_user.username = username
                 new_user.password = None
                 new_user.api_key = generate_api_key(username)
                 new_user.email = attrs['email']
                 new_user.active = attrs.get('active', True)
                 new_user.name = attrs['name'] or generate_email(username)
                 new_user.lastname = attrs['lastname']
                 self.sa.add(new_user)
                 return new_user
             except (DatabaseError,):
                 log.error(traceback.format_exc())
                 self.sa.rollback()
                 raise
         log.debug('User %s already exists. Skipping creation of account'
                   ' for container auth.', username)
         return None
     def create_ldap(self, username, password, user_dn, attrs):
         """
         Checks if user is in database, if not creates this user marked
         as ldap user
         :param username:
         :param password:
         :param user_dn:
         :param attrs:
         """
         from rhodecode.lib.auth import get_crypt_password
         log.debug('Checking for such ldap account in RhodeCode database')
         if self.get_by_username(username, case_insensitive=True) is None:
             # autogenerate email for ldap account without one
             generate_email = lambda usr: '%s@ldap.account' % usr
             try:
                 new_user = User()
                 username = username.lower()
                 # add ldap account always lowercase
                 new_user.username = username
                 new_user.password = get_crypt_password(password)
                 new_user.api_key = generate_api_key(username)
                 new_user.email = attrs['email'] or generate_email(username)
                 new_user.active = attrs.get('active', True)
                 new_user.ldap_dn = safe_unicode(user_dn)
                 new_user.name = attrs['name']
                 new_user.lastname = attrs['lastname']
                 self.sa.add(new_user)
                 return new_user
             except (DatabaseError,):
                 log.error(traceback.format_exc())
                 self.sa.rollback()
                 raise
         log.debug('this %s user exists skipping creation of ldap account',
                   username)
         return None
     def create_registration(self, form_data):
         from rhodecode.model.notification import NotificationModel
         try:
             form_data['admin'] = False
             new_user = self.create(form_data)
             self.sa.add(new_user)
             self.sa.flush()
             # notification to admins
             subject = _('New user registration')
             body = ('New user registration\n'
                     '---------------------\n'
                     '- Username: %s\n'
                     '- Full Name: %s\n'
                     '- Email: %s\n')
             body = body % (new_user.username, new_user.full_name,
                            new_user.email)
             edit_url = url('edit_user', id=new_user.user_id, qualified=True)
             kw = {'registered_user_url': edit_url}
             NotificationModel().create(created_by=new_user, subject=subject,
                                        body=body, recipients=None,
                                        type_=Notification.TYPE_REGISTRATION,
                                        email_kwargs=kw)
         except Exception:
             log.error(traceback.format_exc())
             raise
     def update(self, user_id, form_data, skip_attrs=[]):
         from rhodecode.lib.auth import get_crypt_password
         try:
             user = self.get(user_id, cache=False)
             if user.username == 'default':
                 raise DefaultUserException(
                                 _("You can't Edit this user since it's"
                                   " crucial for entire application"))
             for k, v in form_data.items():
                 if k in skip_attrs:
                     continue
                 if k == 'new_password' and v:
                     user.password = get_crypt_password(v)
                     user.api_key = generate_api_key(user.username)
                 else:
                     if k == 'firstname':
                         k = 'name'
                     setattr(user, k, v)
             self.sa.add(user)
         except Exception:
             log.error(traceback.format_exc())
             raise
     def update_user(self, user, **kwargs):
         from rhodecode.lib.auth import get_crypt_password
         try:
             user = self._get_user(user)
             if user.username == 'default':
                 raise DefaultUserException(
                     _("You can't Edit this user since it's"
                       " crucial for entire application")
+                )
             for k, v in kwargs.items():
                 if k == 'password' and v:
                     v = get_crypt_password(v)
                     user.api_key = generate_api_key(user.username)
                 setattr(user, k, v)
             self.sa.add(user)
             return user
         except Exception:
             log.error(traceback.format_exc())
             raise
     def delete(self, user):
         user = self._get_user(user)
         try:
             if user.username == 'default':
                 raise DefaultUserException(
                     _(u"You can't remove this user since it's"
                       " crucial for entire application")
+                )
             if user.repositories:
                 repos = [x.repo_name for x in user.repositories]
                 raise UserOwnsReposException(
                     _(u'user "%s" still owns %s repositories and cannot be '
                       'removed. Switch owners or remove those repositories. %s')
                     % (user.username, len(repos), ', '.join(repos))
+                )
             self.sa.delete(user)
         except Exception:
             log.error(traceback.format_exc())
             raise
     def reset_password_link(self, data):
         from rhodecode.lib.celerylib import tasks, run_task
         from rhodecode.model.notification import EmailNotificationModel
         user_email = data['email']
         try:
             user = User.get_by_email(user_email)
             if user:
                 log.debug('password reset user found %s' % user)
                 link = url('reset_password_confirmation', key=user.api_key,
                            qualified=True)
                 reg_type = EmailNotificationModel.TYPE_PASSWORD_RESET

rhodecode/tests/api/api_base.py

➞

Show inline comments

@@ @@ -288,384 +288,405 @@ class BaseTestApi(object): @@
     def test_api_lock_repo_lock_aquire(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = {
             'repo': self.REPO,
             'locked': True,
             'locked_since': None,
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                     % (TEST_USER_ADMIN_LOGIN, self.REPO, True))
+        }
         expected['locked_since'] = json.loads(response.body)['result']['locked_since']
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_aquire_by_non_admin(self):
         repo_name = 'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                             cur_user=self.TEST_USER_LOGIN)
         try:
             id_, params = _build_data(self.apikey_regular, 'lock',
                                       repoid=repo_name,
                                       locked=True)
             response = api_call(self, params)
             expected = {
                 'repo': repo_name,
                 'locked': True,
                 'locked_since': None,
                 'locked_by': self.TEST_USER_LOGIN,
                 'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                         % (self.TEST_USER_LOGIN, repo_name, True))
+            }
             expected['locked_since'] = json.loads(response.body)['result']['locked_since']
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_lock_repo_lock_aquire_non_admin_with_userid(self):
         repo_name = 'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                             cur_user=self.TEST_USER_LOGIN)
         try:
             id_, params = _build_data(self.apikey_regular, 'lock',
                                       userid=TEST_USER_ADMIN_LOGIN,
                                       repoid=repo_name,
                                       locked=True)
             response = api_call(self, params)
             expected = 'userid is not the same as your user'
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_lock_repo_lock_aquire_non_admin_not_his_repo(self):
         id_, params = _build_data(self.apikey_regular, 'lock',
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = 'repository `%s` does not exist' % (self.REPO)
         self._compare_error(id_, expected, given=response.body)
     def test_api_lock_repo_lock_release(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=False)
         response = api_call(self, params)
         expected = {
             'repo': self.REPO,
             'locked': False,
             'locked_since': None,
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                     % (TEST_USER_ADMIN_LOGIN, self.REPO, False))
+        }
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_aquire_optional_userid(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = {
             'repo': self.REPO,
             'locked': True,
             'locked_since': None,
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                     % (TEST_USER_ADMIN_LOGIN, self.REPO, True))
+        }
         expected['locked_since'] = json.loads(response.body)['result']['locked_since']
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_optional_locked(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   repoid=self.REPO)
         response = api_call(self, params)
         time_ = json.loads(response.body)['result']['locked_since']
         expected = {
             'repo': self.REPO,
             'locked': True,
             'locked_since': None,
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'msg': ('Repo `%s` locked by `%s`. '
                             % (self.REPO,
                                json.dumps(time_to_datetime(time_))))
+        }
         expected['locked_since'] = time_
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(Repository, 'lock', crash)
     def test_api_lock_error(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = 'Error occurred locking repository `%s`' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_locks_regular_user(self):
         id_, params = _build_data(self.apikey_regular, 'get_locks')
         response = api_call(self, params)
         expected = []
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_locks_with_userid_regular_user(self):
         id_, params = _build_data(self.apikey_regular, 'get_locks',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         expected = 'userid is not the same as your user'
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_locks(self):
         id_, params = _build_data(self.apikey, 'get_locks')
         response = api_call(self, params)
         expected = []
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_locks_with_userid(self):
         id_, params = _build_data(self.apikey, 'get_locks',
                                   userid=TEST_USER_REGULAR_LOGIN)
         response = api_call(self, params)
         expected = []
         self._compare_ok(id_, expected, given=response.body)
     def test_api_create_existing_user(self):
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=TEST_USER_ADMIN_LOGIN,
                                   email='test@foo.com',
                                   password='trololo')
         response = api_call(self, params)
         expected = "user `%s` already exist" % TEST_USER_ADMIN_LOGIN
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_user_with_existing_email(self):
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=TEST_USER_ADMIN_LOGIN + 'new',
                                   email=TEST_USER_REGULAR_EMAIL,
                                   password='trololo')
         response = api_call(self, params)
         expected = "email `%s` already exist" % TEST_USER_REGULAR_EMAIL
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_user(self):
         username = 'test_new_api_user'
         email = username + "@foo.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email,
                                   password='trololo')
         response = api_call(self, params)
         usr = UserModel().get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
+        )
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         UserModel().delete(usr.user_id)
         Session().commit()
     def test_api_create_user_without_password(self):
         username = 'test_new_api_user_passwordless'
         email = username + "@foo.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email)
         response = api_call(self, params)
         usr = UserModel().get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
+        )
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         UserModel().delete(usr.user_id)
         Session().commit()
     @mock.patch.object(UserModel, 'create_or_update', crash)
     def test_api_create_user_when_exception_happened(self):
         username = 'test_new_api_user'
         email = username + "@foo.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email,
                                   password='trololo')
         response = api_call(self, params)
         expected = 'failed to create user `%s`' % username
         self._compare_error(id_, expected, given=response.body)
     def test_api_delete_user(self):
         usr = UserModel().create_or_update(username=u'test_user',
                                            password=u'qweqwe',
                                            email=u'u232@rhodecode.org',
                                            firstname=u'u1', lastname=u'u1')
         Session().commit()
         username = usr.username
         email = usr.email
         usr_id = usr.user_id
         ## DELETE THIS USER NOW
         id_, params = _build_data(self.apikey, 'delete_user',
                                   userid=username,)
         response = api_call(self, 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'u232@rhodecode.org',
                                            firstname=u'u1', lastname=u'u1')
         Session().commit()
         username = usr.username
         id_, params = _build_data(self.apikey, 'delete_user',
                                   userid=username,)
         response = api_call(self, 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 = api_call(self, params)
         ret = {
         'msg': 'updated user ID:%s %s' % (usr.user_id, self.TEST_USER_LOGIN),
         'user': jsonify(UserModel()\
                             .get_by_username(self.TEST_USER_LOGIN)\
                             .get_api_data())
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_update_user_no_changed_params(self):
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         ret = {
         'msg': 'updated user ID:%s %s' % (usr.user_id, TEST_USER_ADMIN_LOGIN),
         'user': ret
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_update_user_by_user_id(self):
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
         response = api_call(self, params)
         ret = {
         'msg': 'updated user ID:%s %s' % (usr.user_id, TEST_USER_ADMIN_LOGIN),
         'user': ret
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(UserModel, 'update_user', crash)
     def test_api_update_user_when_exception_happens(self):
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
         response = api_call(self, params)
         ret = 'failed to update user `%s`' % usr.user_id
         expected = ret
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo(self):
         new_group = 'some_new_group'
         make_users_group(new_group)
         RepoModel().grant_users_group_permission(repo=self.REPO,
                                                  group_name=new_group,
                                                  perm='repository.read')
         Session().commit()
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(self.REPO)
         ret = repo.get_api_data()
         members = []
         followers = []
         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)
         for user in repo.followers:
             followers.append(user.user.get_api_data())
         ret['members'] = members
         ret['followers'] = followers
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         destroy_users_group(new_group)
     def test_api_get_repo_by_non_admin(self):
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(self.REPO)
         ret = repo.get_api_data()
         members = []
         followers = []
         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)
         for user in repo.followers:
             followers.append(user.user.get_api_data())
         ret['members'] = members

0 comments (0 inline, 0 general)