Produits Entr'ouvert » Authentic 2

  <a href="{% url 'a2-manager-users-imports' %}">{% trans 'Imports' %}</a>

      <form method="post" id="action-form">

          <button name="modify">{% trans "Modify" %}</button>

      <form action="download/{{ user_import.filename }}" id="download-form">

        <tr data-uuid="{{ report.uuid }}">

          <td class="created">{% if report.state != 'running' %}<a href="{% url "a2-manager-users-import-report" import_uuid=user_import.uuid report_uuid=report.uuid %}">{{ report.created }}</a>{% else %}{{ report.created }}{% endif %}</td>

          <td class="state"><span>{{ report.state }}</span> {% if report.state == 'error' %}"{{ report.exception }}"{% endif %}</td>

          <td class="applied">{% if not report.simulate %}<span class="icon-check"></span>{% endif %}</td>

          <td class="delete-action">{% if report.simulate %}<form method="post" id="delete-form-{{ report.uuid }}">{% csrf_token %}<button name="delete" value="{{ report.uuid }}">{% trans "Delete" %}</button></form>{% endif %}</td>

  <a href="{% url 'a2-manager-users-imports' %}">{% trans 'Imports' %}</a>

   {% if extra_actions %}

       {% for extra_action in extra_actions %}

         <li><a href="{{ extra_action.url }}">{{ extra_action.label }}</a></li>

       {% endfor %}

   {% endif %}

        t.daemon = True

                    PermissionMixin, MediaMixin)

        extra_actions = ctx['extra_actions'] = []

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

            extra_actions.append({

                'url': reverse('a2-manager-users-imports'),

                'label': _('Import users'),

            })

class UserImportsView(MediaMixin, PermissionMixin, FormView):

    permissions = ['custom_user.admin_user']

    permissions_global = True

class UserImportView(MediaMixin, PermissionMixin, FormView):

    permissions_global = True

class UserImportReportView(MediaMixin, PermissionMixin, TemplateView):

    permissions_global = True

    permissions_global = False

            if self.permissions:

                if self.permissions_global and not request.user.has_perms(self.permissions):

                    raise PermissionDenied

                if not self.permissions_global and not request.user.has_perm_any(self.permissions):

                    raise PermissionDenied

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

import time

from webtest import Upload

from authentic2.manager import user_import

def test_user_import(transactional_db, app, admin, ou1, admin_ou1, media):

    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('utf-8'),

                          'application/octet-stream'))

    response.form.set('encoding', 'utf-8')

    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='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('User Import')

        if new_resp.pyquery('tr[data-uuid="%s"] td.state span' % uuid).text() == 'finished':

            return new_resp

    simulate = reports[0]

    assert simulate.simulate

    response = assert_timeout(20, wait_finished)

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

    assert len(response.pyquery('table.main tbody tr')) == 3

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

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

    assert User.objects.count() == user_count

    response = response.click('User 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(20, 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)