Produits Entr'ouvert » Combo

import hmac

from django.conf import settings

class IsAuthenticatedOrSignedPermission(object):

    @classmethod

    def signature(self, url):

        return hmac.HMAC(settings.SECRET_KEY, url).hexdigest()

    @classmethod

    def sign_url(cls, url):

        signature = cls.signature(url)

        separator = '&' if '?' in url else '?'

        return '%s%ssignature=%s' % (url, separator, signature)

    def has_permission(self, request, view):

        if request.user and request.user.is_authenticated:

            if 'user_id' in view.kwargs and str(request.user.id) == view.kwargs['user_id']:

                return True

        full_path = request.build_absolute_uri()

        if not ('&signature=' in full_path or '?signature=' in full_path):

            return False

        payload, signature = full_path.rsplit('signature=', 1)

        return self.signature(payload[:-1]) == signature

class Get(GenericAPIView):

    permission_classes = (permissions.IsAuthenticated,)

    serializer_class = NotificationSerializer

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

        notifications = Notification.objects.visible(request.user)

        data = []

        response = {'err': 0, 'data': data}

        for notification in notifications:

            id = notification.public_id,

            data.append({

                'id': id,

                'acked': notification.acked,

                'summary': notification.summary,

                'body': notification.body,

                'url': notification.url,

                'origin': notification.origin,

                'end_timestamp': notification.end_timestamp,

                'ack_url': IsAuthenticatedOrSignedPermission.sign_url(

                    request.build_absolute_uri(notification.ack_url)),

                'forget_url': IsAuthenticatedOrSignedPermission.sign_url(

                    request.build_absolute_uri(notification.forget_url)),

            })

        return Response(response)

get = Get.as_view()

            return Response(response, status.HTTP_404_NOT_FOUND)

class CORSApi(object):

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

        response = super(CORSApi, self).options(request, *args, **kwargs)

        response['Access-Control-Request-Method'] = 'GET'

        response['Access-Control-Allow-Origin'] = '*'

        return response

class Ack(CORSApi, GenericAPIView):

    permission_classes = (IsAuthenticatedOrSignedPermission,)

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

        Notification.objects.find(user_id, notification_id).ack()

class Forget(CORSApi, GenericAPIView):

    permission_classes = (IsAuthenticatedOrSignedPermission,)

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

        Notification.objects.find(user_id, notification_id).forget()

from django.core.urlresolvers import reverse

    @property

    def ack_url(self):

        return reverse('api-notification-ack', kwargs={

            'user_id': self.user_id, 'notification_id': self.public_id})

    @property

    def forget_url(self):

        return reverse('api-notification-forget', kwargs={

            'user_id': self.user_id, 'notification_id': self.public_id})

      data-combo-notification-id="{{ notification.public_id }}"

      data-combo-notification-ack-url="{{ notification.ack_url }}"

      data-combo-notification-forget-url="{{ notification.forget_url }}">

from .api_views import add, get, ack, forget

    url('^api/notification/get/$', get,

        name='api-notification-get'),

    url(r'^api/notification/ack/(?P<user_id>\d{1,15})/(?P<notification_id>[\w-]+)/$', ack,

    url(r'^api/notification/forget/(?P<user_id>\d{1,15})/(?P<notification_id>[\w-]+)/$', forget,

    result1 = notify(notif, '5', 5)

    assert result1.end_timestamp.isoformat()[:19] == '2016-11-11T12:11:00'

    result2 = notify(notif, '6', 6)

    assert result2.end_timestamp.isoformat()[:19] == '2016-11-11T12:11:00'

    resp = client.get(result2.ack_url)

    resp = client.get(result1.forget_url)

                              kwargs={'user_id': 1, 'notification_id': '1'})).status_code == 403

                              kwargs={'user_id': 1, 'notification_id': '1'})).status_code == 403

                   kwargs={'user_id': '1', 'notification_id': 'noti1'}) == '/api/notification/ack/1/noti1/'

                   kwargs={'user_id': '2', 'notification_id': 'noti1'}) == '/api/notification/forget/2/noti1/'

def test_notification_ws_get(app, user):

    Notification.notify(user, 'foo')

    Notification.notify(user, 'bar')

    app.set_user(user)

    response = app.get(reverse('api-notification-get'))

    result = response.json

    assert result['err'] == 0

    assert len(result['data']) == 2

    app.set_user(None)

    app.session.flush()

    app.get(reverse('api-notification-get'), status=403)

    assert Notification.objects.visible(user).new().count() == 2

    for notif in result['data']:

        response = app.options(notif['ack_url'])

        assert response['Access-Control-Allow-Origin'] == '*'

        assert response['Access-Control-Request-Method'] == 'GET'

        assert app.get(notif['ack_url'])

        assert app.get(notif['ack_url'])

    assert Notification.objects.visible(user).new().count() == 0

    assert Notification.objects.visible(user).count() == 2

    for notif in result['data']:

        response = app.options(notif['forget_url'])

        assert response['Access-Control-Allow-Origin'] == '*'

        assert response['Access-Control-Request-Method'] == 'GET'

        assert app.get(notif['forget_url'])

        assert app.get(notif['forget_url'])

    assert Notification.objects.visible(user).count() == 0