add notification system (#13812)
This commit is contained in:
parent
8e948a6ab3
commit
963bc1faf7
|
|
@ -0,0 +1,29 @@
|
|||
# combo - content management system
|
||||
# Copyright (C) 2016 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/>.
|
||||
|
||||
import django.apps
|
||||
from django.utils.translation import ugettext_lazy as _
|
||||
|
||||
class AppConfig(django.apps.AppConfig):
|
||||
name = 'combo.apps.notifications'
|
||||
verbose_name = _('Notification')
|
||||
|
||||
def get_before_urls(self):
|
||||
from . import urls
|
||||
return urls.urlpatterns
|
||||
|
||||
|
||||
default_app_config = 'combo.apps.notifications.AppConfig'
|
||||
|
|
@ -0,0 +1,78 @@
|
|||
# combo - content management system
|
||||
# Copyright (C) 2016 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 rest_framework import serializers, permissions, status
|
||||
from rest_framework.generics import GenericAPIView
|
||||
from rest_framework.response import Response
|
||||
|
||||
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)
|
||||
body = serializers.CharField(required=False, allow_blank=False)
|
||||
url = serializers.URLField(required=False, allow_blank=True)
|
||||
start_timestamp = serializers.DateTimeField(required=False, allow_null=True)
|
||||
end_timestamp = serializers.DateTimeField(required=False, allow_null=True)
|
||||
duration = serializers.IntegerField(required=False, allow_null=True, min_value=0)
|
||||
|
||||
|
||||
class Add(GenericAPIView):
|
||||
permission_classes = (permissions.IsAuthenticated,)
|
||||
serializer_class = NotificationSerializer
|
||||
|
||||
def post(self, request, *args, **kwargs):
|
||||
serializer = self.get_serializer(data=request.data)
|
||||
if not serializer.is_valid():
|
||||
response = {'err': 1, 'err_desc': serializer.errors}
|
||||
return Response(response, status.HTTP_400_BAD_REQUEST)
|
||||
data = serializer.validated_data
|
||||
|
||||
notification_id = Notification.notify(
|
||||
user=request.user,
|
||||
summary=data['summary'],
|
||||
id=data.get('id'),
|
||||
body=data.get('body'),
|
||||
url=data.get('url'),
|
||||
start_timestamp=data.get('start_timestamp'),
|
||||
end_timestamp=data.get('end_timestamp'),
|
||||
duration=data.get('duration')
|
||||
)
|
||||
response = {'err': 0, 'data': {'id': notification_id}}
|
||||
return Response(response)
|
||||
|
||||
add = Add.as_view()
|
||||
|
||||
|
||||
class Ack(GenericAPIView):
|
||||
permission_classes = (permissions.IsAuthenticated,)
|
||||
|
||||
def get(self, request, notification_id):
|
||||
Notification.ack(request.user, notification_id)
|
||||
return Response({'err': 0})
|
||||
|
||||
ack = Ack.as_view()
|
||||
|
||||
|
||||
class Forget(GenericAPIView):
|
||||
permission_classes = (permissions.IsAuthenticated,)
|
||||
|
||||
def get(self, request, notification_id, *args, **kwargs):
|
||||
Notification.forget(request.user, notification_id)
|
||||
return Response({'err': 0})
|
||||
|
||||
forget = Forget.as_view()
|
||||
|
|
@ -0,0 +1,51 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from django.db import migrations, models
|
||||
from django.conf import settings
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('data', '0020_auto_20160928_1152'),
|
||||
('auth', '0006_require_contenttypes_0002'),
|
||||
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='Notification',
|
||||
fields=[
|
||||
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
|
||||
('summary', models.CharField(max_length=140, verbose_name='Label')),
|
||||
('body', models.TextField(default=b'', verbose_name='Body', blank=True)),
|
||||
('url', models.URLField(default=b'', verbose_name='URL', blank=True)),
|
||||
('start_timestamp', models.DateTimeField(verbose_name='Start date and time')),
|
||||
('end_timestamp', models.DateTimeField(verbose_name='End date and time')),
|
||||
('acked', models.BooleanField(default=False, verbose_name='Acked')),
|
||||
('external_id', models.SlugField(null=True, verbose_name='External identifier')),
|
||||
('user', models.ForeignKey(to=settings.AUTH_USER_MODEL)),
|
||||
],
|
||||
options={
|
||||
'verbose_name': 'Notification',
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='NotificationsCell',
|
||||
fields=[
|
||||
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
|
||||
('placeholder', models.CharField(max_length=20)),
|
||||
('order', models.PositiveIntegerField()),
|
||||
('slug', models.SlugField(verbose_name='Slug', blank=True)),
|
||||
('extra_css_class', models.CharField(max_length=100, verbose_name='Extra classes for CSS styling', blank=True)),
|
||||
('public', models.BooleanField(default=True, verbose_name='Public')),
|
||||
('restricted_to_unlogged', models.BooleanField(default=False, verbose_name='Restrict to unlogged users')),
|
||||
('groups', models.ManyToManyField(to='auth.Group', verbose_name='Groups', blank=True)),
|
||||
('page', models.ForeignKey(to='data.Page')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': 'User Notifications',
|
||||
},
|
||||
),
|
||||
]
|
||||
|
|
@ -0,0 +1,141 @@
|
|||
# combo - content management system
|
||||
# Copyright (C) 2014-2016 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 django.conf import settings
|
||||
from django.db import models
|
||||
from django.utils.translation import ugettext_lazy as _
|
||||
from django.utils.timezone import now, localtime, timedelta
|
||||
from django.db.models import Q
|
||||
|
||||
from combo.data.models import CellBase
|
||||
from combo.data.library import register_cell_class
|
||||
from combo.utils import NothingInCacheException
|
||||
|
||||
|
||||
class NotificationManager(models.Manager):
|
||||
def filter_by_id(self, id):
|
||||
try:
|
||||
int(id)
|
||||
except (ValueError, TypeError):
|
||||
search_id = Q(external_id=id)
|
||||
else:
|
||||
search_id = Q(pk=id) | Q(external_id=id)
|
||||
return self.filter(search_id)
|
||||
|
||||
|
||||
class Notification(models.Model):
|
||||
objects = NotificationManager()
|
||||
|
||||
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
|
||||
summary = models.CharField(_('Label'), max_length=140)
|
||||
body = models.TextField(_('Body'), default='', blank=True)
|
||||
url = models.URLField(_('URL'), default='', blank=True)
|
||||
start_timestamp = models.DateTimeField(_('Start date and time'))
|
||||
end_timestamp = models.DateTimeField(_('End date and time'))
|
||||
acked = models.BooleanField(_('Acked'), default=False)
|
||||
external_id = models.SlugField(_('External identifier'), null=True)
|
||||
|
||||
class Meta:
|
||||
verbose_name = _('Notification')
|
||||
|
||||
def __unicode__(self):
|
||||
return self.summary
|
||||
|
||||
def public_id(self):
|
||||
return self.external_id or str(self.pk)
|
||||
|
||||
@classmethod
|
||||
def notify(cls, user, summary, id=None, body=None, url=None,
|
||||
start_timestamp=None, duration=None, end_timestamp=None):
|
||||
'''
|
||||
Create a new notification:
|
||||
Notification.notify(user, 'summary') -> id
|
||||
Create a notification with a duration of one day:
|
||||
Notification.notify(user, 'summary', duration=3600*24)
|
||||
Renew an existing notification, or create a new one, with an external_id:
|
||||
Notification.notify(user, 'summary', id='id')
|
||||
'''
|
||||
# get ...
|
||||
notification = Notification.objects.filter_by_id(id).filter(user=user).first() if id else None
|
||||
if not notification: # ... or create
|
||||
notification = Notification(user=user, summary=summary,
|
||||
body=body or '', url=url or '',
|
||||
external_id=id)
|
||||
notification.summary = summary
|
||||
notification.body = body or ''
|
||||
notification.url = url or ''
|
||||
notification.start_timestamp = start_timestamp or now()
|
||||
if duration:
|
||||
if not isinstance(duration, timedelta):
|
||||
duration = timedelta(seconds=duration)
|
||||
notification.end_timestamp = notification.start_timestamp + duration
|
||||
else:
|
||||
notification.end_timestamp = end_timestamp or notification.start_timestamp + timedelta(3)
|
||||
|
||||
notification.save()
|
||||
|
||||
if notification.external_id is None:
|
||||
return '%s' % notification.pk
|
||||
else:
|
||||
return notification.external_id
|
||||
|
||||
@classmethod
|
||||
def ack(cls, user, id):
|
||||
Notification.objects.filter_by_id(id).filter(user=user).update(acked=True)
|
||||
|
||||
@classmethod
|
||||
def forget(cls, user, id):
|
||||
past = now() - timedelta(seconds=5)
|
||||
Notification.objects.filter_by_id(id).filter(user=user).update(end_timestamp=past,
|
||||
acked=True)
|
||||
|
||||
|
||||
@register_cell_class
|
||||
class NotificationsCell(CellBase):
|
||||
user_dependant = True
|
||||
|
||||
class Meta:
|
||||
verbose_name = _('User Notifications')
|
||||
|
||||
def is_relevant(self, context):
|
||||
if not (getattr(context['request'], 'user', None) and context['request'].user.is_authenticated()):
|
||||
return False
|
||||
return True
|
||||
|
||||
def get_cell_extra_context(self, context):
|
||||
extra_context = super(NotificationsCell, self).get_cell_extra_context(context)
|
||||
user = getattr(context.get('request'), 'user', None)
|
||||
if user and user.is_authenticated():
|
||||
extra_context['notifications'] = Notification.objects.filter(user=user,
|
||||
start_timestamp__lte=now(), end_timestamp__gt=now()).order_by('-start_timestamp')
|
||||
return extra_context
|
||||
|
||||
def get_badge(self, context):
|
||||
user = getattr(context.get('request'), 'user', None)
|
||||
if not user or not user.is_authenticated():
|
||||
return
|
||||
notifs = Notification.objects.filter(user=user, start_timestamp__lte=now(),
|
||||
end_timestamp__gt=now())
|
||||
new = notifs.filter(acked=False).count()
|
||||
if not new:
|
||||
return
|
||||
return {'badge': '%s/%s' % (new, notifs.count())}
|
||||
|
||||
def render(self, context):
|
||||
self.context = context
|
||||
if not context.get('synchronous'):
|
||||
raise NothingInCacheException()
|
||||
return super(NotificationsCell, self).render(context)
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
{% load i18n %}
|
||||
<h2>{% trans "Notifications" %}</h2>
|
||||
{% if notifications %}
|
||||
<ul>
|
||||
{% for notification in notifications %}
|
||||
<li class="combo-notification {% if notification.acked %}combo-notification-acked{% endif %}"
|
||||
data-combo-notification-id="{{ notification.public_id }}">
|
||||
<a href="{{ notification.url|default:"#" }}">{{ notification.summary }}</a>
|
||||
{% if notification.body %}
|
||||
<div class="description">
|
||||
{{ notification.body|linebreaks }}
|
||||
</div>
|
||||
{% endif %}
|
||||
</li>
|
||||
{% endfor %}
|
||||
</ul>
|
||||
{% else %}
|
||||
<p>{% trans 'No notifications.' %}</p>
|
||||
{% endif %}
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
# combo - content management system
|
||||
# Copyright (C) 2016 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 django.conf.urls import patterns, url
|
||||
|
||||
from .api_views import add, ack, forget
|
||||
|
||||
urlpatterns = patterns('',
|
||||
url('^api/notification/add/$', add,
|
||||
name='api-notification-add'),
|
||||
url('^api/notification/ack/(?P<notification_id>[\w-]+)/$', ack,
|
||||
name='api-notification-ack'),
|
||||
url('^api/notification/forget/(?P<notification_id>[\w-]+)/$', forget,
|
||||
name='api-notification-forget'),
|
||||
)
|
||||
|
|
@ -55,6 +55,7 @@ INSTALLED_APPS = (
|
|||
'django.contrib.sessions',
|
||||
'django.contrib.messages',
|
||||
'django.contrib.staticfiles',
|
||||
'rest_framework',
|
||||
'ckeditor',
|
||||
'gadjo',
|
||||
'cmsplugin_blurp',
|
||||
|
|
@ -69,6 +70,7 @@ INSTALLED_APPS = (
|
|||
'combo.apps.momo',
|
||||
'combo.apps.newsletters',
|
||||
'combo.apps.fargo',
|
||||
'combo.apps.notifications',
|
||||
'combo.apps.usersearch',
|
||||
'xstatic.pkg.chartnew_js',
|
||||
)
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ Package: python-combo
|
|||
Architecture: all
|
||||
Depends: ${misc:Depends}, ${python:Depends},
|
||||
python-django (>= 1.7),
|
||||
python-djangorestframework (>= 3.3),
|
||||
python-gadjo,
|
||||
python-requests,
|
||||
python-feedparser,
|
||||
|
|
|
|||
|
|
@ -8,3 +8,4 @@ requests
|
|||
XStatic-ChartNew.js
|
||||
eopayment>=1.13
|
||||
python-dateutil
|
||||
djangorestframework>=3.3
|
||||
|
|
|
|||
1
setup.py
1
setup.py
|
|
@ -113,6 +113,7 @@ setup(
|
|||
'XStatic-ChartNew.js',
|
||||
'eopayment>=1.13',
|
||||
'python-dateutil',
|
||||
'djangorestframework>=3.3',
|
||||
],
|
||||
zip_safe=False,
|
||||
cmdclass={
|
||||
|
|
|
|||
|
|
@ -0,0 +1,225 @@
|
|||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from django.contrib.auth.models import User
|
||||
from django.test.client import RequestFactory
|
||||
from django.template import Context
|
||||
from django.utils.timezone import timedelta, now
|
||||
from django.core.urlresolvers import reverse
|
||||
|
||||
from django.test import Client
|
||||
|
||||
from combo.data.models import Page
|
||||
from combo.apps.notifications.models import Notification, NotificationsCell
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
client = Client()
|
||||
|
||||
@pytest.fixture
|
||||
def user():
|
||||
try:
|
||||
admin = User.objects.get(username='admin')
|
||||
except User.DoesNotExist:
|
||||
admin = User.objects.create_user('admin', email=None, password='admin')
|
||||
admin.email = 'admin@example.net'
|
||||
admin.save()
|
||||
return admin
|
||||
|
||||
@pytest.fixture
|
||||
def user2():
|
||||
try:
|
||||
admin2 = User.objects.get(username='admin2')
|
||||
except User.DoesNotExist:
|
||||
admin2 = User.objects.create_user('admin2', email=None, password='admin2')
|
||||
return admin2
|
||||
|
||||
def login(username='admin', password='admin'):
|
||||
resp = client.post('/login/', {'username': username, 'password': password})
|
||||
assert resp.status_code == 302
|
||||
|
||||
|
||||
def test_notification_api(user, user2):
|
||||
id_notifoo = Notification.notify(user, 'notifoo')
|
||||
assert Notification.objects.all().count() == 1
|
||||
noti = Notification.objects.filter_by_id(id_notifoo).filter(user=user).first()
|
||||
assert noti.pk == int(id_notifoo)
|
||||
assert noti.summary == 'notifoo'
|
||||
assert noti.body == ''
|
||||
assert noti.url == ''
|
||||
assert noti.external_id is None
|
||||
assert noti.end_timestamp - noti.start_timestamp == timedelta(3)
|
||||
assert noti.acked is False
|
||||
Notification.ack(user, id_notifoo)
|
||||
noti = Notification.objects.filter_by_id(id_notifoo).filter(user=user).first()
|
||||
assert noti.acked is True
|
||||
|
||||
Notification.notify(user, 'notirefoo', id=id_notifoo)
|
||||
assert Notification.objects.all().count() == 1
|
||||
noti = Notification.objects.filter_by_id(id_notifoo).filter(user=user).first()
|
||||
assert noti.pk == int(id_notifoo)
|
||||
assert noti.summary == 'notirefoo'
|
||||
|
||||
Notification.notify(user, 'notirefoo', id=id_notifoo, duration=3600)
|
||||
noti = Notification.objects.filter_by_id(id_notifoo).filter(user=user).first()
|
||||
assert noti.end_timestamp - noti.start_timestamp == timedelta(seconds=3600)
|
||||
|
||||
Notification.notify(user, 'notibar', id='notibar')
|
||||
assert Notification.objects.all().count() == 2
|
||||
Notification.notify(user, 'notirebar', id='notibar')
|
||||
assert Notification.objects.all().count() == 2
|
||||
|
||||
id2 = Notification.notify(user2, 'notiother')
|
||||
Notification.forget(user2, id2)
|
||||
noti = Notification.objects.filter_by_id(id2).filter(user=user2).first()
|
||||
assert noti.end_timestamp < now()
|
||||
assert noti.acked is True
|
||||
|
||||
def test_notification_cell(user, user2):
|
||||
page = Page(title='notif', slug='test_notification_cell', template_name='standard')
|
||||
page.save()
|
||||
cell = NotificationsCell(page=page, placeholder='content', order=0)
|
||||
|
||||
context = Context({'request': RequestFactory().get('/')})
|
||||
context['synchronous'] = True # to get fresh content
|
||||
|
||||
context['request'].user = None
|
||||
assert cell.is_relevant(context) is False
|
||||
context['request'].user = user
|
||||
assert cell.is_relevant(context) is True
|
||||
assert cell.get_badge(context) is None
|
||||
|
||||
id_noti1 = Notification.notify(user, 'notibar')
|
||||
id_noti2 = Notification.notify(user, 'notifoo')
|
||||
content = cell.render(context)
|
||||
assert 'notibar' in content
|
||||
assert 'notifoo' in content
|
||||
assert cell.get_badge(context) == {'badge': '2/2'}
|
||||
|
||||
Notification.forget(user, id_noti2)
|
||||
content = cell.render(context)
|
||||
assert 'notibar' in content
|
||||
assert 'notifoo' not in content
|
||||
assert cell.get_badge(context) == {'badge': '1/1'}
|
||||
|
||||
Notification.notify(user, 'notirebar', id=id_noti1)
|
||||
content = cell.render(context)
|
||||
assert 'notirebar' in content
|
||||
assert 'notibar' not in content
|
||||
|
||||
Notification.notify(user, 'notiurl', id=id_noti1, url='https://www.example.net/')
|
||||
content = cell.render(context)
|
||||
assert 'notiurl' in content
|
||||
assert 'https://www.example.net/' in content
|
||||
|
||||
ackme = Notification.notify(user, 'ackme')
|
||||
Notification.ack(user, id=ackme)
|
||||
content = cell.render(context)
|
||||
assert 'acked' in content
|
||||
assert cell.get_badge(context) == {'badge': '1/2'}
|
||||
|
||||
Notification.ack(user, id=id_noti1)
|
||||
content = cell.render(context)
|
||||
assert cell.get_badge(context) is None
|
||||
|
||||
Notification.notify(user2, 'notiother')
|
||||
content = cell.render(context)
|
||||
assert 'notiurl' in content
|
||||
assert 'notiother' not in content
|
||||
assert cell.get_badge(context) is None
|
||||
context['request'].user = user2
|
||||
content = cell.render(context)
|
||||
assert 'notiurl' not in content
|
||||
assert 'notiother' in content
|
||||
assert cell.get_badge(context) == {'badge': '1/1'}
|
||||
|
||||
def test_notification_ws(user):
|
||||
|
||||
def notify(data, check_id, count):
|
||||
resp = client.post(reverse('api-notification-add'), json.dumps(data),
|
||||
content_type='application/json')
|
||||
assert resp.status_code == 200
|
||||
result = json.loads(resp.content)
|
||||
assert result == {'data': {'id': check_id}, 'err': 0}
|
||||
assert Notification.objects.filter(user=user).count() == count
|
||||
return Notification.objects.filter_by_id(check_id).last()
|
||||
|
||||
login()
|
||||
notify({'summary': 'foo'}, '1', 1)
|
||||
notify({'summary': 'bar'}, '2', 2)
|
||||
notify({'summary': 'bar', 'id': 'noti3'}, 'noti3', 3)
|
||||
notif = {
|
||||
'summary': 'bar',
|
||||
'url': 'http://www.example.net',
|
||||
'body': 'foobar',
|
||||
'start_timestamp': '2016-11-11T11:11',
|
||||
'end_timestamp': '2016-12-12T12:12',
|
||||
}
|
||||
result = notify(notif, '4', 4)
|
||||
assert result.summary == notif['summary']
|
||||
assert result.url == notif['url']
|
||||
assert result.body == notif['body']
|
||||
assert result.start_timestamp.isoformat()[:19] == '2016-11-11T11:11:00'
|
||||
assert result.end_timestamp.isoformat()[:19] == '2016-12-12T12:12:00'
|
||||
|
||||
del notif['end_timestamp']
|
||||
notif['duration'] = 3600
|
||||
result = notify(notif, '5', 5)
|
||||
assert result.end_timestamp.isoformat()[:19] == '2016-11-11T12:11:00'
|
||||
|
||||
notif['duration'] = '3600'
|
||||
result = notify(notif, '6', 6)
|
||||
assert result.end_timestamp.isoformat()[:19] == '2016-11-11T12:11:00'
|
||||
|
||||
resp = client.get(reverse('api-notification-ack', kwargs={'notification_id': '6'}))
|
||||
assert resp.status_code == 200
|
||||
assert Notification.objects.filter(acked=True).count() == 1
|
||||
assert Notification.objects.filter(acked=True).first().public_id() == '6'
|
||||
|
||||
resp = client.get(reverse('api-notification-forget', kwargs={'notification_id': '5'}))
|
||||
assert resp.status_code == 200
|
||||
assert Notification.objects.filter(acked=True).count() == 2
|
||||
notif = Notification.objects.filter_by_id('5').filter(user=user).first()
|
||||
assert notif.public_id() == '5'
|
||||
assert notif.acked is True
|
||||
assert notif.end_timestamp < now()
|
||||
|
||||
def test_notification_ws_badrequest(user):
|
||||
|
||||
def check_error(data, message):
|
||||
resp = client.post(reverse('api-notification-add'),
|
||||
json.dumps(data) if data else None,
|
||||
content_type='application/json')
|
||||
assert resp.status_code == 400
|
||||
result = json.loads(resp.content)
|
||||
assert result['err'] == 1
|
||||
assert message in result['err_desc'].values()[0][0]
|
||||
|
||||
login()
|
||||
check_error(None, 'required')
|
||||
check_error('blahblah', 'Invalid data')
|
||||
check_error({'summary': ''}, 'may not be blank')
|
||||
check_error({'summary': 'x'*1000}, 'no more than 140 char')
|
||||
check_error({'summary': 'ok', 'url': 'xx'}, 'valid URL')
|
||||
check_error({'summary': 'ok', 'start_timestamp': 'foo'}, 'wrong format')
|
||||
check_error({'summary': 'ok', 'end_timestamp': 'bar'}, 'wrong format')
|
||||
check_error({'summary': 'ok', 'duration': 'xx'}, 'valid integer is required')
|
||||
check_error({'summary': 'ok', 'duration': 4.01}, 'valid integer is required')
|
||||
check_error({'summary': 'ok', 'duration': -4}, 'greater than')
|
||||
|
||||
def test_notification_ws_deny():
|
||||
assert client.post(reverse('api-notification-add'),
|
||||
json.dumps({'summary': 'ok'}),
|
||||
content_type='application/json').status_code == 403
|
||||
assert client.get(reverse('api-notification-ack',
|
||||
kwargs={'notification_id': '1'})).status_code == 403
|
||||
assert client.get(reverse('api-notification-forget',
|
||||
kwargs={'notification_id': '1'})).status_code == 403
|
||||
|
||||
def test_notification_ws_check_urls():
|
||||
assert reverse('api-notification-add') == '/api/notification/add/'
|
||||
assert reverse('api-notification-ack',
|
||||
kwargs={'notification_id': 'noti1'}) == '/api/notification/ack/noti1/'
|
||||
assert reverse('api-notification-forget',
|
||||
kwargs={'notification_id': 'noti1'}) == '/api/notification/forget/noti1/'
|
||||
Loading…
Reference in New Issue