Produits Entr'ouvert » Corbo

import csv

from django.core import validators

from django.core.exceptions import ValidationError

class SubscriptionsImportForm(forms.Form):

    subscribers = forms.FileField(_('Subscribers'),

                    help_text=_('utf-8 encoded, comma separated file with email addresses on first column'))

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

        subscribers = []

        reader = csv.reader(self.cleaned_data['subscribers'])

        for idx, row in enumerate(reader, 1):

            if not row or not row[0]:

                continue

            try:

                validators.validate_email(row[0])

            except ValidationError:

                raise ValidationError(_('Invalid email address at line %d' % idx))

            subscribers.append(row[0])

        return subscribers

    subscriptions_import, menu_json

            url(r'^category/(?P<slug>[\w-]+)/import-subscriptions/$', subscriptions_import,

                name='subscriptions-import'),

<a class="extra-actions-menu-opener">☰</a>

<ul class="extra-actions-menu">

  <li><a href="{% url 'subscriptions-import' slug=category.slug %}" rel="popup">{% trans 'Import subscriptions' %}</a></li>

</ul>

{% extends "corbo/category_detail.html" %}

{% load i18n static %}

{% block breadcrumb %}

{{ block.super }}

<a href='{% url "view_category" slug=category.slug %}'>{{ category }}</a>

{% endblock %}

{% block appbar %}

<h2>{% trans "Import subscriptions" %}</h2>

{% endblock %}

{% block content %}

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

  {% csrf_token %}

  {{ form.as_p }}

<div class="buttons">

  <button>{% trans "Save" %}</button>

  <a href="{% url "view_category" slug=category.slug %}" class="cancel">{% trans "Cancel" %}</a>

</div>

</form>

{% endblock %}

    ListView, TemplateView, RedirectView, DetailView, FormView

from django.contrib import messages

from django.utils.translation import ngettext

from .forms import AnnounceForm, CategoryForm, SubscriptionsImportForm

class SubscriptionsImportView(FormView):

    form_class = SubscriptionsImportForm

    template_name = 'corbo/subscriptions_import_form.html'

    def get_context_data(self, **kwargs):

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

        context['category'] = models.Category.objects.get(slug=self.kwargs['slug'])

        return context

    def get_success_url(self):

        category = models.Category.objects.get(slug=self.kwargs['slug'])

        return reverse('view_category', kwargs={'slug': category.slug})

    def form_valid(self, form):

        new = 0

        c = models.Category.objects.get(slug=self.kwargs['slug'])

        for email in form.cleaned_data['subscribers']:

            obj, created = models.Subscription.objects.get_or_create(category=c, identifier='mailto:%s' % email)

            if created:

               new += 1

        messages.info(self.request, ngettext('%(new)d subscriber added', '%(new)d subscribers added', new) % {'new': new})

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

subscriptions_import = SubscriptionsImportView.as_view()

import pytest

from webtest import TestApp, Upload

from django.core.wsgi import get_wsgi_application

from django.utils.text import slugify

from django.core.urlresolvers import reverse

from django.contrib.auth import get_user_model

from corbo.models import Category, Subscription

pytestmark = pytest.mark.django_db

CATEGORIES = (u'Alerts',)

CSV_CONTENT = """foo@example.net, Foo,

john.doe@example.net, John Doe,

bar@localhost, Bar,

"""

@pytest.fixture

def categories():

    categories = []

    for category in CATEGORIES:

        c, created = Category.objects.get_or_create(name=category, slug=slugify(category))

        categories.append(c)

    return categories

@pytest.fixture

def admin():

    User = get_user_model()

    admin = User.objects.create_superuser(username='admin',

                password='password', email='admin@example.net')

    return admin

def login(app, username='admin', password='password'):

    login_page = app.get('/login/')

    login_form = login_page.forms[0]

    login_form['username'] = username

    login_form['password'] = password

    resp = login_form.submit()

    assert resp.status_int == 302

    return app

def test_subscribe_from_csv(admin, categories):

    app = login(TestApp(get_wsgi_application()))

    for c in categories:

        page = app.get(reverse('subscriptions-import', kwargs={'slug': c.slug}))

        form = page.form

        form['subscribers'] = Upload('users.csv', CSV_CONTENT)

        res = form.submit()

        assert res.status_code == 302

        assert Subscription.objects.filter(category=c).count() == len(CSV_CONTENT.splitlines())

def test_subscribe_from_csv_with_empty_lines(admin, categories):

    app = login(TestApp(get_wsgi_application()))

    content = CSV_CONTENT + '\n\n\n'

    for c in categories:

        page = app.get(reverse('subscriptions-import', kwargs={'slug': c.slug}))

        form = page.form

        form['subscribers'] = Upload('users.csv', content)

        res = form.submit()

        assert res.status_code == 302

        assert Subscription.objects.filter(category=c).count() == len(CSV_CONTENT.splitlines())

def test_subscribe_with_invalid_email(admin, categories):

    app = login(TestApp(get_wsgi_application()))

    content = CSV_CONTENT + '\nwrong, Wrong user,'

    for category in categories:

        page = app.get(reverse('subscriptions-import', kwargs={'slug': category.slug}))

        form = page.form

        form['subscribers'] = Upload('users.csv', content)

        page = form.submit()

        assert page.status_code == 200

        page.mustcontain('Invalid email address at line')