Produits Entr'ouvert » Authentic 2

include media/misc/users_csv_import_example.csv

source_domain,source_id,first_name,last_name,username,email,ou_slug

domain.nowhere.null,123,Hervé,Dupont,hdup,foo@bar.null,foo

domain.nowhere.null,456,René,Durant,rdur,boo@far.null,bar

import logging

from django.contrib.auth import get_user_model

from django.db import transaction

from authentic2.models import Attribute, UserExternalId

def import_users_from_csv_rows(rows, configuration, attributes_configuration,

        dry_run=False, erase=False):

    """

    This is the backend csv import function called by the users csv import

    management form.

    This function takes as mandatory input (i) the csv rows (i.e. a list of

    lists, each one of the latter representing a split csv line), whose first

    row must be the named columns of the csv ; (ii) the column names for the

    user-attributes ; and (iii)

    If the `dry_run` optional argument is set, the import will be simulated

    only.

    If the `erase`, the presence of csv entries mapping to already-existing

    users will result in the deletion of such user accounts. Newly-created user-

    accounts will be replacing them.

    The `results` dictionary object returned by this function provides global as

    well as ou- and source-wise metrics about the import process.

    """

    logger = logging.getLogger(__name__)

    class UsersCsvImportDryRun(Exception):

        """

        Will be used to simulate the user import. If the import is simulated,

        this exception at the end of the transaction.

        """

        pass

    def get_ou(configuration, user_data):

        """

        Retrieve the organizational unit of the user depending on its entry in

        the csv file. If the entry doesn't mention any valid OU, the (mandatory)

        fallback OU is chosen.

        """

        OrganizationalUnit = get_ou_model()

        ou_column_name = configuration.get('ou_column_name', '')

        ou = None

        qs_keyword = None

        if ou_column_name is not None:

            for candidate_keyword in ('id', 'name', 'slug'):

                if ou_column_name.endswith(candidate_keyword):

                    qs_keyword = candidate_keyword

                    break

        if qs_keyword:

            qs_filter = {

                qs_keyword: user_data.get(configuration.get('ou_column_name'))

            }

            try:

                ou = OrganizationalUnit.objects.get(**qs_filter)

            except (OrganizationalUnit.DoesNotExist,

                    OrganizationalUnit.MultipleObjectsReturned):

                logger.error('No OU found for filter %s' % qs_filter)

                pass

        if not ou:

            ou = OrganizationalUnit.objects.get(id=configuration['fallback_ou'])

        return ou

    def create_user_from_data(user_data, attributes_configuration, ou,

            source_id, source_domain):

        """

        Create the user according to the csv entry line, the organizational

        and user-attribute configuration defined in the users management

        csv import form.

        """

        User = get_user_model()

        user = User.objects.create()

        user_attributes = {}

        for key, value in attributes_configuration.items():

            if not value:

                continue

            if hasattr(user, key):

                setattr(user, key, user_data.get(value))

            elif hasattr(user.attributes, key):

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

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

                try:

                    value = deserialize(user_data.get(value))

                except:

                    logger.error(

                        'invalid value %s for attribute %s' % (

                        user_data.get(value), key))

                    continue

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

            user.ou = ou

            user.save()

        return user

    def update_results(results, ou, source, modified_entry):

        # update ou local results

        if ou.slug not in results['by_ou']:

            results['by_ou'][ou.slug] = {'name': ou.name}

        counter = results['by_ou'][ou.slug].get(modified_entry, 0)

        results['by_ou'][ou.slug].update({modified_entry: counter+1})

        # update source local results

        if source not in results['by_source']:

            results['by_source'][source] = {}

        counter = results['by_source'][source].get(modified_entry, 0)

        results['by_source'][source].update({modified_entry: counter+1})

        # update global results

        results[modified_entry] = results[modified_entry]+1

    results = {

        # global results

        'nb_created': 0, 'nb_erased': 0, 'nb_ignored': 0,

        # local results

        'by_ou': {}, 'by_source': {},

        # error message for rendering in template

        'error': None,

    }

    try:

        source_id_column = configuration.get('source_id_column_name')

        source_domain_column = configuration.get('source_domain_column_name')

        if not (source_id_column and source_domain_column):

            error_msg = 'Missing external id or source domain column names.'

            logger.warning(error_msg)

            results['error'] = error_msg

            return results

        with transaction.atomic():

            User = get_user_model()

            for row in rows[1:]:

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

                ou = get_ou(configuration, user_data)

                external_id = user_data.get(source_id_column)

                source_domain = user_data.get(source_domain_column)

                if not (external_id and source_domain):

                    logger.warning('missing external id or source domain for '

                            'user %r' % user_data)

                    continue

                try:

                    user_external_id = UserExternalId.objects.get(

                        external_id=user_data.get(source_id_column),

                        source=user_data.get(source_domain_column))

                except UserExternalId.MultipleObjectsReturned:

                    logger.error('''multiple user external ids returned for

                        external id value '%s' and source %s.''' % (

                        external_id, source_domain))

                    results['nb_ignored'] = results['nb_ignored']+1

                    continue

                except UserExternalId.DoesNotExist:

                    exists = False

                else:

                    exists = True

                if erase:

                    erased_user_ou = user_external_id.user.ou

                    user_external_id.user.delete()

                    user_external_id.delete()

                    update_results(results, erased_user_ou, source_domain, 'nb_erased')

                if erase or not exists:

                    user = create_user_from_data(

                        user_data, attributes_configuration, ou,

                        external_id, source_domain)

                    UserExternalId.objects.get_or_create(

                        user=user, external_id=external_id,

                        source=source_domain)

                    update_results(results, ou, source_domain, 'nb_created')

                if exists and not erase:

                    update_results(results, ou, source_domain, 'nb_ignored')

            if dry_run:

                raise UsersCsvImportDryRun()

    except UsersCsvImportDryRun as e:

        logger.info('Performed Import. Now cancelling the database transaction')

        pass

    except Exception as e:

        results['error'] = error = 'Error while performing the import: %s' % e

        logger.error(error)

    return results

from authentic2.models import PasswordReset, Attribute

class UsersCsvImportUploadFileForm(forms.Form):

    csv_file = forms.FileField(

        label=_('CSV file'),

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

class UsersCsvImportConfigureForm(forms.Form):

    ou_column_name = forms.CharField(

        label=_('Organizational unit column name'),

        max_length=63, required=False)

    fallback_ou = forms.ChoiceField(

        label=_('Fallback organizational unit'),

        help_text=_('''The newly created users will be registered to this chosen

            fallback OU if their entry line do not mention any OU.'''),

        choices=utils.get_ou_choices)

    source_domain_column_name = forms.CharField(

        label=_('Source domain column name'),

        max_length=63)

    source_id_column_name = forms.CharField(

        label=_('Source identifier column name'),

        max_length=63)

class UsersCsvImportConfigureAttributesForm(forms.Form):

    first_name = forms.CharField(label=_('First name'), max_length=63,

        required=False)

    last_name = forms.CharField(label=_('Last name'), max_length=63,

        required=False)

    username = forms.CharField(label=_('Username'), max_length=63,

        required=False)

    email = forms.CharField(label=_('Email address'), max_length=63,

        required=False)

    email_verified = forms.CharField(

        label=_('Has a verified email address'), max_length=63, required=False)

    is_staff = forms.CharField(label=_('Is staff'), max_length=63,

        required=False)

    is_active = forms.CharField(label=_('Is active'), max_length=63,

        required=False)

    def __init__(self, *args, **kwargs):

        """

            TODO: provide more subtlety in initial value retrieval

        """

        header = kwargs.pop('header', [])

        super(UsersCsvImportConfigureAttributesForm, self).__init__(*args, **kwargs)

        # provide initial for standard attribute fields

        for entry in header:

            if self.fields.get(entry):

                self.fields[entry].initial = entry

        # instantiate extended profile attribute fields

        for attribute in Attribute.objects.all():

            if self.fields.get(attribute.name):

                # avoid attribute redundancy introduced by default in authentic

                continue

            field = self.fields['%s' % attribute.name] = forms.CharField(

                label=attribute.label, max_length=63, required=False)

            if not header:

                # no way to provide any initial value for this field

                continue

            if attribute.name in header: # XXX too simplistical

                field.initial = attribute.name

   {% if import_users %}

     <a

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

        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 %}

  <p>{% blocktrans %}{{ nb_lines }} user account data lines in input file.{% endblocktrans %}</p>

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

    {% csrf_token %}

    <p>{% trans "Please fill the csv column names for the organization information:" %}</p>

    {{ form.as_p }}

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

  </form>

{% endblock %}

{% 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 %}

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

    {% csrf_token %}

    <p>{% trans "Please fill the column names for the following user attributes:" %}</p>

    {{ form.as_p }}

    <p>{% trans "An empty field means that the corresponding attribute won't be provisioned." %}</p>

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

  </form>

{% endblock %}

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

{% load i18n %}

{% load authentic2 %}

{% 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 %}

  {% if results|get_item:"nb_created" or results|get_item:"nb_erased" or results|get_item:"nb_ignored" %}

    <div id="global" class="import-results">

    <h2>{% trans "Global results :" %}</h2>

    <ul>

    {% include "authentic2/manager/users_csv_import_perform_include.html" with results=results %}

    </ul>

    </div>

    {% if results.by_ou.items %}

    <div id="by-ou" class="import-results">

      <p>{% trans "Results by organizational unit:" %}</p>

      <ul>

      {% for ou, values in results.by_ou.items %}

        <li>{{ values|get_item:"name" }}

          <ul>

            {% include "authentic2/manager/users_csv_import_perform_include.html" with results=values %}

          </ul>

        </li>

      {% endfor %}

      </ul>

    </div>

    {% endif %}

    {% if results.by_source.items %}

    <div id="by-source" class="import-results">

      <span>{% trans "Results by source:" %}</span>

      <ul>

      {% for source, values in results.by_source.items %}

        <li>{{ source }}

          <ul>

            {% include "authentic2/manager/users_csv_import_perform_include.html" with results=values %}

          </ul>

        </li>

      {% endfor %}

    </div>

    {% endif %}

    <a class="button" href="{% url "a2-manager-users" %}">{% trans "Go back to the users page" %}</a>

  {% else %}

    <div id="global" class="import-results">

    {% trans "No import operation has been performed." %}

    {% if results|get_item:"error" %}{% blocktrans with error=results|get_item:"error" %}Reason: {{ error }}{% endblocktrans %} {% endif %}

  {% endif %}

{% endblock %}

{% load i18n %}

{% load authentic2 %}

{% if results|get_item:"nb_created" %}

<li>{% blocktrans with counter=results|get_item:"nb_created" plural=results|get_item:"nb_created"|pluralize %}{{ counter }} user{{ plural }} have been created.{% endblocktrans %}</li>

{% endif %}

{% if results|get_item:"nb_erased" %}

<li>{% blocktrans with counter=results|get_item:"nb_erased" plural=results|get_item:"nb_erased"|pluralize %}{{ counter }} user{{ plural }} have been erased.{% endblocktrans %}</li>

{% endif %}

{% if results|get_item:"nb_ignored" %}

<li>{% blocktrans with counter=results|get_item:"nb_ignored" plural=results|get_item:"nb_ignored"|pluralize %}{{ counter }} user{{ plural }} have been ignored.{% endblocktrans %}</li>

{% endif %}

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

{% load i18n %}

{% load authentic2 %}

{% 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 %}

  {% if results|get_item:"nb_created" or results|get_item:"nb_erased" or results|get_item:"nb_ignored" %}

    <div id="global" class="import-results">

    <h2>{% trans "Global results :" %}</h2>

    <ul>

    {% include "authentic2/manager/users_csv_import_simulate_include.html" with results=results %}

    </ul>

    </div>

    {% if results.by_ou.items %}

    <div id="by-ou" class="import-results">

      <h2>{% trans "Results by organizational unit:" %}</h2>

      <ul>

      {% for ou, values in results.by_ou.items %}

        <li>{{ values|get_item:"name" }}

          <ul>

            {% include "authentic2/manager/users_csv_import_simulate_include.html" with results=values %}

          </ul>

        </li>

      {% endfor %}

      </ul>

    </div>

    {% endif %}

    {% if results.by_source.items %}

    <div id="by-source" class="import-results">

      <h2>{% trans "Results by source:" %}</h2>

      <ul>

      {% for source, values in results.by_source.items %}

        <li>{{ source }}

          <ul>

            {% include "authentic2/manager/users_csv_import_simulate_include.html" with results=values %}

          </ul>

        </li>

      {% endfor %}

    </div>

    {% endif %}

    <a class="button" href="{% url "a2-manager-users-csv-import-perform" %}">{% trans "Perform the import" %}</a>

    <a class="button" href="{% url "a2-manager-users" %}">{% trans "Cancel" %}</a>

  {% else %}

    <div id="global" class="import-results">

    {% trans "No import operation would be performed." %}

    {% if results|get_item:"error" %}{% blocktrans with error=results|get_item:"error" %}Reason: {{ error }}{% endblocktrans %} {% endif %}

  {% endif %}

{% endblock %}

{% load i18n %}

{% load authentic2 %}

{% if results|get_item:"nb_created" %}

<li>{% blocktrans with counter=results|get_item:"nb_created" plural=results|get_item:"nb_created"|pluralize %}{{ counter }} user{{ plural }} would be created.{% endblocktrans %}</li>

{% endif %}

{% if results|get_item:"nb_erased" %}

<li>{% blocktrans with counter=results|get_item:"nb_erased" plural=results|get_item:"nb_erased"|pluralize %}{{ counter }} user{{ plural }} would be erased.{% endblocktrans %}</li>

{% endif %}

{% if results|get_item:"nb_ignored" %}

<li>{% blocktrans with counter=results|get_item:"nb_ignored" plural=results|get_item:"nb_ignored"|pluralize %}{{ counter }} user{{ plural }} would be ignored.{% endblocktrans %}</li>

{% 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 %}

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

    {% csrf_token %}

    {{ form.as_p }}

    <p><a href="{{ example_csv_href }}">{% trans "Download an example csv file." %}</a></p>

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

  </form>

{% endblock %}

        url(r'^users/csv-import/upload-file/$',

            user_views.users_csv_import_upload_file,

            name='a2-manager-users-csv-import-upload-file'),

        url(r'^users/csv-import/configure/$',

            user_views.users_csv_import_configure,

            name='a2-manager-users-csv-import-configure'),

        url(r'^users/csv-import/configurea-attributes/$',

            user_views.users_csv_import_configure_attributes,

            name='a2-manager-users-csv-import-configure-attributes'),

        url(r'^users/csv-import/simulate/$',

            user_views.users_csv_import_simulate,

            name='a2-manager-users-csv-import-simulate'),

        url(r'^users/csv-import/perform/$',

            user_views.users_csv_import_perform,

            name='a2-manager-users-csv-import-perform'),

from django import forms

from django.views.generic import View, FormView, TemplateView

import csv

from authentic2.data_transfer import import_users_from_csv_rows

    UserChangePasswordForm, ChooseUserRoleForm, UserRoleSearchForm, UserChangeEmailForm,

    UsersCsvImportUploadFileForm, UsersCsvImportConfigureForm,

    UsersCsvImportConfigureAttributesForm)

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

            ctx['import_users'] = True

class UsersCsvImportUploadFileView(FormView):

    template_name = 'authentic2/manager/users_csv_import_upload_file.html'

    form_class = UsersCsvImportUploadFileForm

    title = _('Upload CSV file for user import')

    permissions = ['custom_user.add_user']

    example_csv_href = '/media/misc/users_csv_import_example.csv'

    def get_context_data(self, **kwargs):

        context = super(UsersCsvImportUploadFileView, self).get_context_data(

            **kwargs)

        context['example_csv_href'] = self.example_csv_href

        return context

    def get_success_url(self):

        return reverse('a2-manager-users-csv-import-configure')

    def form_valid(self, form):

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

        reader = csv.reader(csv_file)

        rows = list(reader)

        self.request.session['rows'] = rows

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

users_csv_import_upload_file = UsersCsvImportUploadFileView.as_view()

class UsersCsvImportConfigureView(FormView):

    template_name = 'authentic2/manager/users_csv_import_configure.html'

    form_class = UsersCsvImportConfigureForm

    title = _('Organization configuration')

    permissions = ['custom_user.add_user']

    def get_initial(self):

        initial = super(UsersCsvImportConfigureView, self).get_initial()

        # put number of data lines from the csv

        rows = self.request.session['rows'] # XXX dialect # XXX bom

        header = rows[0]

        # inspect csv_file for ou column name

        for ou_column_name in ['ou__slug', 'ou_slug', 'ou__name', 'ou_name',

                'ou__id', 'ou_id']:

            if ou_column_name in header:

                initial['ou_column_name'] = ou_column_name

                break

        # inspect csv_file for source_id column name

        for source_id_column_name in ['source_id', 'app_id']:

            if source_id_column_name in header:

                initial['source_id_column_name'] = source_id_column_name

                break

        # inspect csv_file for source_domain column name

        if 'source_domain' in header:

            initial['source_domain_column_name'] = 'source_domain'

        return initial

    def get_context_data(self, **kwargs):

        context = super(UsersCsvImportConfigureView, self).get_context_data(**kwargs)

        rows = self.request.session['rows'] # XXX dialect # XXX bom

        header = rows[0]

        context['nb_lines'] = max(len(rows), 0)

        return context

    def get_success_url(self):

        return reverse('a2-manager-users-csv-import-configure-attributes')

    def form_valid(self, form):

        # put configuration in session

        configuration = self.request.session['configuration'] = {}

        configuration.update(form.cleaned_data)

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

users_csv_import_configure = UsersCsvImportConfigureView.as_view()

class UsersCsvImportConfigureAttributesView(FormView):

    form_class = UsersCsvImportConfigureAttributesForm

    template_name = 'authentic2/manager/users_csv_import_configure_attributes.html'

    title = _('User attribute configuration')

    permissions = ['custom_user.add_user']

    def get_form_kwargs(self):

        kwargs = super(UsersCsvImportConfigureAttributesView, self).get_form_kwargs()

        if self.request.method == 'GET':

            rows = self.request.session['rows']

            header = rows[0]

            kwargs.update({'header': header})

        return kwargs

    def form_valid(self, form):

        # put attribute configuration in session

        attribute_configuration = self.request.session['attribute_configuration'] = {}

        attribute_configuration.update(form.cleaned_data)

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

    def get_success_url(self):

        return reverse('a2-manager-users-csv-import-simulate')

users_csv_import_configure_attributes = UsersCsvImportConfigureAttributesView.as_view()

class UsersCsvImportSimulateView(TemplateView):

    template_name = 'authentic2/manager/users_csv_import_simulate.html'

    title = _('Simulation of the user import from CSV file')

    permissions = ['custom_user.add_user']

    results = None

    def get(self, request, *args, **kwargs):

        rows = request.session.get('rows')

        # retrieve configuration from request

        attribute_configuration = request.session.get('attribute_configuration')

        configuration = request.session.get('configuration')

        self.results = import_users_from_csv_rows(

            rows, configuration, attribute_configuration, dry_run=True)

        return super(UsersCsvImportSimulateView, self).get(

            request, *args, **kwargs)

    def get_context_data(self, **kwargs):

        context = super(UsersCsvImportSimulateView, self).get_context_data(

            **kwargs)

        context['results'] = self.results

        return context

users_csv_import_simulate = UsersCsvImportSimulateView.as_view()

class UsersCsvImportPerformView(TemplateView):

    template_name = 'authentic2/manager/users_csv_import_perform.html'

    title = _('Peform the user import from CSV file')

    permissions = ['custom_user.add_user']

    results = None

    def get(self, request, *args, **kwargs):

        rows = request.session.pop('rows')

        # retrieve configuration from request

        attribute_configuration = request.session.pop('attribute_configuration')

        configuration = request.session.pop('configuration')

        self.results = import_users_from_csv_rows(

            rows, configuration, attribute_configuration, dry_run=False)

        return super(UsersCsvImportPerformView, self).get(

            request, *args, **kwargs)

    def get_context_data(self, **kwargs):

        context = super(UsersCsvImportPerformView, self).get_context_data(

            **kwargs)

        context['results'] = self.results

        return context

users_csv_import_perform = UsersCsvImportPerformView.as_view()

@GlobalCache(timeout=10)

def get_ou_choices():

    Ou = get_ou_model()

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

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

from __future__ import unicode_literals

from django.db import models, migrations

class Migration(migrations.Migration):

    dependencies = [

        ('authentic2', '0023_auto_20181031_0900'),

    ]

    operations = [

        migrations.AlterUniqueTogether(

            name='userexternalid',

            unique_together=set([('user', 'source'), ('external_id', 'source')]),

        ),

    ]

        unique_together = [['user', 'source'], ['external_id', 'source']]

from django import template

register = template.Library()

@register.filter(name='get_item')

def get_item(dictionary, value):

    return dictionary.get(value)

source_domain,source_id,first_name,last_name,username,email,ou_slug

domain.nowhere.null,123,Hervé,Dupont,hdup,foo@bar.null,foo

domain.nowhere.null,456,René,Durant,rdur,boo@far.null,bar

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

domain.nowhere.null,123,Hervé,Dupont,Herbie,1,2006-06-06,+12345,75001,hdup@bar.null

domain.nowhere.null,456,René,Durant,Ronnie,0,2007-07-07,+12356,13001,rdur@bar.null

source_domain,source_id,first_name,last_name,username,email

domain.nowhere.null,123,Hervé,Dupont,hdup,foo@bar.null

domain.nowhere.null,456,René,Durant,rdur,boo@far.null

source_domain,app_id,first_name,last_name,nickname,freemason

whatever.nowhere,123,Hervé,Dupont,Herv,0

whatever.nowhere,456,René,Durant,RR,1

app_id,first_name,last_name,email

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

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

from authentic2.models import UserExternalId, Attribute

def test_import_users_from_csv_rows(db):

    ou = OU.objects.create(name='Foo', slug='foo')

    ou = OU.objects.create(name='Bar', slug='bar')

    User = get_user_model()

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

        reader = csv.reader(csv_file)

        rows = list(reader)

    # a dictionary defining the column names of the organizational information:

    organization_configuration = {

        'ou_column_name': 'ou_slug',

        'fallback_ou': get_default_ou().id,

        'source_domain_column_name': 'source_domain',

        'source_id_column_name': 'source_id'

    }

    # a dictionary defining the column names of the user attributes:

    attributes_configuration = {

        'username': 'username',

        'first_name': 'first_name',

        'last_name': 'last_name',

        'email': 'email',

    }

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

    results = import_users_from_csv_rows(

        rows, organization_configuration, attributes_configuration,

        dry_run=False, erase=False)

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

    for i in range(2):

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

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

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

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

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

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

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

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

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

    assert not results['error']

    assert results['nb_created'] == 2

    assert results['nb_ignored'] == 0

    assert results['nb_erased'] == 0

    assert results['by_ou']['foo']['nb_created'] == 1

    assert 'nb_erased' not in results['by_ou']['foo']

    assert 'nb_ignored' not in results['by_ou']['foo']

    assert results['by_ou']['bar']['nb_created'] == 1

    assert 'nb_erased' not in results['by_ou']['bar']

    assert 'nb_ignored' not in results['by_ou']['bar']

    assert results['by_source']['domain.nowhere.null']['nb_created'] == 2

    assert 'nb_erased' not in results['by_source']['domain.nowhere.null']

    assert 'nb_ignored' not in results['by_source']['domain.nowhere.null']

def test_import_users_from_csv_rows_dry_run(db):

    ou = OU.objects.create(name='Foo', slug='foo')

    ou = OU.objects.create(name='Bar', slug='bar')

    User = get_user_model()

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

        reader = csv.reader(csv_file)

        rows = list(reader)

    # a dictionary defining the column names of the organizational information:

    organization_configuration = {

        'ou_column_name': 'ou_slug',

        'fallback_ou': get_default_ou().id,

        'source_domain_column_name': 'source_domain',

        'source_id_column_name': 'source_id'

    }

    # a dictionary defining the column names of the user attributes:

    attributes_configuration = {

        'username': 'username',

        'first_name': 'first_name',

        'last_name': 'last_name',

        'email': 'email',

    }

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

    results = import_users_from_csv_rows(

        rows, organization_configuration, attributes_configuration,

        dry_run=True, erase=False)

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

    assert not results['error']

    assert results['nb_created'] == 2

    assert results['nb_ignored'] == 0

    assert results['nb_erased'] == 0

    assert results['by_ou']['foo']['nb_created'] == 1

    assert 'nb_erased' not in results['by_ou']['foo']

    assert 'nb_ignored' not in results['by_ou']['foo']

    assert results['by_ou']['bar']['nb_created'] == 1

    assert 'nb_erased' not in results['by_ou']['bar']

    assert 'nb_ignored' not in results['by_ou']['bar']

    assert results['by_source']['domain.nowhere.null']['nb_created'] == 2

    assert 'nb_erased' not in results['by_source']['domain.nowhere.null']

    assert 'nb_ignored' not in results['by_source']['domain.nowhere.null']

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

    )

    User = get_user_model()

    # a dictionary defining the column names of the organizational information:

    organization_configuration = {

        'fallback_ou': get_default_ou().id,

        'source_domain_column_name': 'source_domain',

        'source_id_column_name': 'app_id'

    }

    # a dictionary defining the column names of the user attributes:

    attributes_configuration = {

        'username': 'username',

        'first_name': 'first_name',

        'last_name': 'last_name',

        'email': 'email',

        'nickname': 'nickname',

        'is_a_spy': 'is_a_spy',

        'enrolled_on': 'enrolled_on',

        'secret_phone_number': 'secret_phone_number',

        'headquarters_postcode': 'headquarters_postcode',

    }

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

        reader = csv.reader(csv_file)

        rows = list(reader)

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

    results = import_users_from_csv_rows(

        rows, organization_configuration, attributes_configuration,

        dry_run=False, erase=False)

    assert not results['error']

    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][4].decode('utf-8')

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

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

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

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

def test_import_users_from_csv_ignore(db):

    ou = get_default_ou()

    User = get_user_model()

    # a dictionary defining the column names of the organizational information:

    organization_configuration = {

        'fallback_ou': get_default_ou().id,

        'source_domain_column_name': 'source_domain',

        'source_id_column_name': 'source_id'

    }

    # a dictionary defining the column names of the user attributes:

    attributes_configuration = {

        'username': 'username',

        'first_name': 'first_name',

        'last_name': 'last_name',

        'email': 'email',

    }

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

        reader = csv.reader(csv_file)

        rows = list(reader)

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

    results = import_users_from_csv_rows(

        rows, organization_configuration, attributes_configuration,

        dry_run=False, erase=False)

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

    assert not results['error']

    assert results['nb_created'] == 2

    assert results['by_ou']['default']['nb_created'] == 2

    assert results['by_source']['domain.nowhere.null']['nb_created'] == 2

    results = import_users_from_csv_rows(

        rows, organization_configuration, attributes_configuration,

        dry_run=False, erase=False)

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

    assert not results['error']

    assert results['nb_ignored'] == 2

    assert results['by_ou']['default']['nb_ignored'] == 2

    assert results['by_source']['domain.nowhere.null']['nb_ignored'] == 2

    results = import_users_from_csv_rows(

        rows, organization_configuration, attributes_configuration,

        dry_run=False, erase=False)

def test_import_users_from_csv_erase(db):

    ou = get_default_ou()

    User = get_user_model()

    # a dictionary defining the column names of the organizational information:

    organization_configuration = {

        'fallback_ou': get_default_ou().id,

        'source_domain_column_name': 'source_domain',

        'source_id_column_name': 'source_id'

    }

    # a dictionary defining the column names of the user attributes:

    attributes_configuration = {

        'username': 'username',

        'first_name': 'first_name',

        'last_name': 'last_name',

        'email': 'email',

    }

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

        reader = csv.reader(csv_file)

        rows = list(reader)

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

    results = import_users_from_csv_rows(

        rows, organization_configuration, attributes_configuration,

        dry_run=False, erase=False)

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

    assert not results['error']

    assert results['nb_created'] == 2

    assert results['by_ou']['default']['nb_created'] == 2

    assert results['by_source']['domain.nowhere.null']['nb_created'] == 2

    results = import_users_from_csv_rows(

        rows, organization_configuration, attributes_configuration,

        dry_run=False, erase=True)

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

    assert not results['error']

    assert results['nb_erased'] == 2

    assert results['by_ou']['default']['nb_erased'] == 2

    assert results['by_source']['domain.nowhere.null']['nb_erased'] == 2

    results = import_users_from_csv_rows(

        rows, organization_configuration, attributes_configuration,

        dry_run=False, erase=False)

from django import VERSION as django_version

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/csv-import/upload-file/'

    response = app.get(url)

    # page 1: upload file

    assert 'csv_file' in response.text

    form = response.form

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

    response = form.submit().follow()

    # page 2: organizational columns configuration

    assert 'ou_column_name' in response.text

    assert 'fallback_ou' in response.text

    assert 'source_domain_column_name' in response.text

    assert 'source_id_column_name' in response.text

    form = response.form

    form.set('source_domain_column_name', 'source_domain')

    form.set('source_id_column_name', 'app_id')

    response = form.submit().follow()

    # page 3: profile attribute column configuration

    # # standard profile attributes

    assert 'first_name' in response.text

    assert 'last_name' in response.text

    assert 'username' in response.text

    assert 'email' in response.text

    assert 'email_verified' in response.text

    assert 'is_staff' in response.text

    assert 'is_active' in response.text

    # # extended profile attributes

    assert 'nickname' in response.text

    assert 'freemason' in response.text

    form = response.form

    for directly_mapped_element in [

            'first_name', 'last_name', 'nickname', 'freemason']:

        form.set(directly_mapped_element, directly_mapped_element)

    response = form.submit().follow()

    # page 4: simulated import results

    assert '2 user(s) would be created.' in response.text

    assert 'Global results' in response.text

    assert 'Results by organizational unit' in response.text

    assert 'Results by source' in response.text

    response = response.click(description='Perform the import')

    # page5: import results

    assert '2 user(s) have been created.' in response.text

    assert 'Global results' in response.text

    assert 'Results by organizational unit' in response.text

    assert 'Results by source' in response.text

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