def get_by_key(cls, key):

        return cls.query().filter(cls.ui_key == key).scalar()

    @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]))

        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('/').ui_value

    @classmethod

    def create_or_update_hook(cls, key, val):

        new_ui = cls.get_by_key(key) or cls()

        new_ui.ui_section = 'hooks'

        new_ui.ui_active = True

        new_ui.ui_key = key

        new_ui.ui_value = val

        Session().add(new_ui)

    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__ = (

        UniqueConstraint('username'), UniqueConstraint('email'),

        Index('u_username_idx', 'username'),

        Index('u_email_idx', 'email'),

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8', 'sqlite_autoincrement': True}

    )

    DEFAULT_USER = 'default'

    DEFAULT_GRAVATAR_URL = 'https://secure.gravatar.com/avatar/{md5email}?d=identicon&s={size}'

    user_id = Column("user_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)

    username = Column("username", String(255, convert_unicode=False), nullable=True, unique=None, default=None)

    password = Column("password", String(255, convert_unicode=False), nullable=True, unique=None, default=None)

    active = Column("active", Boolean(), nullable=True, unique=None, default=True)

    admin = Column("admin", Boolean(), nullable=True, unique=None, default=False)

    name = Column("firstname", String(255, convert_unicode=False), nullable=True, unique=None, default=None)

    lastname = Column("lastname", String(255, convert_unicode=False), nullable=True, unique=None, default=None)

    _email = Column("email", String(255, convert_unicode=False), nullable=True, unique=None, default=None)

    last_login = Column("last_login", DateTime(timezone=False), nullable=True, unique=None, default=None)

    extern_type = Column("extern_type", String(255, convert_unicode=False), nullable=True, unique=None, default=None)

    extern_name = Column("extern_name", String(255, convert_unicode=False), nullable=True, unique=None, default=None)

    api_key = Column("api_key", String(255, convert_unicode=False), nullable=True, unique=None, default=None)

    inherit_default_permissions = Column("inherit_default_permissions", Boolean(), nullable=False, unique=None, default=True)

    created_on = Column('created_on', DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    _user_data = Column("user_data", LargeBinary(), nullable=True)  # JSON data

    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 username_and_name(self):

        return '%s (%s %s)' % (self.username, self.firstname, self.lastname)

    @property

    def full_name(self):

        return '%s %s' % (self.firstname, self.lastname)

    @property

    def full_name_or_username(self):

        return ('%s %s' % (self.firstname, self.lastname)

                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(user_id=self.user_id, api_key=self.api_key,

                        username=self.username)

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

        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)

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

            if user:

                return user

        # Maybe we can match by username?

        _author = author_name(author)

        user = cls.get_by_username(_author, case_insensitive=True)

        if user:

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

        UniqueConstraint('api_key'),

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8', 'sqlite_autoincrement': True}

    )

    __mapper_args__ = {}

    user_api_key_id = Column("user_api_key_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)

    user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=True, unique=None, default=None)

    api_key = Column("api_key", String(255, convert_unicode=False), nullable=False, unique=True)

    description = Column('description', UnicodeText(1024))

    expires = Column('expires', Float(53), nullable=False)

    created_on = Column('created_on', 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'),

        UniqueConstraint('email'),

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8', 'sqlite_autoincrement': True}

    )

    __mapper_args__ = {}

    email_id = Column("email_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)

    user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=True, unique=None, default=None)

    _email = Column("email", String(255, convert_unicode=False), nullable=True, unique=False, default=None)

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

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8', 'sqlite_autoincrement': True}

    )

    __mapper_args__ = {}

    ip_id = Column("ip_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)

    user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=True, unique=None, default=None)

                                             'password_confirmation': 'test12',

                                             'email': 'TesT_Admin@mail.COM',

                                             'firstname': 'test',

                                             'lastname': 'test'})

        msg = validators.UniqSystemEmail()()._messages['email_taken']

        response.mustcontain(msg)

    def test_register_err_wrong_data(self):

        response = self.app.post(url(controller='login', action='register'),

                                            {'username': 'xs',

                                             'password': 'test',

                                             'password_confirmation': 'test',

                                             'email': 'goodmailm',

                                             'firstname': 'test',

                                             'lastname': 'test'})

        self.assertEqual(response.status, '200 OK')

        response.mustcontain('An email address must contain a single @')

        response.mustcontain('Enter a value 6 characters long or more')

    def test_register_err_username(self):

        response = self.app.post(url(controller='login', action='register'),

                                            {'username': 'error user',

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': 'goodmailm',

                                             'firstname': 'test',

                                             'lastname': 'test'})

        response.mustcontain('An email address must contain a single @')

        response.mustcontain('Username may only contain '

                'alphanumeric characters underscores, '

                'periods or dashes and must begin with '

                'alphanumeric character')

    def test_register_err_case_sensitive(self):

        usr = 'Test_Admin'

        response = self.app.post(url(controller='login', action='register'),

                                            {'username': usr,

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': 'goodmailm',

                                             'firstname': 'test',

                                             'lastname': 'test'})

        response.mustcontain('An email address must contain a single @')

        msg = validators.ValidUsername()._messages['username_exists']

        msg = h.html_escape(msg % {'username': usr})

        response.mustcontain(msg)

    def test_register_special_chars(self):

        response = self.app.post(url(controller='login', action='register'),

                                        {'username': 'xxxaxn',

                                         'password': 'ąćźżąśśśś',

                                         'password_confirmation': 'ąćźżąśśśś',

                                         'email': 'goodmailm@test.plx',

                                         'firstname': 'test',

                                         'lastname': 'test'})

        msg = validators.ValidPassword()._messages['invalid_password']

        response.mustcontain(msg)

    def test_register_password_mismatch(self):

        response = self.app.post(url(controller='login', action='register'),

                                            {'username': 'xs',

                                             'password': '123qwe',

                                             'password_confirmation': 'qwe123',

                                             'email': 'goodmailm@test.plxa',

                                             'firstname': 'test',

                                             'lastname': 'test'})

        msg = validators.ValidPasswordsMatch()._messages['password_mismatch']

        response.mustcontain(msg)

    def test_register_ok(self):

        username = 'test_regular4'

        password = 'qweqwe'

        email = 'marcin@test.com'

        name = 'testname'

        lastname = 'testlastname'

        response = self.app.post(url(controller='login', action='register'),

                                            {'username': username,

                                             'password': password,

                                             'password_confirmation': password,

                                             'email': email,

                                             'firstname': name,

                                             'lastname': lastname,

                                             'admin': True})  # This should be overriden

        self.assertEqual(response.status, '302 Found')

        self.checkSessionFlash(response, 'You have successfully registered into Kallithea')

        ret = Session().query(User).filter(User.username == 'test_regular4').one()

        self.assertEqual(ret.username, username)

        self.assertEqual(check_password(password, ret.password), True)

        self.assertEqual(ret.email, email)

        self.assertEqual(ret.name, name)

        self.assertEqual(ret.lastname, lastname)

        self.assertNotEqual(ret.api_key, None)

        self.assertEqual(ret.admin, False)

    def test_forgot_password_wrong_mail(self):

        bad_email = 'marcin@wrongmail.org'

        response = self.app.post(

                        url(controller='login', action='password_reset'),

                            {'email': bad_email, }

        )

        msg = validators.ValidSystemEmail()._messages['non_existing_email']

        msg = h.html_escape(msg % {'email': bad_email})

        response.mustcontain()

    def test_forgot_password(self):

        response = self.app.get(url(controller='login',

                                    action='password_reset'))

        self.assertEqual(response.status, '200 OK')

        username = 'test_password_reset_1'

        password = 'qweqwe'

        email = 'marcin@python-works.com'

        name = 'passwd'

        lastname = 'reset'

        new = User()

        new.username = username

        new.password = password

        new.email = email

        new.name = name

        new.lastname = lastname

        new.api_key = generate_api_key()

        Session().add(new)

        Session().commit()

        response = self.app.post(url(controller='login',

                                     action='password_reset'),

                                 {'email': email, })

        self.checkSessionFlash(response, 'Your password reset link was sent')

        response = response.follow()

        # BAD KEY

        key = "bad"

        response = self.app.get(url(controller='login',

                                    action='password_reset_confirmation',

                                    key=key))

        self.assertEqual(response.status, '302 Found')

        self.assertTrue(response.location.endswith(url('reset_password')))

        # GOOD KEY

        key = User.get_by_username(username).api_key

        response = self.app.get(url(controller='login',

                                    action='password_reset_confirmation',

                                    key=key))

        self.assertEqual(response.status, '302 Found')

        self.assertTrue(response.location.endswith(url('login_home')))

        self.checkSessionFlash(response,

                               ('Your password reset was successful, '

                                'new password has been sent to your email'))

        response = response.follow()

    def _get_api_whitelist(self, values=None):

        config = {'api_access_controllers_whitelist': values or []}

        return config

    @parameterized.expand([

        ('none', None),

        ('empty_string', ''),

        ('fake_number', '123456'),

        ('proper_api_key', None)

    ])

    def test_access_not_whitelisted_page_via_api_key(self, test_name, api_key):

        whitelist = self._get_api_whitelist([])

        with mock.patch('kallithea.CONFIG', whitelist):

            self.assertEqual([],

                             whitelist['api_access_controllers_whitelist'])

            if test_name == 'proper_api_key':

                #use builtin if api_key is None

                api_key = User.get_first_admin().api_key

            with fixture.anon_access(False):

                self.app.get(url(controller='changeset',

                                 action='changeset_raw',

                                 repo_name=HG_REPO, revision='tip', api_key=api_key),

                             status=302)

    @parameterized.expand([

        ('none', None, 302),

        ('empty_string', '', 302),

        ('fake_number', '123456', 302),

        ('proper_api_key', None, 200)

    ])

    def test_access_whitelisted_page_via_api_key(self, test_name, api_key, code):

        whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])

        with mock.patch('kallithea.CONFIG', whitelist):

            self.assertEqual(['ChangesetController:changeset_raw'],

                             whitelist['api_access_controllers_whitelist'])

            if test_name == 'proper_api_key':

                api_key = User.get_first_admin().api_key

            with fixture.anon_access(False):

                self.app.get(url(controller='changeset',

                                 action='changeset_raw',

                                 repo_name=HG_REPO, revision='tip', api_key=api_key),

                             status=code)

    def test_access_page_via_extra_api_key(self):

        whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])

        with mock.patch('kallithea.CONFIG', whitelist):

            self.assertEqual(['ChangesetController:changeset_raw'],

                             whitelist['api_access_controllers_whitelist'])

            new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')

            Session().commit()

            with fixture.anon_access(False):

                self.app.get(url(controller='changeset',

                                 action='changeset_raw',

                                 repo_name=HG_REPO, revision='tip', api_key=new_api_key.api_key),

                             status=200)

    def test_access_page_via_expired_api_key(self):

        whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])

        with mock.patch('kallithea.CONFIG', whitelist):

            self.assertEqual(['ChangesetController:changeset_raw'],

                             whitelist['api_access_controllers_whitelist'])

            new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')

            Session().commit()

            #patch the api key and make it expired

            new_api_key.expires = 0

            Session().add(new_api_key)

            Session().commit()

            with fixture.anon_access(False):

                self.app.get(url(controller='changeset',

                                 action='changeset_raw',

                                 repo_name=HG_REPO, revision='tip',

                                 api_key=new_api_key.api_key),

                             status=302)