Produits Entr'ouvert » Authentic 2

    A2_USERS_CSV_IMPORT_APP_ID_LABEL=Setting(default='app_id', definition='Label of the applicative identifier used when importing users from csv'),

import csv

import logging

from django.contrib.auth import get_user_model

from django.db import transaction

from authentic2 import app_settings

from authentic2.models import UserExternalId, Attribute

logger = logging.getLogger(__name__)

def create_user_from_data(user_data, app_id):

    User = get_user_model()

    user = User.objects.create()

    for name, value in user_data.items():

        if hasattr(user, name):

            setattr(user, name, value)

        elif hasattr(user.attributes, name):

            attribute = Attribute.objects.get(name=name)

            deserialize = attribute.get_kind()['deserialize']

            try:

                value = deserialize(value)

            except:

                logger.error(

                    'invalid value %s for attribute %s' % (value, name))

                continue

            attribute.set_value(owner=user, value=value)

    user.save()

    UserExternalId.objects.create(

        user=user, external_id=app_id, source='import')

    return user

def import_users_from_csv(csv_file, ou, erase=False):

    reader = csv.reader(csv_file) # XXX dialect # XXX bom

    rows = list(reader)

    app_id_label = app_settings.A2_USERS_CSV_IMPORT_APP_ID_LABEL

    try:

        rows[0].index(app_id_label)

    except ValueError:

        logger.error(

            "csv file doesn't contain a column named %s" % app_id_label)

        return 0, 0, 0

    nb_created = 0

    nb_erased = 0

    nb_ignored = 0

    with transaction.atomic():

        User = get_user_model()

        for row in rows[1:]:

            user_data = dict(zip(rows[0], row))

            user_data.update({'ou': ou})

            app_id = user_data.pop(app_id_label)

            try:

                user_external_id = UserExternalId.objects.get(

                    external_id=app_id, source='import')

            except UserExternalId.MultipleObjectsReturned:

                logger.error(

                    '''multiple user external ids return for external id value

                    '%s' and source 'import'.''' % app_id)

                nb_ignored = nb_ignored+1

                continue

            except UserExternalId.DoesNotExist:

                exists = False

            else:

                exists = True

            if erase:

                user_external_id.user.delete()

                user_external_id.delete()

                nb_erased = nb_erased+1

            if erase or not exists:

                create_user_from_data(user_data, app_id)

                nb_created = nb_created+1

            if exists and not erase:

                nb_ignored = nb_ignored+1

    return nb_created, nb_erased, nb_ignored

class UsersCsvImportForm(forms.Form):

    csv_file = forms.FileField(

        label=_('CSV file'),

        help_text=_('CSV file for the import.')

    )

    ou = forms.ChoiceField(

        label=_('Organizational unit'),

        help_text=_('The newly created users will be registered to the chosen OU.'),

        choices=utils.get_ou_choices,

    )

    erase = forms.BooleanField(

        label=_('Erase existing users'),

        help_text=_('''Defines whether to erase already existing users. If

            unchecked, these users won't undergo any deletion.'''),

        required=False,

    )

   {% if import_users %}

     <a

        href="{% url "a2-manager-users-import" %}"

        id="import-users-btn">

         {% trans "Import users from csv" %}

     </a>

   {% else %}

     <a

        href="#"

        class="disabled"

        id="import-users-btn">

         {% trans "Import users from csv" %}

     </a>

   {% endif %}

{% extends "authentic2/manager/form.html" %}

{% load i18n %}

{% block page-title %}

  {{ block.super }} - {{ view.title }}

{% endblock %}

{% block breadcrumb %}

  {{ block.super }}

  <a href="..">{% trans "Users" %}</a>

  <a href="">{{ view.title }}</a>

{% endblock %}

{% block content %}

  {% trans "Comma-separated value file defining optional standard user attributes as well as the extended profile attributes below:" %}

  <ul>

    <li><b>{{ app_id_label }}</b> ({% trans "mandatory" %})</li>

  {% for attr in attributes %}

    <li>{{ attr.name }}</li>

  {% endfor %}

  </ul>

  <form enctype="multipart/form-data" method="post">

    {% csrf_token %}

    {{ form.as_p }}

    <button class="submit-button">{% trans "Create" %}</button>

  </form>

{% endblock %}

        url(r'^users/import/$',

            user_views.users_csv_import, name='a2-manager-users-import'),

from django.views.generic import View, FormView

from authentic2 import hooks, app_settings as a2_app_settings

    UserChangePasswordForm, ChooseUserRoleForm, UserRoleSearchForm, UserChangeEmailForm,

    UsersCsvImportForm)

        if self.request.user.has_perm('custom_user.add_user'):

            ctx['import_users'] = True

class UsersCsvImportView(FormView):

    form_class = UsersCsvImportForm

    template_name = 'authentic2/manager/users_csv_import.html'

    title = _('CSV import')

    success_url = '..'

    permissions = ['custom_user.add_user']

    def get_context_data(self, **kwargs):

        from authentic2.models import Attribute

        ctx = super(UsersCsvImportView, self).get_context_data()

        ctx.update({

            'attributes': Attribute.objects.all(),

            'app_id_label': a2_app_settings.A2_USERS_CSV_IMPORT_APP_ID_LABEL

        })

        return ctx

    def form_valid(self, form):

        from authentic2.data_transfer import import_users_from_csv

        ou_id = form.cleaned_data.get('ou')

        ou = get_ou_model().objects.get(id=ou_id) or get_default_ou()

        csv_file = form.cleaned_data.get('csv_file')

        erase = form.cleaned_data.get('erase')

        if not csv_file:

            messages.warning(

                self.request,

                _('No import performed. Please provide a csv file.')

            )

        else:

            try:

                nb_created, nb_modified, nb_ignored = import_users_from_csv(

                        csv_file, ou, modify)

            except ValueError as e:

                messages.error(

                    self.request,

                    _('Value error while importing the users: %s') % e

                )

            else:

                message = _('Created %s users.') % nb_created

                if nb_modified:

                    message = message + (' Modified %s users.') % nb_modified

                if nb_ignored:

                    message = message + (' Ignored %s users.') % nb_ignored

                messages.info(

                    self.request,

                    message

                )

        return super(UsersCsvImportView, self).form_valid(form)

users_csv_import = UsersCsvImportView.as_view()

def get_ou_choices():

    Ou = get_ou_model()

    return [(ou.id, ou.name) for ou in Ou.objects.all()]

from django.contrib.auth import get_user_model

import csv

from authentic2.a2_rbac.utils import get_default_ou

    import_ou,

    import_users_from_csv)

from authentic2.models import UserExternalId, Attribute

def test_import_users_from_csv(db):

    ou = OU.objects.create(name='ou')

    User = get_user_model()

    assert not len(User.objects.all())

    with open('tests/users_csv_import_testfile.csv', 'r') as csv_file:

        assert import_users_from_csv(csv_file, ou) == (2, 0, 0)

    with open('tests/users_csv_import_testfile.csv', 'r') as csv_file:

        reader = csv.reader(csv_file)

        rows = list(reader)

        assert len(User.objects.all()) == len(rows)-1

        for i in range(2):

            user = User.objects.all()[i]

            assert user.first_name == rows[i+1][1].decode('utf-8')

            assert user.last_name == rows[i+1][2].decode('utf-8')

            assert user.email == rows[i+1][3].decode('utf-8')

            assert user.ou == ou

            external_id = UserExternalId.objects.get(user=user)

            assert external_id.source == 'import'

            assert external_id.external_id == rows[i+1][0].decode('utf-8')

def test_import_users_from_csv_profile_attributes(db):

    Attribute.objects.create(

        label='Nickname',

        name='nickname',

        kind='string',

        user_visible=True,

        user_editable=True

    )

    Attribute.objects.create(

        label='Is a spy',

        name='is_a_spy',

        kind='boolean',

        user_visible=True,

        user_editable=True

    )

    Attribute.objects.create(

        label='Enrolled on',

        name='enrolled_on',

        kind='date',

        user_visible=True,

        user_editable=True

    )

    Attribute.objects.create(

        label='Secret phone number',

        name='secret_phone_number',

        kind='phone_number',

        user_visible=True,

        user_editable=True

    )

    Attribute.objects.create(

        label='Headquarters postcode',

        name='headquarters_postcode',

        kind='fr_postcode',

        user_visible=True,

        user_editable=True

    )

    ou = OU.objects.create(name='ou')

    User = get_user_model()

    assert not len(User.objects.all())

    with open('tests/users_csv_import_testfile2.csv', 'r') as csv_file:

        assert import_users_from_csv(csv_file, ou) == (2, 0, 0)

    with open('tests/users_csv_import_testfile2.csv', 'r') as csv_file:

        reader = csv.reader(csv_file)

        rows = list(reader)

        assert len(User.objects.all()) == len(rows)-1

        for i in range(2):

            user = User.objects.all()[i]

            assert user.attributes.nickname == rows[i+1][3].decode('utf-8')

            assert user.attributes.is_a_spy == bool(int(rows[i+1][4].decode('utf-8')))

            assert user.attributes.enrolled_on.isoformat() == rows[i+1][5].decode('utf-8')

            assert user.attributes.secret_phone_number == rows[i+1][6].decode('utf-8')

            assert user.attributes.headquarters_postcode == rows[i+1][7].decode('utf-8')

def test_import_users_from_csv_modify(db):

    ou = get_default_ou()

    with open('tests/users_csv_import_testfile_modify1.csv', 'r') as csv_file:

        assert import_users_from_csv(csv_file, ou) == (1, 0, 0) # 1 created

    with open('tests/users_csv_import_testfile_modify2.csv', 'r') as csv_file:

        assert import_users_from_csv(csv_file, ou, True) == (1, 1, 0) # 1 created and 1 erased

    with open('tests/users_csv_import_testfile_modify1.csv', 'r') as csv_file:

        assert import_users_from_csv(csv_file, ou, False) == (0, 0, 1) # 1 ignored

from django import VERSION as django_version

from authentic2 import app_settings

from authentic2.models import Attribute

@pytest.mark.skipif(django_version < (1, 11),

        reason='Requires Django v1.11 or higher.')

def test_manager_import_users_from_csv(admin, app, db):

    Attribute.objects.create(

        label='Nickname',

        name='nickname',

        kind='string',

        user_visible=True,

        user_editable=True

    )

    Attribute.objects.create(

        label='Freemason',

        name='freemason',

        kind='boolean',

        user_visible=True,

        user_editable=True

    )

    User = get_user_model()

    nb_initial = len(User.objects.all())

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

    url = u'/manage/users/import/'

    ou_foo = get_ou_model().objects.create(name='Foo', slug='foo')

    ou_foo.save()

    ou_bar = get_ou_model().objects.create(name='Bar', slug='bar')

    ou_bar.save()

    ous = [

        ou_foo,

        ou_bar,

        get_default_ou()

    ]

    response = app.get(url)

    assert 'nickname' in response.text

    assert 'freemason' in response.text

    assert app_settings.A2_USERS_CSV_IMPORT_APP_ID_LABEL in response.text

    for ou in ous:

        assert 'value="%s"' % ou.id in response.text

    form = response.form

    form.set('csv_file', Upload('tests/users_csv_import_testfile3.csv'))

    form.set('ou', str(ou_bar.id))

    response = form.submit().follow()

    assert len(User.objects.all()) == nb_initial+2

    for last_name in ('Dupont', 'Durant'):

        assert User.objects.get(last_name=last_name).ou == ou_bar

app_id,first_name,last_name,email

123,Hervé,Dupont,foo@bar.null

456,René,Durant,boo@far.null

app_id,first_name,last_name,nickname,is_a_spy,enrolled_on,secret_phone_number,headquarters_postcode

123,Hervé,Dupont,Herbie,1,2006-06-06,+12345,75001

456,René,Durant,Ronnie,0,2007-07-07,+12356,13001

app_id,first_name,last_name,nickname,freemason

123,Hervé,Dupont,Herv,0

456,René,Durant,RR,1

app_id,first_name,last_name,email

123,Hervé,Dupont,foo@bar.null

app_id,first_name,last_name,email

123,Hervé,Durant,foo@bar.null