authentic/tests/test_user_manager.py

# -*- coding: utf-8 -*-
# authentic2 - versatile identity manager
# Copyright (C) 2010-2019 Entr'ouvert
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from __future__ import unicode_literals

import csv
import datetime
import re
import time
from urllib.parse import urlparse

import pytest
from webtest import Upload

from django.contrib.auth import get_user_model
from django.contrib.contenttypes.models import ContentType
from django.urls import reverse
from django.utils.six import text_type

from django_rbac.models import VIEW_OP
from django_rbac.utils import (
        get_operation,
        get_ou_model,
        get_permission_model,
        get_role_model,
)

from authentic2.custom_user.models import User
from authentic2.models import Attribute, AttributeValue
from authentic2.a2_rbac.utils import get_default_ou
from authentic2.a2_rbac.utils import get_view_user_perm
from authentic2.manager import user_import
from authentic2_idp_oidc.models import OIDCAuthorization, OIDCClient


from .utils import login, get_link_from_mail

OU = get_ou_model()


def visible_users(response):
    return set(elt.text for elt in response.pyquery('td.username'))


def test_create_user(app, superuser):
    response = login(app, superuser, '/manage/users/')
    response = response.click('Add user')
    assert User.objects.count() == 1
    response.form.set('username', 'john.doe')
    response.form.set('email', 'john.doe@example.com')
    response.form.set('first_name', 'Jôhn')
    response.form.set('last_name', 'Döe')
    response.form.set('password1', '1234Password')
    response.form.set('password2', '1234Password')
    response.form.set('send_password_reset', False)
    response = response.form.submit(status=302)
    assert User.objects.count() == 2
    user = User.objects.exclude(id=superuser.id).get()
    assert user.ou == get_default_ou()
    assert user.username == 'john.doe'
    assert user.email == 'john.doe@example.com'
    assert user.first_name == 'Jôhn'
    assert user.last_name == 'Döe'
    assert user.check_password('1234Password')


def test_create_user_permission_denied(app, simple_user, ou1, ou2):
    ou1.get_admin_role().members.add(simple_user)
    response = login(app, simple_user, '/manage/users/%s/add/' % ou1.id)

    assert 'You are not authorized to see this page.' not in response.text

    response = app.get('/manage/users/%s/add/' % ou2.id, status=403)
    assert 'You are not authorized to see this page.' in response.text


def test_create_user_only_name(app, superuser):
    response = login(app, superuser, '/manage/users/')
    response = response.click('Add user')
    assert User.objects.count() == 1
    response.form.set('first_name', 'Jôhn')
    response.form.set('last_name', 'Döe')
    response.form.set('password1', '1234Password')
    response.form.set('password2', '1234Password')
    response.form.set('send_password_reset', False)
    response = response.form.submit(status=302)
    assert User.objects.count() == 2


def test_create_user_only_email(app, superuser):
    Attribute.objects.update(required=False)
    response = login(app, superuser, '/manage/users/')
    response = response.click('Add user')
    assert User.objects.count() == 1
    response.form.set('email', 'john.doe@example.com')
    response.form.set('password1', '1234Password')
    response.form.set('password2', '1234Password')
    response.form.set('send_password_reset', False)
    response = response.form.submit(status=302)
    assert User.objects.count() == 2


def test_create_user_only_username(app, superuser):
    Attribute.objects.update(required=False)
    response = login(app, superuser, '/manage/users/')
    response = response.click('Add user')
    assert User.objects.count() == 1
    response.form.set('username', 'john.doe')
    response.form.set('password1', '1234Password')
    response.form.set('password2', '1234Password')
    response.form.set('send_password_reset', False)
    response = response.form.submit(status=302)
    assert User.objects.count() == 2


def test_create_user_no_identifier(app, superuser):
    response = login(app, superuser, '/manage/users/')
    response = response.click('Add user')
    assert User.objects.count() == 1
    response.form.set('password1', '1234Password')
    response.form.set('password2', '1234Password')
    response.form.set('send_password_reset', False)
    response = response.form.submit(status=200)
    assert User.objects.count() == 1
    assert 'An account needs at least one identifier: ' in response


def test_create_user_username_is_unique(app, superuser, settings):
    settings.A2_USERNAME_IS_UNIQUE = True
    Attribute.objects.update(required=False)
    response = login(app, superuser, '/manage/users/')
    response = response.click('Add user')
    assert User.objects.count() == 1
    response.form.set('username', 'john.doe')
    response.form.set('password1', '1234Password')
    response.form.set('password2', '1234Password')
    response.form.set('send_password_reset', False)
    response = response.form.submit(status=302)
    assert User.objects.count() == 2

    # try again
    response = app.get('/manage/users/')
    response = response.click('Add user')
    response.form.set('username', 'john.doe')
    response.form.set('password1', '1234Password')
    response.form.set('password2', '1234Password')
    response.form.set('send_password_reset', False)
    response = response.form.submit(status=200)
    assert User.objects.count() == 2
    assert 'This username is already in use' in response


def test_create_user_email_is_unique(app, superuser, settings):
    settings.A2_EMAIL_IS_UNIQUE = True
    Attribute.objects.update(required=False)
    response = login(app, superuser, '/manage/users/')
    response = response.click('Add user')
    assert User.objects.count() == 1
    response.form.set('email', 'john.doe@example.com')
    response.form.set('password1', '1234Password')
    response.form.set('password2', '1234Password')
    response.form.set('send_password_reset', False)
    response = response.form.submit(status=302)
    assert User.objects.count() == 2

    # try again
    response = app.get('/manage/users/')
    response = response.click('Add user')
    response.form.set('email', 'john.doe@example.com')
    response.form.set('password1', '1234Password')
    response.form.set('password2', '1234Password')
    response.form.set('send_password_reset', False)
    response = response.form.submit(status=200)
    assert User.objects.count() == 2
    assert 'This email address is already in use' in response


def test_manager_user_change_email(app, superuser_or_admin, simple_user, mailoutbox):
    ou = get_default_ou()
    ou.validate_emails = True
    ou.save()

    NEW_EMAIL = 'john.doe@example.com'

    assert NEW_EMAIL != simple_user.email

    response = login(app, superuser_or_admin,
                     reverse('a2-manager-user-by-uuid-detail',
                             kwargs={'slug': text_type(simple_user.uuid)}))
    assert 'Change user email' in response.text
    # cannot click it's a submit button :/
    response = app.get(reverse('a2-manager-user-by-uuid-change-email',
                               kwargs={'slug': text_type(simple_user.uuid)}))
    assert response.form['new_email'].value == simple_user.email
    response.form.set('new_email', NEW_EMAIL)
    assert len(mailoutbox) == 0
    response = response.form.submit().follow()
    assert 'A mail was sent to john.doe@example.com to verify it.' in response.text
    assert 'Change user email' in response.text
    # cannot click it's a submit button :/
    assert len(mailoutbox) == 1
    assert simple_user.email in mailoutbox[0].body
    assert NEW_EMAIL in mailoutbox[0].body

    # logout
    app.session.flush()

    link = get_link_from_mail(mailoutbox[0])
    response = app.get(link).maybe_follow()
    assert (
        'your request for changing your email for john.doe@example.com is successful'
        in response.text)
    simple_user.refresh_from_db()
    assert simple_user.email == NEW_EMAIL


def test_manager_user_change_email_no_change(app, superuser_or_admin, simple_user, mailoutbox):
    ou = get_default_ou()
    ou.validate_emails = True
    ou.save()

    NEW_EMAIL = 'john.doe@example.com'

    assert NEW_EMAIL != simple_user.email

    response = login(app, superuser_or_admin,
                     reverse('a2-manager-user-by-uuid-detail',
                             kwargs={'slug': text_type(simple_user.uuid)}))
    assert 'Change user email' in response.text
    # cannot click it's a submit button :/
    response = app.get(reverse('a2-manager-user-by-uuid-change-email',
                               kwargs={'slug': text_type(simple_user.uuid)}))
    assert response.form['new_email'].value == simple_user.email
    assert len(mailoutbox) == 0
    response = response.form.submit().follow()
    assert 'A mail was sent to john.doe@example.com to verify it.' not in response.text


def test_search_by_attribute(app, simple_user, admin):
    Attribute.objects.create(name='adresse', searchable=True, kind='string')

    simple_user.attributes.adresse = 'avenue du revestel'
    response = login(app, admin, '/manage/users/')

    # all users are visible
    assert visible_users(response) == {simple_user.username, admin.username}

    response.form['search-text'] = 'impasse'
    response = response.form.submit()
    # now all users are hidden
    assert not visible_users(response) & {simple_user.username, admin.username}

    response.form['search-text'] = 'avenue'
    response = response.form.submit()

    # now we see only simple_user
    assert visible_users(response) == {simple_user.username}


def test_export_csv(settings, app, superuser, django_assert_num_queries):
    AT_COUNT = 30
    USER_COUNT = 2000
    DEFAULT_BATCH_SIZE = 1000

    ats = [Attribute(name='at%s' % i, label='At%s' % i, kind='string') for i in range(AT_COUNT)]
    Attribute.objects.bulk_create(ats)

    ats = list(Attribute.objects.all())
    users = [User(username='user%s' % i) for i in range(USER_COUNT)]
    User.objects.bulk_create(users)
    users = list(User.objects.filter(username__startswith='user'))

    ContentType.objects.get_for_model(User)
    atvs = []
    for i in range(USER_COUNT):
        atvs.extend([AttributeValue(
            owner=users[i], attribute=ats[j], content='value-%s-%s' % (i, j)) for j in range(AT_COUNT)])
    AttributeValue.objects.bulk_create(atvs)

    response = login(app, superuser, reverse('a2-manager-users'))
    settings.A2_CACHE_ENABLED = True
    user_count = User.objects.count()
    # queries should be batched to keep prefetching working without
    # overspending memory for the queryset cache, 4 queries by batches
    num_queries = int(4 + 4 * (user_count / DEFAULT_BATCH_SIZE + bool(user_count % DEFAULT_BATCH_SIZE)))
    with django_assert_num_queries(num_queries):
        response = response.click('CSV')
    table = list(csv.reader(response.text.splitlines()))
    assert len(table) == (user_count + 1)
    assert len(table[0]) == (15 + AT_COUNT)


def test_export_csv_disabled_attribute(settings, app, superuser):
    attr = Attribute.objects.create(name='attr', label='Attr', kind='string')
    attr_d = Attribute.objects.create(name='attrd', label='Attrd', kind='string')

    user = User.objects.create(username='user-foo')
    AttributeValue.objects.create(owner=user, attribute=attr, content='attr-value')
    AttributeValue.objects.create(owner=user, attribute=attr_d, content='attrd-value')

    attr_d.disabled = True
    attr_d.save()

    response = login(app, superuser, reverse('a2-manager-users'))
    settings.A2_CACHE_ENABLED = True
    response = response.click('CSV')

    user_count = User.objects.count()
    table = list(csv.reader(response.text.splitlines()))
    assert len(table) == (user_count + 1)
    num_col = 15 + 1  # 1 is the number active attributes,
    # disabled attribute should not show up
    for line in table:
        assert len(line) == num_col


def test_user_table(app, admin, user_ou1, ou1):
    from authentic2.manager.utils import has_show_username

    # base state, username are shown
    response = login(app, admin, '/manage/users/')
    assert response.pyquery('td.username')

    # hide all usernames, from specific and general view
    OU.objects.update(show_username=False)
    has_show_username.cache.clear()

    response = app.get('/manage/users/')
    assert not response.pyquery('td.username')

    response = app.get('/manage/users/?search-ou=%s' % get_default_ou().id)
    assert not response.pyquery('td.username')

    response = app.get('/manage/users/?search-ou=%s' % ou1.id)
    assert not response.pyquery('td.username')

    # hide username except in OU1
    ou1.show_username = True
    ou1.save()
    has_show_username.cache.clear()

    response = app.get('/manage/users/')
    assert not response.pyquery('td.username')

    response = app.get('/manage/users/?search-ou=%s' % get_default_ou().id)
    assert not response.pyquery('td.username')

    response = app.get('/manage/users/?search-ou=%s' % ou1.id)
    assert response.pyquery('td.username')


@pytest.mark.parametrize('encoding', ['utf-8-sig', 'cp1252', 'iso-8859-15'])
def test_user_import(encoding, transactional_db, app, admin, ou1, admin_ou1):
    Attribute.objects.create(name='phone', kind='phone_number', label='Numéro de téléphone')

    user_count = User.objects.count()

    assert Attribute.objects.count() == 3

    response = login(app, admin, '/manage/users/')

    response = response.click('Import users')
    response.form.set('import_file',
                      Upload(
                          'users.csv',
                          u'''email key verified,first_name,last_name,phone
tnoel@entrouvert.com,Thomas,Noël,1234
fpeters@entrouvert.com,Frédéric,Péters,5678
x,x,x,x'''.encode(encoding),
                          'application/octet-stream'))
    response.form.set('encoding', encoding)
    response.form.set('ou', str(get_default_ou().pk))
    response = response.form.submit()

    imports = list(user_import.UserImport.all())
    assert len(imports) == 1
    _import_uuid = response.location.split('/')[-2]
    _import = user_import.UserImport(uuid=_import_uuid)
    assert _import.exists()

    response = response.follow()

    response = response.forms['action-form'].submit(name='modify').follow()

    response = response.forms['action-form'].submit(name='simulate')

    reports = list(_import.reports)
    assert len(reports) == 1
    uuid = reports[0].uuid

    response = response.follow()

    def assert_timeout(duration, wait_function):
        start = time.time()
        while True:
            result = wait_function()
            if result is not None:
                return result
            assert time.time() - start < duration, '%s timed out after %s seconds' % (wait_function, duration)
            time.sleep(0.001)

    def wait_finished():
        new_resp = response.click('Users Import')
        if new_resp.pyquery('tr[data-uuid="%s"] td.state' % uuid).text() == 'Finished':
            return new_resp

    simulate = reports[0]
    assert simulate.simulate

    response = assert_timeout(2, wait_finished)

    response = response.click(href=simulate.uuid)

    assert len(response.pyquery('table.main tbody tr')) == 4
    assert len(response.pyquery('table.main tbody tr.row-valid')) == 2
    assert len(response.pyquery('table.main tbody tr.row-invalid')) == 2

    assert len(response.pyquery('tr.row-errors')) == 0
    assert len(response.pyquery('tr.row-cells-errors')) == 1
    assert sum(bool(response.pyquery(td).text()) for td in response.pyquery('tr.row-cells-errors td li')) == 2
    assert 'Enter a valid email address' in response.pyquery('tr.row-cells-errors td.cell-email li').text()
    assert 'Phone number can start with' in response.pyquery('tr.row-cells-errors td.cell-phone li').text()

    assert User.objects.count() == user_count

    response = response.click('Users Import')
    response = response.forms['action-form'].submit(name='execute')

    execute = list(report for report in _import.reports if not report.simulate)[0]
    uuid = execute.uuid

    response = response.follow()
    response = assert_timeout(2, wait_finished)

    assert User.objects.count() == user_count + 2
    assert User.objects.filter(
        email='tnoel@entrouvert.com',
        first_name=u'Thomas',
        last_name=u'Noël',
        attribute_values__content='1234').count() == 1
    assert User.objects.filter(
        email='fpeters@entrouvert.com',
        first_name=u'Frédéric',
        last_name=u'Péters',
        attribute_values__content='5678').count() == 1

    # logout
    app.session.flush()
    response = login(app, admin_ou1, '/manage/users/')

    app.get('/manage/users/import/', status=403)
    app.get('/manage/users/import/%s/' % _import.uuid, status=403)
    app.get('/manage/users/import/%s/%s/' % (_import.uuid, simulate.uuid), status=403)
    app.get('/manage/users/import/%s/%s/' % (_import.uuid, execute.uuid), status=403)


def test_su_permission(app, admin, simple_user):
    resp = login(app, admin, '/manage/users/%s/' % simple_user.pk)
    assert len(resp.pyquery('button[name="su"]')) == 0
    assert app.get('/manage/users/%s/su/' % simple_user.pk, status=403)


def test_su_superuser_post(app, app_factory, superuser, simple_user):
    resp = login(app, superuser, '/manage/users/%s/' % simple_user.pk)
    assert len(resp.pyquery('button[name="su"]')) == 1
    su_resp = resp.forms['object-actions'].submit(name='su')

    new_app = app_factory()
    new_app.get(su_resp.location).maybe_follow()
    assert new_app.session['_auth_user_id'] == str(simple_user.pk)


def test_su_superuser_dialog(app, app_factory, superuser, simple_user):
    resp = login(app, superuser, '/manage/users/%s/' % simple_user.pk)
    assert len(resp.pyquery('button[name="su"]')) == 1

    su_view_url = resp.pyquery('button[name="su"]')[0].get('data-url')

    resp = app.get(su_view_url)

    anchors = resp.pyquery('a#su-link')
    assert len(anchors) == 1

    su_url = anchors[0].get('href')

    new_app = app_factory()
    new_app.get(su_url).maybe_follow()
    assert new_app.session['_auth_user_id'] == str(simple_user.pk)


def import_csv(csv_content, app):
    response = app.get('/manage/users/')
    response = response.click('Import users')
    index = [i for i in response.forms if 'import_file' in response.forms[i].fields][0]
    response.forms[index].set(
        'import_file',
        Upload('users.csv', csv_content.encode('utf-8'), 'application/octet-stream'))
    response.forms[index].set('encoding', 'utf-8-sig')
    response.forms[index].set('ou', str(get_default_ou().pk))
    response = response.forms[index].submit().follow()
    response = response.forms['action-form'].submit(name='execute').follow()

    start = time.time()
    response = response.click('Users Import')
    while 'Running' in response.text:
        response = response.click('Users Import')
        assert time.time() - start < 2
        time.sleep(.1)

    # report
    urls = re.findall('<a href="(/manage/users/import/[^/]+/[^/]+/)">', response.text)
    response = app.get(urls[0])
    return response


def test_user_import_attributes(transactional_db, app, admin):
    Attribute.objects.create(name='more', kind='string', label='Signe particulier')
    Attribute.objects.create(name='title', kind='title', label='Titre')
    Attribute.objects.create(name='bike', kind='boolean', label='Vélo')
    Attribute.objects.create(name='saintsday', kind='date', label='Fête')
    Attribute.objects.create(name='birthdate', kind='birthdate', label='Date de naissance')
    Attribute.objects.create(name='zip', kind='fr_postcode', label='Code postal (français)')
    Attribute.objects.create(name='phone', kind='phone_number', label='Numéro de téléphone')
    assert Attribute.objects.count() == 9
    user_count = User.objects.count()
    login(app, admin, '/manage/users/')

    csv_lines = [
        u"email key verified,first_name,last_name,more,title,bike,saintsday,birthdate,zip,phone",
        u"elliot@universalpictures.com,Elliott,Thomas,petit,Mr,True,2019-7-20,1972-05-26,75014,1234",
        u"et@universalpictures.com,ET,the Extra-Terrestrial,long,??,False,1/2/3/4,0002-2-22,42,home"]
    response = import_csv('\n'.join(csv_lines), app)
    urls = re.findall('<a href="(/manage/users/import/[^/]+/[^/]+/)">', response.text)
    response = app.get(urls[0])
    assert 'Select a valid choice. ?? is not one of the available choices.' in response.text
    assert 'Enter a valid date.' in response.text
    assert 'birthdate must be in the past and greater or equal than 1900-01-01.' in response.text
    assert 'The value must be a valid french postcode' in response.text
    assert 'Phone number can start with a + and must contain only digits' in response.text

    assert User.objects.count() == user_count + 1
    elliot = User.objects.filter(email='elliot@universalpictures.com')[0]
    assert elliot.attributes.values['more'].content == 'petit'
    assert elliot.attributes.values['title'].content == 'Mr'
    assert elliot.attributes.values['bike'].content == '1'
    assert elliot.attributes.values['saintsday'].content == '2019-07-20'
    assert elliot.attributes.values['birthdate'].content == '1972-05-26'
    assert elliot.attributes.values['zip'].content == '75014'
    assert elliot.attributes.values['phone'].content == '1234'

    csv_lines[2] = \
        u"et@universalpictures.com,ET,the Extra-Terrestrial,,,,,,42000,+888 5678"
    response = import_csv('\n'.join(csv_lines), app)
    assert '0 rows have errors' in response.text

    assert User.objects.count() == user_count + 2
    et = User.objects.filter(email='et@universalpictures.com')[0]
    assert et.attributes.values['more'].content == ''
    assert et.attributes.values['title'].content == ''
    assert et.attributes.values['bike'].content == '0'
    assert 'saintsday' not in et.attributes.values
    assert 'birthdate' not in et.attributes.values
    assert et.attributes.values['zip'].content == '42000'
    assert et.attributes.values['phone'].content == '+8885678'


def test_detail_view(app, admin, simple_user):
    url = '/manage/users/{user.pk}/'.format(user=simple_user)
    login(app, admin, url)


def test_detail_view_deleted(app, admin, simple_user):
    url = '/manage/users/{user.pk}/'.format(user=simple_user)
    login(app, admin, url)
    simple_user.mark_as_deleted()
    app.get(url, status=404)


def test_user_import_row_error_display(transactional_db, app, admin):
    User.objects.create(first_name='Elliott', last_name='1', ou=get_default_ou())
    User.objects.create(first_name='Elliott', last_name='2', ou=get_default_ou())
    content = '''first_name key,last_name
Elliott,3'''
    login(app, admin, '/manage/users/')
    response = import_csv(content, app)

    assert len(response.pyquery('table.main tbody tr.row-invalid')) == 2
    assert len(response.pyquery('table.main tbody tr.row-errors')) == 1
    assert 'matches too many user' in response.pyquery('tr.row-errors').text()


def test_manager_create_user_next(superuser_or_admin, app, ou1):
    login(app, superuser_or_admin, '/manage/')

    next_url = u'/example.nowhere.null/'
    url = u'/manage/users/%s/add/?next=%s' % (ou1.pk, next_url)
    response = app.get(url)

    # cancel is not handled through form submission, it's a link
    # next without cancel, no cancel button
    assert response.pyquery.remove_namespaces()('a.cancel').attr('href') == '../..'
    assert response.pyquery.remove_namespaces()('input[name="next"]').attr('value') == next_url

    next_url = u'/example.nowhere.null/$UUID/'
    cancel_url = u'/example.nowhere.cancel/'
    url = u'/manage/users/%s/add/?next=%s&cancel=%s' % (ou1.pk, next_url, cancel_url)
    response = app.get(url)

    assert response.pyquery.remove_namespaces()('a.cancel').attr('href') == cancel_url
    assert response.pyquery.remove_namespaces()('input[name="next"]').attr('value') == next_url

    form = response.form
    form.set('first_name', 'John')
    form.set('last_name', 'Doe')
    form.set('email', 'john.doe@gmail.com')
    form.set('password1', 'ABcd1234')
    form.set('password2', 'ABcd1234')
    response = form.submit()
    user = User.objects.latest('id')
    assert urlparse(response.location).path == next_url.replace('$UUID', str(user.uuid))


def test_manager_create_user_next_form_error(superuser_or_admin, app, ou1):
    next_url = u'/example.nowhere.null/'
    url = u'/manage/users/%s/add/?next=%s' % (ou1.pk, next_url)
    login(app, superuser_or_admin, '/manage/')
    response = app.get(url)
    form = response.form
    form.set('first_name', 'John')
    form.set('last_name', 'Doe')
    form.set('email', 'jd')  # erroneous
    form.set('password1', 'notvalid')  # erroneous
    assert '<input type="hidden" name="next" value="%s">' % next_url in form.submit().text


def test_manager_add_user_querystring(superuser_or_admin, app, ou1):
    querystring = u'stay_here=true'
    url = u'/manage/users/add/?%s' % querystring
    login(app, superuser_or_admin, '/manage/')
    response = app.get(url)

    assert querystring in response.location


def test_manager_edit_user_next(app, simple_user, superuser_or_admin):
    next_url = u'/example.nowhere.null/'
    url = u'/manage/users/%s/edit/?next=%s' % (simple_user.pk, next_url)
    login(app, superuser_or_admin, '/manage/')
    response = app.get(url)

    # cancel if not handled through form submission
    assert response.pyquery.remove_namespaces()('a.cancel').attr('href') == next_url

    form = response.form
    form.set('last_name', 'New name')
    assert urlparse(form.submit().location).path == next_url


def test_manager_edit_user_next_form_error(superuser_or_admin, app, ou1, simple_user):
    next_url = u'/example.nowhere.null/'
    url = u'/manage/users/%s/edit/?next=%s' % (simple_user.pk, next_url)
    login(app, superuser_or_admin, '/manage/')
    response = app.get(url)
    form = response.form
    form.set('email', 'jd')  # erroneous
    resp = form.submit()
    assert '<input type="hidden" name="next" value="%s">' % next_url in resp.ubody


def test_user_add_settings(settings, admin, app, db):
    passwd_options = ('generate_password', 'reset_password_at_next_login',
                      'send_mail', 'send_password_reset')
    for policy in [choice[0] for choice in OU.USER_ADD_PASSWD_POLICY_CHOICES]:
        ou = get_default_ou()
        ou.user_add_password_policy = policy
        ou.save()
        user_add = login(app, admin, '/manage/users/add/').follow()
        for option, i in zip(passwd_options, range(4)):
            assert (user_add.form.get(option).value
                    == {False: None, True: 'on'}.get(OU.USER_ADD_PASSWD_POLICY_VALUES[policy][i]))
        app.get('/logout/').form.submit()


def test_ou_hide_username(admin, app, db):
    some_ou = OU.objects.create(name=u'Some Ou', show_username=False)

    login(app, admin, '/manage/')
    url = u'/manage/users/%s/add/' % some_ou.pk
    response = app.get(url)
    q = response.pyquery.remove_namespaces()
    assert len(q('p[id="id_username_p"]')) == 0

    form = response.form
    form.set('first_name', 'John')
    form.set('last_name', 'Doe')
    form.set('email', 'john.doe@gmail.com')
    form.set('password1', 'ABcd1234')
    form.set('password2', 'ABcd1234')
    form.submit()

    assert User.objects.get(email='john.doe@gmail.com')


def test_manager_edit_user_email_verified(app, simple_user, superuser_or_admin):
    simple_user.email_verified = True
    simple_user.save()

    url = u'/manage/users/%s/edit/' % simple_user.pk
    login(app, superuser_or_admin, '/manage/')

    user = User.objects.get(id=simple_user.id)
    assert user.email_verified

    response = app.get(url)
    form = response.form
    form.set('email', 'new.email@gmail.net')
    response = form.submit().follow()

    user = User.objects.get(id=simple_user.id)
    assert not user.email_verified


def test_manager_edit_user_address_autocomplete(app, simple_user, superuser_or_admin):
    url = u'/manage/users/%s/edit/' % simple_user.pk
    login(app, superuser_or_admin, '/manage/')

    Attribute.objects.create(
        name='address_autocomplete', label='Address (autocomplete)',
        kind='address_auto', user_visible=True, user_editable=True)

    resp = app.get(url)
    assert resp.html.find('select', {'name': 'address_autocomplete'})
    assert resp.html.find('input', {'id': 'manual-address'})


def test_manager_email_verified_column_user(app, simple_user, superuser_or_admin):
    login(app, superuser_or_admin, '/manage/')

    resp = app.get('/manage/users/')
    assert not resp.html.find('span', {'class': 'verified'})

    simple_user.email_verified = True
    simple_user.save()
    resp = app.get('/manage/users/')
    assert resp.html.find('span', {'class': 'verified'}).text == simple_user.email


def test_manager_user_link_column_is_active(app, simple_user, superuser_or_admin):
    login(app, superuser_or_admin, '/manage/')

    resp = app.get('/manage/users/')
    assert not resp.html.find('span', {'class': 'disabled'})

    simple_user.is_active = False
    simple_user.save()
    resp = app.get('/manage/users/')
    assert resp.html.find('span', {'class': 'disabled'}).text == 'Jôhn Dôe (disabled)'


def test_manager_user_username_field(app, superuser, simple_user):
    login(app, superuser, '/manage/')

    resp = app.get(reverse('a2-manager-user-detail', kwargs={'pk': simple_user.id}))
    assert resp.html.find('input', {'name': 'username'})
    resp = app.get(reverse('a2-manager-user-edit', kwargs={'pk': simple_user.id}))
    assert resp.html.find('input', {'name': 'username'})

    # remove username from user
    simple_user.username = ''
    simple_user.save()

    resp = app.get(reverse('a2-manager-user-detail', kwargs={'pk': simple_user.id}))
    assert resp.html.find('input', {'name': 'username'})
    resp = app.get(reverse('a2-manager-user-edit', kwargs={'pk': simple_user.id}))
    assert resp.html.find('input', {'name': 'username'})

    # disable usernames on organizational unit
    simple_user.ou.show_username = False
    simple_user.ou.save()

    resp = app.get(reverse('a2-manager-user-detail', kwargs={'pk': simple_user.id}))
    assert not resp.html.find('input', {'name': 'username'})
    resp = app.get(reverse('a2-manager-user-edit', kwargs={'pk': simple_user.id}))
    assert not resp.html.find('input', {'name': 'username'})

    # but it's still displayed if it was set
    simple_user.username = 'user'
    simple_user.save()

    resp = app.get(reverse('a2-manager-user-detail', kwargs={'pk': simple_user.id}))
    assert resp.html.find('input', {'name': 'username'})
    resp = app.get(reverse('a2-manager-user-edit', kwargs={'pk': simple_user.id}))
    assert resp.html.find('input', {'name': 'username'})


def test_manager_user_address_autocomplete_field(app, superuser, simple_user):
    login(app, superuser, '/manage/')
    Attribute.objects.create(
        name='address_autocomplete', label='Address (autocomplete)',
        kind='address_auto', user_visible=True, user_editable=True)
    resp = app.get(reverse('a2-manager-user-detail', kwargs={'pk': simple_user.id}))
    assert not resp.html.find('select', {'name': 'address_autocomplete'})
    assert not resp.html.find('input', {'id': 'manual-address'})


def test_manager_user_roles_visibility(app, simple_user, admin, ou1, ou2):
    Role = get_role_model()
    role1 = Role.objects.create(name='Role 1', slug='role1', ou=ou1)
    role2 = Role.objects.create(name='Role 2', slug='role2', ou=ou2)
    simple_user.roles.add(role1)
    simple_user.roles.add(role2)
    simple_user.save()

    login(app, admin, '/manage/')

    resp = app.get(reverse('a2-manager-user-detail', kwargs={'pk': simple_user.id}))
    assert '/manage/roles/%s/' % role1.pk in resp.text
    assert 'Role 1' in resp.text
    assert '/manage/roles/%s/' % role2.pk in resp.text
    assert 'Role 2' in resp.text

    app.get('/logout/').form.submit()

    other_user = get_user_model().objects.create(
        username='other_user', ou=ou1)
    other_user.set_password('auietsrn')
    other_role = Role.objects.create(name='Other role', slug='other-role', ou=ou1)
    view_role1_perm = get_permission_model().objects.create(
            operation=get_operation(VIEW_OP),
            target_ct=ContentType.objects.get_for_model(Role),
            target_id=role1.pk)
    other_role.permissions.add(get_view_user_perm())
    other_role.permissions.add(view_role1_perm)
    other_role.save()
    other_user.roles.add(other_role)
    other_user.save()

    login(app, other_user, '/manage/', 'auietsrn')
    resp = app.get(reverse('a2-manager-user-detail', kwargs={'pk': simple_user.id}))
    assert '/manage/roles/%s/' % role1.pk in resp.text
    assert 'Role 1' in resp.text
    assert '/manage/roles/%s/' % role2.pk not in resp.text
    assert 'Role 2' in resp.text
    app.get('/manage/roles/%s/' % role2.pk, status=403)


def test_manager_user_authorizations(app, superuser, simple_user):
    '''
    for 3 kind of users:
    * check if a button is provided on user detail page
    * access user service consents page
    * try to remove a service consent
    '''
    from django_rbac.utils import get_role_model, get_operation, get_permission_model
    from django_rbac.models import VIEW_OP
    from authentic2.a2_rbac.models import MANAGE_AUTHORIZATIONS_OP
    from tests.conftest import create_user
    Role = get_role_model()
    user_detail_url = reverse('a2-manager-user-detail', kwargs={'pk': simple_user.id})
    user_authorizations_url = reverse(
        'a2-manager-user-authorizations', kwargs={'pk': simple_user.id})

    resp = login(app, superuser)
    resp = app.get(user_detail_url, status=200)
    assert user_authorizations_url not in [
        x['href'] for x in resp.html.find('ul', {'class': 'extra-actions-menu'}).find_all('a')]

    # add a service consent to simple_user
    oidc_client = OIDCClient.objects.create(
        name='client',
        slug='client',
        ou=simple_user.ou,
        redirect_uris='https://example.com/')

    resp = app.get(user_detail_url, status=200)
    assert user_authorizations_url in [
        x['href'] for x in resp.html.find('ul', {'class': 'extra-actions-menu'}).find_all('a')]

    auth = OIDCAuthorization.objects.create(
        client=oidc_client, user=simple_user, scopes='openid',
        expired='2020-01-01T12:01:01Z')
    assert OIDCAuthorization.objects.count() == 1

    view_user_perm = get_permission_model().objects.create(
        operation=get_operation(VIEW_OP),
        target_ct=ContentType.objects.get_for_model(User),
        target_id=simple_user.pk)
    view_user_role = Role.objects.create(name='view_user', ou=simple_user.ou)
    view_user_role.permissions.add(view_user_perm)

    manage_auth_perm = get_permission_model().objects.create(
        operation=get_operation(MANAGE_AUTHORIZATIONS_OP),
        target_ct=ContentType.objects.get_for_model(User),
        target_id=simple_user.pk)
    manage_auth_role = Role.objects.create(name='manage_auth', ou=simple_user.ou)
    manage_auth_role.permissions.add(manage_auth_perm)

    user1 = create_user(username='agent1', ou=simple_user.ou)
    user2 = create_user(username='agent2', ou=simple_user.ou)
    user2.roles.add(view_user_role)
    user3 = create_user(username='agent3', ou=simple_user.ou)
    user3.roles.add(manage_auth_role)

    # user1 without permission
    resp = login(app, user1)
    resp = app.get(user_detail_url, status=403)
    assert 'You are not authorized to see this page' in resp.text
    resp = app.get(user_authorizations_url, status=403)
    assert 'You are not authorized to see this page' in resp.text
    params = {'authorization': auth.pk, 'csrfmiddlewaretoken': '???'}
    resp = app.post(user_authorizations_url, params=params, status=302)
    assert OIDCAuthorization.objects.count() == 1

    # user2 can see auth authorizations
    resp = login(app, user2)
    resp = app.get(user_detail_url, status=200)
    assert user_authorizations_url in [
        x['href'] for x in resp.html.find('ul', {'class': 'extra-actions-menu'}).find_all('a')]
    resp = resp.click('Consents')
    assert resp.html.find('h2').text == 'Consent Management'
    assert resp.html.find('td', {'class': 'remove-icon-column'}).a['class'] == ['disabled']
    # cannot click it's JS :/
    token = str(resp.context['csrf_token'])
    params = {'authorization': auth.pk, 'csrfmiddlewaretoken': token}
    resp = app.post(user_authorizations_url, params=params, status=302)
    assert OIDCAuthorization.objects.count() == 1

    # user3 can remove auth authorizations
    resp = login(app, user3)
    resp = app.get(user_detail_url, status=200)
    assert user_authorizations_url in [
        x['href'] for x in resp.html.find('ul', {'class': 'extra-actions-menu'}).find_all('a')]
    resp = resp.click('Consents')
    resp = app.get(user_authorizations_url, status=200)
    assert resp.html.find('h2').text == 'Consent Management'
    assert resp.html.find('td', {'class': 'remove-icon-column'}).a['class'] == ['js-remove-object']
    # cannot click it's JS :/
    token = str(resp.context['csrf_token'])
    params = {'authorization': auth.pk, 'csrfmiddlewaretoken': token}
    resp = app.post(user_authorizations_url, params=params, status=302)
    assert OIDCAuthorization.objects.count() == 0
    resp = resp.follow()
    assert resp.html.find('td').text == \
        'This user has not granted profile data access to any service yet.'


def test_manager_user_authorizations_breadcrumb(app, superuser, simple_user):
    resp = login(app, superuser)
    user_authorizations_url = reverse(
        'a2-manager-user-authorizations', kwargs={'pk': simple_user.id})
    resp = app.get(user_authorizations_url, status=200)
    assert [x.text for x in resp.html.find('span', {'id': 'breadcrumb'}).find_all('a')] == [
        'Homepage', 'Administration', 'Users', 'Default organizational unit',
        'Jôhn Dôe', 'Consent Management']
    user_authorizations_url = reverse(
        'a2-manager-user-authorizations', kwargs={'pk': superuser.id})
    resp = app.get(user_authorizations_url, status=200)
    assert [x.text for x in resp.html.find('span', {'id': 'breadcrumb'}).find_all('a')] == [
        'Homepage', 'Administration', 'Users',
        'super user', 'Consent Management']


def test_manager_user_roles_breadcrumb(app, superuser, simple_user):
    resp = login(app, superuser)
    user_roles_url = reverse(
        'a2-manager-user-roles', kwargs={'pk': simple_user.id})
    resp = app.get(user_roles_url, status=200)
    assert [x.text for x in resp.html.find('span', {'id': 'breadcrumb'}).find_all('a')] == [
        'Homepage', 'Administration', 'Users', 'Default organizational unit',
        'Jôhn Dôe', 'Roles']
    user_roles_url = reverse(
        'a2-manager-user-roles', kwargs={'pk': superuser.id})
    resp = app.get(user_roles_url, status=200)
    assert [x.text for x in resp.html.find('span', {'id': 'breadcrumb'}).find_all('a')] == [
        'Homepage', 'Administration', 'Users',
        'super user', 'Roles']


def test_manager_create_user_duplicates(admin, app, ou1, settings):
    settings.A2_MANAGER_CHECK_DUPLICATE_USERS = True
    Attribute.objects.create(
        kind='birthdate', name='birthdate', label='birthdate', required=False, searchable=True
    )

    user = User.objects.create(
        first_name='Alexander', last_name='Longname', email='alexandre.longname@entrouvert.com'
    )
    user.attributes.birthdate = datetime.date(1980, 1, 2)
    user2 = User.objects.create(first_name='Alexandra', last_name='Longname')
    user3 = User.objects.create(first_name='Alex', last_name='Shortname')

    login(app, admin)
    resp = app.get('/manage/users/%s/add/' % ou1.pk)

    form = resp.form
    form.set('first_name', 'Alexandre')
    form.set('last_name', 'Longname')
    form.set('email', 'alex@entrouvert.com')
    form.set('password1', 'ABcd1234')
    form.set('password2', 'ABcd1234')
    resp = form.submit()

    assert 'user may already exist' in resp.text
    assert 'Alexander Longname' in resp.text
    assert '- alexandre.longname@entrouvert.com' in resp.text
    assert '- 1980-01-02' in resp.text
    assert '/users/%s/' % user.pk in resp.text
    assert 'Alexandra Longname' in resp.text
    assert '/users/%s/' % user2.pk in resp.text

    # This user was in fact duplicate. Agent reuses the form to fill details on another user
    form = resp.form
    form.set('first_name', 'Alexa')
    form.set('last_name', 'Shortname')
    form.set('email', 'ashortname@entrouvert.com')
    resp = form.submit()

    assert 'user may already exist' in resp.text
    assert '/users/%s/' % user3.pk in resp.text

    # Not a duplicate this time. Simply submitting again creates user
    resp = resp.form.submit().follow()
    assert User.objects.filter(first_name='Alexa').count() == 1