fixture = Fixture()

class TestAdminUsersController(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_index(self):

        self.log_user()

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

        # Test response...

    def test_create(self):

        self.log_user()

        username = 'newtestuser'

        password = 'test12'

        password_confirmation = password

        name = 'name'

        lastname = 'lastname'

        email = 'mail@mail.com'

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

            {'username': username,

             'password': password,

             'password_confirmation': password_confirmation,

             'firstname': name,

             'active': True,

             'lastname': lastname,

             'extern_name': 'internal',

             'extern_type': 'internal',

             'email': email})

        self.checkSessionFlash(response, '''Created user <a href="/_admin/users/''')

        self.checkSessionFlash(response, '''/edit">%s</a>''' % (username))

        new_user = Session().query(User).\

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

        self.assertEqual(new_user.username, username)

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

        self.assertEqual(new_user.name, name)

        self.assertEqual(new_user.lastname, lastname)

        self.assertEqual(new_user.email, email)

        response.follow()

        response = response.follow()

        response.mustcontain("""newtestuser""")

    def test_create_err(self):

        self.log_user()

        username = 'new_user'

        password = ''

        name = 'name'

        lastname = 'lastname'

        email = 'errmail.com'

        response = self.app.post(url('users'), {'username': username,

                                               'password': password,

                                               'name': name,

                                               'active': False,

                                               'lastname': lastname,

                                               'email': email})

        msg = validators.ValidUsername(False, {})._messages['system_invalid_username']

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

        response.mustcontain("""<span class="error-message">%s</span>""" % msg)

        response.mustcontain("""<span class="error-message">Please enter a value</span>""")

        response.mustcontain("""<span class="error-message">An email address must contain a single @</span>""")

        def get_user():

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

        self.assertRaises(NoResultFound, get_user), 'found user in database'

    def test_new(self):

        self.log_user()

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

    @parameterized.expand(

        [('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': 'some@email.com'}),

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

        #                   'password_confirmation': 'foobar123'})

        ])

    def test_update(self, name, attrs):

        self.log_user()

        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)

        Session().commit()

        params = usr.get_api_data()

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

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

        params.update(attrs)

        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'] = self.test_user_1

            # special case since this user is not

                                          # logged in yet his data is not filled

                                          # so we use creation data

        response = self.app.put(url('user', id=usr.user_id), params)

        self.checkSessionFlash(response, 'User updated successfully')

        updated_user = User.get_by_username(self.test_user_1)

        updated_params = updated_user.get_api_data()

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

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

        self.assertEqual(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.delete(url('user', id=new_user.user_id))

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

    def test_delete_repo_err(self):

        self.log_user()

        username = 'repoerr'

        reponame = '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.delete(url('user', id=new_user.user_id))

        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.delete(url('repo', repo_name=reponame))

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

        response = self.app.delete(url('user', id=new_user.user_id))

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

    def test_delete_repo_group_err(self):

        self.log_user()

        username = 'repogrouperr'

        groupname = 'repogroup_fail'

        fixture.create_user(name=username)

        fixture.create_repo_group(name=groupname, cur_user=username)

        new_user = Session().query(User)\

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

        response = self.app.delete(url('user', id=new_user.user_id))

        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.delete(url('delete_repo_group', group_name=groupname))

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

        response = self.app.delete(url('user', id=new_user.user_id))

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

    def test_show(self):

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

    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='a',

                                            lastname='b')

        Session().commit()

        uid = user.user_id

        try:

            #User should have None permission on creation repository

            self.assertEqual(UserModel().has_perm(user, perm_none), False)

            self.assertEqual(UserModel().has_perm(user, perm_create), False)

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

                                     params=dict(_method='put',

                                                 create_repo_perm=True))

            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

            self.assertEqual(UserModel().has_perm(uid, perm_none), False)

            self.assertEqual(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='a',

                                            lastname='b')

        Session().commit()

        uid = user.user_id

        try:

            #User should have None permission on creation repository

            self.assertEqual(UserModel().has_perm(user, perm_none), False)

            self.assertEqual(UserModel().has_perm(user, perm_create), False)

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

                                     params=dict(_method='put'))

            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

            self.assertEqual(UserModel().has_perm(uid, perm_none), True)

            self.assertEqual(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='a',

                                            lastname='b')

        Session().commit()

        uid = user.user_id

        try:

            #User should have None permission on creation repository

            self.assertEqual(UserModel().has_perm(user, perm_none), False)

            self.assertEqual(UserModel().has_perm(user, perm_fork), False)

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

                                     params=dict(_method='put',

                                                 create_repo_perm=True))

            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

            self.assertEqual(UserModel().has_perm(uid, perm_none), False)

            self.assertEqual(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='a',

                                            lastname='b')

        Session().commit()

        uid = user.user_id

        try:

            #User should have None permission on creation repository

            self.assertEqual(UserModel().has_perm(user, perm_none), False)

            self.assertEqual(UserModel().has_perm(user, perm_fork), False)

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

                                     params=dict(_method='put'))

            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

            self.assertEqual(UserModel().has_perm(uid, perm_none), True)

            self.assertEqual(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')

    @parameterized.expand([

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

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        user_id = user.user_id

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

                                params=dict(new_ip=ip))

        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)

        ## cleanup

        for del_ip in UserIpMap.query().filter(UserIpMap.user_id == user_id).all():

            Session().delete(del_ip)

            Session().commit()

    def test_delete_ip(self):

        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'

        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', id=user_id),

                      params=dict(_method='delete', del_ip_id=new_ip_id))

        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)