kallithea Changeset - b3a51c3987be

Changeset - b3a51c3987be

Parent rev.

Child rev.

[Not reviewed]

default

0 5 0

Andrew Shadura - 10 years ago 2016-01-30 12:15:23
andrew@shadura.me

db: always match user emails case insensitively

This commit removes case-sensitive email matching.
It also adds a couple of tests which fail, to demonstrate a defect in the
current implementation (using ILIKE matching instead of case-insensitive
equality comparison).

5 files changed with 18 insertions and 14 deletions:

kallithea/controllers/api/api.py

kallithea/lib/helpers.py

kallithea/model/db.py

kallithea/model/validators.py

kallithea/tests/models/test_users.py

0 comments (0 inline, 0 general)

kallithea/controllers/api/api.py

➞

Show inline comments

@@ @@ -293,769 +293,769 @@ class ApiController(JSONRPCController): @@
         try:
             ScmModel().mark_for_invalidation(repo.repo_name)
             return dict(
                 msg='Cache for repository `%s` was invalidated' % (repoid,),
                 repository=repo.repo_name
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Error occurred during cache invalidation action'
+            )
     # permission check inside
     def lock(self, apiuser, repoid, locked=Optional(None),
              userid=Optional(OAttr('apiuser'))):
         """
         Set locking state on given repository by given user. If userid param
         is skipped, then it is set to id of user who is 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.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param repoid: repository name or repository id
         :type repoid: str or int
         :param locked: lock state to be set
         :type locked: Optional(bool)
         :param userid: set lock as user
         :type userid: Optional(str or int)
         OUTPUT::
           id : <id_given_in_input>
           result : {
             'repo': '<reponame>',
             'locked': <bool: lock state>,
             'locked_since': <int: lock timestamp>,
             'locked_by': <username of person who made the lock>,
             'lock_state_changed': <bool: True if lock state has been changed in this request>,
             'msg': 'Repo `<reponame>` locked by `<username>` on <timestamp>.'
             or
             'msg': 'Repo `<repository name>` not locked.'
             or
             'msg': 'User `<user name>` set lock state for repo `<repository name>` to `<new lock state>`'
+          }
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             'Error occurred locking repository `<reponame>`
+          }
         """
         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,
                     'lock_state_changed': False,
                     'msg': 'Repo `%s` not locked.' % repo.repo_name
+                }
                 return _d
             else:
                 userid, time_ = lockobj
                 lock_user = get_user_or_error(userid)
                 _d = {
                     'repo': repo.repo_name,
                     'locked': True,
                     'locked_since': time_,
                     'locked_by': lock_user.username,
                     'lock_state_changed': False,
                     'msg': ('Repo `%s` locked by `%s` on `%s`.'
                             % (repo.repo_name, lock_user.username,
                                json.dumps(time_to_datetime(time_))))
+                }
                 return _d
         # force locked state through a flag
         else:
             locked = str2bool(locked)
             try:
                 if locked:
                     lock_time = time.time()
                     Repository.lock(repo, user.user_id, lock_time)
                 else:
                     lock_time = None
                     Repository.unlock(repo)
                 _d = {
                     'repo': repo.repo_name,
                     'locked': locked,
                     'locked_since': lock_time,
                     'locked_by': user.username,
                     'lock_state_changed': True,
                     'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                             % (user.username, repo.repo_name, locked))
+                }
                 return _d
             except Exception:
                 log.error(traceback.format_exc())
                 raise JSONRPCError(
                     'Error occurred locking repository `%s`' % repo.repo_name
+                )
     def get_locks(self, apiuser, userid=Optional(OAttr('apiuser'))):
         """
         Get all repositories with locks for given userid, if
         this command is run by non-admin account userid is set to user
         who is calling this method, thus returning locks for himself.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param userid: User to get locks for
         :type userid: Optional(str or int)
         OUTPUT::
           id : <id_given_in_input>
           result : {
             [repo_object, repo_object,...]
+          }
           error :  null
         """
         if not HasPermissionAnyApi('hg.admin')(user=apiuser):
             # 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 is not None:
                     if safe_int(userid) == user.user_id:
                         ret.append(_api_data)
                 else:
                     ret.append(_api_data)
         return ret
     @HasPermissionAllDecorator('hg.admin')
     def get_ip(self, apiuser, userid=Optional(OAttr('apiuser'))):
         """
         Shows IP address as seen from Kallithea server, together with all
         defined IP addresses for given user. If userid is not passed data is
         returned for user who's calling this function.
         This command can be executed only using api_key belonging to user with
         admin rights.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param userid: username to show ips for
         :type userid: Optional(str or int)
         OUTPUT::
             id : <id_given_in_input>
             result : {
                          "server_ip_addr": "<ip_from_clien>",
                          "user_ips": [
+                                        {
                                            "ip_addr": "<ip_with_mask>",
                                            "ip_range": ["<start_ip>", "<end_ip>"],
                                         },
                                         ...
+                                     ]
+            }
         """
         if isinstance(userid, Optional):
             userid = apiuser.user_id
         user = get_user_or_error(userid)
         ips = UserIpMap.query().filter(UserIpMap.user == user).all()
         return dict(
             server_ip_addr=self.ip_addr,
             user_ips=ips
+        )
     # alias for old
     show_ip = get_ip
     @HasPermissionAllDecorator('hg.admin')
     def get_server_info(self, apiuser):
         """
         return server info, including Kallithea version and installed packages
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         OUTPUT::
           id : <id_given_in_input>
           result : {
             'modules': [<module name>,...]
             'py_version': <python version>,
             'platform': <platform type>,
             'kallithea_version': <kallithea version>
+          }
           error :  null
         """
         return Setting.get_server_info()
     def get_user(self, apiuser, userid=Optional(OAttr('apiuser'))):
         """
         Gets a user by username or user_id, Returns empty result if user is
         not found. If userid param is skipped it is set to id of user who is
         calling this method. This command can be executed only using api_key
         belonging to user with admin rights, or regular users that cannot
         specify different userid than theirs
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param userid: user to get data for
         :type userid: Optional(str or int)
         OUTPUT::
             id : <id_given_in_input>
             result: None if user does not exist or
+                    {
                         "user_id" :     "<user_id>",
                         "api_key" :     "<api_key>",
                         "api_keys":     "[<list of all API keys including additional ones>]"
                         "username" :    "<username>",
                         "firstname":    "<firstname>",
                         "lastname" :    "<lastname>",
                         "email" :       "<email>",
                         "emails":       "[<list of all emails including additional ones>]",
                         "ip_addresses": "[<ip_address_for_user>,...]",
                         "active" :      "<bool: user active>",
                         "admin" :       "<bool: user is admin>",
                         "extern_name" : "<extern_name>",
                         "extern_type" : "<extern type>
                         "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
         """
         if not HasPermissionAnyApi('hg.admin')(user=apiuser):
             # 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):
         """
         Lists all existing users. This command can be executed only using api_key
         belonging to user with admin rights.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         OUTPUT::
             id : <id_given_in_input>
             result: [<user_object>, ...]
             error:  null
         """
         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=Optional(''),
                     firstname=Optional(''), lastname=Optional(''),
                     active=Optional(True), admin=Optional(False),
                     extern_name=Optional(EXTERN_TYPE_INTERNAL),
                     extern_type=Optional(EXTERN_TYPE_INTERNAL)):
         """
         Creates new user. Returns new user object. This command can
         be executed only using api_key belonging to user with admin rights.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param username: new username
         :type username: str or int
         :param email: email
         :type email: str
         :param password: password
         :type password: Optional(str)
         :param firstname: firstname
         :type firstname: Optional(str)
         :param lastname: lastname
         :type lastname: Optional(str)
         :param active: active
         :type active: Optional(bool)
         :param admin: admin
         :type admin: Optional(bool)
         :param extern_name: name of extern
         :type extern_name: Optional(str)
         :param extern_type: extern_type
         :type extern_type: Optional(str)
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg" : "created new user `<username>`",
                       "user": <user_obj>
+                    }
             error:  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "user `<username>` already exist"
             or
             "email `<email>` already exist"
             or
             "failed to create user `<username>`"
+          }
         """
         if User.get_by_username(username):
             raise JSONRPCError("user `%s` already exist" % (username,))
-        if User.get_by_email(email, case_insensitive=True):
         if User.get_by_email(email):
             raise JSONRPCError("email `%s` already exist" % (email,))
         if Optional.extract(extern_name):
             # generate temporary password if user is external
             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),
                 extern_type=Optional.extract(extern_type),
                 extern_name=Optional.extract(extern_name)
+            )
             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), password=Optional(None),
                     firstname=Optional(None), lastname=Optional(None),
                     active=Optional(None), admin=Optional(None),
                     extern_type=Optional(None), extern_name=Optional(None)):
         """
         updates given user if such user exists. This command can
         be executed only using api_key belonging to user with admin rights.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param userid: userid to update
         :type userid: str or int
         :param username: new username
         :type username: str or int
         :param email: email
         :type email: str
         :param password: password
         :type password: Optional(str)
         :param firstname: firstname
         :type firstname: Optional(str)
         :param lastname: lastname
         :type lastname: Optional(str)
         :param active: active
         :type active: Optional(bool)
         :param admin: admin
         :type admin: Optional(bool)
         :param extern_name:
         :type extern_name: Optional(str)
         :param extern_type:
         :type extern_type: Optional(str)
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg" : "updated user ID:<userid> <username>",
                       "user": <user_object>,
+                    }
             error:  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to update user `<username>`"
+          }
         """
         user = get_user_or_error(userid)
         # only non optional arguments will be stored in updates
         updates = {}
         try:
             store_update(updates, username, 'username')
             store_update(updates, password, 'password')
             store_update(updates, email, 'email')
             store_update(updates, firstname, 'name')
             store_update(updates, lastname, 'lastname')
             store_update(updates, active, 'active')
             store_update(updates, admin, 'admin')
             store_update(updates, extern_name, 'extern_name')
             store_update(updates, extern_type, 'extern_type')
             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 given user if such user exists. This command can
         be executed only using api_key belonging to user with admin rights.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param userid: user to delete
         :type userid: str or int
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg" : "deleted user ID:<userid> <username>",
                       "user": null
+                    }
             error:  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to delete user ID:<userid> <username>"
+          }
         """
         user = get_user_or_error(userid)
         try:
             UserModel().delete(userid)
             Session().commit()
             return dict(
                 msg='deleted user ID:%s %s' % (user.user_id, user.username),
                 user=None
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to delete user ID:%s %s'
                                % (user.user_id, user.username))
     # permission check inside
     def get_user_group(self, apiuser, usergroupid):
         """
         Gets an existing user group. This command can be executed only using api_key
         belonging to user with admin rights or user who has at least
         read access to user group.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param usergroupid: id of user_group to edit
         :type usergroupid: str or int
         OUTPUT::
             id : <id_given_in_input>
             result : None if group not exist
+                     {
                        "users_group_id" : "<id>",
                        "group_name" :     "<groupname>",
                        "active":          "<bool>",
                        "members" :  [<user_obj>,...]
+                     }
             error : null
         """
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAnyApi('hg.admin')(user=apiuser):
             # check if we have at least read permission for this user group !
             _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
             if not HasUserGroupPermissionAny(*_perms)(
                     user=apiuser, user_group_name=user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
         data = user_group.get_api_data()
         return data
     # permission check inside
     def get_user_groups(self, apiuser):
         """
         Lists all existing user groups. This command can be executed only using
         api_key belonging to user with admin rights or user who has at least
         read access to user group.
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         OUTPUT::
             id : <id_given_in_input>
             result : [<user_group_obj>,...]
             error : null
         """
         result = []
         _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
         extras = {'user': apiuser}
         for user_group in UserGroupList(UserGroupModel().get_all(),
                                         perm_set=_perms, extra_kwargs=extras):
             result.append(user_group.get_api_data())
         return result
     @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')
     def create_user_group(self, apiuser, group_name, description=Optional(''),
                           owner=Optional(OAttr('apiuser')), active=Optional(True)):
         """
         Creates new user group. This command can be executed only using api_key
         belonging to user with admin rights or an user who has create user group
         permission
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param group_name: name of new user group
         :type group_name: str
         :param description: group description
         :type description: str
         :param owner: owner of group. If not passed apiuser is the owner
         :type owner: Optional(str or int)
         :param active: group is active
         :type active: Optional(bool)
         OUTPUT::
             id : <id_given_in_input>
             result: {
                       "msg": "created new user group `<groupname>`",
                       "user_group": <user_group_object>
+                    }
             error:  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "user group `<group name>` already exist"
             or
             "failed to create group `<group name>`"
+          }
         """
         if UserGroupModel().get_by_name(group_name):
             raise JSONRPCError("user group `%s` already exist" % (group_name,))
         try:
             if isinstance(owner, Optional):
                 owner = apiuser.user_id
             owner = get_user_or_error(owner)
             active = Optional.extract(active)
             description = Optional.extract(description)
             ug = UserGroupModel().create(name=group_name, description=description,
                                          owner=owner, active=active)
             Session().commit()
             return dict(
                 msg='created new user group `%s`' % group_name,
                 user_group=ug.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create group `%s`' % (group_name,))
     # permission check inside
     def update_user_group(self, apiuser, usergroupid, group_name=Optional(''),
                           description=Optional(''), owner=Optional(None),
                           active=Optional(True)):
         """
         Updates given usergroup.  This command can be executed only using api_key
         belonging to user with admin rights or an admin of given user group
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param usergroupid: id of user group to update
         :type usergroupid: str or int
         :param group_name: name of new user group
         :type group_name: str
         :param description: group description
         :type description: str
         :param owner: owner of group.
         :type owner: Optional(str or int)
         :param active: group is active
         :type active: Optional(bool)
         OUTPUT::
           id : <id_given_in_input>
           result : {
             "msg": 'updated user group ID:<user group id> <user group name>',
             "user_group": <user_group_object>
+          }
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to update user group `<user group name>`"
+          }
         """
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAnyApi('hg.admin')(user=apiuser):
             # check if we have admin permission for this user group !
             _perms = ('usergroup.admin',)
             if not HasUserGroupPermissionAny(*_perms)(
                     user=apiuser, user_group_name=user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
         if not isinstance(owner, Optional):
             owner = get_user_or_error(owner)
         updates = {}
         store_update(updates, group_name, 'users_group_name')
         store_update(updates, description, 'user_group_description')
         store_update(updates, owner, 'user')
         store_update(updates, active, 'users_group_active')
         try:
             UserGroupModel().update(user_group, updates)
             Session().commit()
             return dict(
                 msg='updated user group ID:%s %s' % (user_group.users_group_id,
                                                      user_group.users_group_name),
                 user_group=user_group.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update user group `%s`' % (usergroupid,))
     # permission check inside
     def delete_user_group(self, apiuser, usergroupid):
         """
         Delete given user group by user group id or name.
         This command can be executed only using api_key
         belonging to user with admin rights or an admin of given user group
         :param apiuser: filled automatically from apikey
         :type apiuser: AuthUser
         :param usergroupid:
         :type usergroupid: int
         OUTPUT::
           id : <id_given_in_input>
           result : {
             "msg": "deleted user group ID:<user_group_id> <user_group_name>"
+          }
           error :  null
         ERROR OUTPUT::
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to delete user group ID:<user_group_id> <user_group_name>"
             or
             "RepoGroup assigned to <repo_groups_list>"
+          }
         """
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAnyApi('hg.admin')(user=apiuser):
             # check if we have admin permission for this user group !
             _perms = ('usergroup.admin',)
             if not HasUserGroupPermissionAny(*_perms)(
                     user=apiuser, user_group_name=user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
         try:
             UserGroupModel().delete(user_group)

kallithea/lib/helpers.py

➞

Show inline comments

@@ @@ -99,769 +99,769 @@ def html_escape(s): @@
         .replace('"', "&quot;")
         .replace("'", "&apos;")
+        )
 def shorter(s, size=20):
     postfix = '...'
     if len(s) > size:
         return s[:size - len(postfix)] + postfix
     return s
 def _reset(name, value=None, id=NotGiven, type="reset", **attrs):
     """
     Reset button
     """
     _set_input_attrs(attrs, type, name, value)
     _set_id_attr(attrs, id, name)
     convert_boolean_attrs(attrs, ["disabled"])
     return HTML.input(**attrs)
 reset = _reset
 safeid = _make_safe_id_component
 def FID(raw_id, path):
     """
     Creates a unique ID for filenode based on it's hash of path and revision
     it's safe to use in urls
     :param raw_id:
     :param path:
     """
     return 'C-%s-%s' % (short_id(raw_id), hashlib.md5(safe_str(path)).hexdigest()[:12])
 class _GetError(object):
     """Get error from form_errors, and represent it as span wrapped error
     message
     :param field_name: field to fetch errors for
     :param form_errors: form errors dict
     """
     def __call__(self, field_name, form_errors):
         tmpl = """<span class="error_msg">%s</span>"""
         if form_errors and field_name in form_errors:
             return literal(tmpl % form_errors.get(field_name))
 get_error = _GetError()
 class _FilesBreadCrumbs(object):
     def __call__(self, repo_name, rev, paths):
         if isinstance(paths, str):
             paths = safe_unicode(paths)
         url_l = [link_to(repo_name, url('files_home',
                                         repo_name=repo_name,
                                         revision=rev, f_path=''),
                          class_='ypjax-link')]
         paths_l = paths.split('/')
         for cnt, p in enumerate(paths_l):
             if p != '':
                 url_l.append(link_to(p,
                                      url('files_home',
                                          repo_name=repo_name,
                                          revision=rev,
                                          f_path='/'.join(paths_l[:cnt + 1])
                                          ),
                                      class_='ypjax-link'
+                                     )
+                             )
         return literal('/'.join(url_l))
 files_breadcrumbs = _FilesBreadCrumbs()
 class CodeHtmlFormatter(HtmlFormatter):
     """
     My code Html Formatter for source codes
     """
     def wrap(self, source, outfile):
         return self._wrap_div(self._wrap_pre(self._wrap_code(source)))
     def _wrap_code(self, source):
         for cnt, it in enumerate(source):
             i, t = it
             t = '<div id="L%s">%s</div>' % (cnt + 1, t)
             yield i, t
     def _wrap_tablelinenos(self, inner):
         dummyoutfile = StringIO.StringIO()
         lncount = 0
         for t, line in inner:
             if t:
                 lncount += 1
             dummyoutfile.write(line)
         fl = self.linenostart
         mw = len(str(lncount + fl - 1))
         sp = self.linenospecial
         st = self.linenostep
         la = self.lineanchors
         aln = self.anchorlinenos
         nocls = self.noclasses
         if sp:
             lines = []
             for i in range(fl, fl + lncount):
                 if i % st == 0:
                     if i % sp == 0:
                         if aln:
                             lines.append('<a href="#%s%d" class="special">%*d</a>' %
                                          (la, i, mw, i))
                         else:
                             lines.append('<span class="special">%*d</span>' % (mw, i))
                     else:
                         if aln:
                             lines.append('<a href="#%s%d">%*d</a>' % (la, i, mw, i))
                         else:
                             lines.append('%*d' % (mw, i))
                 else:
                     lines.append('')
             ls = '\n'.join(lines)
         else:
             lines = []
             for i in range(fl, fl + lncount):
                 if i % st == 0:
                     if aln:
                         lines.append('<a href="#%s%d">%*d</a>' % (la, i, mw, i))
                     else:
                         lines.append('%*d' % (mw, i))
                 else:
                     lines.append('')
             ls = '\n'.join(lines)
         # in case you wonder about the seemingly redundant <div> here: since the
         # content in the other cell also is wrapped in a div, some browsers in
         # some configurations seem to mess up the formatting...
         if nocls:
             yield 0, ('<table class="%stable">' % self.cssclass +
                       '<tr><td><div class="linenodiv" '
                       'style="background-color: #f0f0f0; padding-right: 10px">'
                       '<pre style="line-height: 125%">' +
                       ls + '</pre></div></td><td id="hlcode" class="code">')
         else:
             yield 0, ('<table class="%stable">' % self.cssclass +
                       '<tr><td class="linenos"><div class="linenodiv"><pre>' +
                       ls + '</pre></div></td><td id="hlcode" class="code">')
         yield 0, dummyoutfile.getvalue()
         yield 0, '</td></tr></table>'
 _whitespace_re = re.compile(r'(\t)|( )(?=\n|</div>)')
 def _markup_whitespace(m):
     groups = m.groups()
     if groups[0]:
         return '<u>\t</u>'
     if groups[1]:
         return ' <i></i>'
 def markup_whitespace(s):
     return _whitespace_re.sub(_markup_whitespace, s)
 def pygmentize(filenode, **kwargs):
     """
     pygmentize function using pygments
     :param filenode:
     """
     lexer = get_custom_lexer(filenode.extension) or filenode.lexer
     return literal(markup_whitespace(
         code_highlight(filenode.content, lexer, CodeHtmlFormatter(**kwargs))))
 def pygmentize_annotation(repo_name, filenode, **kwargs):
     """
     pygmentize function for annotation
     :param filenode:
     """
     color_dict = {}
     def gen_color(n=10000):
         """generator for getting n of evenly distributed colors using
         hsv color and golden ratio. It always return same order of colors
         :returns: RGB tuple
         """
         def hsv_to_rgb(h, s, v):
             if s == 0.0:
                 return v, v, v
             i = int(h * 6.0)  # XXX assume int() truncates!
             f = (h * 6.0) - i
             p = v * (1.0 - s)
             q = v * (1.0 - s * f)
             t = v * (1.0 - s * (1.0 - f))
             i = i % 6
             if i == 0:
                 return v, t, p
             if i == 1:
                 return q, v, p
             if i == 2:
                 return p, v, t
             if i == 3:
                 return p, q, v
             if i == 4:
                 return t, p, v
             if i == 5:
                 return v, p, q
         golden_ratio = 0.618033988749895
         h = 0.22717784590367374
         for _unused in xrange(n):
             h += golden_ratio
             h %= 1
             HSV_tuple = [h, 0.95, 0.95]
             RGB_tuple = hsv_to_rgb(*HSV_tuple)
             yield map(lambda x: str(int(x * 256)), RGB_tuple)
     cgenerator = gen_color()
     def get_color_string(cs):
         if cs in color_dict:
             col = color_dict[cs]
         else:
             col = color_dict[cs] = cgenerator.next()
         return "color: rgb(%s)! important;" % (', '.join(col))
     def url_func(repo_name):
         def _url_func(changeset):
             author = escape(changeset.author)
             date = changeset.date
             message = escape(changeset.message)
             tooltip_html = ("<div style='font-size:0.8em'><b>Author:</b>"
                             " %s<br/><b>Date:</b> %s</b><br/><b>Message:"
                             "</b> %s<br/></div>") % (author, date, message)
             lnk_format = show_id(changeset)
             uri = link_to(
                     lnk_format,
                     url('changeset_home', repo_name=repo_name,
                         revision=changeset.raw_id),
                     style=get_color_string(changeset.raw_id),
                     class_='tooltip safe-html-title',
                     title=tooltip_html
+                  )
             uri += '\n'
             return uri
         return _url_func
     return literal(markup_whitespace(annotate_highlight(filenode, url_func(repo_name), **kwargs)))
 def is_following_repo(repo_name, user_id):
     from kallithea.model.scm import ScmModel
     return ScmModel().is_following_repo(repo_name, user_id)
 class _Message(object):
     """A message returned by ``Flash.pop_messages()``.
     Converting the message to a string returns the message text. Instances
     also have the following attributes:
     * ``message``: the message text.
     * ``category``: the category specified when the message was created.
     """
     def __init__(self, category, message):
         self.category = category
         self.message = message
     def __str__(self):
         return self.message
     __unicode__ = __str__
     def __html__(self):
         return escape(safe_unicode(self.message))
 class Flash(_Flash):
     def __call__(self, message, category=None, ignore_duplicate=False, logf=None):
         """
         Show a message to the user _and_ log it through the specified function
         category: notice (default), warning, error, success
         logf: a custom log function - such as log.debug
         logf defaults to log.info, unless category equals 'success', in which
         case logf defaults to log.debug.
         """
         if logf is None:
             logf = log.info
             if category == 'success':
                 logf = log.debug
         logf('Flash %s: %s', category, message)
         super(Flash, self).__call__(message, category, ignore_duplicate)
     def pop_messages(self):
         """Return all accumulated messages and delete them from the session.
         The return value is a list of ``Message`` objects.
         """
         from pylons import session
         messages = session.pop(self.session_key, [])
         session.save()
         return [_Message(*m) for m in messages]
 flash = Flash()
 #==============================================================================
 # SCM FILTERS available via h.
 #==============================================================================
 from kallithea.lib.vcs.utils import author_name, author_email
 from kallithea.lib.utils2 import credentials_filter, age as _age
 from kallithea.model.db import User, ChangesetStatus, PullRequest
 age = lambda  x, y=False: _age(x, y)
 capitalize = lambda x: x.capitalize()
 email = author_email
 short_id = lambda x: x[:12]
 hide_credentials = lambda x: ''.join(credentials_filter(x))
 def show_id(cs):
     """
     Configurable function that shows ID
     by default it's r123:fffeeefffeee
     :param cs: changeset instance
     """
     from kallithea import CONFIG
     def_len = safe_int(CONFIG.get('show_sha_length', 12))
     show_rev = str2bool(CONFIG.get('show_revision_number', False))
     raw_id = cs.raw_id[:def_len]
     if show_rev:
         return 'r%s:%s' % (cs.revision, raw_id)
     else:
         return raw_id
 def fmt_date(date):
     if date:
         return date.strftime("%Y-%m-%d %H:%M:%S").decode('utf8')
     return ""
 def is_git(repository):
     if hasattr(repository, 'alias'):
         _type = repository.alias
     elif hasattr(repository, 'repo_type'):
         _type = repository.repo_type
     else:
         _type = repository
     return _type == 'git'
 def is_hg(repository):
     if hasattr(repository, 'alias'):
         _type = repository.alias
     elif hasattr(repository, 'repo_type'):
         _type = repository.repo_type
     else:
         _type = repository
     return _type == 'hg'
 def user_or_none(author):
     email = author_email(author)
     if email:
-        user = User.get_by_email(email, case_insensitive=True, cache=True)
         user = User.get_by_email(email, cache=True)
         if user is not None:
             return user
     return None
 def email_or_none(author):
     if not author:
         return None
     user = user_or_none(author)
     if user is not None:
         return user.email # always use main email address - not necessarily the one used to find user
     # extract email from the commit string
     email = author_email(author)
     if email:
         return email
     # No valid email, not a valid user in the system, none!
     return None
 def person(author, show_attr="username"):
     """Find the user identified by 'author', return one of the users attributes,
     default to the username attribute, None if there is no user"""
     # attr to return from fetched user
     person_getter = lambda usr: getattr(usr, show_attr)
     # if author is already an instance use it for extraction
     if isinstance(author, User):
         return person_getter(author)
     user = user_or_none(author)
     if user is not None:
         return person_getter(user)
     # Still nothing?  Just pass back the author name if any, else the email
     return author_name(author) or email(author)
 def person_by_id(id_, show_attr="username"):
     # attr to return from fetched user
     person_getter = lambda usr: getattr(usr, show_attr)
     #maybe it's an ID ?
     if str(id_).isdigit() or isinstance(id_, int):
         id_ = int(id_)
         user = User.get(id_)
         if user is not None:
             return person_getter(user)
     return id_
 def desc_stylize(value):
     """
     converts tags from value into html equivalent
     :param value:
     """
     if not value:
         return ''
     value = re.sub(r'\[see\ \=&gt;\ *([a-zA-Z0-9\/\=\?\&\ \:\/\.\-]*)\]',
                    '<div class="metatag" tag="see">see =&gt; \\1 </div>', value)
     value = re.sub(r'\[license\ \=&gt;\ *([a-zA-Z0-9\/\=\?\&\ \:\/\.\-]*)\]',
                    '<div class="metatag" tag="license"><a href="http:\/\/www.opensource.org/licenses/\\1">\\1</a></div>', value)
     value = re.sub(r'\[(requires|recommends|conflicts|base)\ \=&gt;\ *([a-zA-Z0-9\-\/]*)\]',
                    '<div class="metatag" tag="\\1">\\1 =&gt; <a href="/\\2">\\2</a></div>', value)
     value = re.sub(r'\[(lang|language)\ \=&gt;\ *([a-zA-Z\-\/\#\+]*)\]',
                    '<div class="metatag" tag="lang">\\2</div>', value)
     value = re.sub(r'\[([a-z]+)\]',
                   '<div class="metatag" tag="\\1">\\1</div>', value)
     return value
 def boolicon(value):
     """Returns boolean value of a value, represented as small html image of true/false
     icons
     :param value: value
     """
     if value:
         return HTML.tag('i', class_="icon-ok")
     else:
         return HTML.tag('i', class_="icon-minus-circled")
 def action_parser(user_log, feed=False, parse_cs=False):
     """
     This helper will action_map the specified string action into translated
     fancy names with icons and links
     :param user_log: user log instance
     :param feed: use output for feeds (no html and fancy icons)
     :param parse_cs: parse Changesets into VCS instances
     """
     action = user_log.action
     action_params = ' '
     x = action.split(':')
     if len(x) > 1:
         action, action_params = x
     def get_cs_links():
         revs_limit = 3  # display this amount always
         revs_top_limit = 50  # show upto this amount of changesets hidden
         revs_ids = action_params.split(',')
         deleted = user_log.repository is None
         if deleted:
             return ','.join(revs_ids)
         repo_name = user_log.repository.repo_name
         def lnk(rev, repo_name):
             lazy_cs = False
             title_ = None
             url_ = '#'
             if isinstance(rev, BaseChangeset) or isinstance(rev, AttributeDict):
                 if rev.op and rev.ref_name:
                     if rev.op == 'delete_branch':
                         lbl = _('Deleted branch: %s') % rev.ref_name
                     elif rev.op == 'tag':
                         lbl = _('Created tag: %s') % rev.ref_name
                     else:
                         lbl = 'Unknown operation %s' % rev.op
                 else:
                     lazy_cs = True
                     lbl = rev.short_id[:8]
                     url_ = url('changeset_home', repo_name=repo_name,
                                revision=rev.raw_id)
             else:
                 # changeset cannot be found - it might have been stripped or removed
                 lbl = rev[:12]
                 title_ = _('Changeset not found')
             if parse_cs:
                 return link_to(lbl, url_, title=title_, class_='tooltip')
             return link_to(lbl, url_, raw_id=rev.raw_id, repo_name=repo_name,
                            class_='lazy-cs' if lazy_cs else '')
         def _get_op(rev_txt):
             _op = None
             _name = rev_txt
             if len(rev_txt.split('=>')) == 2:
                 _op, _name = rev_txt.split('=>')
             return _op, _name
         revs = []
         if len(filter(lambda v: v != '', revs_ids)) > 0:
             repo = None
             for rev in revs_ids[:revs_top_limit]:
                 _op, _name = _get_op(rev)
                 # we want parsed changesets, or new log store format is bad
                 if parse_cs:
                     try:
                         if repo is None:
                             repo = user_log.repository.scm_instance
                         _rev = repo.get_changeset(rev)
                         revs.append(_rev)
                     except ChangesetDoesNotExistError:
                         log.error('cannot find revision %s in this repo', rev)
                         revs.append(rev)
                 else:
                     _rev = AttributeDict({
                         'short_id': rev[:12],
                         'raw_id': rev,
                         'message': '',
                         'op': _op,
                         'ref_name': _name
                     })
                     revs.append(_rev)
         cs_links = [" " + ', '.join(
             [lnk(rev, repo_name) for rev in revs[:revs_limit]]
         )]
         _op1, _name1 = _get_op(revs_ids[0])
         _op2, _name2 = _get_op(revs_ids[-1])
         _rev = '%s...%s' % (_name1, _name2)
         compare_view = (
             ' <div class="compare_view tooltip" title="%s">'
             '<a href="%s">%s</a> </div>' % (
                 _('Show all combined changesets %s->%s') % (
                     revs_ids[0][:12], revs_ids[-1][:12]
                 ),
                 url('changeset_home', repo_name=repo_name,
                     revision=_rev
                 ),
                 _('Compare view')
+            )
+        )
         # if we have exactly one more than normally displayed
         # just display it, takes less space than displaying
         # "and 1 more revisions"
         if len(revs_ids) == revs_limit + 1:
             cs_links.append(", " + lnk(revs[revs_limit], repo_name))
         # hidden-by-default ones
         if len(revs_ids) > revs_limit + 1:
             uniq_id = revs_ids[0]
             html_tmpl = (
                 '<span> %s <a class="show_more" id="_%s" '
                 'href="#more">%s</a> %s</span>'
+            )
             if not feed:
                 cs_links.append(html_tmpl % (
                       _('and'),
                       uniq_id, _('%s more') % (len(revs_ids) - revs_limit),
                       _('revisions')
+                    )
+                )
             if not feed:
                 html_tmpl = '<span id="%s" style="display:none">, %s </span>'
             else:
                 html_tmpl = '<span id="%s"> %s </span>'
             morelinks = ', '.join(
               [lnk(rev, repo_name) for rev in revs[revs_limit:]]
+            )
             if len(revs_ids) > revs_top_limit:
                 morelinks += ', ...'
             cs_links.append(html_tmpl % (uniq_id, morelinks))
         if len(revs) > 1:
             cs_links.append(compare_view)
         return ''.join(cs_links)
     def get_fork_name():
         repo_name = action_params
         url_ = url('summary_home', repo_name=repo_name)
         return _('Fork name %s') % link_to(action_params, url_)
     def get_user_name():
         user_name = action_params
         return user_name
     def get_users_group():
         group_name = action_params
         return group_name
     def get_pull_request():
         pull_request_id = action_params
         nice_id = PullRequest.make_nice_id(pull_request_id)
         deleted = user_log.repository is None
         if deleted:
             repo_name = user_log.repository_name
         else:
             repo_name = user_log.repository.repo_name
         return link_to(_('Pull request %s') % nice_id,
                     url('pullrequest_show', repo_name=repo_name,
                     pull_request_id=pull_request_id))
     def get_archive_name():
         archive_name = action_params
         return archive_name
     # action : translated str, callback(extractor), icon
     action_map = {
     'user_deleted_repo':           (_('[deleted] repository'),
                                     None, 'icon-trashcan'),
     'user_created_repo':           (_('[created] repository'),
                                     None, 'icon-plus'),
     'user_created_fork':           (_('[created] repository as fork'),
                                     None, 'icon-fork'),
     'user_forked_repo':            (_('[forked] repository'),
                                     get_fork_name, 'icon-fork'),
     'user_updated_repo':           (_('[updated] repository'),
                                     None, 'icon-pencil'),
     'user_downloaded_archive':      (_('[downloaded] archive from repository'),
                                     get_archive_name, 'icon-download-cloud'),
     'admin_deleted_repo':          (_('[delete] repository'),
                                     None, 'icon-trashcan'),
     'admin_created_repo':          (_('[created] repository'),
                                     None, 'icon-plus'),
     'admin_forked_repo':           (_('[forked] repository'),
                                     None, 'icon-fork'),
     'admin_updated_repo':          (_('[updated] repository'),
                                     None, 'icon-pencil'),
     'admin_created_user':          (_('[created] user'),
                                     get_user_name, 'icon-user'),
     'admin_updated_user':          (_('[updated] user'),
                                     get_user_name, 'icon-user'),
     'admin_created_users_group':   (_('[created] user group'),
                                     get_users_group, 'icon-pencil'),
     'admin_updated_users_group':   (_('[updated] user group'),
                                     get_users_group, 'icon-pencil'),
     'user_commented_revision':     (_('[commented] on revision in repository'),
                                     get_cs_links, 'icon-comment'),
     'user_commented_pull_request': (_('[commented] on pull request for'),
                                     get_pull_request, 'icon-comment'),
     'user_closed_pull_request':    (_('[closed] pull request for'),
                                     get_pull_request, 'icon-ok'),
     'push':                        (_('[pushed] into'),
                                     get_cs_links, 'icon-move-up'),
     'push_local':                  (_('[committed via Kallithea] into repository'),
                                     get_cs_links, 'icon-pencil'),
     'push_remote':                 (_('[pulled from remote] into repository'),
                                     get_cs_links, 'icon-move-up'),
     'pull':                        (_('[pulled] from'),
                                     None, 'icon-move-down'),
     'started_following_repo':      (_('[started following] repository'),
                                     None, 'icon-heart'),
     'stopped_following_repo':      (_('[stopped following] repository'),
                                     None, 'icon-heart-empty'),
+    }
     action_str = action_map.get(action, action)
     if feed:
         action = action_str[0].replace('[', '').replace(']', '')
     else:
         action = action_str[0] \
             .replace('[', '<span class="journal_highlight">') \
             .replace(']', '</span>')
     action_params_func = lambda: ""
     if callable(action_str[1]):
         action_params_func = action_str[1]
     def action_parser_icon():
         action = user_log.action
         action_params = None
         x = action.split(':')
         if len(x) > 1:
             action, action_params = x
         tmpl = """<i class="%s" alt="%s"></i>"""
         ico = action_map.get(action, ['', '', ''])[2]
         return literal(tmpl % (ico, action))
     # returned callbacks we need to call to get
     return [lambda: literal(action), action_params_func, action_parser_icon]
 #==============================================================================
 # PERMS
 #==============================================================================
 from kallithea.lib.auth import HasPermissionAny, HasPermissionAll, \
 HasRepoPermissionAny, HasRepoPermissionAll, HasRepoGroupPermissionAll, \
 HasRepoGroupPermissionAny
 #==============================================================================
 # GRAVATAR URL
 #==============================================================================
 def gravatar(email_address, cls='', size=30, ssl_enabled=True):
     """return html element of the gravatar
     This method will return an <img> with the resolution double the size (for
     retina screens) of the image. If the url returned from gravatar_url is
     empty then we fallback to using an icon.
     """
     src = gravatar_url(email_address, size*2, ssl_enabled)
     #  here it makes sense to use style="width: ..." (instead of, say, a
     # stylesheet) because we using this to generate a high-res (retina) size
     tmpl = '<img alt="" class="{cls}" style="width: {size}px; height: {size}px" src="{src}"/>'
     # if src is empty then there was no gravatar, so we use a font icon
     if not src:
         tmpl = """<i class="icon-user {cls}" style="font-size: {size}px;"></i>"""
     tmpl = tmpl.format(cls=cls, size=size, src=src)
     return literal(tmpl)
 def gravatar_url(email_address, size=30, ssl_enabled=True):
     # doh, we need to re-import those to mock it later
     from pylons import url
     from pylons import tmpl_context as c
     _def = 'anonymous@kallithea-scm.org'  # default gravatar
     _use_gravatar = c.visual.use_gravatar
     _gravatar_url = c.visual.gravatar_url or User.DEFAULT_GRAVATAR_URL
     email_address = email_address or _def

kallithea/model/db.py

➞

Show inline comments

@@ @@ -210,806 +210,800 @@ class Setting(Base, BaseModel): @@
         """
         self._app_settings_value = safe_unicode(val)
     @hybrid_property
     def app_settings_type(self):
         return self._app_settings_type
     @app_settings_type.setter
     def app_settings_type(self, val):
         if val not in self.SETTINGS_TYPES:
             raise Exception('type must be one of %s got %s'
                             % (self.SETTINGS_TYPES.keys(), val))
         self._app_settings_type = val
     def __unicode__(self):
         return u"<%s('%s:%s[%s]')>" % (
             self.__class__.__name__,
             self.app_settings_name, self.app_settings_value, self.app_settings_type
+        )
     @classmethod
     def get_by_name(cls, key):
         return cls.query() \
             .filter(cls.app_settings_name == key).scalar()
     @classmethod
     def get_by_name_or_create(cls, key, val='', type='unicode'):
         res = cls.get_by_name(key)
         if res is None:
             res = cls(key, val, type)
         return res
     @classmethod
     def create_or_update(cls, key, val=Optional(''), type=Optional('unicode')):
         """
         Creates or updates Kallithea setting. If updates are triggered, it will only
         update parameters that are explicitly set. Optional instance will be skipped.
         :param key:
         :param val:
         :param type:
         :return:
         """
         res = cls.get_by_name(key)
         if res is None:
             val = Optional.extract(val)
             type = Optional.extract(type)
             res = cls(key, val, type)
         else:
             res.app_settings_name = key
             if not isinstance(val, Optional):
                 # update if set
                 res.app_settings_value = val
             if not isinstance(type, Optional):
                 # update if set
                 res.app_settings_type = type
         return res
     @classmethod
     def get_app_settings(cls, cache=False):
         ret = cls.query()
         if cache:
             ret = ret.options(FromCache("sql_cache_short", "get_hg_settings"))
         if ret is None:
             raise Exception('Could not get application settings !')
         settings = {}
         for each in ret:
             settings[each.app_settings_name] = \
                 each.app_settings_value
         return settings
     @classmethod
     def get_auth_plugins(cls, cache=False):
         auth_plugins = cls.get_by_name("auth_plugins").app_settings_value
         return auth_plugins
     @classmethod
     def get_auth_settings(cls, cache=False):
         ret = cls.query() \
                 .filter(cls.app_settings_name.startswith('auth_')).all()
         fd = {}
         for row in ret:
             fd[row.app_settings_name] = row.app_settings_value
         return fd
     @classmethod
     def get_default_repo_settings(cls, cache=False, strip_prefix=False):
         ret = cls.query() \
                 .filter(cls.app_settings_name.startswith('default_')).all()
         fd = {}
         for row in ret:
             key = row.app_settings_name
             if strip_prefix:
                 key = remove_prefix(key, prefix='default_')
             fd.update({key: row.app_settings_value})
         return fd
     @classmethod
     def get_server_info(cls):
         import pkg_resources
         import platform
         import kallithea
         from kallithea.lib.utils import check_git_version
         mods = [(p.project_name, p.version) for p in pkg_resources.working_set]
         info = {
             'modules': sorted(mods, key=lambda k: k[0].lower()),
             'py_version': platform.python_version(),
             'platform': safe_unicode(platform.platform()),
             'kallithea_version': kallithea.__version__,
             'git_version': safe_unicode(check_git_version()),
             'git_path': kallithea.CONFIG.get('git_path')
+        }
         return info
 class Ui(Base, BaseModel):
     __tablename__ = DB_PREFIX + 'ui'
     __table_args__ = (
         # FIXME: ui_key as key is wrong and should be removed when the corresponding
         # Ui.get_by_key has been replaced by the composite key
         UniqueConstraint('ui_key'),
         UniqueConstraint('ui_section', 'ui_key'),
         _table_args_default_dict,
+    )
     HOOK_UPDATE = 'changegroup.update'
     HOOK_REPO_SIZE = 'changegroup.repo_size'
     HOOK_PUSH = 'changegroup.push_logger'
     HOOK_PRE_PUSH = 'prechangegroup.pre_push'
     HOOK_PULL = 'outgoing.pull_logger'
     HOOK_PRE_PULL = 'preoutgoing.pre_pull'
     ui_id = Column(Integer(), unique=True, primary_key=True)
     ui_section = Column(String(255), nullable=False)
     ui_key = Column(String(255), nullable=False)
     ui_value = Column(String(255), nullable=True) # FIXME: not nullable?
     ui_active = Column(Boolean(), nullable=False, default=True)
     @classmethod
     def get_by_key(cls, section, key):
         """ Return specified Ui object, or None if not found. """
         return cls.query().filter_by(ui_section=section, ui_key=key).scalar()
     @classmethod
     def get_or_create(cls, section, key):
         """ Return specified Ui object, creating it if necessary. """
         setting = cls.get_by_key(section, key)
         if setting is None:
             setting = cls(ui_section=section, ui_key=key)
             Session().add(setting)
         return setting
     @classmethod
     def get_builtin_hooks(cls):
         q = cls.query()
         q = q.filter(cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE,
                                      cls.HOOK_PUSH, cls.HOOK_PRE_PUSH,
                                      cls.HOOK_PULL, cls.HOOK_PRE_PULL]))
         q = q.filter(cls.ui_section == 'hooks')
         return q.all()
     @classmethod
     def get_custom_hooks(cls):
         q = cls.query()
         q = q.filter(~cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE,
                                       cls.HOOK_PUSH, cls.HOOK_PRE_PUSH,
                                       cls.HOOK_PULL, cls.HOOK_PRE_PULL]))
         q = q.filter(cls.ui_section == 'hooks')
         return q.all()
     @classmethod
     def get_repos_location(cls):
         return cls.get_by_key('paths', '/').ui_value
     @classmethod
     def create_or_update_hook(cls, key, val):
         new_ui = cls.get_or_create('hooks', key)
         new_ui.ui_active = True
         new_ui.ui_value = val
     def __repr__(self):
         return '<%s[%s]%s=>%s]>' % (self.__class__.__name__, self.ui_section,
                                     self.ui_key, self.ui_value)
 class User(Base, BaseModel):
     __tablename__ = 'users'
     __table_args__ = (
         Index('u_username_idx', 'username'),
         Index('u_email_idx', 'email'),
         _table_args_default_dict,
+    )
     DEFAULT_USER = 'default'
     DEFAULT_GRAVATAR_URL = 'https://secure.gravatar.com/avatar/{md5email}?d=identicon&s={size}'
     user_id = Column(Integer(), unique=True, primary_key=True)
     username = Column(String(255), nullable=False, unique=True)
     password = Column(String(255), nullable=False)
     active = Column(Boolean(), nullable=False, default=True)
     admin = Column(Boolean(), nullable=False, default=False)
     name = Column("firstname", Unicode(255), nullable=False)
     lastname = Column(Unicode(255), nullable=False)
     _email = Column("email", String(255), nullable=True, unique=True) # FIXME: not nullable?
     last_login = Column(DateTime(timezone=False), nullable=True)
     extern_type = Column(String(255), nullable=True) # FIXME: not nullable?
     extern_name = Column(String(255), nullable=True) # FIXME: not nullable?
     api_key = Column(String(255), nullable=False)
     inherit_default_permissions = Column(Boolean(), nullable=False, default=True)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     _user_data = Column("user_data", LargeBinary(), nullable=True)  # JSON data # FIXME: not nullable?
     user_log = relationship('UserLog')
     user_perms = relationship('UserToPerm', primaryjoin="User.user_id==UserToPerm.user_id", cascade='all')
     repositories = relationship('Repository')
     repo_groups = relationship('RepoGroup')
     user_groups = relationship('UserGroup')
     user_followers = relationship('UserFollowing', primaryjoin='UserFollowing.follows_user_id==User.user_id', cascade='all')
     followings = relationship('UserFollowing', primaryjoin='UserFollowing.user_id==User.user_id', cascade='all')
     repo_to_perm = relationship('UserRepoToPerm', primaryjoin='UserRepoToPerm.user_id==User.user_id', cascade='all')
     repo_group_to_perm = relationship('UserRepoGroupToPerm', primaryjoin='UserRepoGroupToPerm.user_id==User.user_id', cascade='all')
     group_member = relationship('UserGroupMember', cascade='all')
     notifications = relationship('UserNotification', cascade='all')
     # notifications assigned to this user
     user_created_notifications = relationship('Notification', cascade='all')
     # comments created by this user
     user_comments = relationship('ChangesetComment', cascade='all')
     #extra emails for this user
     user_emails = relationship('UserEmailMap', cascade='all')
     #extra API keys
     user_api_keys = relationship('UserApiKeys', cascade='all')
     @hybrid_property
     def email(self):
         return self._email
     @email.setter
     def email(self, val):
         self._email = val.lower() if val else None
     @property
     def firstname(self):
         # alias for future
         return self.name
     @property
     def emails(self):
         other = UserEmailMap.query().filter(UserEmailMap.user==self).all()
         return [self.email] + [x.email for x in other]
     @property
     def api_keys(self):
         other = UserApiKeys.query().filter(UserApiKeys.user==self).all()
         return [self.api_key] + [x.api_key for x in other]
     @property
     def ip_addresses(self):
         ret = UserIpMap.query().filter(UserIpMap.user == self).all()
         return [x.ip_addr for x in ret]
     @property
     def full_name(self):
         return '%s %s' % (self.firstname, self.lastname)
     @property
     def full_name_or_username(self):
         """
         Show full name.
         If full name is not set, fall back to username.
         """
         return ('%s %s' % (self.firstname, self.lastname)
                 if (self.firstname and self.lastname) else self.username)
     @property
     def full_name_and_username(self):
         """
         Show full name and username as 'Firstname Lastname (username)'.
         If full name is not set, fall back to username.
         """
         return ('%s %s (%s)' % (self.firstname, self.lastname, self.username)
                 if (self.firstname and self.lastname) else self.username)
     @property
     def full_contact(self):
         return '%s %s <%s>' % (self.firstname, self.lastname, self.email)
     @property
     def short_contact(self):
         return '%s %s' % (self.firstname, self.lastname)
     @property
     def is_admin(self):
         return self.admin
     @property
     def AuthUser(self):
         """
         Returns instance of AuthUser for this user
         """
         from kallithea.lib.auth import AuthUser
         return AuthUser(dbuser=self)
     @hybrid_property
     def user_data(self):
         if not self._user_data:
             return {}
         try:
             return json.loads(self._user_data)
         except TypeError:
             return {}
     @user_data.setter
     def user_data(self, val):
         try:
             self._user_data = json.dumps(val)
         except Exception:
             log.error(traceback.format_exc())
     def __unicode__(self):
         return u"<%s('id:%s:%s')>" % (self.__class__.__name__,
                                       self.user_id, self.username)
     @classmethod
     def get_or_404(cls, id_, allow_default=True):
         '''
         Overridden version of BaseModel.get_or_404, with an extra check on
         the default user.
         '''
         user = super(User, cls).get_or_404(id_)
         if allow_default == False:
             if user.username == User.DEFAULT_USER:
                 raise DefaultUserException
         return user
     @classmethod
     def get_by_username(cls, username, case_insensitive=False, cache=False):
         if case_insensitive:
             q = cls.query().filter(cls.username.ilike(username))
         else:
             q = cls.query().filter(cls.username == username)
         if cache:
             q = q.options(FromCache(
                             "sql_cache_short",
                             "get_user_%s" % _hash_key(username)
+                          )
+            )
         return q.scalar()
     @classmethod
     def get_by_api_key(cls, api_key, cache=False, fallback=True):
         if len(api_key) != 40 or not api_key.isalnum():
             return None
         q = cls.query().filter(cls.api_key == api_key)
         if cache:
             q = q.options(FromCache("sql_cache_short",
                                     "get_api_key_%s" % api_key))
         res = q.scalar()
         if fallback and not res:
             #fallback to additional keys
             _res = UserApiKeys.query() \
                 .filter(UserApiKeys.api_key == api_key) \
                 .filter(or_(UserApiKeys.expires == -1,
                             UserApiKeys.expires >= time.time())) \
                 .first()
             if _res:
                 res = _res.user
         return res
     @classmethod
     def get_by_email(cls, email, case_insensitive=False, cache=False):
         if case_insensitive:
             q = cls.query().filter(cls.email.ilike(email))
         else:
             q = cls.query().filter(cls.email == email)
     def get_by_email(cls, email, cache=False):
         q = cls.query().filter(cls.email.ilike(email))
         if cache:
             q = q.options(FromCache("sql_cache_short",
                                     "get_email_key_%s" % email))
         ret = q.scalar()
         if ret is None:
             q = UserEmailMap.query()
             # try fetching in alternate email map
             if case_insensitive:
                 q = q.filter(UserEmailMap.email.ilike(email))
             else:
                 q = q.filter(UserEmailMap.email == email)
             q = q.filter(UserEmailMap.email.ilike(email))
             q = q.options(joinedload(UserEmailMap.user))
             if cache:
                 q = q.options(FromCache("sql_cache_short",
                                         "get_email_map_key_%s" % email))
             ret = getattr(q.scalar(), 'user', None)
         return ret
     @classmethod
     def get_from_cs_author(cls, author):
         """
         Tries to get User objects out of commit author string
         :param author:
         """
         from kallithea.lib.helpers import email, author_name
         # Valid email in the attribute passed, see if they're in the system
         _email = email(author)
         if _email:
-            user = cls.get_by_email(_email, case_insensitive=True)
             user = cls.get_by_email(_email)
             if user is not None:
                 return user
         # Maybe we can match by username?
         _author = author_name(author)
         user = cls.get_by_username(_author, case_insensitive=True)
         if user is not None:
             return user
     def update_lastlogin(self):
         """Update user lastlogin"""
         self.last_login = datetime.datetime.now()
         Session().add(self)
         log.debug('updated user %s lastlogin', self.username)
     @classmethod
     def get_first_admin(cls):
         user = User.query().filter(User.admin == True).first()
         if user is None:
             raise Exception('Missing administrative account!')
         return user
     @classmethod
     def get_default_user(cls, cache=False):
         user = User.get_by_username(User.DEFAULT_USER, cache=cache)
         if user is None:
             raise Exception('Missing default account!')
         return user
     def get_api_data(self, details=False):
         """
         Common function for generating user related data for API
         """
         user = self
         data = dict(
             user_id=user.user_id,
             username=user.username,
             firstname=user.name,
             lastname=user.lastname,
             email=user.email,
             emails=user.emails,
             active=user.active,
             admin=user.admin,
+        )
         if details:
             data.update(dict(
                 extern_type=user.extern_type,
                 extern_name=user.extern_name,
                 api_key=user.api_key,
                 api_keys=user.api_keys,
                 last_login=user.last_login,
                 ip_addresses=user.ip_addresses
                 ))
         return data
     def __json__(self):
         data = dict(
             full_name=self.full_name,
             full_name_or_username=self.full_name_or_username,
             short_contact=self.short_contact,
             full_contact=self.full_contact
+        )
         data.update(self.get_api_data())
         return data
 class UserApiKeys(Base, BaseModel):
     __tablename__ = 'user_api_keys'
     __table_args__ = (
         Index('uak_api_key_idx', 'api_key'),
         Index('uak_api_key_expires_idx', 'api_key', 'expires'),
         _table_args_default_dict,
+    )
     __mapper_args__ = {}
     user_api_key_id = Column(Integer(), unique=True, primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     api_key = Column(String(255), nullable=False, unique=True)
     description = Column(UnicodeText(1024), nullable=False)
     expires = Column(Float(53), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     user = relationship('User')
     @property
     def expired(self):
         if self.expires == -1:
             return False
         return time.time() > self.expires
 class UserEmailMap(Base, BaseModel):
     __tablename__ = 'user_email_map'
     __table_args__ = (
         Index('uem_email_idx', 'email'),
         _table_args_default_dict,
+    )
     __mapper_args__ = {}
     email_id = Column(Integer(), unique=True, primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     _email = Column("email", String(255), nullable=False, unique=True)
     user = relationship('User')
     @validates('_email')
     def validate_email(self, key, email):
         # check if this email is not main one
         main_email = Session().query(User).filter(User.email == email).scalar()
         if main_email is not None:
             raise AttributeError('email %s is present is user table' % email)
         return email
     @hybrid_property
     def email(self):
         return self._email
     @email.setter
     def email(self, val):
         self._email = val.lower() if val else None
 class UserIpMap(Base, BaseModel):
     __tablename__ = 'user_ip_map'
     __table_args__ = (
         UniqueConstraint('user_id', 'ip_addr'),
         _table_args_default_dict,
+    )
     __mapper_args__ = {}
     ip_id = Column(Integer(), unique=True, primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     ip_addr = Column(String(255), nullable=False)
     active = Column(Boolean(), nullable=False, default=True)
     user = relationship('User')
     @classmethod
     def _get_ip_range(cls, ip_addr):
         from kallithea.lib import ipaddr
         net = ipaddr.IPNetwork(address=ip_addr)
         return [str(net.network), str(net.broadcast)]
     def __json__(self):
         return dict(
           ip_addr=self.ip_addr,
           ip_range=self._get_ip_range(self.ip_addr)
+        )
     def __unicode__(self):
         return u"<%s('user_id:%s=>%s')>" % (self.__class__.__name__,
                                             self.user_id, self.ip_addr)
 class UserLog(Base, BaseModel):
     __tablename__ = 'user_logs'
     __table_args__ = (
         _table_args_default_dict,
+    )
     user_log_id = Column(Integer(), unique=True, primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=True)
     username = Column(String(255), nullable=False)
     repository_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=True)
     repository_name = Column(Unicode(255), nullable=False)
     user_ip = Column(String(255), nullable=True)
     action = Column(UnicodeText(1200000), nullable=False)
     action_date = Column(DateTime(timezone=False), nullable=False)
     def __unicode__(self):
         return u"<%s('id:%s:%s')>" % (self.__class__.__name__,
                                       self.repository_name,
                                       self.action)
     @property
     def action_as_day(self):
         return datetime.date(*self.action_date.timetuple()[:3])
     user = relationship('User')
     repository = relationship('Repository', cascade='')
 class UserGroup(Base, BaseModel):
     __tablename__ = 'users_groups'
     __table_args__ = (
         _table_args_default_dict,
+    )
     users_group_id = Column(Integer(), unique=True, primary_key=True)
     users_group_name = Column(Unicode(255), nullable=False, unique=True)
     user_group_description = Column(Unicode(10000), nullable=True) # FIXME: not nullable?
     users_group_active = Column(Boolean(), nullable=False)
     inherit_default_permissions = Column("users_group_inherit_default_permissions", Boolean(), nullable=False, default=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     _group_data = Column("group_data", LargeBinary(), nullable=True)  # JSON data # FIXME: not nullable?
     members = relationship('UserGroupMember', cascade="all, delete-orphan")
     users_group_to_perm = relationship('UserGroupToPerm', cascade='all')
     users_group_repo_to_perm = relationship('UserGroupRepoToPerm', cascade='all')
     users_group_repo_group_to_perm = relationship('UserGroupRepoGroupToPerm', cascade='all')
     user_user_group_to_perm = relationship('UserUserGroupToPerm ', cascade='all')
     user_group_user_group_to_perm = relationship('UserGroupUserGroupToPerm ', primaryjoin="UserGroupUserGroupToPerm.target_user_group_id==UserGroup.users_group_id", cascade='all')
     user = relationship('User')
     @hybrid_property
     def group_data(self):
         if not self._group_data:
             return {}
         try:
             return json.loads(self._group_data)
         except TypeError:
             return {}
     @group_data.setter
     def group_data(self, val):
         try:
             self._group_data = json.dumps(val)
         except Exception:
             log.error(traceback.format_exc())
     def __unicode__(self):
         return u"<%s('id:%s:%s')>" % (self.__class__.__name__,
                                       self.users_group_id,
                                       self.users_group_name)
     @classmethod
     def get_by_group_name(cls, group_name, cache=False,
                           case_insensitive=False):
         if case_insensitive:
             q = cls.query().filter(cls.users_group_name.ilike(group_name))
         else:
             q = cls.query().filter(cls.users_group_name == group_name)
         if cache:
             q = q.options(FromCache(
                             "sql_cache_short",
                             "get_group_%s" % _hash_key(group_name)
+                          )
+            )
         return q.scalar()
     @classmethod
     def get(cls, user_group_id, cache=False):
         user_group = cls.query()
         if cache:
             user_group = user_group.options(FromCache("sql_cache_short",
                                     "get_users_group_%s" % user_group_id))
         return user_group.get(user_group_id)
     def get_api_data(self, with_members=True):
         user_group = self
         data = dict(
             users_group_id=user_group.users_group_id,
             group_name=user_group.users_group_name,
             group_description=user_group.user_group_description,
             active=user_group.users_group_active,
             owner=user_group.user.username,
+        )
         if with_members:
             members = []
             for user in user_group.members:
                 user = user.user
                 members.append(user.get_api_data())
             data['members'] = members
         return data
 class UserGroupMember(Base, BaseModel):
     __tablename__ = 'users_groups_members'
     __table_args__ = (
         _table_args_default_dict,
+    )
     users_group_member_id = Column(Integer(), unique=True, primary_key=True)
     users_group_id = Column(Integer(), ForeignKey('users_groups.users_group_id'), nullable=False)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     user = relationship('User')
     users_group = relationship('UserGroup')
     def __init__(self, gr_id='', u_id=''):
         self.users_group_id = gr_id
         self.user_id = u_id
 class RepositoryField(Base, BaseModel):
     __tablename__ = 'repositories_fields'
     __table_args__ = (
         UniqueConstraint('repository_id', 'field_key'),  # no-multi field
         _table_args_default_dict,
+    )
     PREFIX = 'ex_'  # prefix used in form to not conflict with already existing fields
     repo_field_id = Column(Integer(), unique=True, primary_key=True)
     repository_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)
     field_key = Column(String(250), nullable=False)
     field_label = Column(String(1024), nullable=False)
     field_value = Column(String(10000), nullable=False)
     field_desc = Column(String(1024), nullable=False)
     field_type = Column(String(255), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     repository = relationship('Repository')
     @property
     def field_key_prefixed(self):
         return 'ex_%s' % self.field_key
     @classmethod
     def un_prefix_key(cls, key):
         if key.startswith(cls.PREFIX):
             return key[len(cls.PREFIX):]
         return key
     @classmethod
     def get_by_key_name(cls, key, repo):
         row = cls.query() \
                 .filter(cls.repository == repo) \
                 .filter(cls.field_key == key).scalar()
         return row
 class Repository(Base, BaseModel):
     __tablename__ = 'repositories'
     __table_args__ = (
         Index('r_repo_name_idx', 'repo_name'),
         _table_args_default_dict,
+    )
     DEFAULT_CLONE_URI = '{scheme}://{user}@{netloc}/{repo}'
     DEFAULT_CLONE_URI_ID = '{scheme}://{user}@{netloc}/_{repoid}'
     STATE_CREATED = 'repo_state_created'
     STATE_PENDING = 'repo_state_pending'
     STATE_ERROR = 'repo_state_error'
     repo_id = Column(Integer(), unique=True, primary_key=True)
     repo_name = Column(Unicode(255), nullable=False, unique=True)
     repo_state = Column(String(255), nullable=False)
     clone_uri = Column(String(255), nullable=True) # FIXME: not nullable?
     repo_type = Column(String(255), nullable=False)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     private = Column(Boolean(), nullable=False)
     enable_statistics = Column("statistics", Boolean(), nullable=False, default=True)
     enable_downloads = Column("downloads", Boolean(), nullable=False, default=True)
     description = Column(Unicode(10000), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     updated_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     _landing_revision = Column("landing_revision", String(255), nullable=False)
     enable_locking = Column(Boolean(), nullable=False, default=False)
     _locked = Column("locked", String(255), nullable=True) # FIXME: not nullable?
     _changeset_cache = Column("changeset_cache", LargeBinary(), nullable=True) # JSON data # FIXME: not nullable?
     fork_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=True)
     group_id = Column(Integer(), ForeignKey('groups.group_id'), nullable=True)
     user = relationship('User')
     fork = relationship('Repository', remote_side=repo_id)
     group = relationship('RepoGroup')
     repo_to_perm = relationship('UserRepoToPerm', cascade='all', order_by='UserRepoToPerm.repo_to_perm_id')
     users_group_to_perm = relationship('UserGroupRepoToPerm', cascade='all')
     stats = relationship('Statistics', cascade='all', uselist=False)
     followers = relationship('UserFollowing',
                              primaryjoin='UserFollowing.follows_repo_id==Repository.repo_id',
                              cascade='all')
     extra_fields = relationship('RepositoryField',
                                 cascade="all, delete-orphan")
     logs = relationship('UserLog')
     comments = relationship('ChangesetComment', cascade="all, delete-orphan")
     pull_requests_org = relationship('PullRequest',
                     primaryjoin='PullRequest.org_repo_id==Repository.repo_id',
                     cascade="all, delete-orphan")
     pull_requests_other = relationship('PullRequest',
                     primaryjoin='PullRequest.other_repo_id==Repository.repo_id',
                     cascade="all, delete-orphan")
     def __unicode__(self):
         return u"<%s('%s:%s')>" % (self.__class__.__name__, self.repo_id,

kallithea/model/validators.py

➞

Show inline comments

@@ @@ -338,534 +338,534 @@ def ValidAuthToken(): @@
         def validate_python(self, value, state):
             if value != authentication_token():
                 msg = M(self, 'invalid_token', state)
                 raise formencode.Invalid(msg, value, state)
     return _validator
 def ValidRepoName(edit=False, old_data=None):
     old_data = old_data or {}
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_repo_name':
                 _('Repository name %(repo)s is not allowed'),
             'repository_exists':
                 _('Repository named %(repo)s already exists'),
             'repository_in_group_exists': _('Repository "%(repo)s" already '
                                             'exists in group "%(group)s"'),
             'same_group_exists': _('Repository group with name "%(repo)s" '
                                    'already exists')
+        }
         def _to_python(self, value, state):
             repo_name = repo_name_slug(value.get('repo_name', ''))
             repo_group = value.get('repo_group')
             if repo_group:
                 gr = RepoGroup.get(repo_group)
                 group_path = gr.full_path
                 group_name = gr.group_name
                 # value needs to be aware of group name in order to check
                 # db key This is an actual just the name to store in the
                 # database
                 repo_name_full = group_path + RepoGroup.url_sep() + repo_name
             else:
                 group_name = group_path = ''
                 repo_name_full = repo_name
             value['repo_name'] = repo_name
             value['repo_name_full'] = repo_name_full
             value['group_path'] = group_path
             value['group_name'] = group_name
             return value
         def validate_python(self, value, state):
             repo_name = value.get('repo_name')
             repo_name_full = value.get('repo_name_full')
             group_path = value.get('group_path')
             group_name = value.get('group_name')
             if repo_name in [ADMIN_PREFIX, '']:
                 msg = M(self, 'invalid_repo_name', state, repo=repo_name)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(repo_name=msg)
+                )
             rename = old_data.get('repo_name') != repo_name_full
             create = not edit
             if rename or create:
                 if group_path != '':
                     if Repository.get_by_repo_name(repo_name_full):
                         msg = M(self, 'repository_in_group_exists', state,
                                 repo=repo_name, group=group_name)
                         raise formencode.Invalid(msg, value, state,
                             error_dict=dict(repo_name=msg)
+                        )
                 elif RepoGroup.get_by_group_name(repo_name_full):
                         msg = M(self, 'same_group_exists', state,
                                 repo=repo_name)
                         raise formencode.Invalid(msg, value, state,
                             error_dict=dict(repo_name=msg)
+                        )
                 elif Repository.get_by_repo_name(repo_name_full):
                         msg = M(self, 'repository_exists', state,
                                 repo=repo_name)
                         raise formencode.Invalid(msg, value, state,
                             error_dict=dict(repo_name=msg)
+                        )
             return value
     return _validator
 def ValidForkName(*args, **kwargs):
     return ValidRepoName(*args, **kwargs)
 def SlugifyName():
     class _validator(formencode.validators.FancyValidator):
         def _to_python(self, value, state):
             return repo_name_slug(value)
         def validate_python(self, value, state):
             pass
     return _validator
 def ValidCloneUri():
     from kallithea.lib.utils import make_ui
     def url_handler(repo_type, url, ui):
         if repo_type == 'hg':
             from kallithea.lib.vcs.backends.hg.repository import MercurialRepository
             if url.startswith('http') or url.startswith('ssh'):
                 # initially check if it's at least the proper URL
                 # or does it pass basic auth
                 MercurialRepository._check_url(url, ui)
             elif url.startswith('svn+http'):
                 from hgsubversion.svnrepo import svnremoterepo
                 svnremoterepo(ui, url).svn.uuid
             elif url.startswith('git+http'):
                 raise NotImplementedError()
             else:
                 raise Exception('clone from URI %s not allowed' % (url,))
         elif repo_type == 'git':
             from kallithea.lib.vcs.backends.git.repository import GitRepository
             if url.startswith('http') or url.startswith('git'):
                 # initially check if it's at least the proper URL
                 # or does it pass basic auth
                 GitRepository._check_url(url)
             elif url.startswith('svn+http'):
                 raise NotImplementedError()
             elif url.startswith('hg+http'):
                 raise NotImplementedError()
             else:
                 raise Exception('clone from URI %s not allowed' % (url))
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'clone_uri': _('Invalid repository URL'),
             'invalid_clone_uri': _('Invalid repository URL. It must be a '
                                    'valid http, https, ssh, svn+http or svn+https URL'),
+        }
         def validate_python(self, value, state):
             repo_type = value.get('repo_type')
             url = value.get('clone_uri')
             if url and url != value.get('clone_uri_hidden'):
                 try:
                     url_handler(repo_type, url, make_ui('db', clear_session=False))
                 except Exception:
                     log.exception('URL validation failed')
                     msg = M(self, 'clone_uri')
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(clone_uri=msg)
+                    )
     return _validator
 def ValidForkType(old_data=None):
     old_data = old_data or {}
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_fork_type': _('Fork has to be the same type as parent')
+        }
         def validate_python(self, value, state):
             if old_data['repo_type'] != value:
                 msg = M(self, 'invalid_fork_type', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(repo_type=msg)
+                )
     return _validator
 def CanWriteGroup(old_data=None):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'permission_denied': _("You don't have permissions "
                                    "to create repository in this group"),
             'permission_denied_root': _("no permission to create repository "
                                         "in root location")
+        }
         def _to_python(self, value, state):
             #root location
             if value == -1:
                 return None
             return value
         def validate_python(self, value, state):
             gr = RepoGroup.get(value)
             gr_name = gr.group_name if gr is not None else None # None means ROOT location
             # create repositories with write permission on group is set to true
             create_on_write = HasPermissionAny('hg.create.write_on_repogroup.true')()
             group_admin = HasRepoGroupPermissionAny('group.admin')(gr_name,
                                             'can write into group validator')
             group_write = HasRepoGroupPermissionAny('group.write')(gr_name,
                                             'can write into group validator')
             forbidden = not (group_admin or (group_write and create_on_write))
             can_create_repos = HasPermissionAny('hg.admin', 'hg.create.repository')
             gid = (old_data['repo_group'].get('group_id')
                    if (old_data and 'repo_group' in old_data) else None)
             value_changed = gid != value
             new = not old_data
             # do check if we changed the value, there's a case that someone got
             # revoked write permissions to a repository, he still created, we
             # don't need to check permission if he didn't change the value of
             # groups in form box
             if value_changed or new:
                 #parent group need to be existing
                 if gr and forbidden:
                     msg = M(self, 'permission_denied', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(repo_type=msg)
+                    )
                 ## check if we can write to root location !
                 elif gr is None and not can_create_repos():
                     msg = M(self, 'permission_denied_root', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(repo_type=msg)
+                    )
     return _validator
 def CanCreateGroup(can_create_in_root=False):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'permission_denied': _("You don't have permissions "
                                    "to create a group in this location")
+        }
         def to_python(self, value, state):
             #root location
             if value == -1:
                 return None
             return value
         def validate_python(self, value, state):
             gr = RepoGroup.get(value)
             gr_name = gr.group_name if gr is not None else None # None means ROOT location
             if can_create_in_root and gr is None:
                 #we can create in root, we're fine no validations required
                 return
             forbidden_in_root = gr is None and not can_create_in_root
             val = HasRepoGroupPermissionAny('group.admin')
             forbidden = not val(gr_name, 'can create group validator')
             if forbidden_in_root or forbidden:
                 msg = M(self, 'permission_denied', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(group_parent_id=msg)
+                )
     return _validator
 def ValidPerms(type_='repo'):
     if type_ == 'repo_group':
         EMPTY_PERM = 'group.none'
     elif type_ == 'repo':
         EMPTY_PERM = 'repository.none'
     elif type_ == 'user_group':
         EMPTY_PERM = 'usergroup.none'
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'perm_new_member_name':
                 _('This username or user group name is not valid')
+        }
         def to_python(self, value, state):
             perms_update = OrderedSet()
             perms_new = OrderedSet()
             # build a list of permission to update and new permission to create
             #CLEAN OUT ORG VALUE FROM NEW MEMBERS, and group them using
             new_perms_group = defaultdict(dict)
             for k, v in value.copy().iteritems():
                 if k.startswith('perm_new_member'):
                     del value[k]
                     _type, part = k.split('perm_new_member_')
                     args = part.split('_')
                     if len(args) == 1:
                         new_perms_group[args[0]]['perm'] = v
                     elif len(args) == 2:
                         _key, pos = args
                         new_perms_group[pos][_key] = v
             # fill new permissions in order of how they were added
             for k in sorted(map(int, new_perms_group.keys())):
                 perm_dict = new_perms_group[str(k)]
                 new_member = perm_dict.get('name')
                 new_perm = perm_dict.get('perm')
                 new_type = perm_dict.get('type')
                 if new_member and new_perm and new_type:
                     perms_new.add((new_member, new_perm, new_type))
             for k, v in value.iteritems():
                 if k.startswith('u_perm_') or k.startswith('g_perm_'):
                     member = k[7:]
                     t = {'u': 'user',
                          'g': 'users_group'
                     }[k[0]]
                     if member == User.DEFAULT_USER:
                         if str2bool(value.get('repo_private')):
                             # set none for default when updating to
                             # private repo protects against form manipulation
                             v = EMPTY_PERM
                     perms_update.add((member, v, t))
             value['perms_updates'] = list(perms_update)
             value['perms_new'] = list(perms_new)
             # update permissions
             for k, v, t in perms_new:
                 try:
                     if t is 'user':
                         self.user_db = User.query() \
                             .filter(User.active == True) \
                             .filter(User.username == k).one()
                     if t is 'users_group':
                         self.user_db = UserGroup.query() \
                             .filter(UserGroup.users_group_active == True) \
                             .filter(UserGroup.users_group_name == k).one()
                 except Exception:
                     log.exception('Updated permission failed')
                     msg = M(self, 'perm_new_member_type', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(perm_new_member_name=msg)
+                    )
             return value
     return _validator
 def ValidSettings():
     class _validator(formencode.validators.FancyValidator):
         def _to_python(self, value, state):
             # settings  form for users that are not admin
             # can't edit certain parameters, it's extra backup if they mangle
             # with forms
             forbidden_params = [
                 'user', 'repo_type', 'repo_enable_locking',
                 'repo_enable_downloads', 'repo_enable_statistics'
+            ]
             for param in forbidden_params:
                 if param in value:
                     del value[param]
             return value
         def validate_python(self, value, state):
             pass
     return _validator
 def ValidPath():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_path': _('This is not a valid path')
+        }
         def validate_python(self, value, state):
             if not os.path.isdir(value):
                 msg = M(self, 'invalid_path', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(paths_root_path=msg)
+                )
     return _validator
 def UniqSystemEmail(old_data=None):
     old_data = old_data or {}
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'email_taken': _('This email address is already in use')
+        }
         def _to_python(self, value, state):
             return value.lower()
         def validate_python(self, value, state):
             if (old_data.get('email') or '').lower() != value:
-                user = User.get_by_email(value, case_insensitive=True)
                 user = User.get_by_email(value)
                 if user is not None:
                     msg = M(self, 'email_taken', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(email=msg)
+                    )
     return _validator
 def ValidSystemEmail():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'non_existing_email': _('Email address "%(email)s" not found')
+        }
         def _to_python(self, value, state):
             return value.lower()
         def validate_python(self, value, state):
-            user = User.get_by_email(value, case_insensitive=True)
             user = User.get_by_email(value)
             if user is None:
                 msg = M(self, 'non_existing_email', state, email=value)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(email=msg)
+                )
     return _validator
 def LdapLibValidator():
     class _validator(formencode.validators.FancyValidator):
         messages = {
+        }
         def validate_python(self, value, state):
             try:
                 import ldap
                 ldap  # pyflakes silence !
             except ImportError:
                 raise LdapImportError()
     return _validator
 def AttrLoginValidator():
     class _validator(formencode.validators.UnicodeString):
         messages = {
             'invalid_cn':
                   _('The LDAP Login attribute of the CN must be specified - '
                     'this is the name of the attribute that is equivalent '
                     'to "username"')
+        }
         messages['empty'] = messages['invalid_cn']
     return _validator
 def ValidIp():
     class _validator(CIDR):
         messages = dict(
             badFormat=_('Please enter a valid IPv4 or IPv6 address'),
             illegalBits=_('The network size (bits) must be within the range'
                 ' of 0-32 (not %(bits)r)')
+        )
         def to_python(self, value, state):
             v = super(_validator, self).to_python(value, state)
             v = v.strip()
             net = ipaddr.IPNetwork(address=v)
             if isinstance(net, ipaddr.IPv4Network):
                 #if IPv4 doesn't end with a mask, add /32
                 if '/' not in value:
                     v += '/32'
             if isinstance(net, ipaddr.IPv6Network):
                 #if IPv6 doesn't end with a mask, add /128
                 if '/' not in value:
                     v += '/128'
             return v
         def validate_python(self, value, state):
             try:
                 addr = value.strip()
                 #this raises an ValueError if address is not IPv4 or IPv6
                 ipaddr.IPNetwork(address=addr)
             except ValueError:
                 raise formencode.Invalid(self.message('badFormat', state),
                                          value, state)
     return _validator
 def FieldKey():
     class _validator(formencode.validators.FancyValidator):
         messages = dict(
             badFormat=_('Key name can only consist of letters, '
                         'underscore, dash or numbers')
+        )
         def validate_python(self, value, state):
             if not re.match('[a-zA-Z0-9_-]+$', value):
                 raise formencode.Invalid(self.message('badFormat', state),
                                          value, state)
     return _validator
 def BasePath():
     class _validator(formencode.validators.FancyValidator):
         messages = dict(
             badPath=_('Filename cannot be inside a directory')
+        )
         def _to_python(self, value, state):
             return value
         def validate_python(self, value, state):
             if value != os.path.basename(value):
                 raise formencode.Invalid(self.message('badPath', state),
                                          value, state)
     return _validator
 def ValidAuthPlugins():
     class _validator(formencode.validators.FancyValidator):
         messages = dict(
             import_duplicate=_('Plugins %(loaded)s and %(next_to_load)s both export the same name')
+        )
         def _to_python(self, value, state):
             # filter empty values
             return filter(lambda s: s not in [None, ''], value)
         def validate_python(self, value, state):
             from kallithea.lib import auth_modules
             module_list = value
             unique_names = {}
             try:
                 for module in module_list:
                     plugin = auth_modules.loadplugin(module)
                     plugin_name = plugin.name
                     if plugin_name in unique_names:
                         msg = M(self, 'import_duplicate', state,
                                 loaded=unique_names[plugin_name],
                                 next_to_load=plugin_name)
                         raise formencode.Invalid(msg, value, state)
                     unique_names[plugin_name] = plugin
             except (ImportError, AttributeError, TypeError) as e:
                 raise formencode.Invalid(str(e), value, state)
     return _validator

kallithea/tests/models/test_users.py

➞

Show inline comments

 from kallithea.tests import *
 from kallithea.model.db import User, UserGroup, UserGroupMember, UserEmailMap, \
     Permission
 from kallithea.model.user import UserModel
 from kallithea.model.meta import Session
 from kallithea.model.user_group import UserGroupModel
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 class TestUser(BaseTestCase):
     def __init__(self, methodName='runTest'):
         Session.remove()
         super(TestUser, self).__init__(methodName=methodName)
     def tearDown(self):
         Session.remove()
     def test_create_and_remove(self):
         usr = UserModel().create_or_update(username=u'test_user',
                                            password=u'qweqwe',
                                            email=u'u232@example.com',
                                            firstname=u'u1', lastname=u'u1')
         Session().commit()
         self.assertEqual(User.get_by_username(u'test_user'), usr)
         # make user group
         user_group = fixture.create_user_group(u'some_example_group')
         Session().commit()
         UserGroupModel().add_user_to_group(user_group, usr)
         Session().commit()
         self.assertEqual(UserGroup.get(user_group.users_group_id), user_group)
         self.assertEqual(UserGroupMember.query().count(), 1)
         UserModel().delete(usr.user_id)
         Session().commit()
         self.assertEqual(UserGroupMember.query().all(), [])
     def test_additonal_email_as_main(self):
         usr = UserModel().create_or_update(username=u'test_user',
                                            password=u'qweqwe',
                                      email=u'main_email@example.com',
                                      firstname=u'u1', lastname=u'u1')
         Session().commit()
         def do():
             m = UserEmailMap()
             m.email = u'main_email@example.com'
             m.user = usr
             Session().add(m)
             Session().commit()
         self.assertRaises(AttributeError, do)
         UserModel().delete(usr.user_id)
         Session().commit()
     def test_extra_email_map(self):
         usr = UserModel().create_or_update(username=u'test_user',
                                            password=u'qweqwe',
                                      email=u'main_email@example.com',
                                      firstname=u'u1', lastname=u'u1')
         Session().commit()
         m = UserEmailMap()
         m.email = u'main_email2@example.com'
         m.user = usr
         Session().add(m)
         Session().commit()
         u = User.get_by_email(email='MAIN_email@example.com')
         self.assertEqual(usr.user_id, u.user_id)
         self.assertEqual(usr.username, u.username)
         u = User.get_by_email(email='main_email@example.com')
         self.assertEqual(usr.user_id, u.user_id)
         self.assertEqual(usr.username, u.username)
         u = User.get_by_email(email='main_email2@example.com')
         self.assertEqual(usr.user_id, u.user_id)
         self.assertEqual(usr.username, u.username)
         u = User.get_by_email(email='main_email3@example.com')
         self.assertEqual(None, u)
         u = User.get_by_email(email='main_e%ail@example.com')
         self.assertEqual(None, u)
         u = User.get_by_email(email='main_emai_@example.com')
         self.assertEqual(None, u)
         UserModel().delete(usr.user_id)
         Session().commit()
 class TestUsers(BaseTestCase):
     def __init__(self, methodName='runTest'):
         super(TestUsers, self).__init__(methodName=methodName)
     def setUp(self):
         self.u1 = UserModel().create_or_update(username=u'u1',
                                         password=u'qweqwe',
                                         email=u'u1@example.com',
                                         firstname=u'u1', lastname=u'u1')
     def tearDown(self):
         perm = Permission.query().all()
         for p in perm:
             UserModel().revoke_perm(self.u1, p)
         UserModel().delete(self.u1)
         Session().commit()
         Session.remove()
     def test_add_perm(self):
         perm = Permission.query().all()[0]
         UserModel().grant_perm(self.u1, perm)
         Session().commit()
         self.assertEqual(UserModel().has_perm(self.u1, perm), True)
     def test_has_perm(self):
         perm = Permission.query().all()
         for p in perm:
             has_p = UserModel().has_perm(self.u1, p)
             self.assertEqual(False, has_p)
     def test_revoke_perm(self):
         perm = Permission.query().all()[0]
         UserModel().grant_perm(self.u1, perm)
         Session().commit()
         self.assertEqual(UserModel().has_perm(self.u1, perm), True)
         #revoke
         UserModel().revoke_perm(self.u1, perm)
         Session().commit()
         self.assertEqual(UserModel().has_perm(self.u1, perm), False)

0 comments (0 inline, 0 general)