# -*- coding: utf-8 -*-

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""

kallithea.model.ssh_key

~~~~~~~~~~~~~~~~~~~~~~~

SSH key model for Kallithea

"""

import logging

from tg.i18n import ugettext as _

from kallithea.lib.utils2 import safe_str

from kallithea.model.db import UserSshKeys, User

from kallithea.model.meta import Session

from kallithea.lib import ssh

log = logging.getLogger(__name__)

class SshKeyModelException(Exception):

    """Exception raised by SshKeyModel methods to report errors"""

class SshKeyModel(object):

    def create(self, user, description, public_key):

        """

        :param user: user or user_id

        :param description: description of SshKey

        :param publickey: public key text

        Will raise SshKeyModelException on errors

        """

        try:

        except ssh.SshKeyParseError as e:

            raise SshKeyModelException(_('SSH key %r is invalid: %s') % (safe_str(public_key), e.message))

        user = User.guess_instance(user)

        new_ssh_key = UserSshKeys()

        new_ssh_key.user_id = user.user_id

        new_ssh_key.description = description

        new_ssh_key.public_key = public_key

        for ssh_key in UserSshKeys.query().filter(UserSshKeys.fingerprint == new_ssh_key.fingerprint).all():

            raise SshKeyModelException(_('SSH key %s is already used by %s') %

                                       (new_ssh_key.fingerprint, ssh_key.user.username))

        Session().add(new_ssh_key)

        return new_ssh_key

    def delete(self, public_key, user=None):

        """

        Deletes given public_key, if user is set it also filters the object for

        deletion by given user.

        Will raise SshKeyModelException on errors

        """

        ssh_key = UserSshKeys.query().filter(UserSshKeys._public_key == public_key)

        if user:

            user = User.guess_instance(user)

            ssh_key = ssh_key.filter(UserSshKeys.user_id == user.user_id)

        ssh_key = ssh_key.scalar()

        if ssh_key is None:

            raise SshKeyModelException(_('SSH key %r not found') % safe_str(public_key))

        Session().delete(ssh_key)

    def get_ssh_keys(self, user):

        user = User.guess_instance(user)

        user_ssh_keys = UserSshKeys.query() \

            .filter(UserSshKeys.user_id == user.user_id).all()

        return user_ssh_keys

        params.pop('_authentication_token')

        updated_user = User.get_by_username(self.test_user_1)

        updated_params = updated_user.get_api_data(True)

        updated_params.update({'password_confirmation': ''})

        updated_params.update({'new_password': ''})

        assert params == updated_params

    def test_delete(self):

        self.log_user()

        username = 'newtestuserdeleteme'

        fixture.create_user(name=username)

        new_user = Session().query(User) \

            .filter(User.username == username).one()

        response = self.app.post(url('delete_user', id=new_user.user_id),

            params={'_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'Successfully deleted user')

    def test_delete_repo_err(self):

        self.log_user()

        username = 'repoerr'

        reponame = u'repoerr_fail'

        fixture.create_user(name=username)

        fixture.create_repo(name=reponame, cur_user=username)

        new_user = Session().query(User) \

            .filter(User.username == username).one()

        response = self.app.post(url('delete_user', id=new_user.user_id),

            params={'_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'User "%s" still '

                               'owns 1 repositories and cannot be removed. '

                               'Switch owners or remove those repositories: '

                               '%s' % (username, reponame))

        response = self.app.post(url('delete_repo', repo_name=reponame),

            params={'_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'Deleted repository %s' % reponame)

        response = self.app.post(url('delete_user', id=new_user.user_id),

            params={'_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'Successfully deleted user')

    def test_delete_repo_group_err(self, user_and_repo_group_fail):

        new_user, repo_group = user_and_repo_group_fail

        username = new_user.username

        groupname = repo_group.group_name

        self.log_user()

        response = self.app.post(url('delete_user', id=new_user.user_id),

            params={'_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'User "%s" still '

                               'owns 1 repository groups and cannot be removed. '

                               'Switch owners or remove those repository groups: '

                               '%s' % (username, groupname))

        # Relevant _if_ the user deletion succeeded to make sure we can render groups without owner

        # rg = RepoGroup.get_by_group_name(group_name=groupname)

        # response = self.app.get(url('repos_groups', id=rg.group_id))

        response = self.app.post(url('delete_repo_group', group_name=groupname),

            params={'_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'Removed repository group %s' % groupname)

        response = self.app.post(url('delete_user', id=new_user.user_id),

            params={'_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'Successfully deleted user')

    def test_delete_user_group_err(self):

        self.log_user()

        username = 'usergrouperr'

        groupname = u'usergroup_fail'

        fixture.create_user(name=username)

        ug = fixture.create_user_group(name=groupname, cur_user=username)

        new_user = Session().query(User) \

            .filter(User.username == username).one()

        response = self.app.post(url('delete_user', id=new_user.user_id),

            params={'_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'User "%s" still '

                               'owns 1 user groups and cannot be removed. '

                               'Switch owners or remove those user groups: '

                               '%s' % (username, groupname))

        # TODO: why do this fail?

        #response = self.app.delete(url('delete_users_group', id=groupname))

        #self.checkSessionFlash(response, 'Removed user group %s' % groupname)

        fixture.destroy_user_group(ug.users_group_id)

        response = self.app.post(url('delete_user', id=new_user.user_id),

            params={'_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'Successfully deleted user')

    def test_edit(self):

        self.log_user()

        user = User.get_by_username(TEST_USER_ADMIN_LOGIN)

        response = self.app.get(url('edit_user', id=user.user_id))

    def test_add_perm_create_repo(self):

        self.log_user()

        perm_none = Permission.get_by_key('hg.create.none')

        perm_create = Permission.get_by_key('hg.create.repository')

        user = UserModel().create_or_update(username='dummy', password='qwe',

                                            email='dummy', firstname=u'a',

                                            lastname=u'b')

        Session().commit()

        uid = user.user_id

        try:

            # User should have None permission on creation repository

            assert UserModel().has_perm(user, perm_none) == False

            assert UserModel().has_perm(user, perm_create) == False

            response = self.app.post(url('edit_user_perms_update', id=uid),

                                     params=dict(create_repo_perm=True,

                                                 _authentication_token=self.authentication_token()))

            perm_none = Permission.get_by_key('hg.create.none')

            perm_create = Permission.get_by_key('hg.create.repository')

            # User should have None permission on creation repository

            assert UserModel().has_perm(uid, perm_none) == False

            assert UserModel().has_perm(uid, perm_create) == True

        finally:

            UserModel().delete(uid)

            Session().commit()

    def test_revoke_perm_create_repo(self):

        self.log_user()

        perm_none = Permission.get_by_key('hg.create.none')

        perm_create = Permission.get_by_key('hg.create.repository')

        user = UserModel().create_or_update(username='dummy', password='qwe',

                                            email='dummy', firstname=u'a',

                                            lastname=u'b')

        Session().commit()

        uid = user.user_id

        try:

            # User should have None permission on creation repository

            assert UserModel().has_perm(user, perm_none) == False

            assert UserModel().has_perm(user, perm_create) == False

            response = self.app.post(url('edit_user_perms_update', id=uid),

                                     params=dict(_authentication_token=self.authentication_token()))

            perm_none = Permission.get_by_key('hg.create.none')

            perm_create = Permission.get_by_key('hg.create.repository')

            # User should have None permission on creation repository

            assert UserModel().has_perm(uid, perm_none) == True

            assert UserModel().has_perm(uid, perm_create) == False

        finally:

            UserModel().delete(uid)

            Session().commit()

    def test_add_perm_fork_repo(self):

        self.log_user()

        perm_none = Permission.get_by_key('hg.fork.none')

        perm_fork = Permission.get_by_key('hg.fork.repository')

        user = UserModel().create_or_update(username='dummy', password='qwe',

                                            email='dummy', firstname=u'a',

                                            lastname=u'b')

        Session().commit()

        uid = user.user_id

        try:

            # User should have None permission on creation repository

            assert UserModel().has_perm(user, perm_none) == False

            assert UserModel().has_perm(user, perm_fork) == False

            response = self.app.post(url('edit_user_perms_update', id=uid),

                                     params=dict(create_repo_perm=True,

                                                 _authentication_token=self.authentication_token()))

            perm_none = Permission.get_by_key('hg.create.none')

            perm_create = Permission.get_by_key('hg.create.repository')

            # User should have None permission on creation repository

            assert UserModel().has_perm(uid, perm_none) == False

            assert UserModel().has_perm(uid, perm_create) == True

        finally:

            UserModel().delete(uid)

            Session().commit()

    def test_revoke_perm_fork_repo(self):

        self.log_user()

        perm_none = Permission.get_by_key('hg.fork.none')

        perm_fork = Permission.get_by_key('hg.fork.repository')

        user = UserModel().create_or_update(username='dummy', password='qwe',

                                            email='dummy', firstname=u'a',

                                            lastname=u'b')

        Session().commit()

        uid = user.user_id

        try:

            # User should have None permission on creation repository

            assert UserModel().has_perm(user, perm_none) == False

            assert UserModel().has_perm(user, perm_fork) == False

            response = self.app.post(url('edit_user_perms_update', id=uid),

                                     params=dict(_authentication_token=self.authentication_token()))

            perm_none = Permission.get_by_key('hg.create.none')

            perm_create = Permission.get_by_key('hg.create.repository')

            # User should have None permission on creation repository

            assert UserModel().has_perm(uid, perm_none) == True

            assert UserModel().has_perm(uid, perm_create) == False

        finally:

            UserModel().delete(uid)

            Session().commit()

    def test_ips(self):

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        response = self.app.get(url('edit_user_ips', id=user.user_id))

        response.mustcontain('All IP addresses are allowed')

    @parametrize('test_name,ip,ip_range,failure', [

        ('127/24', '127.0.0.1/24', '127.0.0.0 - 127.0.0.255', False),

        ('10/32', '10.0.0.10/32', '10.0.0.10 - 10.0.0.10', False),

        ('0/16', '0.0.0.0/16', '0.0.0.0 - 0.0.255.255', False),

        ('0/8', '0.0.0.0/8', '0.0.0.0 - 0.255.255.255', False),

        ('127_bad_mask', '127.0.0.1/99', '127.0.0.1 - 127.0.0.1', True),

        ('127_bad_ip', 'foobar', 'foobar', True),

    ])

    def test_add_ip(self, test_name, ip, ip_range, failure, auto_clear_ip_permissions):

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        user_id = user.user_id

        response = self.app.post(url('edit_user_ips_update', id=user_id),

                                 params=dict(new_ip=ip, _authentication_token=self.authentication_token()))

        if failure:

            self.checkSessionFlash(response, 'Please enter a valid IPv4 or IPv6 address')

            response = self.app.get(url('edit_user_ips', id=user_id))

            response.mustcontain(no=[ip])

            response.mustcontain(no=[ip_range])

        else:

            response = self.app.get(url('edit_user_ips', id=user_id))

            response.mustcontain(ip)

            response.mustcontain(ip_range)

    def test_delete_ip(self, auto_clear_ip_permissions):

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        user_id = user.user_id

        ip = '127.0.0.1/32'

        ip_range = '127.0.0.1 - 127.0.0.1'

        with test_context(self.app):

            new_ip = UserModel().add_extra_ip(user_id, ip)

            Session().commit()

        new_ip_id = new_ip.ip_id

        response = self.app.get(url('edit_user_ips', id=user_id))

        response.mustcontain(ip)

        response.mustcontain(ip_range)

        self.app.post(url('edit_user_ips_delete', id=user_id),

                      params=dict(del_ip_id=new_ip_id, _authentication_token=self.authentication_token()))

        response = self.app.get(url('edit_user_ips', id=user_id))

        response.mustcontain('All IP addresses are allowed')

        response.mustcontain(no=[ip])

        response.mustcontain(no=[ip_range])

    def test_api_keys(self):

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        response = self.app.get(url('edit_user_api_keys', id=user.user_id))

        response.mustcontain(user.api_key)

        response.mustcontain('Expires: Never')

    @parametrize('desc,lifetime', [

        ('forever', -1),

        ('5mins', 60*5),

        ('30days', 60*60*24*30),

    ])

    def test_add_api_keys(self, desc, lifetime):

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        user_id = user.user_id

        response = self.app.post(url('edit_user_api_keys_update', id=user_id),

                 {'description': desc, 'lifetime': lifetime, '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'API key successfully created')

        try:

            response = response.follow()

            user = User.get(user_id)

            for api_key in user.api_keys:

                response.mustcontain(api_key)

        finally:

            for api_key in UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all():

                Session().delete(api_key)

                Session().commit()

    def test_remove_api_key(self):

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        user_id = user.user_id

        response = self.app.post(url('edit_user_api_keys_update', id=user_id),

                {'description': 'desc', 'lifetime': -1, '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'API key successfully created')

        response = response.follow()

        # now delete our key

        keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()

        assert 1 == len(keys)

        response = self.app.post(url('edit_user_api_keys_delete', id=user_id),

                 {'del_api_key': keys[0].api_key, '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'API key successfully deleted')

        keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()

        assert 0 == len(keys)

    def test_reset_main_api_key(self):

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        user_id = user.user_id

        api_key = user.api_key

        response = self.app.get(url('edit_user_api_keys', id=user_id))

        response.mustcontain(api_key)

        response.mustcontain('Expires: Never')

        response = self.app.post(url('edit_user_api_keys_delete', id=user_id),

                 {'del_api_key_builtin': api_key, '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'API key successfully reset')

        response = response.follow()

        response.mustcontain(no=[api_key])

    def test_add_ssh_key(self):

        description = u'something'

        public_key = u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUQ== me@localhost'

        fingerprint = u'Ke3oUCNJM87P0jJTb3D+e3shjceP2CqMpQKVd75E9I8'

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        user_id = user.user_id

        response = self.app.post(url('edit_user_ssh_keys', id=user_id),

                                 {'description': description,

                                  'public_key': public_key,

                                  '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'SSH key %s successfully added' % fingerprint)

        response = response.follow()

        response.mustcontain(fingerprint)

        ssh_key = UserSshKeys.query().filter(UserSshKeys.user_id == user_id).one()

        assert ssh_key.fingerprint == fingerprint

        assert ssh_key.description == description

        Session().delete(ssh_key)

        Session().commit()

    def test_remove_ssh_key(self):

        description = u''

        public_key = u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUQ== me@localhost'

        fingerprint = u'Ke3oUCNJM87P0jJTb3D+e3shjceP2CqMpQKVd75E9I8'

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        user_id = user.user_id

        response = self.app.post(url('edit_user_ssh_keys', id=user_id),

                                 {'description': description,

                                  'public_key': public_key,

                                  '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'SSH key %s successfully added' % fingerprint)

        response.follow()

        ssh_key = UserSshKeys.query().filter(UserSshKeys.user_id == user_id).one()

        response = self.app.post(url('edit_user_ssh_keys_delete', id=user_id),

                                 {'del_public_key': ssh_key.public_key,

                                  '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'SSH key successfully deleted')

        keys = UserSshKeys.query().all()

        assert 0 == len(keys)

class TestAdminUsersController_unittest(TestController):

    """ Unit tests for the users controller """

    def test_get_user_or_raise_if_default(self, monkeypatch, test_context_fixture):

        # flash complains about an non-existing session

        def flash_mock(*args, **kwargs):

            pass

        monkeypatch.setattr(h, 'flash', flash_mock)

        u = UsersController()

        # a regular user should work correctly

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        assert u._get_user_or_raise_if_default(user.user_id) == user

        # the default user should raise

        with pytest.raises(HTTPNotFound):

            u._get_user_or_raise_if_default(User.get_default_user().user_id)

class TestAdminUsersControllerForDefaultUser(TestController):

    """

    Edit actions on the default user are not allowed.

    Validate that they throw a 404 exception.

    """

    def test_edit_default_user(self):

        self.log_user()

        user = User.get_default_user()

        response = self.app.get(url('edit_user', id=user.user_id), status=404)

    def test_edit_advanced_default_user(self):

        self.log_user()

        user = User.get_default_user()

        response = self.app.get(url('edit_user_advanced', id=user.user_id), status=404)

    # API keys

    def test_edit_api_keys_default_user(self):

        self.log_user()

        user = User.get_default_user()

        response = self.app.get(url('edit_user_api_keys', id=user.user_id), status=404)

    def test_add_api_keys_default_user(self):

        self.log_user()

        user = User.get_default_user()

        response = self.app.post(url('edit_user_api_keys_update', id=user.user_id),

                 {'_authentication_token': self.authentication_token()}, status=404)

    def test_delete_api_keys_default_user(self):

        self.log_user()

        user = User.get_default_user()

        response = self.app.post(url('edit_user_api_keys_delete', id=user.user_id),

                 {'_authentication_token': self.authentication_token()}, status=404)

    # Permissions

    def test_edit_perms_default_user(self):

        self.log_user()

        user = User.get_default_user()

        response = self.app.get(url('edit_user_perms', id=user.user_id), status=404)

    def test_update_perms_default_user(self):

        self.log_user()

        user = User.get_default_user()

        response = self.app.post(url('edit_user_perms_update', id=user.user_id),

                 {'_authentication_token': self.authentication_token()}, status=404)

    # Emails

    def test_edit_emails_default_user(self):

        self.log_user()

        user = User.get_default_user()

        response = self.app.get(url('edit_user_emails', id=user.user_id), status=404)

    def test_add_emails_default_user(self):

        self.log_user()

        user = User.get_default_user()

        response = self.app.post(url('edit_user_emails_update', id=user.user_id),

                 {'_authentication_token': self.authentication_token()}, status=404)

    def test_delete_emails_default_user(self):

        self.log_user()

        user = User.get_default_user()

        response = self.app.post(url('edit_user_emails_delete', id=user.user_id),

                 {'_authentication_token': self.authentication_token()}, status=404)

    # IP addresses

    # Add/delete of IP addresses for the default user is used to maintain

    # the global IP whitelist and thus allowed. Only 'edit' is forbidden.

    def test_edit_ip_default_user(self):

        self.log_user()

        user = User.get_default_user()

        response = self.app.get(url('edit_user_ips', id=user.user_id), status=404)

# -*- coding: utf-8 -*-

from kallithea.model.db import User, UserFollowing, Repository, UserApiKeys, UserSshKeys

from kallithea.tests.base import *

from kallithea.tests.fixture import Fixture

from kallithea.lib import helpers as h

from kallithea.model.user import UserModel

from kallithea.model.meta import Session

from tg.util.webtest import test_context

fixture = Fixture()

class TestMyAccountController(TestController):

    test_user_1 = 'testme'

    @classmethod

    def teardown_class(cls):

        if User.get_by_username(cls.test_user_1):

            UserModel().delete(cls.test_user_1)

            Session().commit()

    def test_my_account(self):

        self.log_user()

        response = self.app.get(url('my_account'))

        response.mustcontain('value="%s' % TEST_USER_ADMIN_LOGIN)

    def test_my_account_my_repos(self):

        self.log_user()

        response = self.app.get(url('my_account_repos'))

        cnt = Repository.query().filter(Repository.owner ==

                           User.get_by_username(TEST_USER_ADMIN_LOGIN)).count()

        response.mustcontain('"raw_name": "%s"' % HG_REPO)

        response.mustcontain('"just_name": "%s"' % GIT_REPO)

    def test_my_account_my_watched(self):

        self.log_user()

        response = self.app.get(url('my_account_watched'))

        cnt = UserFollowing.query().filter(UserFollowing.user ==

                            User.get_by_username(TEST_USER_ADMIN_LOGIN)).count()

        response.mustcontain('"raw_name": "%s"' % HG_REPO)

        response.mustcontain('"just_name": "%s"' % GIT_REPO)

    def test_my_account_my_emails(self):

        self.log_user()

        response = self.app.get(url('my_account_emails'))

        response.mustcontain('No additional emails specified')

    def test_my_account_my_emails_add_existing_email(self):

        self.log_user()

        response = self.app.get(url('my_account_emails'))

        response.mustcontain('No additional emails specified')

        response = self.app.post(url('my_account_emails'),

                                 {'new_email': TEST_USER_REGULAR_EMAIL, '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'This email address is already in use')

    def test_my_account_my_emails_add_missing_email_in_form(self):

        self.log_user()

        response = self.app.get(url('my_account_emails'))

        response.mustcontain('No additional emails specified')

        response = self.app.post(url('my_account_emails'),

            {'_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'Please enter an email address')

    def test_my_account_my_emails_add_remove(self):

        self.log_user()

        response = self.app.get(url('my_account_emails'))

        response.mustcontain('No additional emails specified')

        response = self.app.post(url('my_account_emails'),

                                 {'new_email': 'barz@example.com', '_authentication_token': self.authentication_token()})

        response = self.app.get(url('my_account_emails'))

        from kallithea.model.db import UserEmailMap

        email_id = UserEmailMap.query() \

            .filter(UserEmailMap.user == User.get_by_username(TEST_USER_ADMIN_LOGIN)) \

            .filter(UserEmailMap.email == 'barz@example.com').one().email_id

        response.mustcontain('barz@example.com')

        response.mustcontain('<input id="del_email_id" name="del_email_id" type="hidden" value="%s" />' % email_id)

        response = self.app.post(url('my_account_emails_delete'),

                                 {'del_email_id': email_id, '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'Removed email from user')

        response = self.app.get(url('my_account_emails'))

        response.mustcontain('No additional emails specified')

    @parametrize('name,attrs',

        [('firstname', {'firstname': 'new_username'}),

         ('lastname', {'lastname': 'new_username'}),

         ('admin', {'admin': True}),

         ('admin', {'admin': False}),

         ('extern_type', {'extern_type': 'ldap'}),

         ('extern_type', {'extern_type': None}),

         #('extern_name', {'extern_name': 'test'}),

         #('extern_name', {'extern_name': None}),

         ('active', {'active': False}),

         ('active', {'active': True}),

         ('email', {'email': 'someemail@example.com'}),

        # ('new_password', {'new_password': 'foobar123',

        #                   'password_confirmation': 'foobar123'})

        ])

    def test_my_account_update(self, name, attrs):

        usr = fixture.create_user(self.test_user_1, password='qweqwe',

                                  email='testme@example.com',

                                  extern_type='internal',

                                  extern_name=self.test_user_1,

                                  skip_if_exists=True)

        params = usr.get_api_data(True)  # current user data

        user_id = usr.user_id

        self.log_user(username=self.test_user_1, password='qweqwe')

        params.update({'password_confirmation': ''})

        params.update({'new_password': ''})

        params.update({'extern_type': 'internal'})

        params.update({'extern_name': self.test_user_1})

        params.update({'_authentication_token': self.authentication_token()})

        params.update(attrs)

        response = self.app.post(url('my_account'), params)

        self.checkSessionFlash(response,

                               'Your account was updated successfully')

        updated_user = User.get_by_username(self.test_user_1)

        updated_params = updated_user.get_api_data(True)

        updated_params.update({'password_confirmation': ''})

        updated_params.update({'new_password': ''})

        params['last_login'] = updated_params['last_login']

        if name == 'email':

            params['emails'] = [attrs['email']]

        if name == 'extern_type':

            # cannot update this via form, expected value is original one

            params['extern_type'] = "internal"

        if name == 'extern_name':

            # cannot update this via form, expected value is original one

            params['extern_name'] = str(user_id)

        if name == 'active':

            # my account cannot deactivate account

            params['active'] = True

        if name == 'admin':

            # my account cannot make you an admin !

            params['admin'] = False

        params.pop('_authentication_token')

        assert params == updated_params

    def test_my_account_update_err_email_exists(self):

        self.log_user()

        new_email = TEST_USER_REGULAR_EMAIL  # already existing email

        response = self.app.post(url('my_account'),

                                params=dict(

                                    username=TEST_USER_ADMIN_LOGIN,

                                    new_password=TEST_USER_ADMIN_PASS,

                                    password_confirmation='test122',

                                    firstname=u'NewName',

                                    lastname=u'NewLastname',

                                    email=new_email,

                                    _authentication_token=self.authentication_token())

                                )

        response.mustcontain('This email address is already in use')

    def test_my_account_update_err(self):

        self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)

        new_email = 'newmail.pl'

        response = self.app.post(url('my_account'),

                                 params=dict(

                                            username=TEST_USER_ADMIN_LOGIN,

                                            new_password=TEST_USER_ADMIN_PASS,

                                            password_confirmation='test122',

                                            firstname=u'NewName',

                                            lastname=u'NewLastname',

                                            email=new_email,

                                            _authentication_token=self.authentication_token()))

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

        from kallithea.model import validators

        with test_context(self.app):

            msg = validators.ValidUsername(edit=False, old_data={}) \

                    ._messages['username_exists']

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

        response.mustcontain(msg)

    def test_my_account_api_keys(self):

        usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)

        user = User.get(usr['user_id'])

        response = self.app.get(url('my_account_api_keys'))

        response.mustcontain(user.api_key)

        response.mustcontain('Expires: Never')

    @parametrize('desc,lifetime', [

        ('forever', -1),

        ('5mins', 60*5),

        ('30days', 60*60*24*30),

    ])

    def test_my_account_add_api_keys(self, desc, lifetime):

        usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)

        user = User.get(usr['user_id'])

        response = self.app.post(url('my_account_api_keys'),

                                 {'description': desc, 'lifetime': lifetime, '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'API key successfully created')

        try:

            response = response.follow()

            user = User.get(usr['user_id'])

            for api_key in user.api_keys:

                response.mustcontain(api_key)

        finally:

            for api_key in UserApiKeys.query().all():

                Session().delete(api_key)

                Session().commit()

    def test_my_account_remove_api_key(self):

        usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)

        user = User.get(usr['user_id'])

        response = self.app.post(url('my_account_api_keys'),

                                 {'description': 'desc', 'lifetime': -1, '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'API key successfully created')

        response = response.follow()

        # now delete our key

        keys = UserApiKeys.query().all()

        assert 1 == len(keys)

        response = self.app.post(url('my_account_api_keys_delete'),

                 {'del_api_key': keys[0].api_key, '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'API key successfully deleted')

        keys = UserApiKeys.query().all()

        assert 0 == len(keys)

    def test_my_account_reset_main_api_key(self):

        usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)

        user = User.get(usr['user_id'])

        api_key = user.api_key

        response = self.app.get(url('my_account_api_keys'))

        response.mustcontain(api_key)

        response.mustcontain('Expires: Never')

        response = self.app.post(url('my_account_api_keys_delete'),

                 {'del_api_key_builtin': api_key, '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'API key successfully reset')

        response = response.follow()

        response.mustcontain(no=[api_key])

    def test_my_account_add_ssh_key(self):

        description = u'something'

        public_key = u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUQ== me@localhost'

        fingerprint = u'Ke3oUCNJM87P0jJTb3D+e3shjceP2CqMpQKVd75E9I8'

        self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)

        response = self.app.post(url('my_account_ssh_keys'),

                                 {'description': description,

                                  'public_key': public_key,

                                  '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'SSH key %s successfully added' % fingerprint)

        response = response.follow()

        response.mustcontain(fingerprint)

        user_id = response.session['authuser']['user_id']

        ssh_key = UserSshKeys.query().filter(UserSshKeys.user_id == user_id).one()

        assert ssh_key.fingerprint == fingerprint

        assert ssh_key.description == description

        Session().delete(ssh_key)

        Session().commit()

    def test_my_account_remove_ssh_key(self):

        description = u''

        public_key = u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUQ== me@localhost'

        fingerprint = u'Ke3oUCNJM87P0jJTb3D+e3shjceP2CqMpQKVd75E9I8'

        self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)

        response = self.app.post(url('my_account_ssh_keys'),

                                 {'description': description,

                                  'public_key': public_key,

                                  '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'SSH key %s successfully added' % fingerprint)

        response.follow()

        user_id = response.session['authuser']['user_id']

        ssh_key = UserSshKeys.query().filter(UserSshKeys.user_id == user_id).one()

        response = self.app.post(url('my_account_ssh_keys_delete'),

                                 {'del_public_key': ssh_key.public_key,

                                  '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'SSH key successfully deleted')