mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-22 00:11:16 -05:00
1145 lines
49 KiB
Python
1145 lines
49 KiB
Python
from datetime import datetime
|
|
import uuid
|
|
import app.celery.tasks
|
|
from tests import create_authorization_header
|
|
from tests.app.conftest import sample_notification, sample_job, sample_service, sample_email_template, sample_template
|
|
from flask import json
|
|
from app.models import Service
|
|
from app.dao.templates_dao import dao_get_all_templates_for_service
|
|
from app.dao.services_dao import dao_update_service
|
|
from app.dao.notifications_dao import get_notification_by_id
|
|
from freezegun import freeze_time
|
|
from tests.app import load_example_ses
|
|
|
|
|
|
def test_get_notification_by_id(notify_api, sample_notification):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_notification.service_id,
|
|
path='/notifications/{}'.format(sample_notification.id),
|
|
method='GET')
|
|
|
|
response = client.get(
|
|
'/notifications/{}'.format(sample_notification.id),
|
|
headers=[auth_header])
|
|
|
|
notification = json.loads(response.get_data(as_text=True))['notification']
|
|
assert notification['status'] == 'sent'
|
|
assert notification['template'] == sample_notification.template.id
|
|
assert notification['to'] == '+447700900855'
|
|
assert notification['service'] == str(sample_notification.service_id)
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_get_notifications_empty_result(notify_api, sample_api_key):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
missing_notification_id = uuid.uuid4()
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_api_key.service_id,
|
|
path='/notifications/{}'.format(missing_notification_id),
|
|
method='GET')
|
|
|
|
response = client.get(
|
|
path='/notifications/{}'.format(missing_notification_id),
|
|
headers=[auth_header])
|
|
|
|
notification = json.loads(response.get_data(as_text=True))
|
|
assert notification['result'] == "error"
|
|
assert notification['message'] == "not found"
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_get_all_notifications(notify_api, sample_notification):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_notification.service_id,
|
|
path='/notifications',
|
|
method='GET')
|
|
|
|
response = client.get(
|
|
'/notifications',
|
|
headers=[auth_header])
|
|
|
|
notifications = json.loads(response.get_data(as_text=True))
|
|
assert notifications['notifications'][0]['status'] == 'sent'
|
|
assert notifications['notifications'][0]['template'] == sample_notification.template.id
|
|
assert notifications['notifications'][0]['to'] == '+447700900855'
|
|
assert notifications['notifications'][0]['service'] == str(sample_notification.service_id)
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_get_all_notifications_newest_first(notify_api, notify_db, notify_db_session, sample_email_template):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
notification_1 = sample_notification(notify_db, notify_db_session, sample_email_template.service)
|
|
notification_2 = sample_notification(notify_db, notify_db_session, sample_email_template.service)
|
|
notification_3 = sample_notification(notify_db, notify_db_session, sample_email_template.service)
|
|
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_email_template.service_id,
|
|
path='/notifications',
|
|
method='GET')
|
|
|
|
response = client.get(
|
|
'/notifications',
|
|
headers=[auth_header])
|
|
|
|
notifications = json.loads(response.get_data(as_text=True))
|
|
assert len(notifications['notifications']) == 3
|
|
assert notifications['notifications'][0]['to'] == notification_3.to
|
|
assert notifications['notifications'][1]['to'] == notification_2.to
|
|
assert notifications['notifications'][2]['to'] == notification_1.to
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_get_all_notifications_for_service_in_order(notify_api, notify_db, notify_db_session):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
service_1 = sample_service(notify_db, notify_db_session, service_name="1")
|
|
service_2 = sample_service(notify_db, notify_db_session, service_name="2")
|
|
|
|
sample_notification(notify_db, notify_db_session, service=service_2)
|
|
|
|
notification_1 = sample_notification(notify_db, notify_db_session, service=service_1)
|
|
notification_2 = sample_notification(notify_db, notify_db_session, service=service_1)
|
|
notification_3 = sample_notification(notify_db, notify_db_session, service=service_1)
|
|
|
|
auth_header = create_authorization_header(
|
|
path='/service/{}/notifications'.format(service_1.id),
|
|
method='GET')
|
|
|
|
response = client.get(
|
|
path='/service/{}/notifications'.format(service_1.id),
|
|
headers=[auth_header])
|
|
|
|
resp = json.loads(response.get_data(as_text=True))
|
|
assert len(resp['notifications']) == 3
|
|
assert resp['notifications'][0]['to'] == notification_3.to
|
|
assert resp['notifications'][1]['to'] == notification_2.to
|
|
assert resp['notifications'][2]['to'] == notification_1.to
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_get_all_notifications_for_job_in_order(notify_api, notify_db, notify_db_session, sample_service):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
main_job = sample_job(notify_db, notify_db_session, service=sample_service)
|
|
another_job = sample_job(notify_db, notify_db_session, service=sample_service)
|
|
|
|
notification_1 = sample_notification(
|
|
notify_db, notify_db_session, job=main_job, to_field="1", created_at=datetime.utcnow()
|
|
)
|
|
notification_2 = sample_notification(
|
|
notify_db, notify_db_session, job=main_job, to_field="2", created_at=datetime.utcnow()
|
|
)
|
|
notification_3 = sample_notification(
|
|
notify_db, notify_db_session, job=main_job, to_field="3", created_at=datetime.utcnow()
|
|
)
|
|
sample_notification(notify_db, notify_db_session, job=another_job)
|
|
|
|
auth_header = create_authorization_header(
|
|
path='/service/{}/job/{}/notifications'.format(sample_service.id, main_job.id),
|
|
method='GET')
|
|
|
|
response = client.get(
|
|
path='/service/{}/job/{}/notifications'.format(sample_service.id, main_job.id),
|
|
headers=[auth_header])
|
|
|
|
resp = json.loads(response.get_data(as_text=True))
|
|
assert len(resp['notifications']) == 3
|
|
assert resp['notifications'][0]['to'] == notification_3.to
|
|
assert resp['notifications'][1]['to'] == notification_2.to
|
|
assert resp['notifications'][2]['to'] == notification_1.to
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_should_not_get_notifications_by_service_with_client_credentials(notify_api, sample_api_key):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_api_key.service.id,
|
|
path='/service/{}/notifications'.format(sample_api_key.service.id),
|
|
method='GET')
|
|
|
|
response = client.get(
|
|
'/service/{}/notifications'.format(sample_api_key.service.id),
|
|
headers=[auth_header])
|
|
|
|
resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 403
|
|
assert resp['result'] == 'error'
|
|
assert resp['message'] == 'Forbidden, invalid authentication token provided'
|
|
|
|
|
|
def test_should_not_get_notifications_by_job_and_service_with_client_credentials(notify_api, sample_job):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_job.service.id,
|
|
path='/service/{}/job/{}/notifications'.format(sample_job.service.id, sample_job.id),
|
|
method='GET')
|
|
|
|
response = client.get(
|
|
'/service/{}/job/{}/notifications'.format(sample_job.service.id, sample_job.id),
|
|
headers=[auth_header])
|
|
|
|
resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 403
|
|
assert resp['result'] == 'error'
|
|
assert resp['message'] == 'Forbidden, invalid authentication token provided'
|
|
|
|
|
|
def test_should_reject_invalid_page_param(notify_api, sample_email_template):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_email_template.service_id,
|
|
path='/notifications',
|
|
method='GET')
|
|
|
|
response = client.get(
|
|
'/notifications?page=invalid',
|
|
headers=[auth_header])
|
|
|
|
notifications = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 400
|
|
assert notifications['result'] == 'error'
|
|
assert notifications['message'] == 'Invalid page'
|
|
|
|
|
|
def test_should_return_pagination_links(notify_api, notify_db, notify_db_session, sample_email_template):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
notify_api.config['PAGE_SIZE'] = 1
|
|
|
|
sample_notification(notify_db, notify_db_session, sample_email_template.service)
|
|
notification_2 = sample_notification(notify_db, notify_db_session, sample_email_template.service)
|
|
sample_notification(notify_db, notify_db_session, sample_email_template.service)
|
|
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_email_template.service_id,
|
|
path='/notifications',
|
|
method='GET')
|
|
|
|
response = client.get(
|
|
'/notifications?page=2',
|
|
headers=[auth_header])
|
|
|
|
notifications = json.loads(response.get_data(as_text=True))
|
|
assert len(notifications['notifications']) == 1
|
|
assert notifications['links']['last'] == '/notifications?page=3'
|
|
assert notifications['links']['prev'] == '/notifications?page=1'
|
|
assert notifications['links']['next'] == '/notifications?page=3'
|
|
assert notifications['notifications'][0]['to'] == notification_2.to
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_get_all_notifications_returns_empty_list(notify_api, sample_api_key):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_api_key.service.id,
|
|
path='/notifications',
|
|
method='GET')
|
|
|
|
response = client.get(
|
|
'/notifications',
|
|
headers=[auth_header])
|
|
|
|
notifications = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 200
|
|
assert len(notifications['notifications']) == 0
|
|
|
|
|
|
def test_create_sms_should_reject_if_missing_required_fields(notify_api, sample_api_key, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
|
|
|
data = {}
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_api_key.service_id,
|
|
request_body=json.dumps(data),
|
|
path='/notifications/sms',
|
|
method='POST')
|
|
|
|
response = client.post(
|
|
path='/notifications/sms',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
app.celery.tasks.send_sms.apply_async.assert_not_called()
|
|
assert json_resp['result'] == 'error'
|
|
assert 'Missing data for required field.' in json_resp['message']['to'][0]
|
|
assert 'Missing data for required field.' in json_resp['message']['template'][0]
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_should_reject_bad_phone_numbers(notify_api, sample_template, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
|
|
|
data = {
|
|
'to': 'invalid',
|
|
'template': sample_template.id
|
|
}
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_template.service.id,
|
|
request_body=json.dumps(data),
|
|
path='/notifications/sms',
|
|
method='POST')
|
|
|
|
response = client.post(
|
|
path='/notifications/sms',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
app.celery.tasks.send_sms.apply_async.assert_not_called()
|
|
assert json_resp['result'] == 'error'
|
|
assert len(json_resp['message'].keys()) == 1
|
|
assert 'Invalid phone number: Must not contain letters or symbols' in json_resp['message']['to']
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_send_notification_invalid_template_id(notify_api, sample_template, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
|
|
|
data = {
|
|
'to': '+447700900855',
|
|
'template': 9999
|
|
}
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_template.service.id,
|
|
request_body=json.dumps(data),
|
|
path='/notifications/sms',
|
|
method='POST')
|
|
|
|
response = client.post(
|
|
path='/notifications/sms',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
app.celery.tasks.send_sms.apply_async.assert_not_called()
|
|
|
|
assert response.status_code == 404
|
|
assert len(json_resp['message'].keys()) == 1
|
|
test_string = 'Template {} not found for service {}'.format(9999, sample_template.service.id)
|
|
assert test_string in json_resp['message']['template']
|
|
|
|
|
|
@freeze_time("2016-01-01 11:09:00.061258")
|
|
def test_send_notification_with_placeholders_replaced(notify_api, sample_template_with_placeholders, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
|
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
|
|
|
|
data = {
|
|
'to': '+447700900855',
|
|
'template': sample_template_with_placeholders.id,
|
|
'personalisation': {
|
|
'name': 'Jo'
|
|
}
|
|
}
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_template_with_placeholders.service.id,
|
|
request_body=json.dumps(data),
|
|
path='/notifications/sms',
|
|
method='POST')
|
|
|
|
response = client.post(
|
|
path='/notifications/sms',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
notification_id = json.loads(response.data)['notification_id']
|
|
app.celery.tasks.send_sms.apply_async.assert_called_once_with(
|
|
(str(sample_template_with_placeholders.service.id),
|
|
notification_id,
|
|
"something_encrypted",
|
|
"2016-01-01T11:09:00.061258"),
|
|
queue="sms"
|
|
)
|
|
assert response.status_code == 201
|
|
|
|
|
|
def test_send_notification_with_missing_personalisation(notify_api, sample_template_with_placeholders, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
|
|
|
data = {
|
|
'to': '+447700900855',
|
|
'template': sample_template_with_placeholders.id,
|
|
'personalisation': {
|
|
'foo': 'bar'
|
|
}
|
|
}
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_template_with_placeholders.service.id,
|
|
request_body=json.dumps(data),
|
|
path='/notifications/sms',
|
|
method='POST')
|
|
|
|
response = client.post(
|
|
path='/notifications/sms',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
app.celery.tasks.send_sms.apply_async.assert_not_called()
|
|
|
|
assert response.status_code == 400
|
|
assert 'Missing personalisation: name' in json_resp['message']['template']
|
|
|
|
|
|
def test_send_notification_with_too_much_personalisation_data(
|
|
notify_api, sample_template_with_placeholders, mocker
|
|
):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
|
|
|
data = {
|
|
'to': '+447700900855',
|
|
'template': sample_template_with_placeholders.id,
|
|
'personalisation': {
|
|
'name': 'Jo', 'foo': 'bar'
|
|
}
|
|
}
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_template_with_placeholders.service.id,
|
|
request_body=json.dumps(data),
|
|
path='/notifications/sms',
|
|
method='POST')
|
|
|
|
response = client.post(
|
|
path='/notifications/sms',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
app.celery.tasks.send_sms.apply_async.assert_not_called()
|
|
|
|
assert response.status_code == 400
|
|
assert 'Personalisation not needed for template: foo' in json_resp['message']['template']
|
|
|
|
|
|
def test_prevents_sending_to_any_mobile_on_restricted_service(notify_api, sample_template, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
|
|
|
Service.query.filter_by(
|
|
id=sample_template.service.id
|
|
).update(
|
|
{'restricted': True}
|
|
)
|
|
invalid_mob = '+447700900855'
|
|
data = {
|
|
'to': invalid_mob,
|
|
'template': sample_template.id
|
|
}
|
|
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_template.service.id,
|
|
request_body=json.dumps(data),
|
|
path='/notifications/sms',
|
|
method='POST')
|
|
|
|
response = client.post(
|
|
path='/notifications/sms',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
app.celery.tasks.send_sms.apply_async.assert_not_called()
|
|
|
|
assert response.status_code == 400
|
|
assert 'Invalid phone number for restricted service' in json_resp['message']['to']
|
|
|
|
|
|
def test_should_not_allow_template_from_another_service(notify_api, service_factory, sample_user, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
|
|
|
service_1 = service_factory.get('service 1', user=sample_user)
|
|
service_2 = service_factory.get('service 2', user=sample_user)
|
|
|
|
service_2_templates = dao_get_all_templates_for_service(service_id=service_2.id)
|
|
data = {
|
|
'to': sample_user.mobile_number,
|
|
'template': service_2_templates[0].id
|
|
}
|
|
|
|
auth_header = create_authorization_header(
|
|
service_id=service_1.id,
|
|
request_body=json.dumps(data),
|
|
path='/notifications/sms',
|
|
method='POST')
|
|
|
|
response = client.post(
|
|
path='/notifications/sms',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
app.celery.tasks.send_sms.apply_async.assert_not_called()
|
|
|
|
assert response.status_code == 404
|
|
test_string = 'Template {} not found for service {}'.format(service_2_templates[0].id, service_1.id)
|
|
assert test_string in json_resp['message']['template']
|
|
|
|
|
|
@freeze_time("2016-01-01 11:09:00.061258")
|
|
def test_should_allow_valid_sms_notification(notify_api, sample_template, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
|
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
|
|
|
|
data = {
|
|
'to': '07700 900 855',
|
|
'template': sample_template.id
|
|
}
|
|
|
|
auth_header = create_authorization_header(
|
|
request_body=json.dumps(data),
|
|
path='/notifications/sms',
|
|
method='POST',
|
|
service_id=sample_template.service_id
|
|
)
|
|
|
|
response = client.post(
|
|
path='/notifications/sms',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
notification_id = json.loads(response.data)['notification_id']
|
|
assert app.encryption.encrypt.call_args[0][0]['to'] == '+447700900855'
|
|
app.celery.tasks.send_sms.apply_async.assert_called_once_with(
|
|
(str(sample_template.service_id),
|
|
notification_id,
|
|
"something_encrypted",
|
|
"2016-01-01T11:09:00.061258"),
|
|
queue="sms"
|
|
)
|
|
assert response.status_code == 201
|
|
assert notification_id
|
|
|
|
|
|
def test_create_email_should_reject_if_missing_required_fields(notify_api, sample_api_key, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_email.apply_async')
|
|
|
|
data = {}
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_api_key.service_id,
|
|
request_body=json.dumps(data),
|
|
path='/notifications/email',
|
|
method='POST')
|
|
|
|
response = client.post(
|
|
path='/notifications/email',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
app.celery.tasks.send_email.apply_async.assert_not_called()
|
|
assert json_resp['result'] == 'error'
|
|
assert 'Missing data for required field.' in json_resp['message']['to'][0]
|
|
assert 'Missing data for required field.' in json_resp['message']['template'][0]
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_should_reject_email_notification_with_bad_email(notify_api, sample_email_template, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_email.apply_async')
|
|
to_address = "bad-email"
|
|
data = {
|
|
'to': to_address,
|
|
'template': sample_email_template.service.id
|
|
}
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_email_template.service.id,
|
|
request_body=json.dumps(data),
|
|
path='/notifications/email',
|
|
method='POST')
|
|
|
|
response = client.post(
|
|
path='/notifications/email',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
data = json.loads(response.get_data(as_text=True))
|
|
app.celery.tasks.send_email.apply_async.assert_not_called()
|
|
assert response.status_code == 400
|
|
assert data['result'] == 'error'
|
|
assert data['message']['to'][0] == 'Not a valid email address'
|
|
|
|
|
|
def test_should_reject_email_notification_with_template_id_that_cant_be_found(
|
|
notify_api, sample_email_template, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_email.apply_async')
|
|
data = {
|
|
'to': 'ok@ok.com',
|
|
'template': 1234
|
|
}
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_email_template.service.id,
|
|
request_body=json.dumps(data),
|
|
path='/notifications/email',
|
|
method='POST')
|
|
|
|
response = client.post(
|
|
path='/notifications/email',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
data = json.loads(response.get_data(as_text=True))
|
|
app.celery.tasks.send_email.apply_async.assert_not_called()
|
|
assert response.status_code == 404
|
|
assert data['result'] == 'error'
|
|
test_string = 'Template {} not found for service {}'.format(
|
|
1234,
|
|
sample_email_template.service.id
|
|
)
|
|
assert test_string in data['message']['template']
|
|
|
|
|
|
def test_should_not_allow_email_template_from_another_service(notify_api, service_factory, sample_user, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_email.apply_async')
|
|
|
|
service_1 = service_factory.get('service 1', template_type='email', user=sample_user)
|
|
service_2 = service_factory.get('service 2', template_type='email', user=sample_user)
|
|
|
|
service_2_templates = dao_get_all_templates_for_service(service_id=service_2.id)
|
|
|
|
data = {
|
|
'to': sample_user.email_address,
|
|
'template': service_2_templates[0].id
|
|
}
|
|
|
|
auth_header = create_authorization_header(
|
|
service_id=service_1.id,
|
|
request_body=json.dumps(data),
|
|
path='/notifications/email',
|
|
method='POST')
|
|
|
|
response = client.post(
|
|
path='/notifications/email',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
app.celery.tasks.send_email.apply_async.assert_not_called()
|
|
|
|
assert response.status_code == 404
|
|
test_string = 'Template {} not found for service {}'.format(service_2_templates[0].id, service_1.id)
|
|
assert test_string in json_resp['message']['template']
|
|
|
|
|
|
def test_should_not_send_email_if_restricted_and_not_a_service_user(notify_api, sample_email_template, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_email.apply_async')
|
|
|
|
sample_email_template.service.restricted = True
|
|
dao_update_service(sample_email_template)
|
|
|
|
data = {
|
|
'to': "not-someone-we-trust@email-address.com",
|
|
'template': sample_email_template.id
|
|
}
|
|
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_email_template.service.id,
|
|
request_body=json.dumps(data),
|
|
path='/notifications/email',
|
|
method='POST')
|
|
|
|
response = client.post(
|
|
path='/notifications/email',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
app.celery.tasks.send_email.apply_async.assert_not_called()
|
|
|
|
assert response.status_code == 400
|
|
assert 'Email address not permitted for restricted service' in json_resp['message']['to']
|
|
|
|
|
|
def test_should_not_send_email_for_job_if_restricted_and_not_a_service_user(
|
|
notify_api,
|
|
sample_job,
|
|
sample_email_template,
|
|
mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_email.apply_async')
|
|
|
|
sample_email_template.service.restricted = True
|
|
dao_update_service(sample_email_template)
|
|
|
|
data = {
|
|
'to': "not-someone-we-trust@email-address.com",
|
|
'template': sample_job.template.id,
|
|
'job': sample_job.id
|
|
}
|
|
|
|
auth_header = create_authorization_header(
|
|
service_id=sample_job.service.id,
|
|
request_body=json.dumps(data),
|
|
path='/notifications/email',
|
|
method='POST')
|
|
|
|
response = client.post(
|
|
path='/notifications/email',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
app.celery.tasks.send_email.apply_async.assert_not_called()
|
|
|
|
assert response.status_code == 400
|
|
assert 'Email address not permitted for restricted service' in json_resp['message']['to']
|
|
|
|
|
|
@freeze_time("2016-01-01 11:09:00.061258")
|
|
def test_should_allow_valid_email_notification(notify_api, sample_email_template, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_email.apply_async')
|
|
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
|
|
|
|
data = {
|
|
'to': 'ok@ok.com',
|
|
'template': sample_email_template.id
|
|
}
|
|
|
|
auth_header = create_authorization_header(
|
|
request_body=json.dumps(data),
|
|
path='/notifications/email',
|
|
method='POST',
|
|
service_id=sample_email_template.service_id
|
|
)
|
|
|
|
response = client.post(
|
|
path='/notifications/email',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
assert response.status_code == 201
|
|
notification_id = json.loads(response.get_data(as_text=True))['notification_id']
|
|
app.celery.tasks.send_email.apply_async.assert_called_once_with(
|
|
(str(sample_email_template.service_id),
|
|
notification_id,
|
|
"Email Subject",
|
|
"sample.service@test.notify.com",
|
|
"something_encrypted",
|
|
"2016-01-01T11:09:00.061258"),
|
|
queue="email"
|
|
)
|
|
assert response.status_code == 201
|
|
assert notification_id
|
|
|
|
|
|
@freeze_time("2016-01-01 12:00:00.061258")
|
|
def test_should_block_api_call_if_over_day_limit(notify_db, notify_db_session, notify_api, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_email.apply_async')
|
|
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
|
|
|
|
service = sample_service(notify_db, notify_db_session, limit=1)
|
|
email_template = sample_email_template(notify_db, notify_db_session, service=service)
|
|
sample_notification(
|
|
notify_db, notify_db_session, template=email_template, service=service, created_at=datetime.utcnow()
|
|
)
|
|
|
|
data = {
|
|
'to': 'ok@ok.com',
|
|
'template': email_template.id
|
|
}
|
|
|
|
auth_header = create_authorization_header(
|
|
request_body=json.dumps(data),
|
|
path='/notifications/email',
|
|
method='POST',
|
|
service_id=service.id
|
|
)
|
|
|
|
response = client.post(
|
|
path='/notifications/email',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 429
|
|
assert 'Exceeded send limits (1) for today' in json_resp['message']
|
|
|
|
|
|
@freeze_time("2016-01-01 12:00:00.061258")
|
|
def test_should_block_api_call_if_over_day_limit_regardless_of_type(notify_db, notify_db_session, notify_api, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
|
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
|
|
|
|
service = sample_service(notify_db, notify_db_session, limit=1)
|
|
email_template = sample_email_template(notify_db, notify_db_session, service=service)
|
|
sms_template = sample_template(notify_db, notify_db_session, service=service)
|
|
sample_notification(
|
|
notify_db, notify_db_session, template=email_template, service=service, created_at=datetime.utcnow()
|
|
)
|
|
|
|
data = {
|
|
'to': '+447234123123',
|
|
'template': sms_template.id
|
|
}
|
|
|
|
auth_header = create_authorization_header(
|
|
request_body=json.dumps(data),
|
|
path='/notifications/sms',
|
|
method='POST',
|
|
service_id=service.id
|
|
)
|
|
|
|
response = client.post(
|
|
path='/notifications/sms',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 429
|
|
assert 'Exceeded send limits (1) for today' in json_resp['message']
|
|
|
|
|
|
@freeze_time("2016-01-01 12:00:00.061258")
|
|
def test_should_allow_api_call_if_under_day_limit_regardless_of_type(notify_db, notify_db_session, notify_api, mocker):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
|
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
|
|
|
|
service = sample_service(notify_db, notify_db_session, limit=2)
|
|
email_template = sample_email_template(notify_db, notify_db_session, service=service)
|
|
sms_template = sample_template(notify_db, notify_db_session, service=service)
|
|
sample_notification(notify_db, notify_db_session, template=email_template, service=service)
|
|
|
|
data = {
|
|
'to': '+447634123123',
|
|
'template': sms_template.id
|
|
}
|
|
|
|
auth_header = create_authorization_header(
|
|
request_body=json.dumps(data),
|
|
path='/notifications/sms',
|
|
method='POST',
|
|
service_id=service.id
|
|
)
|
|
|
|
response = client.post(
|
|
path='/notifications/sms',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
assert response.status_code == 201
|
|
|
|
|
|
def test_firetext_callback_should_not_need_auth(notify_api):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
response = client.post(
|
|
path='/notifications/sms/firetext',
|
|
data='mobile=441234123123&status=0&reference=send-sms-code&time=2016-03-10 14:17:00',
|
|
headers=[('Content-Type', 'application/x-www-form-urlencoded')])
|
|
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_firetext_callback_should_return_400_if_empty_reference(notify_api):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
response = client.post(
|
|
path='/notifications/sms/firetext',
|
|
data='mobile=441234123123&status=0&reference=&time=2016-03-10 14:17:00',
|
|
headers=[('Content-Type', 'application/x-www-form-urlencoded')])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 400
|
|
assert json_resp['result'] == 'error'
|
|
assert json_resp['message'] == 'Firetext callback failed: reference missing'
|
|
|
|
|
|
def test_firetext_callback_should_return_400_if_no_reference(notify_api):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
response = client.post(
|
|
path='/notifications/sms/firetext',
|
|
data='mobile=441234123123&status=0&time=2016-03-10 14:17:00',
|
|
headers=[('Content-Type', 'application/x-www-form-urlencoded')])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 400
|
|
assert json_resp['result'] == 'error'
|
|
assert json_resp['message'] == 'Firetext callback failed: reference missing'
|
|
|
|
|
|
def test_firetext_callback_should_return_200_if_send_sms_reference(notify_api):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
response = client.post(
|
|
path='/notifications/sms/firetext',
|
|
data='mobile=441234123123&status=0&time=2016-03-10 14:17:00&reference=send-sms-code',
|
|
headers=[('Content-Type', 'application/x-www-form-urlencoded')])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 200
|
|
assert json_resp['result'] == 'success'
|
|
assert json_resp['message'] == 'Firetext callback succeeded: send-sms-code'
|
|
|
|
|
|
def test_firetext_callback_should_return_400_if_no_status(notify_api):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
response = client.post(
|
|
path='/notifications/sms/firetext',
|
|
data='mobile=441234123123&time=2016-03-10 14:17:00',
|
|
headers=[('Content-Type', 'application/x-www-form-urlencoded')])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 400
|
|
assert json_resp['result'] == 'error'
|
|
assert json_resp['message'] == 'Firetext callback failed: status missing'
|
|
|
|
|
|
def test_firetext_callback_should_return_400_if_unknown_status(notify_api):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
response = client.post(
|
|
path='/notifications/sms/firetext',
|
|
data='mobile=441234123123&status=99&time=2016-03-10 14:17:00&reference={}'.format(uuid.uuid4()),
|
|
headers=[('Content-Type', 'application/x-www-form-urlencoded')])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 400
|
|
assert json_resp['result'] == 'error'
|
|
assert json_resp['message'] == 'Firetext callback failed: status 99 not found.'
|
|
|
|
|
|
def test_firetext_callback_should_return_400_if_invalid_guid_notification_id(notify_api):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
response = client.post(
|
|
path='/notifications/sms/firetext',
|
|
data='mobile=441234123123&status=0&time=2016-03-10 14:17:00&reference=1234',
|
|
headers=[('Content-Type', 'application/x-www-form-urlencoded')])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 400
|
|
assert json_resp['result'] == 'error'
|
|
assert json_resp['message'] == 'Firetext callback with invalid reference 1234'
|
|
|
|
|
|
def test_firetext_callback_should_return_404_if_cannot_find_notification_id(notify_db, notify_db_session, notify_api):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
missing_notification_id = uuid.uuid4()
|
|
response = client.post(
|
|
path='/notifications/sms/firetext',
|
|
data='mobile=441234123123&status=0&time=2016-03-10 14:17:00&reference={}'.format(
|
|
missing_notification_id
|
|
),
|
|
headers=[('Content-Type', 'application/x-www-form-urlencoded')])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 404
|
|
assert json_resp['result'] == 'error'
|
|
assert json_resp['message'] == 'Firetext callback failed: notification {} not found. Status {}'.format(
|
|
missing_notification_id,
|
|
'delivered'
|
|
)
|
|
|
|
|
|
def test_firetext_callback_should_update_notification_status(notify_api, sample_notification):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
original = get_notification_by_id(sample_notification.id)
|
|
assert original.status == 'sent'
|
|
|
|
response = client.post(
|
|
path='/notifications/sms/firetext',
|
|
data='mobile=441234123123&status=0&time=2016-03-10 14:17:00&reference={}'.format(
|
|
sample_notification.id
|
|
),
|
|
headers=[('Content-Type', 'application/x-www-form-urlencoded')])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 200
|
|
assert json_resp['result'] == 'success'
|
|
assert json_resp['message'] == 'Firetext callback succeeded. reference {} updated'.format(
|
|
sample_notification.id
|
|
)
|
|
updated = get_notification_by_id(sample_notification.id)
|
|
assert updated.status == 'delivered'
|
|
|
|
|
|
def test_firetext_callback_should_update_notification_status_failed(notify_api, sample_notification):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
original = get_notification_by_id(sample_notification.id)
|
|
assert original.status == 'sent'
|
|
|
|
response = client.post(
|
|
path='/notifications/sms/firetext',
|
|
data='mobile=441234123123&status=1&time=2016-03-10 14:17:00&reference={}'.format(
|
|
sample_notification.id
|
|
),
|
|
headers=[('Content-Type', 'application/x-www-form-urlencoded')])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 200
|
|
assert json_resp['result'] == 'success'
|
|
assert json_resp['message'] == 'Firetext callback succeeded. reference {} updated'.format(
|
|
sample_notification.id
|
|
)
|
|
updated = get_notification_by_id(sample_notification.id)
|
|
assert updated.status == 'failed'
|
|
|
|
|
|
def test_firetext_callback_should_update_notification_status_sent(notify_api, notify_db, notify_db_session):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
notification = sample_notification(notify_db, notify_db_session, status='delivered')
|
|
original = get_notification_by_id(notification.id)
|
|
assert original.status == 'delivered'
|
|
|
|
response = client.post(
|
|
path='/notifications/sms/firetext',
|
|
data='mobile=441234123123&status=2&time=2016-03-10 14:17:00&reference={}'.format(
|
|
notification.id
|
|
),
|
|
headers=[('Content-Type', 'application/x-www-form-urlencoded')])
|
|
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 200
|
|
assert json_resp['result'] == 'success'
|
|
assert json_resp['message'] == 'Firetext callback succeeded. reference {} updated'.format(
|
|
notification.id
|
|
)
|
|
updated = get_notification_by_id(notification.id)
|
|
assert updated.status == 'sent'
|
|
|
|
|
|
def test_ses_callback_should_not_need_auth(notify_api):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
response = client.post(
|
|
path='/notifications/email/ses',
|
|
data=load_example_ses('ses_response'),
|
|
headers=[('Content-Type', 'text/plain; charset=UTF-8')]
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_ses_callback_should_fail_if_invalid_json(notify_api):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
response = client.post(
|
|
path='/notifications/email/ses',
|
|
data="nonsense",
|
|
headers=[('Content-Type', 'text/plain; charset=UTF-8')]
|
|
)
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 400
|
|
assert json_resp['result'] == 'error'
|
|
assert json_resp['message'] == 'SES callback failed: invalid json'
|
|
|
|
|
|
def test_ses_callback_should_fail_if_invalid_notification_type(notify_api):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
ses_response = json.loads(load_example_ses('ses_response'))
|
|
ses_response['Message']['notificationType'] = 'Unknown'
|
|
|
|
response = client.post(
|
|
path='/notifications/email/ses',
|
|
data=json.dumps(ses_response),
|
|
headers=[('Content-Type', 'text/plain; charset=UTF-8')]
|
|
)
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 400
|
|
assert json_resp['result'] == 'error'
|
|
assert json_resp['message'] == 'SES callback failed: status Unknown not found'
|
|
|
|
|
|
def test_ses_callback_should_fail_if_missing_destination(notify_api):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
ses_response = json.loads(load_example_ses('ses_response'))
|
|
del(ses_response['Message']['mail']['destination'])
|
|
|
|
response = client.post(
|
|
path='/notifications/email/ses',
|
|
data=json.dumps(ses_response),
|
|
headers=[('Content-Type', 'text/plain; charset=UTF-8')]
|
|
)
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 400
|
|
assert json_resp['result'] == 'error'
|
|
assert json_resp['message'] == 'SES callback failed: destination missing'
|
|
|
|
|
|
def test_ses_callback_should_fail_if_notification_cannot_be_found(notify_db, notify_db_session, notify_api):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
ses_response = json.loads(load_example_ses('ses_response'))
|
|
ses_response['Message']['mail']['destination'] = ['wont find this']
|
|
|
|
response = client.post(
|
|
path='/notifications/email/ses',
|
|
data=json.dumps(ses_response),
|
|
headers=[('Content-Type', 'text/plain; charset=UTF-8')]
|
|
)
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 404
|
|
assert json_resp['result'] == 'error'
|
|
assert json_resp['message'] == 'SES callback failed: notification not found. Status delivered'
|
|
|
|
|
|
def test_ses_callback_should_update_notification_status(notify_api, sample_notification):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
|
|
assert get_notification_by_id(sample_notification.id).status == 'sent'
|
|
|
|
ses_response = json.loads(load_example_ses('ses_response'))
|
|
ses_response['Message']['mail']['destination'] = [sample_notification.to]
|
|
|
|
response = client.post(
|
|
path='/notifications/email/ses',
|
|
data=json.dumps(ses_response),
|
|
headers=[('Content-Type', 'text/plain; charset=UTF-8')]
|
|
)
|
|
json_resp = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 200
|
|
assert json_resp['result'] == 'success'
|
|
assert json_resp['message'] == 'SES callback succeeded'
|
|
assert get_notification_by_id(sample_notification.id).status == 'delivered'
|