From f4edcd7f869cff86bb11541674b2d98460ccaadc Mon Sep 17 00:00:00 2001 From: Benjamin Dauvergne Date: Fri, 16 Mar 2018 14:41:40 +0100 Subject: [PATCH 6/8] notifications: enforce namespacing on external ids (to be rebased) --- combo/apps/notifications/api_views.py | 34 +++++++++++++++++++--------------- combo/apps/notifications/models.py | 7 +++++++ tests/test_notification.py | 30 +++++++++++++++++++++++++++--- 3 files changed, 53 insertions(+), 18 deletions(-) diff --git a/combo/apps/notifications/api_views.py b/combo/apps/notifications/api_views.py index 640984b..b026af1 100644 --- a/combo/apps/notifications/api_views.py +++ b/combo/apps/notifications/api_views.py @@ -23,7 +23,7 @@ from .models import Notification class NotificationSerializer(serializers.Serializer): summary = serializers.CharField(required=True, allow_blank=False, max_length=140) - id = serializers.SlugField(required=False, allow_null=True) + id = serializers.CharField(required=False, allow_null=True) body = serializers.CharField(allow_blank=False, default='') url = serializers.URLField(allow_blank=True, default='') origin = serializers.CharField(allow_blank=True, default='') @@ -42,20 +42,24 @@ class Add(GenericAPIView): response = {'err': 1, 'err_desc': serializer.errors} return Response(response, status.HTTP_400_BAD_REQUEST) data = serializer.validated_data - - notification, created = Notification.notify( - user=request.user, - summary=data['summary'], - id=data.get('id'), - body=data.get('body'), - url=data.get('url'), - origin=data.get('origin'), - start_timestamp=data.get('start_timestamp'), - end_timestamp=data.get('end_timestamp'), - duration=data.get('duration') - ) - response = {'err': 0, 'data': {'id': notification.public_id, 'created': created}} - return Response(response) + try: + notification, created = Notification.notify( + user=request.user, + summary=data['summary'], + id=data.get('id'), + body=data.get('body'), + url=data.get('url'), + origin=data.get('origin'), + start_timestamp=data.get('start_timestamp'), + end_timestamp=data.get('end_timestamp'), + duration=data.get('duration') + ) + except ValueError as e: + response = {'err': 1, 'err_desc': {'id': [unicode(e)]}} + return Response(response, status.HTTP_400_BAD_REQUEST) + else: + response = {'err': 0, 'data': {'id': notification.public_id, 'created': created}} + return Response(response) add = Add.as_view() diff --git a/combo/apps/notifications/models.py b/combo/apps/notifications/models.py index 90ee11d..e4611e1 100644 --- a/combo/apps/notifications/models.py +++ b/combo/apps/notifications/models.py @@ -14,6 +14,8 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . +import re + from django.conf import settings from django.db import models from django.utils.translation import ugettext_lazy as _ @@ -56,6 +58,8 @@ class NotificationQuerySet(QuerySet): class Notification(models.Model): + ID_RE = r'^[\w-]+:[\w]+$' + objects = NotificationQuerySet.as_manager() user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE) @@ -68,6 +72,7 @@ class Notification(models.Model): acked = models.BooleanField(_('Acked'), default=False) external_id = models.SlugField(_('External identifier'), null=True) + class Meta: verbose_name = _('Notification') unique_together = ( @@ -125,6 +130,8 @@ class Notification(models.Model): id = unicode(id) except Exception as e: raise ValueError('id must be convertible to unicode', e) + if not re.match(cls.ID_RE, id): + raise ValueError('id must match regular expression %s' % cls.ID_RE) notification, created = Notification.objects.update_or_create( user=user, external_id=id, defaults=defaults) diff --git a/tests/test_notification.py b/tests/test_notification.py index 02f44ad..b14f41c 100644 --- a/tests/test_notification.py +++ b/tests/test_notification.py @@ -66,9 +66,9 @@ def test_notification_api(user, user2): noti = Notification.objects.get() assert noti.end_timestamp - noti.start_timestamp == timedelta(seconds=3600) - notification, created = Notification.notify(user, 'notibar', id='notibar') + notification, created = Notification.notify(user, 'notibar', id='ns:notibar') assert Notification.objects.count() == 2 - notification, created = Notification.notify(user, 'notirebar', id='notibar') + notification, created = Notification.notify(user, 'notirebar', id='ns:notibar') assert not created assert Notification.objects.count() == 2 @@ -153,7 +153,7 @@ def test_notification_ws(user): login() notify({'summary': 'foo'}, '1', 1) notify({'summary': 'bar'}, '2', 2) - notify({'summary': 'bar', 'id': 'noti3'}, 'noti3', 3) + notify({'summary': 'bar', 'id': 'ns:noti3'}, 'ns:noti3', 3) notif = { 'summary': 'bar', 'url': 'http://www.example.net', @@ -233,3 +233,27 @@ def test_notification_ws_check_urls(): kwargs={'notification_id': 'noti1'}) == '/api/notification/ack/noti1/' assert reverse('api-notification-forget', kwargs={'notification_id': 'noti1'}) == '/api/notification/forget/noti1/' + + +def test_notification_id_and_origin(user): + login() + + def notify(data): + resp = client.post(reverse('api-notification-add'), json.dumps(data), + content_type='application/json') + + return json.loads(resp.content) + + result = notify({'summary': 'foo', 'id': '1'}) + assert result['err'] == 1 + + notification, created = Notification.notify(user, 'foo') + + result = notify({'summary': 'foo', 'id': str(notification.id)}) + assert result['err'] == 0 + + result = notify({'summary': 'foo', 'id': 'foo'}) + assert result['err'] == 1 + + result = notify({'summary': 'foo', 'id': 'foo:foo', 'origin': 'bar'}) + assert result['err'] == 0 -- 2.14.2