mirror of
https://github.com/GSA/notifications-api.git
synced 2026-07-17 21:20:59 -04:00
1c154c4113
This is being done for the PaaS migration to allow us to keep traffic coming in whilst we migrate the database. uses the same tasks as the CSV uploaded notifications. Simple changes to not persist the notification, and call into a different task.
225 lines
10 KiB
Python
225 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
|
|
from freezegun import freeze_time
|
|
|
|
|
|
@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.tasks.send_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) == 0
|
|
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(resp_json['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.tasks.send_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))
|
|
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(resp_json['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")])
|
|
@freeze_time("2016-01-01 11:00:00.000000")
|
|
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):
|
|
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
|
|
sample = create_sample_template(
|
|
notify_db,
|
|
notify_db_session,
|
|
template_type=notification_type,
|
|
process_type='priority'
|
|
)
|
|
mocked = mocker.patch('app.celery.tasks.send_{}.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']
|
|
print(response.data)
|
|
assert response.status_code == 201
|
|
mocked.assert_called_once_with(
|
|
(str(sample.service_id), notification_id, 'something_encrypted', '2016-01-01T11:00:00.000000Z'), queue='notify'
|
|
)
|