Produits Entr'ouvert » Authentic 2

span.activity {

	background: url(indicator.gif) no-repeat top right;

	padding-right: 30px;

}

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

{% load i18n gadjo staticfiles %}

{% block page-title %}{{ block.super }} - {% trans "User export" %}{% endblock %}

{% block appbar %}

<h2>Users export</h2>

{% endblock %}

{% block breadcrumb %}

  {{ block.super }}

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

  <a href="{% url 'a2-manager-users-export-progress' uuid=uuid %}"></a>

{% endblock %}

{% block main %}

<div class="section">

  <div class="running">

    <p>{% trans "Preparing CSV export file..." %}</p>

    <span class="activity">{% trans "Progress:" %} <span id="progress">0</span>%</span>

  </div>

  <div class="done">

    <p>{% trans "Export completed." %}</p>

    <p><a class="button" href="{% url 'a2-manager-users-export-file' uuid=uuid %}">{% trans "Download CSV" %}</a></p>

  </div>

</div>

<script>

function updateStatus() {

  $('div.done').hide();

  $.get('{% url 'a2-manager-users-export-progress' uuid=uuid %}', null,

    function (text) {

      if(text != 100) {

        $('span#progress').text(text);

        window.setTimeout(updateStatus, 2500);

      } else {

        $('div.running').hide();

        $('div.done').show();

      }

    }

  );

}

$(document).ready(updateStatus);

</script>

{% endblock %}

        url(r'^users/export-progress/(?P<uuid>[a-z0-9-]+)/$',

            user_views.users_export_progress, name='a2-manager-users-export-progress'),

        url(r'^users/export/(?P<uuid>[a-z0-9-]+)/$',

            user_views.users_export_file, name='a2-manager-users-export-file'),

import os

import pickle

import uuid

from django.core.files.storage import default_storage

from authentic2.utils import batch_queryset

class UserExport(object):

    def __init__(self, uuid):

        self.uuid = uuid

        self.path = os.path.join(self.base_path(), self.uuid)

        self.export_path = os.path.join(self.path, 'export.csv')

        self.progress_path = os.path.join(self.path, 'progress')

    @classmethod

    def base_path(self):

        path = default_storage.path('user_exports')

        if not os.path.exists(path):

            os.makedirs(path)

        return path

    @property

    def exists(self):

        return os.path.exists(self.path)

    @classmethod

    def new(cls):

        export = cls(str(uuid.uuid4()))

        os.makedirs(export.path)

        return export

    @property

    def csv(self):

        return open(self.export_path, 'r')

    def set_export_content(self, content):

        with open(self.export_path, 'w') as f:

            f.write(content)

    @property

    def progress(self):

        progress = 0

        if os.path.exists(self.progress_path):

            with open(self.progress_path, 'r') as f:

                progress = f.read()

        return int(progress) if progress else 0

    def set_progress(self, progress):

        with open(self.progress_path, 'w') as f:

            f.write(str(progress))

def export_users_to_file(uuid, query):

    export = UserExport(uuid)

    qs = get_user_model().objects.all()

    qs.query = pickle.loads(query.encode('latin-1'))

    count = qs.count() or 1

    def callback(progress):

        export.set_progress(round(progress / count * 100))

    qs = batch_queryset(qs, progress_callback=callback)

    dataset = get_user_dataset(qs)

    if hasattr(dataset, 'csv'):

        # compatiblity for tablib < 0.11

        csv = dataset.csv

    else:

        csv = dataset.export('csv')

    export.set_export_content(csv)

    export.set_progress(100)

import pickle

import sys

from django.db import models, transaction, connection

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

from django.http import Http404, FileResponse, HttpResponseRedirect, HttpResponse

from .user_export import get_user_dataset, UserExport

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

        if 'uwsgi' in sys.modules:

            from authentic2.utils.spooler import export_users

            export = UserExport.new()

            query = pickle.dumps(self.get_table_data().query)

            # workaround django-uwsgi encoding issue https://github.com/unbit/django-uwsgi/issues/10

            query = query.decode('latin1').encode('utf-8')

            tenant = getattr(connection, 'tenant', None)

            export_users.spool(domain=getattr(tenant, 'domain_url', None), uuid=export.uuid, query=query)

            return redirect(request, 'a2-manager-users-export-progress', kwargs={'uuid': export.uuid})

        else:

            return super().get(request, *args, **kwargs)

class UsersExportFileView(ExportMixin, PermissionMixin, View):

    permissions = ['custom_user.view_user']

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

        self.export = UserExport(kwargs.get('uuid'))

        if not self.export.exists:

            raise Http404()

        return super().get(request, *args, format='csv', **kwargs)

    def get_dataset(self):

        return self.export

users_export_file = UsersExportFileView.as_view()

class UsersExportProgressView(MediaMixin, TemplateView):

    template_name = 'authentic2/manager/user_export.html'

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

        self.uuid = kwargs.get('uuid')

        export = UserExport(self.uuid)

        if not export.exists:

            raise Http404()

        if request.is_ajax():

            return HttpResponse(export.progress)

        return super().get(request, *args, **kwargs)

    def get_context_data(self, **kwargs):

        ctx = super().get_context_data(**kwargs)

        ctx['uuid'] = self.uuid

        return ctx

users_export_progress = UsersExportProgressView.as_view()

def batch_queryset(qs, size=1000, progress_callback=None):

        if progress_callback:

            progress_callback(i * size)

# authentic2 - versatile identity manager

# Copyright (C) 2010-2021 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 uwsgidecorators import spool

from django.db import connection

from authentic2.manager.user_export import export_users_to_file

def set_connection(domain):

    from hobo.multitenant.middleware import TenantMiddleware

    tenant = TenantMiddleware.get_tenant_by_hostname(domain)

    connection.set_tenant(tenant)

@spool

def export_users(args):

    if args.get('domain'):

        # multitenant installation

        set_connection(args['domain'])

    export_users_to_file(args['uuid'], args['query'])

import mock

import sys

@pytest.fixture

def mock_uwsgi():

    from authentic2.manager.user_export import export_users_to_file

    def run_now(f):

        def spool(**kwargs):

            for k, v in kwargs.items():

                # workaround django-uwsgi encoding issue https://github.com/unbit/django-uwsgi/issues/10

                if isinstance(v, bytes):

                    kwargs[k] = v.decode('utf-8')

            f(kwargs)

        f.spool = spool

        return f

    sys.modules['uwsgi'] = mock.MagicMock()

    uwsgidecorators = mock.MagicMock()

    uwsgidecorators.spool = run_now

    with mock.patch.dict('sys.modules', uwsgidecorators=uwsgidecorators):

        yield

    del sys.modules['uwsgi']

def test_export_csv_async(settings, app, superuser, mock_uwsgi):

    users = [User(username='user%s' % i) for i in range(10)]

    User.objects.bulk_create(users)

    resp = login(app, superuser, reverse('a2-manager-users'))

    resp = resp.click('CSV')

    url = resp.url

    resp = resp.follow()

    assert 'Preparing CSV export file...' in resp.text

    assert '<span id="progress">0</span>' in resp.text

    # spooler mock is in fact synchronous, csv file is already there

    resp = resp.click('Download CSV')

    table = list(csv.reader(resp.text.splitlines()))

    assert len(table) == User.objects.count() + 1

    # ajax call returns 100% progress

    resp = app.get(url, xhr=True)

    assert resp.text == '100'

    resp = app.get('/manage/users/?search-text=user1')

    resp = resp.click('CSV').follow()

    resp = resp.click('Download CSV')

    table = list(csv.reader(resp.text.splitlines()))

    assert len(table) == 3  # user1 and superuser match