mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-22 08:21:13 -05:00
c9dafa0372
Notifications don't have a `scheduled_for` attribute, and the test function deleted isn't used anywhere.
724 lines
30 KiB
Python
724 lines
30 KiB
Python
import datetime
|
|
|
|
import pytest
|
|
from flask import json, url_for
|
|
|
|
from app.utils import DATETIME_FORMAT
|
|
from tests import create_service_authorization_header
|
|
from tests.app.db import create_notification, create_template
|
|
|
|
|
|
@pytest.mark.parametrize('billable_units, provider', [
|
|
(1, 'mmg'),
|
|
(0, 'mmg'),
|
|
(1, None)
|
|
])
|
|
def test_get_notification_by_id_returns_200(
|
|
client, billable_units, provider, sample_template
|
|
):
|
|
sample_notification = create_notification(
|
|
template=sample_template,
|
|
billable_units=billable_units,
|
|
sent_by=provider,
|
|
)
|
|
|
|
# another
|
|
create_notification(
|
|
template=sample_template,
|
|
billable_units=billable_units,
|
|
sent_by=provider,
|
|
)
|
|
|
|
auth_header = create_service_authorization_header(service_id=sample_notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications/{}'.format(sample_notification.id),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers['Content-type'] == 'application/json'
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
expected_template_response = {
|
|
'id': '{}'.format(sample_notification.serialize()['template']['id']),
|
|
'version': sample_notification.serialize()['template']['version'],
|
|
'uri': sample_notification.serialize()['template']['uri']
|
|
}
|
|
|
|
expected_response = {
|
|
'id': '{}'.format(sample_notification.id),
|
|
'reference': None,
|
|
'email_address': None,
|
|
'phone_number': '{}'.format(sample_notification.to),
|
|
'line_1': None,
|
|
'line_2': None,
|
|
'line_3': None,
|
|
'line_4': None,
|
|
'line_5': None,
|
|
'line_6': None,
|
|
'postcode': None,
|
|
'type': '{}'.format(sample_notification.notification_type),
|
|
'status': '{}'.format(sample_notification.status),
|
|
'template': expected_template_response,
|
|
'created_at': sample_notification.created_at.strftime(DATETIME_FORMAT),
|
|
'created_by_name': None,
|
|
'body': sample_notification.template.content,
|
|
"subject": None,
|
|
'sent_at': sample_notification.sent_at,
|
|
'completed_at': sample_notification.completed_at(),
|
|
'scheduled_for': None,
|
|
'postage': None,
|
|
}
|
|
|
|
assert json_response == expected_response
|
|
|
|
|
|
def test_get_notification_by_id_with_placeholders_returns_200(
|
|
client, sample_email_template_with_placeholders
|
|
):
|
|
sample_notification = create_notification(
|
|
template=sample_email_template_with_placeholders,
|
|
personalisation={"name": "Bob"}
|
|
)
|
|
|
|
auth_header = create_service_authorization_header(service_id=sample_notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications/{}'.format(sample_notification.id),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers['Content-type'] == 'application/json'
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
expected_template_response = {
|
|
'id': '{}'.format(sample_notification.serialize()['template']['id']),
|
|
'version': sample_notification.serialize()['template']['version'],
|
|
'uri': sample_notification.serialize()['template']['uri']
|
|
}
|
|
|
|
expected_response = {
|
|
'id': '{}'.format(sample_notification.id),
|
|
'reference': None,
|
|
'email_address': '{}'.format(sample_notification.to),
|
|
'phone_number': None,
|
|
'line_1': None,
|
|
'line_2': None,
|
|
'line_3': None,
|
|
'line_4': None,
|
|
'line_5': None,
|
|
'line_6': None,
|
|
'postcode': None,
|
|
'type': '{}'.format(sample_notification.notification_type),
|
|
'status': '{}'.format(sample_notification.status),
|
|
'template': expected_template_response,
|
|
'created_at': sample_notification.created_at.strftime(DATETIME_FORMAT),
|
|
'created_by_name': None,
|
|
'body': "Hello Bob\nThis is an email from GOV.UK",
|
|
"subject": "Bob",
|
|
'sent_at': sample_notification.sent_at,
|
|
'completed_at': sample_notification.completed_at(),
|
|
'scheduled_for': None,
|
|
'postage': None,
|
|
}
|
|
|
|
assert json_response == expected_response
|
|
|
|
|
|
def test_get_notification_by_reference_returns_200(client, sample_template):
|
|
sample_notification_with_reference = create_notification(template=sample_template,
|
|
client_reference='some-client-reference')
|
|
|
|
auth_header = create_service_authorization_header(service_id=sample_notification_with_reference.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications?reference={}'.format(sample_notification_with_reference.client_reference),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers['Content-type'] == 'application/json'
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
assert len(json_response['notifications']) == 1
|
|
|
|
assert json_response['notifications'][0]['id'] == str(sample_notification_with_reference.id)
|
|
assert json_response['notifications'][0]['reference'] == "some-client-reference"
|
|
|
|
|
|
def test_get_notification_by_id_returns_created_by_name_if_notification_created_by_id(
|
|
client,
|
|
sample_user,
|
|
sample_template,
|
|
):
|
|
sms_notification = create_notification(template=sample_template)
|
|
sms_notification.created_by_id = sample_user.id
|
|
|
|
auth_header = create_service_authorization_header(service_id=sms_notification.service_id)
|
|
response = client.get(
|
|
path=url_for('v2_notifications.get_notification_by_id', notification_id=sms_notification.id),
|
|
headers=[('Content-Type', 'application/json'), auth_header]
|
|
)
|
|
|
|
json_response = response.get_json()
|
|
assert json_response['created_by_name'] == 'Test User'
|
|
|
|
|
|
def test_get_notification_by_reference_nonexistent_reference_returns_no_notifications(client, sample_service):
|
|
auth_header = create_service_authorization_header(service_id=sample_service.id)
|
|
response = client.get(
|
|
path='/v2/notifications?reference={}'.format('nonexistent-reference'),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers['Content-type'] == "application/json"
|
|
assert len(json_response['notifications']) == 0
|
|
|
|
|
|
def test_get_notification_by_id_nonexistent_id(client, sample_notification):
|
|
auth_header = create_service_authorization_header(service_id=sample_notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications/dd4b8b9d-d414-4a83-9256-580046bf18f9',
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
assert response.status_code == 404
|
|
assert response.headers['Content-type'] == 'application/json'
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
assert json_response == {
|
|
"errors": [
|
|
{
|
|
"error": "NoResultFound",
|
|
"message": "No result found"
|
|
}
|
|
],
|
|
"status_code": 404
|
|
}
|
|
|
|
|
|
@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_service_authorization_header(service_id=sample_notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications/{}'.format(id),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
assert response.status_code == 400
|
|
assert response.headers['Content-type'] == 'application/json'
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
assert json_response == {"errors": [
|
|
{"error": "ValidationError",
|
|
"message": "notification_id is not a valid UUID"
|
|
}],
|
|
"status_code": 400}
|
|
|
|
|
|
@pytest.mark.parametrize('created_at_month, postage, estimated_delivery', [
|
|
(12, 'second', '2000-12-06T16:00:00.000000Z'), # 4pm GMT in winter
|
|
(6, 'second', '2000-06-05T15:00:00.000000Z'), # 4pm BST in summer
|
|
|
|
(12, 'first', '2000-12-05T16:00:00.000000Z'), # 4pm GMT in winter
|
|
(6, 'first', '2000-06-03T15:00:00.000000Z'), # 4pm BST in summer (two days before 2nd class due to weekends)
|
|
])
|
|
def test_get_notification_adds_delivery_estimate_for_letters(
|
|
client,
|
|
sample_letter_notification,
|
|
created_at_month,
|
|
postage,
|
|
estimated_delivery,
|
|
):
|
|
sample_letter_notification.created_at = datetime.date(2000, created_at_month, 1)
|
|
sample_letter_notification.postage = postage
|
|
|
|
auth_header = create_service_authorization_header(service_id=sample_letter_notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications/{}'.format(sample_letter_notification.id),
|
|
headers=[('Content-Type', 'application/json'), auth_header]
|
|
)
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 200
|
|
assert json_response['postage'] == postage
|
|
assert json_response['estimated_delivery'] == estimated_delivery
|
|
|
|
|
|
@pytest.mark.parametrize('template_type', ['sms', 'email'])
|
|
def test_get_notification_doesnt_have_delivery_estimate_for_non_letters(client, sample_service, template_type):
|
|
template = create_template(service=sample_service, template_type=template_type)
|
|
mocked_notification = create_notification(template=template)
|
|
|
|
auth_header = create_service_authorization_header(service_id=mocked_notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications/{}'.format(mocked_notification.id),
|
|
headers=[('Content-Type', 'application/json'), auth_header]
|
|
)
|
|
assert response.status_code == 200
|
|
assert 'estimated_delivery' not in json.loads(response.get_data(as_text=True))
|
|
|
|
|
|
def test_get_all_notifications_except_job_notifications_returns_200(client, sample_template, sample_job):
|
|
create_notification(template=sample_template, job=sample_job) # should not return this job notification
|
|
notifications = [create_notification(template=sample_template) for _ in range(2)]
|
|
notification = notifications[-1]
|
|
|
|
auth_header = create_service_authorization_header(service_id=notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications',
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers['Content-type'] == "application/json"
|
|
assert json_response['links']['current'].endswith("/v2/notifications")
|
|
assert 'next' in json_response['links'].keys()
|
|
assert len(json_response['notifications']) == 2
|
|
|
|
assert json_response['notifications'][0]['id'] == str(notification.id)
|
|
assert json_response['notifications'][0]['status'] == "created"
|
|
assert json_response['notifications'][0]['template'] == {
|
|
'id': str(notification.template.id),
|
|
'uri': notification.template.get_link(),
|
|
'version': 1
|
|
}
|
|
assert json_response['notifications'][0]['phone_number'] == "+447700900855"
|
|
assert json_response['notifications'][0]['type'] == "sms"
|
|
assert not json_response['notifications'][0]['scheduled_for']
|
|
|
|
|
|
def test_get_all_notifications_with_include_jobs_arg_returns_200(
|
|
client, sample_template, sample_job
|
|
):
|
|
notifications = [
|
|
create_notification(template=sample_template, job=sample_job),
|
|
create_notification(template=sample_template)
|
|
]
|
|
notification = notifications[-1]
|
|
|
|
auth_header = create_service_authorization_header(service_id=notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications?include_jobs=true',
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 200
|
|
assert json_response['links']['current'].endswith("/v2/notifications?include_jobs=true")
|
|
assert 'next' in json_response['links'].keys()
|
|
assert len(json_response['notifications']) == 2
|
|
|
|
assert json_response['notifications'][0]['id'] == str(notification.id)
|
|
assert json_response['notifications'][0]['status'] == notification.status
|
|
assert json_response['notifications'][0]['phone_number'] == notification.to
|
|
assert json_response['notifications'][0]['type'] == notification.template.template_type
|
|
assert not json_response['notifications'][0]['scheduled_for']
|
|
|
|
|
|
def test_get_all_notifications_no_notifications_if_no_notifications(client, sample_service):
|
|
auth_header = create_service_authorization_header(service_id=sample_service.id)
|
|
response = client.get(
|
|
path='/v2/notifications',
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers['Content-type'] == "application/json"
|
|
assert json_response['links']['current'].endswith("/v2/notifications")
|
|
assert 'next' not in json_response['links'].keys()
|
|
assert len(json_response['notifications']) == 0
|
|
|
|
|
|
def test_get_all_notifications_filter_by_template_type(client, sample_service):
|
|
email_template = create_template(service=sample_service, template_type="email")
|
|
sms_template = create_template(service=sample_service, template_type="sms")
|
|
|
|
notification = create_notification(template=email_template, to_field="don.draper@scdp.biz")
|
|
create_notification(template=sms_template)
|
|
|
|
auth_header = create_service_authorization_header(service_id=notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications?template_type=email',
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers['Content-type'] == "application/json"
|
|
assert json_response['links']['current'].endswith("/v2/notifications?template_type=email")
|
|
assert 'next' in json_response['links'].keys()
|
|
assert len(json_response['notifications']) == 1
|
|
|
|
assert json_response['notifications'][0]['id'] == str(notification.id)
|
|
assert json_response['notifications'][0]['status'] == "created"
|
|
assert json_response['notifications'][0]['template'] == {
|
|
'id': str(email_template.id),
|
|
'uri': notification.template.get_link(),
|
|
'version': 1
|
|
}
|
|
assert json_response['notifications'][0]['email_address'] == "don.draper@scdp.biz"
|
|
assert json_response['notifications'][0]['type'] == "email"
|
|
|
|
|
|
def test_get_all_notifications_filter_by_template_type_invalid_template_type(client, sample_notification):
|
|
auth_header = create_service_authorization_header(service_id=sample_notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications?template_type=orange',
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 400
|
|
assert response.headers['Content-type'] == "application/json"
|
|
|
|
assert json_response['status_code'] == 400
|
|
assert len(json_response['errors']) == 1
|
|
assert json_response['errors'][0]['message'] == "template_type orange is not one of [sms, email, letter]"
|
|
|
|
|
|
def test_get_all_notifications_filter_by_single_status(client, sample_template):
|
|
notification = create_notification(template=sample_template, status="pending")
|
|
create_notification(template=sample_template)
|
|
|
|
auth_header = create_service_authorization_header(service_id=notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications?status=pending',
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers['Content-type'] == "application/json"
|
|
assert json_response['links']['current'].endswith("/v2/notifications?status=pending")
|
|
assert 'next' in json_response['links'].keys()
|
|
assert len(json_response['notifications']) == 1
|
|
|
|
assert json_response['notifications'][0]['id'] == str(notification.id)
|
|
assert json_response['notifications'][0]['status'] == "pending"
|
|
|
|
|
|
def test_get_all_notifications_filter_by_status_invalid_status(client, sample_notification):
|
|
auth_header = create_service_authorization_header(service_id=sample_notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications?status=elephant',
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 400
|
|
assert response.headers['Content-type'] == "application/json"
|
|
|
|
assert json_response['status_code'] == 400
|
|
assert len(json_response['errors']) == 1
|
|
assert json_response['errors'][0]['message'] == "status elephant is not one of [cancelled, created, sending, " \
|
|
"sent, delivered, pending, failed, technical-failure, temporary-failure, permanent-failure, " \
|
|
"pending-virus-check, validation-failed, virus-scan-failed, returned-letter, accepted, received]"
|
|
|
|
|
|
def test_get_all_notifications_filter_by_multiple_statuses(client, sample_template):
|
|
notifications = [
|
|
create_notification(template=sample_template, status=_status)
|
|
for _status in ["created", "pending", "sending"]
|
|
]
|
|
failed_notification = create_notification(template=sample_template, status="permanent-failure")
|
|
|
|
auth_header = create_service_authorization_header(service_id=notifications[0].service_id)
|
|
response = client.get(
|
|
path='/v2/notifications?status=created&status=pending&status=sending',
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers['Content-type'] == "application/json"
|
|
assert json_response['links']['current'].endswith("/v2/notifications?status=created&status=pending&status=sending")
|
|
assert 'next' in json_response['links'].keys()
|
|
assert len(json_response['notifications']) == 3
|
|
|
|
returned_notification_ids = [_n['id'] for _n in json_response['notifications']]
|
|
for _id in [_notification.id for _notification in notifications]:
|
|
assert str(_id) in returned_notification_ids
|
|
|
|
assert failed_notification.id not in returned_notification_ids
|
|
|
|
|
|
def test_get_all_notifications_filter_by_failed_status(client, sample_template):
|
|
created_notification = create_notification(template=sample_template, status="created")
|
|
failed_notifications = [
|
|
create_notification(template=sample_template, status=_status)
|
|
for _status in ["technical-failure", "temporary-failure", "permanent-failure"]
|
|
]
|
|
|
|
auth_header = create_service_authorization_header(service_id=created_notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications?status=failed',
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers['Content-type'] == "application/json"
|
|
assert json_response['links']['current'].endswith("/v2/notifications?status=failed")
|
|
assert 'next' in json_response['links'].keys()
|
|
assert len(json_response['notifications']) == 3
|
|
|
|
returned_notification_ids = [n['id'] for n in json_response['notifications']]
|
|
for _id in [_notification.id for _notification in failed_notifications]:
|
|
assert str(_id) in returned_notification_ids
|
|
|
|
assert created_notification.id not in returned_notification_ids
|
|
|
|
|
|
def test_get_all_notifications_filter_by_id(client, sample_template):
|
|
older_notification = create_notification(template=sample_template)
|
|
newer_notification = create_notification(template=sample_template)
|
|
|
|
auth_header = create_service_authorization_header(service_id=newer_notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications?older_than={}'.format(newer_notification.id),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers['Content-type'] == "application/json"
|
|
assert json_response['links']['current'].endswith("/v2/notifications?older_than={}".format(newer_notification.id))
|
|
assert 'next' in json_response['links'].keys()
|
|
assert len(json_response['notifications']) == 1
|
|
|
|
assert json_response['notifications'][0]['id'] == str(older_notification.id)
|
|
|
|
|
|
def test_get_all_notifications_filter_by_id_invalid_id(client, sample_notification):
|
|
auth_header = create_service_authorization_header(service_id=sample_notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications?older_than=1234-badly-formatted-id-7890',
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert json_response['status_code'] == 400
|
|
assert len(json_response['errors']) == 1
|
|
assert json_response['errors'][0]['message'] == "older_than is not a valid UUID"
|
|
|
|
|
|
def test_get_all_notifications_filter_by_id_no_notifications_if_nonexistent_id(client, sample_template):
|
|
notification = create_notification(template=sample_template)
|
|
|
|
auth_header = create_service_authorization_header(service_id=notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications?older_than=dd4b8b9d-d414-4a83-9256-580046bf18f9',
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers['Content-type'] == "application/json"
|
|
assert json_response['links']['current'].endswith(
|
|
"/v2/notifications?older_than=dd4b8b9d-d414-4a83-9256-580046bf18f9")
|
|
assert 'next' not in json_response['links'].keys()
|
|
assert len(json_response['notifications']) == 0
|
|
|
|
|
|
def test_get_all_notifications_filter_by_id_no_notifications_if_last_notification(client, sample_template):
|
|
notification = create_notification(template=sample_template)
|
|
|
|
auth_header = create_service_authorization_header(service_id=notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications?older_than={}'.format(notification.id),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers['Content-type'] == "application/json"
|
|
assert json_response['links']['current'].endswith("/v2/notifications?older_than={}".format(notification.id))
|
|
assert 'next' not in json_response['links'].keys()
|
|
assert len(json_response['notifications']) == 0
|
|
|
|
|
|
def test_get_all_notifications_filter_multiple_query_parameters(client, sample_email_template):
|
|
# this is the notification we are looking for
|
|
older_notification = create_notification(
|
|
template=sample_email_template, status="pending")
|
|
|
|
# wrong status
|
|
create_notification(template=sample_email_template)
|
|
wrong_template = create_template(sample_email_template.service, template_type='sms')
|
|
# wrong template
|
|
create_notification(template=wrong_template, status="pending")
|
|
|
|
# we only want notifications created before this one
|
|
newer_notification = create_notification(template=sample_email_template)
|
|
|
|
# this notification was created too recently
|
|
create_notification(template=sample_email_template, status="pending")
|
|
|
|
auth_header = create_service_authorization_header(service_id=newer_notification.service_id)
|
|
response = client.get(
|
|
path='/v2/notifications?status=pending&template_type=email&older_than={}'.format(newer_notification.id),
|
|
headers=[('Content-Type', 'application/json'), auth_header])
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers['Content-type'] == "application/json"
|
|
# query parameters aren't returned in order
|
|
for url_part in [
|
|
"/v2/notifications?",
|
|
"template_type=email",
|
|
"status=pending",
|
|
"older_than={}".format(newer_notification.id)
|
|
]:
|
|
assert url_part in json_response['links']['current']
|
|
|
|
assert 'next' in json_response['links'].keys()
|
|
assert len(json_response['notifications']) == 1
|
|
|
|
assert json_response['notifications'][0]['id'] == str(older_notification.id)
|
|
|
|
|
|
def test_get_all_notifications_renames_letter_statuses(
|
|
client,
|
|
sample_letter_notification,
|
|
sample_notification,
|
|
sample_email_notification,
|
|
):
|
|
auth_header = create_service_authorization_header(service_id=sample_letter_notification.service_id)
|
|
response = client.get(
|
|
path=url_for('v2_notifications.get_notifications'),
|
|
headers=[('Content-Type', 'application/json'), auth_header]
|
|
)
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 200
|
|
|
|
for noti in json_response['notifications']:
|
|
if noti['type'] == 'sms' or noti['type'] == 'email':
|
|
assert noti['status'] == 'created'
|
|
elif noti['type'] == 'letter':
|
|
assert noti['status'] == 'accepted'
|
|
else:
|
|
pytest.fail()
|
|
|
|
|
|
@pytest.mark.parametrize('db_status,expected_status', [
|
|
('created', 'accepted'),
|
|
('sending', 'accepted'),
|
|
('delivered', 'received'),
|
|
('pending', 'pending'),
|
|
('technical-failure', 'technical-failure')
|
|
])
|
|
def test_get_notifications_renames_letter_statuses(client, sample_letter_template, db_status, expected_status):
|
|
letter_noti = create_notification(
|
|
sample_letter_template,
|
|
status=db_status,
|
|
personalisation={'address_line_1': 'Mr Foo', 'address_line_2': '1 Bar Street', 'postcode': 'N1'}
|
|
)
|
|
auth_header = create_service_authorization_header(service_id=letter_noti.service_id)
|
|
response = client.get(
|
|
path=url_for('v2_notifications.get_notification_by_id', notification_id=letter_noti.id),
|
|
headers=[('Content-Type', 'application/json'), auth_header]
|
|
)
|
|
|
|
json_response = json.loads(response.get_data(as_text=True))
|
|
assert response.status_code == 200
|
|
assert json_response['status'] == expected_status
|
|
|
|
|
|
def test_get_pdf_for_notification_returns_pdf_content(
|
|
client,
|
|
sample_letter_notification,
|
|
mocker,
|
|
):
|
|
mock_get_letter_pdf = mocker.patch(
|
|
'app.v2.notifications.get_notifications.get_letter_pdf_and_metadata', return_value=(b'foo', {
|
|
"message": "",
|
|
"invalid_pages": "",
|
|
"page_count": "1"
|
|
})
|
|
)
|
|
sample_letter_notification.status = 'created'
|
|
|
|
auth_header = create_service_authorization_header(service_id=sample_letter_notification.service_id)
|
|
response = client.get(
|
|
path=url_for('v2_notifications.get_pdf_for_notification', notification_id=sample_letter_notification.id),
|
|
headers=[('Content-Type', 'application/json'), auth_header]
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.get_data() == b'foo'
|
|
mock_get_letter_pdf.assert_called_once_with(sample_letter_notification)
|
|
|
|
|
|
def test_get_pdf_for_notification_returns_400_if_pdf_not_found(
|
|
client,
|
|
sample_letter_notification,
|
|
mocker,
|
|
):
|
|
# if no files are returned get_letter_pdf throws StopIteration as the iterator runs out
|
|
mock_get_letter_pdf = mocker.patch(
|
|
'app.v2.notifications.get_notifications.get_letter_pdf_and_metadata',
|
|
side_effect=(StopIteration, {})
|
|
)
|
|
sample_letter_notification.status = 'created'
|
|
|
|
auth_header = create_service_authorization_header(service_id=sample_letter_notification.service_id)
|
|
response = client.get(
|
|
path=url_for('v2_notifications.get_pdf_for_notification', notification_id=sample_letter_notification.id),
|
|
headers=[('Content-Type', 'application/json'), auth_header]
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json['errors'] == [{
|
|
'error': 'PDFNotReadyError',
|
|
'message': 'PDF not available yet, try again later'
|
|
}]
|
|
mock_get_letter_pdf.assert_called_once_with(sample_letter_notification)
|
|
|
|
|
|
@pytest.mark.parametrize('status, expected_message', [
|
|
('virus-scan-failed', 'File did not pass the virus scan'),
|
|
('technical-failure', 'PDF not available for letters in status technical-failure'),
|
|
])
|
|
def test_get_pdf_for_notification_only_returns_pdf_content_if_right_status(
|
|
client,
|
|
sample_letter_notification,
|
|
mocker,
|
|
status,
|
|
expected_message
|
|
):
|
|
mock_get_letter_pdf = mocker.patch(
|
|
'app.v2.notifications.get_notifications.get_letter_pdf_and_metadata', return_value=(b'foo', {
|
|
"message": "",
|
|
"invalid_pages": "",
|
|
"page_count": "1"
|
|
})
|
|
)
|
|
sample_letter_notification.status = status
|
|
|
|
auth_header = create_service_authorization_header(service_id=sample_letter_notification.service_id)
|
|
response = client.get(
|
|
path=url_for('v2_notifications.get_pdf_for_notification', notification_id=sample_letter_notification.id),
|
|
headers=[('Content-Type', 'application/json'), auth_header]
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json['errors'] == [{
|
|
'error': 'BadRequestError',
|
|
'message': expected_message
|
|
}]
|
|
assert mock_get_letter_pdf.called is False
|
|
|
|
|
|
def test_get_pdf_for_notification_fails_for_non_letters(client, sample_notification):
|
|
auth_header = create_service_authorization_header(service_id=sample_notification.service_id)
|
|
response = client.get(
|
|
path=url_for('v2_notifications.get_pdf_for_notification', notification_id=sample_notification.id),
|
|
headers=[('Content-Type', 'application/json'), auth_header]
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json['errors'] == [{'error': 'BadRequestError', 'message': 'Notification is not a letter'}]
|