226 lines
8.7 KiB
Python
226 lines
8.7 KiB
Python
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/'
|