Produits Entr'ouvert » Corbo

# 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 .api_views import NewslettersView, SubscriptionsView

    def get_subscriptions(self, email, uuid=None):

        subscriptions = defaultdict(dict)

            cat_id = s.category.pk

            subscriptions[cat_id]['id'] = str(cat_id)

            subscriptions[cat_id]['text'] = s.category.name

        if feed_response.ok:

            content = feedparser.parse(feed_response.content)

            for entry in content.get('entries', []):

                published = datetime.fromtimestamp(mktime(entry.published_parsed))

                html_tree = etree.HTML(entry['summary'])

                storage = DefaultStorage()

                    img.attrib['src'] = storage.url(new_image_name)

                announce, created = Announce.objects.get_or_create(identifier=entry['id'],

                announce.title = entry['title']

                announce.text = etree.tostring(html_tree)

                announce.publication_time = published

class Announce(models.Model):

    title = models.CharField(_('title'), max_length=256,

                             help_text=_('maximum 256 characters'))

    identifier = models.CharField(max_length=256, null=True, blank=True)

    mtime = models.DateTimeField(_('modification time'), auto_now=True)

    def __unicode__(self):

    def is_expired(self):

        if self.expiration_time:

        handler = HTML2Text()

        template = loader.get_template('corbo/announce.html')

        message = Message(subject=self.announce.title, mail_from=settings.CORBO_DEFAULT_FROM_EMAIL,

        html_tree = etree.HTML(self.announce.text)

        storage = DefaultStorage()

        for img in html_tree.xpath('//img/@src'):

                continue

            unsubscribe_token = signing.dumps({'category': self.announce.category.pk,

                                               'identifier': s.identifier})

            message.html = message.html.replace(UNSUBSCRIBE_LINK_PLACEHOLDER, unsubscribe_link)

            handler.body_width = 0

            message.text = handler.handle(message.html)

    category = models.ForeignKey('Category', verbose_name=_('Category'))

    uuid = models.CharField(_('User identifier'), max_length=128, blank=True)

    identifier = models.CharField(_('identifier'), max_length=128, blank=True,

    def get_identifier_display(self):

        try:

from django.core import signing

from django.core.urlresolvers import reverse

from django.views.generic import CreateView, UpdateView, DeleteView, \

from django.contrib.syndication.views import Feed

from django.shortcuts import resolve_url

from django.utils.encoding import force_text

from django.utils.feedgenerator import Atom1Feed as DjangoAtom1Feed

from django.http import HttpResponseRedirect, HttpResponse, Http404

from django.contrib.auth import logout as auth_logout

from django.contrib.auth import views as auth_views

except ImportError:

    get_idps = lambda: []

def login(request, *args, **kwargs):

    if any(get_idps()):

            return HttpResponseRedirect(resolve_url('mellon_login'))

    return auth_views.login(request, *args, **kwargs)

def logout(request, next_page=None):

    if any(get_idps()):

        return HttpResponseRedirect(resolve_url('mellon_logout'))

class HomepageView(RedirectView):

    pattern_name = 'manage'

homepage = HomepageView.as_view()

        context['category'] = kwargs['form'].initial['category']

        return context

add_announce = AnnounceCreateView.as_view()

    def get_success_url(self):

        return reverse('view_category', kwargs={'pk': self.object.category.pk})

edit_announce = AnnounceEditView.as_view()

class AnnounceDeleteView(DeleteView):

    model = models.Announce

    def get_success_url(self):

        return reverse('view_category', kwargs={'pk': self.object.category.pk})

delete_announce = AnnounceDeleteView.as_view()

class CategoryCreateView(CreateView):

    form_class = CategoryForm

    template_name = 'corbo/category_form.html'

        form.save()

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

add_category = CategoryCreateView.as_view()

    def get_success_url(self):

        return reverse('view_category', kwargs={'pk': self.object.pk})

edit_category = CategoryEditView.as_view()

        context['announces'] = self.object.announce_set.all()

        return context

view_category = CategoryView.as_view()

    def get_success_url(self):

        return reverse('manage')

delete_category = CategoryDeleteView.as_view()

        data = signing.loads(self.kwargs['unsubscription_token'])

        try:

            return models.Subscription.objects.get(category__pk=data['category'],

        except models.Subscription.DoesNotExist:

            raise Http404

    def get_success_url(self):

        return reverse('unsubscription_done')

unsubscribe = UnsubscribeView.as_view()

class UnsubscriptionDoneView(TemplateView):

unsubscription_done = UnsubscriptionDoneView.as_view()

    template_name = 'corbo/manage.html'

    model = models.Category

manage = ManageView.as_view()

class Atom1Feed(DjangoAtom1Feed):

    def root_attributes(self):

        attrs = super(Atom1Feed, self).root_attributes()

        attrs.update({'xml:base': self.feed['link']})

    def item_pubdate(self, item):

        return item.publication_time or item.mtime

atom = AtomView()

def menu_json(request):

    label = _('Announces')

    json_str = json.dumps([{'label': force_text(label),

    for variable in ('jsonpCallback', 'callback'):

        if variable in request.GET:

            response = HttpResponse(content_type='application/javascript')

import pytest

import django_webtest

@pytest.fixture

def app(request):

    wtm = django_webtest.WebTestMixin()

</feed>

"""

def mocked_request_get(*args, **kwargs):

    storage = DefaultStorage()

    class MockResponse:

        def __init__(self, content):

import pytest

from uuid import uuid4

        categories.append(c)

    return categories

@pytest.fixture

def announces():

    announces = []

        announces.append(a)

    return announces

@pytest.fixture

def user():

    User = get_user_model()

    user.set_password('password')

    user.save()

    return user

    for identifier, name in channel_choices[:1]:

        for category in categories:

            uri = '%s:%s' % (identifier, foo)

            resp = app.get(reverse('subscriptions'), {'email': foo})

            assert 'data' in resp.json

            data = resp.json['data']

            subscriptions = [{'id': category_id,

                              'text': category.name,

                              'transports': transports}]

            if resp.json['data']:

                resp = app.get(subscriptions_url)

                for cat in resp.json['data']:

                    if cat['id'] == category_id:

                        sub_transports = [c['id'] for c in cat['transports']]

import pytest

from uuid import uuid4

import os

import re

import urllib

from django.core.urlresolvers import reverse

from django.core import mail, signing

from django.utils import timezone

from django.core.files.storage import DefaultStorage

pytestmark = pytest.mark.django_db

        categories.append(c)

    return categories

@pytest.fixture

def announces():

    announces = []

        announces.append(a)

    return announces

def test_emailing_with_no_subscriptions(app, categories, announces):

    for announce in announces:

        broadcast = Broadcast.objects.get(announce=announce)

        assert not broadcast.result

    assert not mail.outbox

def test_send_email(app, categories, announces):

    for category in categories:

        uuid = uuid4()

    for announce in announces:

        broadcast.send()

        assert broadcast.result

        assert mail.outbox

def test_check_inline_css(app, categories, announces):

    for announce in announces:

        announce.text = '<style type="text/css">h2 {color: #F00}</style>' + announce.text

        announce.save()

        uuid = uuid4()

        broadcast = Broadcast.objects.get(announce=announce)

        broadcast.send()

        assert broadcast.result

        assert 'h2 style="color:#F00"' in mail.outbox[0].html

        mail.outbox = []

def test_check_inline_images(app, categories, announces):

    storage = DefaultStorage()

    media_path = os.path.join(os.path.dirname(__file__), 'media')

        announce.text = announce.text + '<img src="%s" />' % img_src

        announce.save()

        uuid = uuid4()

        broadcast = Broadcast.objects.get(announce=announce)

        broadcast.send()

        assert broadcast.result

        mail.outbox = []

    storage.delete(image_name)

def test_unsubscription_link(app, categories, announces):

    unsubscription_link_sentinel = ''

    for category in categories:

        uuid = uuid4()

        scheme = 'mailto:'

        uri = scheme + '%s@example.net' % uuid

        for announce in announces:

            if announce.category != category:

                continue

            broadcast.send()

            assert broadcast.result

            assert mail.outbox

            signature = urllib.unquote(re.findall('/unsubscribe/(.*)"', mail.outbox[0].html)[0])

            unsubscription_link = reverse('unsubscribe', kwargs={'unsubscription_token': signature})

            assert signing.loads(signature) == {

            assert mail.outbox[0].subject == announce.title

            assert unsubscription_link in mail.outbox[0].html

            assert unsubscription_link in mail.outbox[0].text