mirror of
https://github.com/GSA/notifications-api.git
synced 2026-01-03 01:40:44 -05:00
227 lines
10 KiB
Python
227 lines
10 KiB
Python
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"])
|
|
def test_post_sms_notification_returns_201(notify_api, sample_template_with_placeholders, mocker, reference):
|
|
with notify_api.test_request_context():
|
|
with notify_api.test_client() as client:
|
|
mocked = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
|
|
data = {
|
|
'phone_number': '+447700900855',
|
|
'template_id': str(sample_template_with_placeholders.id),
|
|
'personalisation': {' Name': 'Jo'}
|
|
}
|
|
if reference:
|
|
data.update({"reference": reference})
|
|
auth_header = create_authorization_header(service_id=sample_template_with_placeholders.service_id)
|
|
|
|
response = client.post(
|
|
path='/v2/notifications/sms',
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
assert response.status_code == 201
|
|
resp_json = json.loads(response.get_data(as_text=True))
|
|
notifications = Notification.query.all()
|
|
assert len(notifications) == 1
|
|
notification_id = notifications[0].id
|
|
assert resp_json['id'] == str(notification_id)
|
|
assert resp_json['reference'] == reference
|
|
assert resp_json['content']['body'] == sample_template_with_placeholders.content.replace("(( Name))", "Jo")
|
|
# conftest fixture service does not have a sms sender, use config default
|
|
assert resp_json['content']['from_number'] == notify_api.config["FROM_NUMBER"]
|
|
assert 'v2/notifications/{}'.format(notification_id) in resp_json['uri']
|
|
assert resp_json['template']['id'] == str(sample_template_with_placeholders.id)
|
|
assert resp_json['template']['version'] == sample_template_with_placeholders.version
|
|
assert 'services/{}/templates/{}'.format(sample_template_with_placeholders.service_id,
|
|
sample_template_with_placeholders.id) \
|
|
in resp_json['template']['uri']
|
|
assert mocked.called
|
|
|
|
|
|
@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/{}'.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_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'}]
|
|
|
|
|
|
@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/{}'.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'}]
|
|
|
|
|
|
@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/{}'.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"
|
|
}]
|
|
|
|
|
|
@pytest.mark.parametrize("reference", [None, "reference_from_client"])
|
|
def test_post_email_notification_returns_201(client, sample_email_template_with_placeholders, mocker, reference):
|
|
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
|
|
data = {
|
|
"email_address": sample_email_template_with_placeholders.service.users[0].email_address,
|
|
"template_id": sample_email_template_with_placeholders.id,
|
|
"personalisation": {"name": "Bob"}
|
|
}
|
|
if reference:
|
|
data.update({"reference": reference})
|
|
auth_header = create_authorization_header(service_id=sample_email_template_with_placeholders.service_id)
|
|
response = client.post(
|
|
path="v2/notifications/email",
|
|
data=json.dumps(data),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
assert response.status_code == 201
|
|
resp_json = json.loads(response.get_data(as_text=True))
|
|
notification = Notification.query.first()
|
|
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 \
|
|
.replace('((name))', 'Bob').replace('GOV.UK', u'GOV.\u200bUK')
|
|
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']
|
|
assert resp_json['template']['id'] == str(sample_email_template_with_placeholders.id)
|
|
assert resp_json['template']['version'] == sample_email_template_with_placeholders.version
|
|
assert 'services/{}/templates/{}'.format(str(sample_email_template_with_placeholders.service_id),
|
|
str(sample_email_template_with_placeholders.id)) \
|
|
in resp_json['template']['uri']
|
|
assert mocked.called
|
|
|
|
|
|
@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))
|
|
|
|
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)
|
|
}
|
|
|
|
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
|
|
|
|
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')
|