darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-07-23 01:20:01 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #794 from alphagov/priority-by-template

Browse Source 
Priority by template
This commit is contained in:
Rebecca Law
2017-01-18 10:28:53 +00:00
committed by GitHub
parent cbe6a20d4e f66670c558
commit d0bea35069
18 changed files with 407 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
tests/app/conftest.py
View File

@@ -156,7 +156,8 @@ def sample_template(notify_db,
subject_line='Subject',
user=None,
service=None,
created_by=None):
created_by=None,
process_type='normal'):
if user is None:
user = create_user()
if service is None:
@@ -169,7 +170,8 @@ def sample_template(notify_db,
'content': content,
'service': service,
'created_by': created_by,
'archived': archived
'archived': archived,
'process_type': process_type
}
if template_type in ['email', 'letter']:
data.update({

1
tests/app/dao/test_templates_dao.py
View File

@@ -31,6 +31,7 @@ def test_create_template(sample_service, sample_user, template_type, subject):
assert Template.query.count() == 1
assert len(dao_get_all_templates_for_service(sample_service.id)) == 1
assert dao_get_all_templates_for_service(sample_service.id)[0].name == 'Sample Template'
assert dao_get_all_templates_for_service(sample_service.id)[0].process_type == 'normal'
def test_update_template(sample_service, sample_user):

37
tests/app/notifications/rest/test_send_notification.py
View File

@@ -710,8 +710,8 @@ def test_should_delete_notification_and_return_error_if_sqs_fails(
@pytest.mark.parametrize('to_email', [
'simulate-delivered@notifications.service.gov.uk',
'simulate-permanent-failure@notifications.service.gov.uk',
'simulate-temporary-failure@notifications.service.gov.uk'
'simulate-delivered-2@notifications.service.gov.uk',
'simulate-delivered-3@notifications.service.gov.uk'
])
def test_should_not_persist_notification_or_send_email_if_simulated_email(
client,
@@ -957,3 +957,36 @@ def test_create_template_raises_invalid_request_when_content_too_large(
if not should_error:
pytest.fail("do not expect an InvalidRequest")
assert e.message == {'content': ['Content has a character count greater than the limit of {}'.format(limit)]}
@pytest.mark.parametrize("notification_type, send_to",
[("sms", "07700 900 855"),
("email", "sample@email.com")])
def test_send_notification_uses_priority_queue_when_template_is_marked_as_priority(client, notify_db,
notify_db_session, mocker,
notification_type, send_to):
sample = create_sample_template(
notify_db,
notify_db_session,
template_type=notification_type,
process_type='priority'
)
mocked = mocker.patch('app.celery.provider_tasks.deliver_{}.apply_async'.format(notification_type))
data = {
'to': send_to,
'template': str(sample.id)
}
auth_header = create_authorization_header(service_id=sample.service_id)
response = client.post(
path='/notifications/{}'.format(notification_type),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
response_data = json.loads(response.data)['data']
notification_id = response_data['notification']['id']
assert response.status_code == 201
mocked.assert_called_once_with([notification_id], queue='notify')

25
tests/app/notifications/test_process_notification.py
View File

@@ -10,7 +10,9 @@ from collections import namedtuple
from app.models import Template, Notification, NotificationHistory
from app.notifications import SendNotificationToQueueError
from app.notifications.process_notifications import (create_content_for_notification,
persist_notification, send_notification_to_queue)
persist_notification,
send_notification_to_queue,
simulated_recipient)
from app.v2.errors import BadRequestError
@@ -193,3 +195,24 @@ def test_send_notification_to_queue_throws_exception_deletes_notification(sample
assert Notification.query.count() == 0
assert NotificationHistory.query.count() == 0
@pytest.mark.parametrize("to_address, notification_type, expected",
[("+447700900000", "sms", True),
("+447700900111", "sms", True),
("+447700900222", "sms", True),
("simulate-delivered@notifications.service.gov.uk", "email", True),
("simulate-delivered-2@notifications.service.gov.uk", "email", True),
("simulate-delivered-3@notifications.service.gov.uk", "email", True),
("07515896969", "sms", False),
("valid_email@test.com", "email", False)])
def test_simulated_recipient(notify_api, to_address, notification_type, expected):
# The values where the expected = 'research-mode' are listed in the config['SIMULATED_EMAIL_ADDRESSES']
# and config['SIMULATED_SMS_NUMBERS']. These values should result in using the research mode queue.
# SIMULATED_EMAIL_ADDRESSES = ('simulate-delivered@notifications.service.gov.uk',
# 'simulate-delivered-2@notifications.service.gov.uk',
# 'simulate-delivered-2@notifications.service.gov.uk')
# SIMULATED_SMS_NUMBERS = ('+447700900000', '+447700900111', '+447700900222')
actual = simulated_recipient(to_address, notification_type)
assert actual == expected

52
tests/app/notifications/test_rest.py
View File

@@ -13,34 +13,44 @@ from tests import create_authorization_header
from tests.app.conftest import sample_notification as create_sample_notification
def test_get_sms_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)
def test_get_sms_notification_by_id(client, sample_notification):
auth_header = create_authorization_header(service_id=sample_notification.service_id)
response = client.get(
'/notifications/{}'.format(sample_notification.id),
headers=[auth_header])
response = client.get(
'/notifications/{}'.format(sample_notification.id),
headers=[auth_header])
assert response.status_code == 200
notification = json.loads(response.get_data(as_text=True))['data']['notification']
assert notification['status'] == 'created'
assert notification['template'] == {
'id': str(sample_notification.template.id),
'name': sample_notification.template.name,
'template_type': sample_notification.template.template_type,
'version': 1
}
assert notification['to'] == '+447700900855'
assert notification['service'] == str(sample_notification.service_id)
assert notification['body'] == "This is a template:\nwith a newline"
assert not notification.get('subject')
assert response.status_code == 200
notification = json.loads(response.get_data(as_text=True))['data']['notification']
assert notification['status'] == 'created'
assert notification['template'] == {
'id': str(sample_notification.template.id),
'name': sample_notification.template.name,
'template_type': sample_notification.template.template_type,
'version': 1
}
assert notification['to'] == '+447700900855'
assert notification['service'] == str(sample_notification.service_id)
assert notification['body'] == "This is a template:\nwith a newline"
assert not notification.get('subject')
@pytest.mark.parametrize("id", ["1234-badly-formatted-id-7890", "0"])
def test_get_sms_notification_by_invalid_id(client, sample_notification, id):
auth_header = create_authorization_header(service_id=sample_notification.service_id)
response = client.get(
'/notifications/{}'.format(id),
headers=[auth_header])
assert response.status_code == 405
def test_get_email_notification_by_id(notify_api, notify_db, notify_db_session, sample_email_template):
email_notification = create_sample_notification(notify_db,
notify_db_session,
to_field="sample_email@example.com",
service=sample_email_template.service,
template=sample_email_template,
status='sending')
@@ -62,7 +72,7 @@ def test_get_email_notification_by_id(notify_api, notify_db, notify_db_session,
'template_type': email_notification.template.template_type,
'version': 1
}
assert notification['to'] == '+447700900855'
assert notification['to'] == 'sample_email@example.com'
assert notification['service'] == str(email_notification.service_id)
assert response.status_code == 200
assert notification['body'] == sample_email_template.content

22
tests/app/template/test_rest.py
View File

@@ -1,4 +1,3 @@
import pytest
import json
import random
import string
@@ -42,11 +41,17 @@ def test_should_create_a_new_template_for_a_service(
assert json_resp['data']['service'] == str(sample_service.id)
assert json_resp['data']['id']
assert json_resp['data']['version'] == 1
assert json_resp['data']['process_type'] == 'normal'
assert json_resp['data']['created_by'] == str(sample_user.id)
if subject:
assert json_resp['data']['subject'] == 'subject'
else:
assert not json_resp['data']['subject']
template = Template.query.get(json_resp['data']['id'])
from app.schemas import template_schema
assert sorted(json_resp['data']) == sorted(template_schema.dump(template).data)
def test_should_be_error_if_service_does_not_exist_on_create(client, sample_user, fake_uuid):
data = {
@@ -335,6 +340,7 @@ def test_should_get_a_single_template(
assert response.status_code == 200
assert data['content'] == content
assert data['subject'] == subject
assert data['process_type'] == 'normal'
@pytest.mark.parametrize(
@@ -516,3 +522,17 @@ def test_update_does_not_create_new_version_when_there_is_no_change(client, samp
template = dao_get_template_by_id(sample_template.id)
assert template.version == 1
def test_update_set_process_type_on_template(client, sample_template):
auth_header = create_authorization_header()
data = {
'process_type': 'priority'
}
resp = client.post('/service/{}/template/{}'.format(sample_template.service_id, sample_template.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 200
template = dao_get_template_by_id(sample_template.id)
assert template.process_type == 'priority'

5
tests/app/v2/notifications/test_get_notifications.py
View File

@@ -164,10 +164,11 @@ def test_get_notification_by_id_nonexistent_id(client, sample_notification):
}
def test_get_notification_by_id_invalid_id(client, sample_notification):
@pytest.mark.parametrize("id", ["1234-badly-formatted-id-7890", "0"])
def test_get_notification_by_id_invalid_id(client, sample_notification, id):
auth_header = create_authorization_header(service_id=sample_notification.service_id)
response = client.get(
path='/v2/notifications/1234-badly-formatted-id-7890',
path='/v2/notifications/{}'.format(id),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 404

208
tests/app/v2/notifications/test_post_notifications.py
View File

@@ -1,9 +1,9 @@
import uuid
import pytest
from flask import json
from app.models import Notification
from tests import create_authorization_header
from tests.app.conftest import sample_template as create_sample_template
@pytest.mark.parametrize("reference", [None, "reference_from_client"])
@@ -43,71 +43,77 @@ def test_post_sms_notification_returns_201(notify_api, sample_template_with_plac
assert mocked.called
def test_post_sms_notification_returns_404_and_missing_template(notify_api, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'phone_number': '+447700900855',
'template_id': str(uuid.uuid4())
}
auth_header = create_authorization_header(service_id=sample_service.id)
@pytest.mark.parametrize("notification_type, key_send_to, send_to",
[("sms", "phone_number", "+447700900855"),
("email", "email_address", "sample@email.com")])
def test_post_sms_notification_returns_404_and_missing_template(client, sample_service,
notification_type, key_send_to, send_to):
data = {
key_send_to: send_to,
'template_id': str(uuid.uuid4())
}
auth_header = create_authorization_header(service_id=sample_service.id)
response = client.post(
path='/v2/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
response = client.post(
path='/v2/notifications/{}'.format(notification_type),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 400
assert response.headers['Content-type'] == 'application/json'
assert response.status_code == 400
assert response.headers['Content-type'] == 'application/json'
error_json = json.loads(response.get_data(as_text=True))
assert error_json['status_code'] == 400
assert error_json['errors'] == [{"error": "BadRequestError",
"message": 'Template not found'}]
error_json = json.loads(response.get_data(as_text=True))
assert error_json['status_code'] == 400
assert error_json['errors'] == [{"error": "BadRequestError",
"message": 'Template not found'}]
def test_post_sms_notification_returns_403_and_well_formed_auth_error(notify_api, sample_template):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'phone_number': '+447700900855',
'template_id': str(sample_template.id)
}
@pytest.mark.parametrize("notification_type, key_send_to, send_to",
[("sms", "phone_number", "+447700900855"),
("email", "email_address", "sample@email.com")])
def test_post_notification_returns_403_and_well_formed_auth_error(client, sample_template,
notification_type, key_send_to, send_to):
data = {
key_send_to: send_to,
'template_id': str(sample_template.id)
}
response = client.post(
path='/v2/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json')])
response = client.post(
path='/v2/notifications/{}'.format(notification_type),
data=json.dumps(data),
headers=[('Content-Type', 'application/json')])
assert response.status_code == 401
assert response.headers['Content-type'] == 'application/json'
error_resp = json.loads(response.get_data(as_text=True))
assert error_resp['status_code'] == 401
assert error_resp['errors'] == [{'error': "AuthError",
'message': 'Unauthorized, authentication token must be provided'}]
assert response.status_code == 401
assert response.headers['Content-type'] == 'application/json'
error_resp = json.loads(response.get_data(as_text=True))
assert error_resp['status_code'] == 401
assert error_resp['errors'] == [{'error': "AuthError",
'message': 'Unauthorized, authentication token must be provided'}]
def test_post_sms_notification_returns_400_and_for_schema_problems(notify_api, sample_template):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'phone_number': '+447700900855',
'template': str(sample_template.id)
}
auth_header = create_authorization_header(service_id=sample_template.service_id)
@pytest.mark.parametrize("notification_type, key_send_to, send_to",
[("sms", "phone_number", "+447700900855"),
("email", "email_address", "sample@email.com")])
def test_notification_returns_400_and_for_schema_problems(client, sample_template, notification_type, key_send_to,
send_to):
data = {
key_send_to: send_to,
'template': str(sample_template.id)
}
auth_header = create_authorization_header(service_id=sample_template.service_id)
response = client.post(
path='/v2/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
response = client.post(
path='/v2/notifications/{}'.format(notification_type),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 400
assert response.headers['Content-type'] == 'application/json'
error_resp = json.loads(response.get_data(as_text=True))
assert error_resp['status_code'] == 400
assert error_resp['errors'] == [{'error': 'ValidationError',
'message': "template_id is a required property"
}]
assert response.status_code == 400
assert response.headers['Content-type'] == 'application/json'
error_resp = json.loads(response.get_data(as_text=True))
assert error_resp['status_code'] == 400
assert error_resp['errors'] == [{'error': 'ValidationError',
'message': "template_id is a required property"
}]
@pytest.mark.parametrize("reference", [None, "reference_from_client"])
@@ -131,9 +137,9 @@ def test_post_email_notification_returns_201(client, sample_email_template_with_
assert resp_json['id'] == str(notification.id)
assert resp_json['reference'] == reference
assert notification.reference is None
assert resp_json['content']['body'] == sample_email_template_with_placeholders.content\
assert resp_json['content']['body'] == sample_email_template_with_placeholders.content \
.replace('((name))', 'Bob').replace('GOV.UK', u'GOV.\u200bUK')
assert resp_json['content']['subject'] == sample_email_template_with_placeholders.subject\
assert resp_json['content']['subject'] == sample_email_template_with_placeholders.subject \
.replace('((name))', 'Bob')
assert resp_json['content']['from_email'] == sample_email_template_with_placeholders.service.email_from
assert 'v2/notifications/{}'.format(notification.id) in resp_json['uri']
@@ -145,24 +151,76 @@ def test_post_email_notification_returns_201(client, sample_email_template_with_
assert mocked.called
def test_post_email_notification_returns_404_and_missing_template(notify_api, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
"email_address": sample_service.users[0].email_address,
'template_id': str(uuid.uuid4())
}
auth_header = create_authorization_header(service_id=sample_service.id)
@pytest.mark.parametrize('recipient, notification_type', [
('simulate-delivered@notifications.service.gov.uk', 'email'),
('simulate-delivered-2@notifications.service.gov.uk', 'email'),
('simulate-delivered-3@notifications.service.gov.uk', 'email'),
('07700 900000', 'sms'),
('07700 900111', 'sms'),
('07700 900222', 'sms')
])
def test_should_not_persist_or_send_notification_if_simulated_recipient(
client,
recipient,
notification_type,
sample_email_template,
sample_template,
mocker):
apply_async = mocker.patch('app.celery.provider_tasks.deliver_{}.apply_async'.format(notification_type))
response = client.post(
path='/v2/notifications/email',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
if notification_type == 'sms':
data = {
'phone_number': recipient,
'template_id': str(sample_template.id)
}
else:
data = {
'email_address': recipient,
'template_id': str(sample_email_template.id)
}
assert response.status_code == 400
assert response.headers['Content-type'] == 'application/json'
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
error_json = json.loads(response.get_data(as_text=True))
assert error_json['status_code'] == 400
assert error_json['errors'] == [{"error": "BadRequestError",
"message": 'Template not found'}]
response = client.post(
path='/v2/notifications/{}'.format(notification_type),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 201
apply_async.assert_not_called()
assert Notification.query.count() == 0
@pytest.mark.parametrize("notification_type, key_send_to, send_to",
[("sms", "phone_number", "07700 900 855"),
("email", "email_address", "sample@email.com")])
def test_send_notification_uses_priority_queue_when_template_is_marked_as_priority(client, notify_db,
notify_db_session,
mocker,
notification_type,
key_send_to,
send_to):
sample = create_sample_template(
notify_db,
notify_db_session,
template_type=notification_type,
process_type='priority'
)
mocked = mocker.patch('app.celery.provider_tasks.deliver_{}.apply_async'.format(notification_type))
data = {
key_send_to: send_to,
'template_id': str(sample.id)
}
auth_header = create_authorization_header(service_id=sample.service_id)
response = client.post(
path='/v2/notifications/{}'.format(notification_type),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
notification_id = json.loads(response.data)['id']
assert response.status_code == 201
mocked.assert_called_once_with([notification_id], queue='notify')