392 lines
14 KiB
Python
392 lines
14 KiB
Python
import logging
|
|
import mock
|
|
import os
|
|
import pytest
|
|
|
|
from django.core.urlresolvers import reverse
|
|
from django.contrib.auth.models import User
|
|
from django.test import override_settings
|
|
|
|
from corbo.models import Broadcast, Subscription
|
|
from corbo.utils import format_phonenumber
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
@pytest.fixture
|
|
def admin_user():
|
|
try:
|
|
user = User.objects.get(username='admin')
|
|
except User.DoesNotExist:
|
|
user = User.objects.create_superuser('admin', email='admin@example.com', password='admin')
|
|
return user
|
|
|
|
def login(app, username='admin', password='admin'):
|
|
login_page = app.get(reverse('auth_login'))
|
|
login_form = login_page.forms[0]
|
|
login_form['username'] = username
|
|
login_form['password'] = password
|
|
resp = login_form.submit()
|
|
assert resp.status_int == 302
|
|
return app
|
|
|
|
def test_unlogged_access(app):
|
|
# connect while not being logged in
|
|
assert app.get('/', status=302).location.endswith(reverse('manage'))
|
|
assert app.get('/manage/', status=302).location.endswith(reverse('auth_login') + '?next=/manage/')
|
|
|
|
def test_access(app, admin_user):
|
|
app = login(app)
|
|
resp = app.get(reverse('manage'), status=200)
|
|
assert 'New category' in resp
|
|
|
|
def test_logout(app, admin_user):
|
|
app = login(app)
|
|
app.get(reverse('auth_logout'))
|
|
assert app.get('/', status=302).location.endswith(reverse('manage'))
|
|
|
|
def test_create_category(app, admin_user):
|
|
app = login(app)
|
|
resp = app.get(reverse('manage'))
|
|
assert 'New category' in resp
|
|
category_page = resp.click('New category')
|
|
category_form = category_page.forms[0]
|
|
category_form['name'] = 'Alerts'
|
|
resp = category_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('manage'))
|
|
resp = resp.follow()
|
|
assert 'Alerts' in resp
|
|
|
|
def test_edit_category(app, admin_user):
|
|
app = login(app)
|
|
resp = app.get(reverse('manage'))
|
|
assert 'New category' in resp
|
|
category_page = resp.click('New category')
|
|
category_form = category_page.forms[0]
|
|
category_form['name'] = 'Alerts'
|
|
resp = category_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('manage'))
|
|
resp = app.get(reverse('manage'))
|
|
assert 'Alerts' in resp
|
|
assert '0 announces' in resp
|
|
assert '0 subscriptions' in resp
|
|
resp = resp.click('Alerts')
|
|
assert 'Edit' in resp
|
|
edit_page = resp.click('Edit')
|
|
edit_form = edit_page.forms[0]
|
|
edit_form['name'] = 'New Alerts'
|
|
resp = edit_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('view_category', kwargs={'slug': 'alerts'}))
|
|
|
|
def test_delete_category(app, admin_user):
|
|
app = login(app)
|
|
resp = app.get(reverse('manage'))
|
|
assert 'New category' in resp
|
|
category_page = resp.click('New category')
|
|
category_form = category_page.forms[0]
|
|
category_form['name'] = 'Alerts'
|
|
resp = category_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('manage'))
|
|
resp = app.get(reverse('manage'))
|
|
assert 'Alerts' in resp
|
|
assert '0 announces' in resp
|
|
assert '0 subscriptions' in resp
|
|
resp = resp.click('Alerts')
|
|
assert 'Delete' in resp
|
|
delete_page = resp.click('Delete')
|
|
delete_form = delete_page.forms[0]
|
|
resp = delete_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('manage'))
|
|
|
|
def test_create_announce(app, admin_user):
|
|
app = login(app)
|
|
resp = app.get('/manage/')
|
|
assert 'New category' in resp
|
|
category_page = resp.click('New category')
|
|
category_form = category_page.forms[0]
|
|
category_form['name'] = 'Alerts'
|
|
resp = category_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('manage'))
|
|
resp = app.get(reverse('manage'))
|
|
assert 'Alerts' in resp
|
|
resp = resp.click('Alerts')
|
|
assert 'New announce' in resp
|
|
announce_page = resp.click('New announce')
|
|
announce_form = announce_page.forms[0]
|
|
announce_form['title'] = 'First announce'
|
|
announce_form['text'] = 'announce content'
|
|
resp = announce_form.submit()
|
|
assert resp.status_int == 302
|
|
category_url = reverse('view_category', kwargs={'slug': 'alerts'})
|
|
assert resp.location.endswith(category_url)
|
|
resp = resp.follow()
|
|
assert 'First announce' in resp
|
|
|
|
def test_edit_announce(app, admin_user):
|
|
app = login(app)
|
|
resp = app.get('/manage/')
|
|
assert 'New category' in resp
|
|
category_page = resp.click('New category')
|
|
category_form = category_page.forms[0]
|
|
category_form['name'] = 'Alerts'
|
|
resp = category_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('manage'))
|
|
resp = app.get(resp.location)
|
|
resp = resp.click('Alerts')
|
|
assert 'New announce' in resp
|
|
announce_page = resp.click('New announce')
|
|
announce_form = announce_page.forms[0]
|
|
announce_form['title'] = 'First announce'
|
|
announce_form['text'] = 'announce content'
|
|
resp = announce_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('view_category', kwargs={'slug': 'alerts'}))
|
|
resp = resp.follow()
|
|
assert 'First announce' in resp
|
|
announce_page = resp.click('First announce')
|
|
assert 'First announce' in announce_page
|
|
assert 'Edit' in announce_page
|
|
assert 'Send test email' in announce_page
|
|
assert 'Delete' in announce_page
|
|
announce_edit_page = announce_page.click('Edit')
|
|
edit_form = announce_edit_page.forms[0]
|
|
edit_form['publication_time'] = '2017-03-03 09:00:00'
|
|
edit_form['expiration_time'] = '2017-12-31 23:00:00'
|
|
resp = edit_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('view_announce', kwargs={'pk': 1}))
|
|
|
|
# simulate announce deliver
|
|
broadcast = Broadcast.objects.get(announce__pk=1)
|
|
broadcast.deliver_time = '2017-03-30 20:00:00'
|
|
broadcast.delivery_count = 1
|
|
broadcast.save()
|
|
|
|
resp = app.get(resp.location)
|
|
|
|
assert '<strong>Publication</strong> March 3, 2017, 9 a.m.' in resp
|
|
assert '<strong>Expiration</strong> Dec. 31, 2017, 11 p.m.' in resp
|
|
assert '<strong>Sent</strong> March 30, 2017, 8 p.m.' in resp
|
|
assert 'to 1 destination' in resp
|
|
|
|
broadcast.delivery_count = 2
|
|
broadcast.save()
|
|
|
|
resp = app.get('http://testserver/manage/announce/1/')
|
|
assert '<strong>Sent</strong> March 30, 2017, 8 p.m.' in resp
|
|
assert 'to 2 destinations' in resp
|
|
|
|
|
|
def test_delete_announce(app, admin_user):
|
|
app = login(app)
|
|
resp = app.get('/manage/')
|
|
assert 'New category' in resp
|
|
category_page = resp.click('New category')
|
|
category_form = category_page.forms[0]
|
|
category_form['name'] = 'Alerts'
|
|
resp = category_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('manage'))
|
|
resp = app.get(reverse('manage'))
|
|
resp = resp.click('Alerts')
|
|
assert 'New announce' in resp
|
|
announce_page = resp.click('New announce')
|
|
announce_form = announce_page.forms[0]
|
|
announce_form['title'] = 'First announce'
|
|
announce_form['text'] = 'announce content'
|
|
resp = announce_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('view_category', kwargs={'slug': 'alerts'}))
|
|
resp = resp.follow()
|
|
assert 'First announce' in resp
|
|
resp = resp.click('First announce')
|
|
assert 'Delete' in resp
|
|
assert 'Send test email' in resp
|
|
announce_delete_page = resp.click('Delete')
|
|
announce_delete_form = announce_delete_page.forms[0]
|
|
resp = announce_delete_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('view_category', kwargs={'slug': 'alerts'}))
|
|
|
|
def test_email_announce(app, admin_user):
|
|
app = login(app)
|
|
resp = app.get('/manage/')
|
|
assert 'New category' in resp
|
|
category_page = resp.click('New category')
|
|
category_form = category_page.forms[0]
|
|
category_form['name'] = 'Alerts'
|
|
resp = category_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('manage'))
|
|
resp = resp.follow()
|
|
resp = resp.click('Alerts')
|
|
assert 'New announce' in resp
|
|
announce_page = resp.click('New announce')
|
|
announce_form = announce_page.forms[0]
|
|
announce_form['title'] = 'First announce'
|
|
announce_form['text'] = 'announce content'
|
|
resp = announce_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('view_category', kwargs={'slug': 'alerts'}))
|
|
resp = resp.follow()
|
|
assert 'First announce' in resp
|
|
resp = resp.click('First announce')
|
|
assert 'Send test email' in resp
|
|
assert 'Send test SMS' not in resp
|
|
resp = resp.click('Send test email')
|
|
send_form = resp.forms[0]
|
|
assert send_form.method == 'post'
|
|
assert 'email' in send_form.fields
|
|
assert send_form.fields['email'][0].value == admin_user.email
|
|
assert 'Send' in resp
|
|
assert 'Cancel' in resp
|
|
resp = send_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('view_announce', kwargs={'pk': '1'}))
|
|
|
|
@mock.patch('corbo.utils.requests.post')
|
|
def test_sms_announce(mocked_post, app, admin_user, settings):
|
|
app = login(app)
|
|
resp = app.get('/manage/')
|
|
assert 'New category' in resp
|
|
category_page = resp.click('New category')
|
|
category_form = category_page.forms[0]
|
|
category_form['name'] = 'Alerts'
|
|
resp = category_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('manage'))
|
|
resp = resp.follow()
|
|
resp = resp.click('Alerts')
|
|
|
|
# create new announce
|
|
assert 'New announce' in resp
|
|
announce_page = resp.click('New announce')
|
|
announce_form = announce_page.forms[0]
|
|
announce_form['title'] = 'First announce'
|
|
announce_form['text'] = 'announce content'
|
|
resp = announce_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('view_category', kwargs={'slug': 'alerts'}))
|
|
resp = resp.follow()
|
|
|
|
# view announce
|
|
assert 'First announce' in resp
|
|
settings.SMS_GATEWAY_URL = 'http:/passerelle.com'
|
|
resp = resp.click('First announce')
|
|
assert 'Send test SMS' in resp
|
|
|
|
# open send sms form
|
|
resp = resp.click('Send test SMS')
|
|
send_form = resp.forms[0]
|
|
assert 'mobile' in send_form.fields
|
|
assert send_form.fields['mobile'][0].value == ''
|
|
# submit with no mobile
|
|
resp = send_form.submit()
|
|
assert resp.status_int == 200
|
|
|
|
# add mellon attribute to web session
|
|
session = app.session
|
|
session['mellon_session'] = {'mobile': ['00000000']}
|
|
session.save()
|
|
app.set_cookie(str(settings.SESSION_COOKIE_NAME), str(session.session_key))
|
|
resp = resp.click('First announce')
|
|
resp = resp.click('Send test SMS')
|
|
send_form = resp.forms[0]
|
|
assert send_form.fields['mobile'][0].value == '00000000'
|
|
|
|
form = resp.forms[0]
|
|
form['mobile'] = '0607080900'
|
|
# simulate response from passerelle
|
|
mocked_response = mock.Mock()
|
|
mocked_response.json.return_value = {'err': 0, 'data': True}
|
|
mocked_post.return_value = mocked_response
|
|
resp = form.submit()
|
|
assert resp.location.endswith(reverse('view_announce', kwargs={'pk': 1}))
|
|
resp = resp.follow()
|
|
# make sure the form informs about the success
|
|
assert 'SMS successfully sent' in resp
|
|
|
|
resp = resp.click('Send test SMS')
|
|
form = resp.forms[0]
|
|
form['mobile'] = '0607080900'
|
|
# simulate error from passerelle
|
|
mocked_response.json.return_value = {'err': 1, 'data': None, 'err_desc': 'Destination error'}
|
|
resp = form.submit()
|
|
resp = resp.follow()
|
|
assert 'Error occured while sending SMS' in resp
|
|
|
|
def test_sms_announce_with_invalid_gateway_url(app, admin_user, settings, caplog):
|
|
app = login(app)
|
|
resp = app.get('/manage/')
|
|
assert 'New category' in resp
|
|
category_page = resp.click('New category')
|
|
category_form = category_page.forms[0]
|
|
category_form['name'] = 'Alerts'
|
|
resp = category_form.submit()
|
|
resp = resp.follow()
|
|
resp = resp.click('Alerts')
|
|
assert 'New announce' in resp
|
|
announce_page = resp.click('New announce')
|
|
announce_form = announce_page.forms[0]
|
|
announce_form['title'] = 'First announce'
|
|
announce_form['text'] = 'announce content'
|
|
resp = announce_form.submit()
|
|
assert resp.status_int == 302
|
|
assert resp.location.endswith(reverse('view_category', kwargs={'slug': 'alerts'}))
|
|
resp = resp.follow()
|
|
assert 'First announce' in resp
|
|
settings.SMS_GATEWAY_URL='invalid_url'
|
|
resp = resp.click('First announce')
|
|
assert 'Send test SMS' in resp
|
|
resp = resp.click('Send test SMS')
|
|
form = resp.forms[0]
|
|
form['mobile'] = '0607080900'
|
|
resp = form.submit()
|
|
records = caplog.records
|
|
assert len(records) == 1
|
|
for record in records:
|
|
assert record.name == 'corbo.utils'
|
|
assert record.levelno == logging.WARNING
|
|
assert 'Invalid URL' in record.getMessage()
|
|
|
|
|
|
@pytest.fixture(params=['foo@example.net', '06 07 08 09 00', '+33 6 07 08 09 00',
|
|
'0033607080900', '+33 (0) 6 07 08 09 00', '0033+607080900+'])
|
|
def search_identifier(request, subscriptions):
|
|
return request.param
|
|
|
|
|
|
def test_subscriptions_search(app, admin_user, search_identifier):
|
|
app = login(app)
|
|
resp = app.get('/manage/')
|
|
resp = resp.click('Search')
|
|
if format_phonenumber(search_identifier):
|
|
search_identifier = format_phonenumber(search_identifier)[-9:]
|
|
# empty search
|
|
resp = resp.form.submit()
|
|
assert resp.html.find('div', {'class': 'subscriptions-search'}) is None
|
|
# unknow identifier
|
|
resp.form['q'] = 'toto'
|
|
resp = resp.form.submit()
|
|
assert resp.html.find('div', {'class': 'subscriptions-search'}) is None
|
|
# search by email
|
|
assert Subscription.objects.filter(identifier__contains=search_identifier).count() == 2
|
|
resp.form['q'] = search_identifier
|
|
resp = resp.form.submit()
|
|
assert len(resp.html.find('div', {'class': 'subscriptions-search'}).find_all('li')) == 2
|
|
# delete sub
|
|
first_sub = Subscription.objects.filter(identifier__contains=search_identifier).first()
|
|
resp = resp.click(href='/manage/subscription/delete/%d/' % first_sub.pk)
|
|
resp = resp.form.submit().follow()
|
|
assert Subscription.objects.filter(identifier__contains=search_identifier).count() == 1
|
|
# ensure we're back on the previous page with less result
|
|
assert len(resp.html.find('div', {'class': 'subscriptions-search'}).find_all('li')) == 1
|
|
assert len(resp.html.find_all('li', {'class': 'success'})) == 1
|
|
assert '%s successfuly unsubscribed from' % search_identifier in resp.html.find('li', {'class': 'success'}).text
|