: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,))

            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

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:

        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_

        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

            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

                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:

            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,

        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:

                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):

            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)

        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_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)

        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)