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

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

    return user

def import_users_from_csv(csv_file, ou):

    reader = csv.reader(csv_file) # XXX dialect

    rows = list(reader)

    app_id_label = app_settings.A2_USERS_CSV_IMPORT_APP_ID_LABEL

    rows[0].index(app_id_label)

    with transaction.atomic():

        User = get_user_model()

        for i in range(1,len(rows)):

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

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

            app_id = user_data.pop(app_id_label)

            user = create_user_from_data(user_data)

            user_external_id = UserExternalId.objects.create(

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

    return len(rows)-1

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

        initial=get_default_ou().uuid

    )

   {% 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 with named columns:" %}

  <ul>

    <li>{{ app_id_label }}</li>

  {% for attr in attributes %}

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

  {% endfor %}

  </ul>

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

    {% csrf_token %}

    {{ form.as_p }}

    {% if form.instance and form.instance.id %}

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

      <button class="cancel-button" name="cancel" formnovalidate>{% trans "Cancel" %}</button>

    {% else %}

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

    {% endif %}

  </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_uuid = form.cleaned_data.get('ou')

        ou = get_ou_model().objects.get(uuid=ou_uuid) or get_default_ou()

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

        if not csv_file:

            messages.warning(

                self.request,

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

            )

        else:

            try:

                nb_created = import_users_from_csv(csv_file, ou)

            except ValueError as e:

                messages.error(

                    self.request,

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

                )

            else:

                messages.info(

                    self.request,

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

                )

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

users_csv_import = UsersCsvImportView.as_view()

@GlobalCache(timeout=10)

def get_ou_choices():

    Ou = get_ou_model()

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

from django.contrib.auth import get_user_model

import csv

    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

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

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

app_id,first_name,last_name

123,Hervé,Dupont

456,René,Durant

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