darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-02 08:45:16 -05:00
Code Issues Packages Projects Releases Wiki Activity

remove v2

Browse Source
This commit is contained in:
Kenneth Kehl
2024-05-30 12:27:07 -07:00
parent 61cbec16a4
commit 22ffc56520
50 changed files with 64 additions and 5277 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -8,6 +8,7 @@ from freezegun import freeze_time
from sqlalchemy.exc import SQLAlchemyError
from app.enums import KeyType, NotificationType, ServicePermissionType, TemplateType
from app.errors import BadRequestError
from app.models import Notification, NotificationHistory
from app.notifications.process_notifications import (
create_content_for_notification,
@@ -16,7 +17,6 @@ from app.notifications.process_notifications import (
simulated_recipient,
)
from app.serialised_models import SerialisedTemplate
from app.v2.errors import BadRequestError
from notifications_utils.recipients import (
validate_and_format_email_address,
validate_and_format_phone_number,

2
tests/app/notifications/test_validators.py
View File

@@ -5,6 +5,7 @@ from freezegun import freeze_time
import app
from app.dao import templates_dao
from app.enums import KeyType, NotificationType, ServicePermissionType, TemplateType
from app.errors import BadRequestError, RateLimitError, TotalRequestsError
from app.notifications.process_notifications import create_content_for_notification
from app.notifications.sns_cert_validator import (
VALID_SNS_TOPICS,
@@ -35,7 +36,6 @@ from app.serialised_models import (
)
from app.service.utils import service_allowed_to_send_to
from app.utils import get_template_instance
from app.v2.errors import BadRequestError, RateLimitError, TotalRequestsError
from notifications_utils import SMS_CHAR_COUNT_LIMIT
from tests.app.db import (
create_api_key,

46
tests/app/public_contracts/test_GET_notification.py
View File

@@ -3,13 +3,9 @@ import pytest
from app.dao.api_key_dao import save_model_api_key
from app.enums import KeyType
from app.models import ApiKey
from app.v2.notifications.notification_schemas import (
get_notification_response,
get_notifications_response,
)
from tests import create_service_authorization_header
from . import return_json_from_response, validate, validate_v0
from . import return_json_from_response, validate_v0
def _get_notification(client, notification, url):
@@ -27,46 +23,6 @@ def _get_notification(client, notification, url):
return client.get(url, headers=[auth_header])
# v2
def test_get_v2_sms_contract(client, sample_notification, mocker):
mock_s3_personalisation = mocker.patch(
"app.v2.notifications.get_notifications.get_personalisation_from_s3"
)
mock_s3_personalisation.return_value = {}
response_json = return_json_from_response(
_get_notification(
client,
sample_notification,
"/v2/notifications/{}".format(sample_notification.id),
)
)
validate(response_json, get_notification_response)
def test_get_v2_email_contract(client, sample_email_notification, mocker):
mock_s3_personalisation = mocker.patch(
"app.v2.notifications.get_notifications.get_personalisation_from_s3"
)
mock_s3_personalisation.return_value = {}
response_json = return_json_from_response(
_get_notification(
client,
sample_email_notification,
"/v2/notifications/{}".format(sample_email_notification.id),
)
)
validate(response_json, get_notification_response)
def test_get_v2_notifications_contract(client, sample_notification):
response_json = return_json_from_response(
_get_notification(client, sample_notification, "/v2/notifications")
)
validate(response_json, get_notifications_response)
# v0

3
tests/app/service/send_notification/test_send_notification.py
View File

@@ -12,10 +12,9 @@ from app.dao.api_key_dao import save_model_api_key
from app.dao.services_dao import dao_update_service
from app.dao.templates_dao import dao_get_all_templates_for_service, dao_update_template
from app.enums import KeyType, NotificationType, TemplateProcessType, TemplateType
from app.errors import InvalidRequest
from app.errors import InvalidRequest, RateLimitError
from app.models import ApiKey, Notification, NotificationHistory, Template
from app.service.send_notification import send_one_off_notification
from app.v2.errors import RateLimitError
from notifications_utils import SMS_CHAR_COUNT_LIMIT
from tests import create_service_authorization_header
from tests.app.db import (

2
tests/app/service/send_notification/test_send_one_off_notification.py
View File

@@ -13,9 +13,9 @@ from app.enums import (
TemplateProcessType,
TemplateType,
)
from app.errors import BadRequestError
from app.models import Notification, ServiceGuestList
from app.service.send_notification import send_one_off_notification
from app.v2.errors import BadRequestError
from notifications_utils import SMS_CHAR_COUNT_LIMIT
from notifications_utils.recipients import InvalidPhoneError
from tests.app.db import (

8
tests/app/user/test_rest_verify.py
View File

@@ -499,10 +499,10 @@ def test_send_user_email_code_with_urlencoded_next_param(
mock_redis_get = mocker.patch("app.celery.scheduled_tasks.redis_store.raw_get")
mock_redis_get.return_value = "foo"
mock_s3_personalisation = mocker.patch(
"app.v2.notifications.get_notifications.get_personalisation_from_s3"
)
mock_s3_personalisation.return_value = {"name": "Bob"}
# mock_s3_personalisation = mocker.patch(
# "app.v2.notifications.get_notifications.get_personalisation_from_s3"
# )
# mock_s3_personalisation.return_value = {"name": "Bob"}
mocker.patch("app.celery.scheduled_tasks.redis_store.raw_set")

0
tests/app/v2/__init__.py
View File

0
tests/app/v2/inbound_sms/__init__.py
View File

222
tests/app/v2/inbound_sms/test_get_inbound_sms.py
View File

@@ -1,222 +0,0 @@
from flask import json, url_for
from tests import create_service_authorization_header
from tests.app.db import (
create_inbound_sms,
create_service_callback_api,
create_service_inbound_api,
)
def test_get_inbound_sms_returns_200(client, sample_service):
all_inbound_sms = [
create_inbound_sms(
service=sample_service, user_number="447700900111", content="Hi"
),
create_inbound_sms(service=sample_service, user_number="447700900112"),
create_inbound_sms(
service=sample_service, user_number="447700900111", content="Bye"
),
create_inbound_sms(service=sample_service, user_number="07700900113"),
]
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.get(
path="/v2/received-text-messages",
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))[
"received_text_messages"
]
reversed_all_inbound_sms = sorted(
all_inbound_sms, key=lambda sms: sms.created_at, reverse=True
)
expected_response = [i.serialize() for i in reversed_all_inbound_sms]
assert json_response == expected_response
def test_get_inbound_sms_returns_200_when_service_has_callbacks(client, sample_service):
create_service_inbound_api(
service=sample_service,
url="https://inbound.example.com",
)
create_service_callback_api(
service=sample_service,
url="https://inbound.example.com",
)
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.get(
path="/v2/received-text-messages",
headers=[("Content-Type", "application/json"), auth_header],
)
assert response.status_code == 200
def test_get_inbound_sms_generate_page_links(client, sample_service, mocker):
mocker.patch.dict(
"app.v2.inbound_sms.get_inbound_sms.current_app.config", {"API_PAGE_SIZE": 2}
)
all_inbound_sms = [
create_inbound_sms(
service=sample_service, user_number="447700900111", content="Hi"
),
create_inbound_sms(service=sample_service, user_number="447700900111"),
create_inbound_sms(
service=sample_service, user_number="447700900111", content="End"
),
]
reversed_inbound_sms = sorted(
all_inbound_sms, key=lambda sms: sms.created_at, reverse=True
)
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.get(
url_for("v2_inbound_sms.get_inbound_sms"),
headers=[("Content-Type", "application/json"), auth_header],
)
assert response.status_code == 200
json_response = json.loads(response.get_data(as_text=True))
expected_inbound_sms_list = [i.serialize() for i in reversed_inbound_sms[:2]]
assert json_response["received_text_messages"] == expected_inbound_sms_list
assert (
url_for("v2_inbound_sms.get_inbound_sms", _external=True)
== json_response["links"]["current"]
)
assert (
url_for(
"v2_inbound_sms.get_inbound_sms",
older_than=reversed_inbound_sms[1].id,
_external=True,
)
== json_response["links"]["next"]
)
def test_get_next_inbound_sms_will_get_correct_inbound_sms_list(
client, sample_service, mocker
):
mocker.patch.dict(
"app.v2.inbound_sms.get_inbound_sms.current_app.config", {"API_PAGE_SIZE": 2}
)
all_inbound_sms = [
create_inbound_sms(
service=sample_service, user_number="447700900111", content="1"
),
create_inbound_sms(
service=sample_service, user_number="447700900111", content="2"
),
create_inbound_sms(
service=sample_service, user_number="447700900111", content="3"
),
create_inbound_sms(
service=sample_service, user_number="447700900111", content="4"
),
]
reversed_inbound_sms = sorted(
all_inbound_sms, key=lambda sms: sms.created_at, reverse=True
)
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.get(
path=url_for(
"v2_inbound_sms.get_inbound_sms", older_than=reversed_inbound_sms[1].id
),
headers=[("Content-Type", "application/json"), auth_header],
)
assert response.status_code == 200
json_response = json.loads(response.get_data(as_text=True))
expected_inbound_sms_list = [i.serialize() for i in reversed_inbound_sms[2:]]
assert json_response["received_text_messages"] == expected_inbound_sms_list
assert (
url_for("v2_inbound_sms.get_inbound_sms", _external=True)
== json_response["links"]["current"]
)
assert (
url_for(
"v2_inbound_sms.get_inbound_sms",
older_than=reversed_inbound_sms[3].id,
_external=True,
)
== json_response["links"]["next"]
)
def test_get_next_inbound_sms_at_end_will_return_empty_inbound_sms_list(
client, sample_service, mocker
):
inbound_sms = create_inbound_sms(service=sample_service)
mocker.patch.dict(
"app.v2.inbound_sms.get_inbound_sms.current_app.config", {"API_PAGE_SIZE": 1}
)
auth_header = create_service_authorization_header(service_id=inbound_sms.service.id)
response = client.get(
path=url_for("v2_inbound_sms.get_inbound_sms", older_than=inbound_sms.id),
headers=[("Content-Type", "application/json"), auth_header],
)
assert response.status_code == 200
json_response = json.loads(response.get_data(as_text=True))
expected_inbound_sms_list = []
assert json_response["received_text_messages"] == expected_inbound_sms_list
assert (
url_for("v2_inbound_sms.get_inbound_sms", _external=True)
== json_response["links"]["current"]
)
assert "next" not in json_response["links"].keys()
def test_get_inbound_sms_for_no_inbound_sms_returns_empty_list(client, sample_service):
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.get(
path="/v2/received-text-messages",
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))[
"received_text_messages"
]
expected_response = []
assert json_response == expected_response
def test_get_inbound_sms_with_invalid_query_string_returns_400(client, sample_service):
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.get(
path="/v2/received-text-messages?user_number=447700900000",
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["status_code"] == 400
assert json_response["errors"][0]["error"] == "ValidationError"
assert (
json_response["errors"][0]["message"]
== "Additional properties are not allowed (user_number was unexpected)"
)

99
tests/app/v2/inbound_sms/test_inbound_sms_schemas.py
View File

@@ -1,99 +0,0 @@
import pytest
from flask import json, url_for
from jsonschema.exceptions import ValidationError
from app.schema_validation import validate
from app.v2.inbound_sms.inbound_sms_schemas import (
get_inbound_sms_request,
get_inbound_sms_response,
get_inbound_sms_single_response,
)
from tests import create_service_authorization_header
from tests.app.db import create_inbound_sms
valid_inbound_sms = {
"user_number": "447700900111",
"created_at": "2017-11-02T15:07:57.197546Z",
"service_id": "a5149c32-f03b-4711-af49-ad6993797d45",
"id": "342786aa-23ce-4695-9aad-7f79e68ee29a",
"notify_number": "testing",
"content": "Hello",
}
valid_inbound_sms_list = {
"received_text_messages": [valid_inbound_sms],
"links": {"current": valid_inbound_sms["id"]},
}
invalid_inbound_sms = {
"user_number": "447700900111",
"created_at": "2017-11-02T15:07:57.197546",
"service_id": "a5149c32-f03b-4711-af49-ad6993797d45",
"id": "342786aa-23ce-4695-9aad-7f79e68ee29a",
"notify_number": "testing",
}
invalid_inbound_sms_list = {"received_text_messages": [invalid_inbound_sms]}
def test_get_inbound_sms_contract(client, sample_service):
all_inbound_sms = [
create_inbound_sms(service=sample_service, user_number="447700900113"),
create_inbound_sms(service=sample_service, user_number="447700900112"),
create_inbound_sms(service=sample_service, user_number="447700900111"),
]
reversed_inbound_sms = sorted(
all_inbound_sms, key=lambda sms: sms.created_at, reverse=True
)
auth_header = create_service_authorization_header(
service_id=all_inbound_sms[0].service_id
)
response = client.get("/v2/received-text-messages", headers=[auth_header])
response_json = json.loads(response.get_data(as_text=True))
validated_resp = validate(response_json, get_inbound_sms_response)
assert validated_resp["received_text_messages"] == [
i.serialize() for i in reversed_inbound_sms
]
assert validated_resp["links"]["current"] == url_for(
"v2_inbound_sms.get_inbound_sms", _external=True
)
assert validated_resp["links"]["next"] == url_for(
"v2_inbound_sms.get_inbound_sms",
older_than=all_inbound_sms[0].id,
_external=True,
)
@pytest.mark.parametrize(
"request_args", [{"older_than": "6ce466d0-fd6a-11e5-82f5-e0accb9d11a6"}, {}]
)
def test_valid_inbound_sms_request_json(client, request_args):
validate(request_args, get_inbound_sms_request)
def test_invalid_inbound_sms_request_json(client):
with pytest.raises(expected_exception=ValidationError):
validate({"user_number": "447700900111"}, get_inbound_sms_request)
def test_valid_inbound_sms_response_json():
assert (
validate(valid_inbound_sms, get_inbound_sms_single_response)
== valid_inbound_sms
)
def test_valid_inbound_sms_list_response_json():
validate(valid_inbound_sms_list, get_inbound_sms_response)
def test_invalid_inbound_sms_response_json():
with pytest.raises(expected_exception=ValidationError):
validate(invalid_inbound_sms, get_inbound_sms_single_response)
def test_invalid_inbound_sms_list_response_json():
with pytest.raises(expected_exception=ValidationError):
validate(invalid_inbound_sms_list, get_inbound_sms_response)

0
tests/app/v2/notifications/__init__.py
View File

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

@@ -1,742 +0,0 @@
import pytest
from flask import json, url_for
from app.enums import NotificationStatus, NotificationType, TemplateType
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, "sns"), (0, "sns"), (1, None)]
)
def test_get_notification_by_id_returns_200(
client, billable_units, provider, sample_template, mocker
):
mock_s3_personalisation = mocker.patch(
"app.v2.notifications.get_notifications.get_personalisation_from_s3"
)
mock_s3_personalisation.return_value = {}
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,
"provider_response": None,
"carrier": None,
}
assert json_response == expected_response
def test_get_notification_by_id_with_placeholders_returns_200(
client, sample_email_template_with_placeholders, mocker
):
mock_s3_personalisation = mocker.patch(
"app.v2.notifications.get_notifications.get_personalisation_from_s3"
)
mock_s3_personalisation.return_value = {"name": "Bob"}
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,
"provider_response": None,
"carrier": None,
}
assert json_response == expected_response
def test_get_notification_by_reference_returns_200(client, sample_template, mocker):
sample_notification_with_reference = create_notification(
template=sample_template, client_reference="some-client-reference"
)
mock_s3_personalisation = mocker.patch(
"app.v2.notifications.get_notifications.get_personalisation_from_s3"
)
mock_s3_personalisation.return_value = {}
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, mocker
):
mock_s3_personalisation = mocker.patch(
"app.v2.notifications.get_notifications.get_personalisation_from_s3"
)
mock_s3_personalisation.return_value = {"name": "Bob"}
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("template_type", [TemplateType.SMS, TemplateType.EMAIL])
def test_get_notification_doesnt_have_delivery_estimate_for_non_letters(
client, sample_service, template_type, mocker
):
mock_s3_personalisation = mocker.patch(
"app.v2.notifications.get_notifications.get_personalisation_from_s3"
)
mock_s3_personalisation.return_value = {"name": "Bob"}
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"] == NotificationStatus.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"] == "1"
assert json_response["notifications"][0]["type"] == NotificationType.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, mocker
):
mock_s3_personalisation = mocker.patch(
"app.v2.notifications.get_notifications.get_personalisation_from_s3"
)
mock_s3_personalisation.return_value = {}
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 "1" == 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=TemplateType.EMAIL
)
sms_template = create_template(
service=sample_service, template_type=TemplateType.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"] == NotificationStatus.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"] == "1"
assert json_response["notifications"][0]["type"] == NotificationType.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
type_str = ", ".join(
[f"<{type(e).__name__}.{e.name}: {e.value}>" for e in TemplateType]
)
assert (
json_response["errors"][0]["message"]
== f"template_type orange is not one of [{type_str}]"
)
def test_get_all_notifications_filter_by_single_status(client, sample_template):
notification = create_notification(
template=sample_template,
status=NotificationStatus.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"] == NotificationStatus.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
type_str = ", ".join(
[f"<{type(e).__name__}.{e.name}: {e.value}>" for e in NotificationStatus]
)
assert (
json_response["errors"][0]["message"]
== f"status elephant is not one of [{type_str}]"
)
def test_get_all_notifications_filter_by_multiple_statuses(client, sample_template):
notifications = [
create_notification(template=sample_template, status=_status)
for _status in [
NotificationStatus.CREATED,
NotificationStatus.PENDING,
NotificationStatus.SENDING,
]
]
failed_notification = create_notification(
template=sample_template,
status=NotificationStatus.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=NotificationStatus.CREATED,
)
failed_notifications = [
create_notification(template=sample_template, status=NotificationStatus.FAILED)
]
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"]) == 1
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
):
# TODO had to change pending to sending. Is that correct?
# this is the notification we are looking for
older_notification = create_notification(
template=sample_email_template,
status=NotificationStatus.SENDING,
)
# wrong status
create_notification(template=sample_email_template)
wrong_template = create_template(
sample_email_template.service, template_type=TemplateType.SMS
)
# wrong template
create_notification(template=wrong_template, status=NotificationStatus.SENDING)
# 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=NotificationStatus.SENDING,
)
auth_header = create_service_authorization_header(
service_id=newer_notification.service_id
)
response = client.get(
path="/v2/notifications?status=sending&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=sending",
"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_notification,
sample_email_notification,
):
auth_header = create_service_authorization_header(
service_id=sample_email_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"] == NotificationType.SMS
or noti["type"] == NotificationType.EMAIL
):
assert noti["status"] == NotificationStatus.CREATED
else:
pytest.fail()

381
tests/app/v2/notifications/test_notification_schemas.py
View File

@@ -1,381 +0,0 @@
import uuid
import pytest
from flask import json
from freezegun import freeze_time
from jsonschema import ValidationError
from app.enums import NotificationStatus, TemplateType
from app.schema_validation import validate
from app.v2.notifications.notification_schemas import get_notifications_request
from app.v2.notifications.notification_schemas import (
post_email_request as post_email_request_schema,
)
from app.v2.notifications.notification_schemas import (
post_sms_request as post_sms_request_schema,
)
valid_get_json = {}
valid_get_with_optionals_json = {
"reference": "test reference",
"status": [NotificationStatus.CREATED],
"template_type": [TemplateType.EMAIL],
"include_jobs": "true",
"older_than": "a5149c32-f03b-4711-af49-ad6993797d45",
}
@pytest.mark.parametrize("input", [valid_get_json, valid_get_with_optionals_json])
def test_get_notifications_valid_json(input):
assert validate(input, get_notifications_request) == input
@pytest.mark.parametrize(
"invalid_statuses, valid_statuses",
[
# one invalid status
(["elephant"], []),
# multiple invalid statuses
(["elephant", "giraffe", "cheetah"], []),
# one bad status and one good status
(["elephant"], [NotificationStatus.CREATED]),
],
)
def test_get_notifications_request_invalid_statuses(invalid_statuses, valid_statuses):
type_str = ", ".join(
[f"<{type(e).__name__}.{e.name}: {e.value}>" for e in NotificationStatus]
)
partial_error_status = f"is not one of [{type_str}]"
with pytest.raises(ValidationError) as e:
validate(
{"status": invalid_statuses + valid_statuses}, get_notifications_request
)
errors = json.loads(str(e.value)).get("errors")
assert len(errors) == len(invalid_statuses)
for index, value in enumerate(invalid_statuses):
assert errors[index]["message"] == f"status {value} {partial_error_status}"
@pytest.mark.parametrize(
"invalid_template_types, valid_template_types",
[
# one invalid template_type
(["orange"], []),
# multiple invalid template_types
(["orange", "avocado", "banana"], []),
# one bad template_type and one good template_type
(["orange"], [TemplateType.SMS]),
],
)
def test_get_notifications_request_invalid_template_types(
invalid_template_types, valid_template_types
):
type_str = ", ".join(
[f"<{type(e).__name__}.{e.name}: {e.value}>" for e in TemplateType]
)
partial_error_template_type = f"is not one of [{type_str}]"
with pytest.raises(ValidationError) as e:
validate(
{"template_type": invalid_template_types + valid_template_types},
get_notifications_request,
)
errors = json.loads(str(e.value)).get("errors")
assert len(errors) == len(invalid_template_types)
for index, value in enumerate(invalid_template_types):
assert errors[index]["message"] == (
f"template_type {value} {partial_error_template_type}"
)
def test_get_notifications_request_invalid_statuses_and_template_types():
with pytest.raises(ValidationError) as e:
validate(
{
"status": [NotificationStatus.CREATED, "elephant", "giraffe"],
"template_type": [TemplateType.SMS, "orange", "avocado"],
},
get_notifications_request,
)
errors = json.loads(str(e.value)).get("errors")
assert len(errors) == 4
error_messages = [error["message"] for error in errors]
type_str = ", ".join(
[f"<{type(e).__name__}.{e.name}: {e.value}>" for e in NotificationStatus]
)
for invalid_status in ["elephant", "giraffe"]:
assert f"status {invalid_status} is not one of [{type_str}]" in error_messages
type_str = ", ".join(
[f"<{type(e).__name__}.{e.name}: {e.value}>" for e in TemplateType]
)
for invalid_template_type in ["orange", "avocado"]:
assert (
f"template_type {invalid_template_type} is not one of [{type_str}]"
in error_messages
)
valid_json = {"phone_number": "2028675309", "template_id": str(uuid.uuid4())}
valid_json_with_optionals = {
"phone_number": "2028675309",
"template_id": str(uuid.uuid4()),
"reference": "reference from caller",
"personalisation": {"key": "value"},
}
@pytest.mark.parametrize("input", [valid_json, valid_json_with_optionals])
def test_post_sms_schema_is_valid(input):
assert validate(input, post_sms_request_schema) == input
@pytest.mark.parametrize(
"template_id",
[
"2ebe4da8-17be-49fe-b02f-dff2760261a0" + "\n",
"2ebe4da8-17be-49fe-b02f-dff2760261a0" + " ",
"2ebe4da8-17be-49fe-b02f-dff2760261a0" + "\r",
"\t" + "2ebe4da8-17be-49fe-b02f-dff2760261a0",
"2ebe4da8-17be-49fe-b02f-dff2760261a0"[4:],
"bad_uuid",
],
)
def test_post_sms_json_schema_bad_uuid(template_id):
j = {"template_id": template_id, "phone_number": "2028675309"}
with pytest.raises(ValidationError) as e:
validate(j, post_sms_request_schema)
error = json.loads(str(e.value))
assert len(error.keys()) == 2
assert error.get("status_code") == 400
assert len(error.get("errors")) == 1
assert {
"error": "ValidationError",
"message": "template_id is not a valid UUID",
} in error["errors"]
def test_post_sms_json_schema_bad_uuid_and_missing_phone_number():
j = {"template_id": "notUUID"}
with pytest.raises(ValidationError) as e:
validate(j, post_sms_request_schema)
error = json.loads(str(e.value))
assert len(error.keys()) == 2
assert error.get("status_code") == 400
assert len(error.get("errors")) == 2
assert {
"error": "ValidationError",
"message": "phone_number is a required property",
} in error["errors"]
assert {
"error": "ValidationError",
"message": "template_id is not a valid UUID",
} in error["errors"]
def test_post_sms_schema_with_personalisation_that_is_not_a_dict():
j = {
"phone_number": "2028675309",
"template_id": str(uuid.uuid4()),
"reference": "reference from caller",
"personalisation": "not_a_dict",
}
with pytest.raises(ValidationError) as e:
validate(j, post_sms_request_schema)
error = json.loads(str(e.value))
assert len(error.get("errors")) == 1
assert error["errors"] == [
{
"error": "ValidationError",
"message": "personalisation not_a_dict is not of type object",
}
]
assert error.get("status_code") == 400
assert len(error.keys()) == 2
@pytest.mark.parametrize(
"invalid_phone_number, err_msg",
[
("08515111111", "phone_number Phone number is not possible"),
("07515111*11", "phone_number Not enough digits"),
(
"notaphoneumber",
"phone_number The string supplied did not seem to be a phone number.",
),
(7700900001, "phone_number 7700900001 is not of type string"),
(None, "phone_number None is not of type string"),
([], "phone_number [] is not of type string"),
({}, "phone_number {} is not of type string"),
],
)
def test_post_sms_request_schema_invalid_phone_number(invalid_phone_number, err_msg):
j = {"phone_number": invalid_phone_number, "template_id": str(uuid.uuid4())}
with pytest.raises(ValidationError) as e:
validate(j, post_sms_request_schema)
errors = json.loads(str(e.value)).get("errors")
assert len(errors) == 1
assert {"error": "ValidationError", "message": err_msg} == errors[0]
def test_post_sms_request_schema_invalid_phone_number_and_missing_template():
j = {
"phone_number": "5558675309",
}
with pytest.raises(ValidationError) as e:
validate(j, post_sms_request_schema)
errors = json.loads(str(e.value)).get("errors")
assert len(errors) == 2
assert {
"error": "ValidationError",
"message": "phone_number Phone number range is not in use",
} in errors
assert {
"error": "ValidationError",
"message": "template_id is a required property",
} in errors
valid_post_email_json = {
"email_address": "test@example.gov.uk",
"template_id": str(uuid.uuid4()),
}
valid_post_email_json_with_optionals = {
"email_address": "test@example.gov.uk",
"template_id": str(uuid.uuid4()),
"reference": "reference from caller",
"personalisation": {"key": "value"},
}
@pytest.mark.parametrize(
"input", [valid_post_email_json, valid_post_email_json_with_optionals]
)
def test_post_email_schema_is_valid(input):
assert validate(input, post_email_request_schema) == input
def test_post_email_schema_bad_uuid_and_missing_email_address():
j = {"template_id": "bad_template"}
with pytest.raises(ValidationError):
validate(j, post_email_request_schema)
@pytest.mark.parametrize(
"email_address, err_msg",
[
("example", "email_address Not a valid email address"),
(12345, "email_address 12345 is not of type string"),
("with(brackets)@example.com", "email_address Not a valid email address"),
(None, "email_address None is not of type string"),
([], "email_address [] is not of type string"),
({}, "email_address {} is not of type string"),
],
)
def test_post_email_schema_invalid_email_address(email_address, err_msg):
j = {"template_id": str(uuid.uuid4()), "email_address": email_address}
with pytest.raises(ValidationError) as e:
validate(j, post_email_request_schema)
errors = json.loads(str(e.value)).get("errors")
assert len(errors) == 1
assert {"error": "ValidationError", "message": err_msg} == errors[0]
def valid_email_response():
return {
"id": str(uuid.uuid4()),
"content": {
"body": "the body of the message",
"subject": "subject of the message",
"from_email": "service@dig.gov.uk",
},
"uri": "http://notify.api/v2/notifications/id",
"template": {
"id": str(uuid.uuid4()),
"version": 1,
"uri": "http://notify.api/v2/template/id",
},
"scheduled_for": "",
}
@pytest.mark.parametrize("schema", [post_email_request_schema, post_sms_request_schema])
@freeze_time("2017-05-12 13:00:00")
def test_post_schema_valid_scheduled_for(schema):
j = {"template_id": str(uuid.uuid4()), "scheduled_for": "2017-05-12 13:15"}
if schema == post_email_request_schema:
j.update({"email_address": "joe@gmail.com"})
else:
j.update({"phone_number": "2028675309"})
assert validate(j, schema) == j
@pytest.mark.parametrize(
"invalid_datetime",
["13:00:00 2017-01-01", "2017-31-12 13:00:00", "01-01-2017T14:00:00.0000Z"],
)
@pytest.mark.parametrize("schema", [post_email_request_schema, post_sms_request_schema])
def test_post_email_schema_invalid_scheduled_for(invalid_datetime, schema):
j = {"template_id": str(uuid.uuid4()), "scheduled_for": invalid_datetime}
if schema == post_email_request_schema:
j.update({"email_address": "joe@gmail.com"})
else:
j.update({"phone_number": "2028675309"})
with pytest.raises(ValidationError) as e:
validate(j, schema)
error = json.loads(str(e.value))
assert error["status_code"] == 400
assert error["errors"] == [
{
"error": "ValidationError",
"message": "scheduled_for datetime format is invalid. "
"It must be a valid ISO8601 date time format, "
"https://en.wikipedia.org/wiki/ISO_8601",
}
]
@freeze_time("2017-05-12 13:00:00")
def test_scheduled_for_raises_validation_error_when_in_the_past():
j = {
"phone_number": "2028675309",
"template_id": str(uuid.uuid4()),
"scheduled_for": "2017-05-12 10:00",
}
with pytest.raises(ValidationError) as e:
validate(j, post_sms_request_schema)
error = json.loads(str(e.value))
assert error["status_code"] == 400
assert error["errors"] == [
{
"error": "ValidationError",
"message": "scheduled_for datetime can not be in the past",
}
]
@freeze_time("2017-05-12 13:00:00")
def test_scheduled_for_raises_validation_error_when_more_than_24_hours_in_the_future():
j = {
"phone_number": "2028675309",
"template_id": str(uuid.uuid4()),
"scheduled_for": "2017-05-13 14:00",
}
with pytest.raises(ValidationError) as e:
validate(j, post_sms_request_schema)
error = json.loads(str(e.value))
assert error["status_code"] == 400
assert error["errors"] == [
{
"error": "ValidationError",
"message": "scheduled_for datetime can only be 24 hours in the future",
}
]

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

File diff suppressed because it is too large Load Diff

0
tests/app/v2/template/__init__.py
View File

146
tests/app/v2/template/test_get_template.py
View File

@@ -1,146 +0,0 @@
import pytest
from flask import json
from app.enums import TemplateType
from app.utils import DATETIME_FORMAT
from tests import create_service_authorization_header
from tests.app.db import create_template
valid_version_params = [None, 1]
@pytest.mark.parametrize(
"tmp_type, expected_name, expected_subject",
[
(TemplateType.SMS, "sms Template Name", None),
(TemplateType.EMAIL, "email Template Name", "Template subject"),
],
)
@pytest.mark.parametrize("version", valid_version_params)
def test_get_template_by_id_returns_200(
client, sample_service, tmp_type, expected_name, expected_subject, version
):
template = create_template(sample_service, template_type=tmp_type)
auth_header = create_service_authorization_header(service_id=sample_service.id)
version_path = "/version/{}".format(version) if version else ""
response = client.get(
path="/v2/template/{}{}".format(template.id, version_path),
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_response = {
"id": "{}".format(template.id),
"type": "{}".format(template.template_type),
"created_at": template.created_at.strftime(DATETIME_FORMAT),
"updated_at": None,
"version": template.version,
"created_by": template.created_by.email_address,
"body": template.content,
"subject": expected_subject,
"name": expected_name,
"personalisation": {},
}
assert json_response == expected_response
@pytest.mark.parametrize(
"create_template_args, expected_personalisation",
[
(
{
"template_type": TemplateType.SMS,
"content": "Hello ((placeholder)) ((conditional??yes))",
},
{
"placeholder": {"required": True},
"conditional": {"required": True},
},
),
(
{
"template_type": TemplateType.EMAIL,
"subject": "((subject))",
"content": "((content))",
},
{
"subject": {"required": True},
"content": {"required": True},
},
),
],
)
@pytest.mark.parametrize("version", valid_version_params)
def test_get_template_by_id_returns_placeholders(
client,
sample_service,
version,
create_template_args,
expected_personalisation,
):
template = create_template(sample_service, **create_template_args)
auth_header = create_service_authorization_header(service_id=sample_service.id)
version_path = "/version/{}".format(version) if version else ""
response = client.get(
path="/v2/template/{}{}".format(template.id, version_path),
headers=[("Content-Type", "application/json"), auth_header],
)
json_response = json.loads(response.get_data(as_text=True))
assert json_response["personalisation"] == expected_personalisation
def test_get_template_with_non_existent_template_id_returns_404(
client, fake_uuid, sample_service
):
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.get(
path="/v2/template/{}".format(fake_uuid),
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("tmp_type", TemplateType)
def test_get_template_with_non_existent_version_returns_404(
client, sample_service, tmp_type
):
template = create_template(sample_service, template_type=tmp_type)
auth_header = create_service_authorization_header(service_id=sample_service.id)
invalid_version = template.version + 1
response = client.get(
path="/v2/template/{}/version/{}".format(template.id, invalid_version),
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,
}

222
tests/app/v2/template/test_post_template.py
View File

@@ -1,222 +0,0 @@
import pytest
from flask import json
from app.models import TemplateType
from tests import create_service_authorization_header
from tests.app.db import create_template
valid_personalisation = {"personalisation": {"Name": "Jo"}}
valid_post = [
(
"Some subject",
"Some content",
None,
"Some subject",
"Some content",
(
'<p style="Margin: 0 0 20px 0; font-size: 19px; line-height: 25px; color: #0B0C0C;">'
"Some content"
"</p>"
),
),
(
"Some subject",
"Dear ((Name)), Hello. Yours Truly, The Government.",
valid_personalisation,
"Some subject",
"Dear Jo, Hello. Yours Truly, The Government.",
(
'<p style="Margin: 0 0 20px 0; font-size: 19px; line-height: 25px; color: #0B0C0C;">'
"Dear Jo, Hello. Yours Truly, The Government."
"</p>"
),
),
(
"Message for ((Name))",
"Dear ((Name)), Hello. Yours Truly, The Government.",
valid_personalisation,
"Message for Jo",
"Dear Jo, Hello. Yours Truly, The Government.",
(
'<p style="Margin: 0 0 20px 0; font-size: 19px; line-height: 25px; color: #0B0C0C;">'
"Dear Jo, Hello. Yours Truly, The Government."
"</p>"
),
),
(
"Message for ((Name))",
"Some content",
valid_personalisation,
"Message for Jo",
"Some content",
(
'<p style="Margin: 0 0 20px 0; font-size: 19px; line-height: 25px; color: #0B0C0C;">'
"Some content"
"</p>"
),
),
]
@pytest.mark.parametrize("tmp_type", (TemplateType.SMS, TemplateType.EMAIL))
@pytest.mark.parametrize(
"subject,content,post_data,expected_subject,expected_content,expected_html",
valid_post,
)
def test_valid_post_template_returns_200(
client,
sample_service,
tmp_type,
subject,
content,
post_data,
expected_subject,
expected_content,
expected_html,
):
template = create_template(
sample_service, template_type=tmp_type, subject=subject, content=content
)
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.post(
path="/v2/template/{}/preview".format(template.id),
data=json.dumps(post_data),
headers=[("Content-Type", "application/json"), auth_header],
)
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
assert resp_json["id"] == str(template.id)
if tmp_type == TemplateType.EMAIL:
assert expected_subject in resp_json["subject"]
assert resp_json["html"] == expected_html
else:
assert resp_json["html"] is None
assert expected_content in resp_json["body"]
def test_email_templates_not_rendered_into_content(client, sample_service):
template = create_template(
sample_service,
template_type=TemplateType.EMAIL,
subject="Test",
content=("Hello\n" "\r\n" "\r\n" "\n" "# This is a heading\n" "\n" "Paragraph"),
)
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.post(
path="/v2/template/{}/preview".format(template.id),
data=json.dumps(None),
headers=[("Content-Type", "application/json"), auth_header],
)
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
assert resp_json["body"] == template.content
@pytest.mark.parametrize("tmp_type", (TemplateType.SMS, TemplateType.EMAIL))
def test_invalid_post_template_returns_400(client, sample_service, tmp_type):
template = create_template(
sample_service,
template_type=tmp_type,
content="Dear ((Name)), Hello ((Missing)). Yours Truly, The Government.",
)
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.post(
path="/v2/template/{}/preview".format(template.id),
data=json.dumps(valid_personalisation),
headers=[("Content-Type", "application/json"), auth_header],
)
assert response.status_code == 400
resp_json = json.loads(response.get_data(as_text=True))
assert resp_json["errors"][0]["error"] == "BadRequestError"
assert "Missing personalisation: Missing" in resp_json["errors"][0]["message"]
def test_post_template_with_non_existent_template_id_returns_404(
client, fake_uuid, sample_service
):
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.post(
path="/v2/template/{}/preview".format(fake_uuid),
data=json.dumps(valid_personalisation),
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,
}
def test_post_template_returns_200_without_personalisation(client, sample_template):
response = client.post(
path="/v2/template/{}/preview".format(sample_template.id),
data=None,
headers=[
("Content-Type", "application/json"),
create_service_authorization_header(service_id=sample_template.service_id),
],
)
assert response.status_code == 200
def test_post_template_returns_200_without_personalisation_and_missing_content_header(
client, sample_template
):
response = client.post(
path="/v2/template/{}/preview".format(sample_template.id),
data=None,
headers=[
create_service_authorization_header(service_id=sample_template.service_id)
],
)
assert response.status_code == 200
def test_post_template_returns_200_without_personalisation_as_valid_json_and_missing_content_header(
client, sample_template
):
response = client.post(
path="/v2/template/{}/preview".format(sample_template.id),
data=json.dumps(None),
headers=[
create_service_authorization_header(service_id=sample_template.service_id)
],
)
assert response.status_code == 200
def test_post_template_returns_200_with_valid_json_and_missing_content_header(
client, sample_template
):
response = client.post(
path="/v2/template/{}/preview".format(sample_template.id),
data=json.dumps(valid_personalisation),
headers=[
create_service_authorization_header(service_id=sample_template.service_id)
],
)
assert response.status_code == 200

159
tests/app/v2/template/test_template_schemas.py
View File

@@ -1,159 +0,0 @@
import uuid
import pytest
from flask import json
from jsonschema.exceptions import ValidationError
from app.enums import TemplateType
from app.schema_validation import validate
from app.v2.template.template_schemas import (
get_template_by_id_request,
get_template_by_id_response,
post_template_preview_request,
post_template_preview_response,
)
valid_json_get_response = {
"id": str(uuid.uuid4()),
"type": TemplateType.SMS,
"created_at": "2017-01-10T18:25:43.511Z",
"updated_at": None,
"version": 1,
"created_by": "someone@test.com",
"body": "some body",
"name": "some name",
}
valid_json_get_response_with_optionals = {
"id": str(uuid.uuid4()),
"type": TemplateType.EMAIL,
"created_at": "2017-01-10T18:25:43.511Z",
"updated_at": None,
"version": 1,
"created_by": "someone",
"body": "some body",
"subject": "some subject",
"name": "some name",
}
valid_request_args = [
{"id": str(uuid.uuid4()), "version": 1},
{"id": str(uuid.uuid4())},
]
invalid_request_args = [
(
{"id": str(uuid.uuid4()), "version": "test"},
["version test is not of type integer, null"],
),
(
{"id": str(uuid.uuid4()), "version": 0},
["version 0 is less than the minimum of 1"],
),
({"version": 1}, ["id is a required property"]),
({"id": "invalid_uuid"}, ["id is not a valid UUID"]),
(
{"id": "invalid_uuid", "version": 0},
["version 0 is less than the minimum of 1", "id is not a valid UUID"],
),
]
valid_json_post_args = {"id": str(uuid.uuid4()), "personalisation": {"key": "value"}}
invalid_json_post_args = [
(
{"id": "invalid_uuid", "personalisation": {"key": "value"}},
["id is not a valid UUID"],
),
(
{"id": str(uuid.uuid4()), "personalisation": ["a", "b"]},
["personalisation [a, b] is not of type object"],
),
(
{"personalisation": "invalid_personalisation"},
[
"id is a required property",
"personalisation invalid_personalisation is not of type object",
],
),
]
valid_json_post_response = {
"id": str(uuid.uuid4()),
"type": TemplateType.EMAIL,
"version": 1,
"body": "some body",
}
valid_json_post_response_with_optionals = {
"id": str(uuid.uuid4()),
"type": TemplateType.EMAIL,
"version": 1,
"body": "some body",
"subject": "some subject",
"html": "<p>some body</p>",
}
@pytest.mark.parametrize("args", valid_request_args)
def test_get_template_request_schema_against_valid_args_is_valid(args):
assert validate(args, get_template_by_id_request) == args
@pytest.mark.parametrize("args,error_message", invalid_request_args)
def test_get_template_request_schema_against_invalid_args_is_invalid(
args, error_message
):
with pytest.raises(ValidationError) as e:
validate(args, get_template_by_id_request)
errors = json.loads(str(e.value))
assert errors["status_code"] == 400
for error in errors["errors"]:
assert error["message"] in error_message
@pytest.mark.parametrize("template_type", TemplateType)
@pytest.mark.parametrize(
"response", [valid_json_get_response, valid_json_get_response_with_optionals]
)
@pytest.mark.parametrize("updated_datetime", [None, "2017-01-11T18:25:43.511Z"])
def test_get_template_response_schema_is_valid(
response, template_type, updated_datetime
):
if updated_datetime:
response["updated_at"] = updated_datetime
response["type"] = template_type
assert validate(response, get_template_by_id_response) == response
def test_post_template_preview_against_valid_args_is_valid():
assert (
validate(valid_json_post_args, post_template_preview_request)
== valid_json_post_args
)
@pytest.mark.parametrize("args,error_messages", invalid_json_post_args)
def test_post_template_preview_against_invalid_args_is_invalid(args, error_messages):
with pytest.raises(ValidationError) as e:
validate(args, post_template_preview_request)
errors = json.loads(str(e.value))
assert errors["status_code"] == 400
assert len(errors["errors"]) == len(error_messages)
for error in errors["errors"]:
assert error["message"] in error_messages
@pytest.mark.parametrize("template_type", TemplateType)
@pytest.mark.parametrize(
"response", [valid_json_post_response, valid_json_post_response_with_optionals]
)
def test_post_template_preview_response_schema_is_valid(response, template_type):
response["type"] = template_type
assert validate(response, post_template_preview_response) == response

130
tests/app/v2/templates/test_get_templates.py
View File

@@ -1,130 +0,0 @@
import pytest
from flask import json
from app.models import TemplateType
from tests import create_service_authorization_header
from tests.app.db import create_template
def test_get_all_templates_returns_200(client, sample_service):
templates = [
create_template(
sample_service,
template_type=tmp_type,
subject=f"subject_{name}" if tmp_type == TemplateType.EMAIL else "",
template_name=name,
)
for name, tmp_type in (("A", TemplateType.SMS), ("B", TemplateType.EMAIL))
]
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.get(
path="/v2/templates",
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["templates"]) == len(templates)
for index, template in enumerate(json_response["templates"]):
assert template["id"] == str(templates[index].id)
assert template["body"] == templates[index].content
assert template["type"] == templates[index].template_type
if templates[index].template_type == TemplateType.EMAIL:
assert template["subject"] == templates[index].subject
@pytest.mark.parametrize("tmp_type", (TemplateType.SMS, TemplateType.EMAIL))
def test_get_all_templates_for_valid_type_returns_200(client, sample_service, tmp_type):
templates = [
create_template(
sample_service,
template_type=tmp_type,
template_name=f"Template {i}",
subject=f"subject_{i}" if tmp_type == TemplateType.EMAIL else "",
)
for i in range(3)
]
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.get(
path=f"/v2/templates?type={tmp_type}",
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["templates"]) == len(templates)
for index, template in enumerate(json_response["templates"]):
assert template["id"] == str(templates[index].id)
assert template["body"] == templates[index].content
assert template["type"] == tmp_type
if templates[index].template_type == TemplateType.EMAIL:
assert template["subject"] == templates[index].subject
@pytest.mark.parametrize("tmp_type", (TemplateType.SMS, TemplateType.EMAIL))
def test_get_correct_num_templates_for_valid_type_returns_200(
client, sample_service, tmp_type
):
num_templates = 3
templates = []
for _ in range(num_templates):
templates.append(create_template(sample_service, template_type=tmp_type))
for other_type in TemplateType:
if other_type != tmp_type:
templates.append(create_template(sample_service, template_type=other_type))
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.get(
path=f"/v2/templates?type={tmp_type}",
headers=[("Content-Type", "application/json"), auth_header],
)
assert response.status_code == 200
json_response = json.loads(response.get_data(as_text=True))
assert len(json_response["templates"]) == num_templates
def test_get_all_templates_for_invalid_type_returns_400(client, sample_service):
auth_header = create_service_authorization_header(service_id=sample_service.id)
invalid_type = "coconut"
response = client.get(
path=f"/v2/templates?type={invalid_type}",
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))
type_str = ", ".join(
[f"<{type(e).__name__}.{e.name}: {e.value}>" for e in TemplateType]
)
assert json_response == {
"status_code": 400,
"errors": [
{
"message": f"type coconut is not one of [{type_str}]",
"error": "ValidationError",
}
],
}

301
tests/app/v2/templates/test_templates_schemas.py
View File

@@ -1,301 +0,0 @@
import uuid
import pytest
from flask import json
from jsonschema.exceptions import ValidationError
from app.enums import TemplateType
from app.schema_validation import validate
from app.v2.templates.templates_schemas import (
get_all_template_request,
get_all_template_response,
)
valid_json_get_all_response = [
{
"templates": [
{
"id": str(uuid.uuid4()),
"type": TemplateType.SMS,
"created_at": "2017-01-10T18:25:43.511Z",
"updated_at": None,
"version": 1,
"created_by": "someone@test.com",
"body": "some body",
"name": "some name",
},
{
"id": str(uuid.uuid4()),
"type": TemplateType.EMAIL,
"created_at": "2017-02-10T18:25:43.511Z",
"updated_at": None,
"version": 2,
"created_by": "someone@test.com",
"subject": "test subject",
"body": "some body",
"name": "some name",
},
]
},
{
"templates": [
{
"id": str(uuid.uuid4()),
"type": TemplateType.SMS,
"created_at": "2017-02-10T18:25:43.511Z",
"updated_at": None,
"version": 2,
"created_by": "someone@test.com",
"body": "some body",
"name": "some name",
}
]
},
{"templates": []},
]
invalid_json_get_all_response = [
(
{
"templates": [
{
"id": "invalid_id",
"type": TemplateType.SMS,
"created_at": "2017-02-10T18:25:43.511Z",
"updated_at": None,
"version": 1,
"created_by": "someone@test.com",
"body": "some body",
"name": "some name",
}
]
},
["templates is not a valid UUID"],
),
(
{
"templates": [
{
"id": str(uuid.uuid4()),
"type": TemplateType.SMS,
"created_at": "2017-02-10T18:25:43.511Z",
"updated_at": None,
"version": "invalid_version",
"created_by": "someone@test.com",
"body": "some body",
"name": "some name",
}
]
},
["templates invalid_version is not of type integer"],
),
(
{
"templates": [
{
"id": str(uuid.uuid4()),
"type": TemplateType.SMS,
"created_at": "invalid_created_at",
"updated_at": None,
"version": 1,
"created_by": "someone@test.com",
"body": "some body",
"name": "some name",
}
]
},
["templates invalid_created_at is not a date-time"],
),
({}, ["templates is a required property"]),
(
{
"templates": [
{
"type": TemplateType.SMS,
"created_at": "2017-02-10T18:25:43.511Z",
"updated_at": None,
"version": 1,
"created_by": "someone@test.com",
"body": "some body",
"name": "some name",
}
]
},
["templates id is a required property"],
),
(
{
"templates": [
{
"id": str(uuid.uuid4()),
"type": TemplateType.SMS,
"created_at": "2017-02-10T18:25:43.511Z",
"updated_at": None,
"version": 1,
"created_by": "someone@test.com",
"body": "some body",
}
]
},
["templates name is a required property"],
),
(
{
"templates": [
{
"id": str(uuid.uuid4()),
"created_at": "2017-02-10T18:25:43.511Z",
"updated_at": None,
"version": 1,
"created_by": "someone@test.com",
"body": "some body",
"name": "some name",
}
]
},
["templates type is a required property"],
),
(
{
"templates": [
{
"id": str(uuid.uuid4()),
"type": TemplateType.SMS,
"updated_at": None,
"version": 1,
"created_by": "someone@test.com",
"body": "some body",
"name": "some name",
}
]
},
["templates created_at is a required property"],
),
(
{
"templates": [
{
"id": str(uuid.uuid4()),
"type": TemplateType.SMS,
"created_at": "2017-02-10T18:25:43.511Z",
"version": 1,
"created_by": "someone@test.com",
"body": "some body",
"name": "some name",
}
]
},
["templates updated_at is a required property"],
),
(
{
"templates": [
{
"id": str(uuid.uuid4()),
"type": TemplateType.SMS,
"created_at": "2017-02-10T18:25:43.511Z",
"updated_at": None,
"created_by": "someone@test.com",
"body": "some body",
"name": "some name",
}
]
},
["templates version is a required property"],
),
(
{
"templates": [
{
"id": str(uuid.uuid4()),
"type": TemplateType.SMS,
"created_at": "2017-02-10T18:25:43.511Z",
"updated_at": None,
"version": 1,
"body": "some body",
"name": "some name",
}
]
},
["templates created_by is a required property"],
),
(
{
"templates": [
{
"id": str(uuid.uuid4()),
"type": TemplateType.SMS,
"created_at": "2017-02-10T18:25:43.511Z",
"updated_at": None,
"version": 1,
"created_by": "someone@test.com",
"name": "some name",
}
]
},
["templates body is a required property"],
),
(
{
"templates": [
{
"type": TemplateType.SMS,
"created_at": "2017-02-10T18:25:43.511Z",
"updated_at": None,
"created_by": "someone@test.com",
"body": "some body",
"name": "some name",
}
]
},
[
"templates id is a required property",
"templates version is a required property",
],
),
]
@pytest.mark.parametrize("template_type", TemplateType)
def test_get_all_template_request_schema_against_no_args_is_valid(template_type):
data = {}
assert validate(data, get_all_template_request) == data
@pytest.mark.parametrize("template_type", TemplateType)
def test_get_all_template_request_schema_against_valid_args_is_valid(template_type):
data = {"type": template_type}
assert validate(data, get_all_template_request) == data
@pytest.mark.parametrize("template_type", TemplateType)
def test_get_all_template_request_schema_against_invalid_args_is_invalid(template_type):
data = {"type": "unknown"}
with pytest.raises(ValidationError) as e:
validate(data, get_all_template_request)
errors = json.loads(str(e.value))
assert errors["status_code"] == 400
assert len(errors["errors"]) == 1
type_str = ", ".join(
[f"<{type(e).__name__}.{e.name}: {e.value}>" for e in TemplateType]
)
assert errors["errors"][0]["message"] == f"type unknown is not one of [{type_str}]"
@pytest.mark.parametrize("response", valid_json_get_all_response)
def test_valid_get_all_templates_response_schema_is_valid(response):
assert validate(response, get_all_template_response) == response
@pytest.mark.parametrize("response,error_messages", invalid_json_get_all_response)
def test_invalid_get_all_templates_response_schema_is_invalid(response, error_messages):
with pytest.raises(ValidationError) as e:
validate(response, get_all_template_response)
errors = json.loads(str(e.value))
assert errors["status_code"] == 400
assert len(errors["errors"]) == len(error_messages)
for error in errors["errors"]:
assert error["message"] in error_messages

159
tests/app/v2/test_errors.py
View File

@@ -1,159 +0,0 @@
import pytest
from flask import url_for
from sqlalchemy.exc import DataError
from app.v2.errors import ValidationError
@pytest.fixture(scope="function")
def app_for_test():
import flask
from flask import Blueprint
from app import init_app
from app.authentication.auth import AuthError
from app.v2.errors import BadRequestError, TooManyRequestsError
app = flask.Flask(__name__)
app.config["TESTING"] = True
init_app(app)
from app.v2.errors import register_errors
blue = Blueprint("v2_under_test", __name__, url_prefix="/v2/under_test")
@blue.route("/raise_auth_error", methods=["GET"])
def raising_auth_error():
raise AuthError("some message", 403)
@blue.route("/raise_bad_request", methods=["GET"])
def raising_bad_request():
raise BadRequestError(message="you forgot the thing")
@blue.route("/raise_too_many_requests", methods=["GET"])
def raising_too_many_requests():
raise TooManyRequestsError(sending_limit="452")
@blue.route("/raise_validation_error", methods=["GET"])
def raising_validation_error():
from app.schema_validation import validate
from app.v2.notifications.notification_schemas import post_sms_request
validate({"template_id": "bad_uuid"}, post_sms_request)
@blue.route("raise_data_error", methods=["GET"])
def raising_data_error():
raise DataError("There was a db problem", "params", "orig")
@blue.route("raise_exception", methods=["GET"])
def raising_exception():
raise AssertionError("Raising any old exception")
register_errors(blue)
app.register_blueprint(blue)
return app
def test_auth_error(app_for_test):
with app_for_test.test_request_context():
with app_for_test.test_client() as client:
response = client.get(url_for("v2_under_test.raising_auth_error"))
assert response.status_code == 403
error = response.json
assert error == {
"status_code": 403,
"errors": [{"error": "AuthError", "message": "some message"}],
}
def test_bad_request_error(app_for_test):
with app_for_test.test_request_context():
with app_for_test.test_client() as client:
response = client.get(url_for("v2_under_test.raising_bad_request"))
assert response.status_code == 400
error = response.json
assert error == {
"status_code": 400,
"errors": [
{"error": "BadRequestError", "message": "you forgot the thing"}
],
}
def test_too_many_requests_error(app_for_test):
with app_for_test.test_request_context():
with app_for_test.test_client() as client:
response = client.get(url_for("v2_under_test.raising_too_many_requests"))
assert response.status_code == 429
error = response.json
assert error == {
"status_code": 429,
"errors": [
{
"error": "TooManyRequestsError",
"message": "Exceeded send limits (452) for today",
}
],
}
def test_validation_error(app_for_test):
with app_for_test.test_request_context():
with app_for_test.test_client() as client:
response = client.get(url_for("v2_under_test.raising_validation_error"))
assert response.status_code == 400
error = response.json
assert len(error.keys()) == 2
assert error["status_code"] == 400
assert len(error["errors"]) == 2
assert {
"error": "ValidationError",
"message": "phone_number is a required property",
} in error["errors"]
assert {
"error": "ValidationError",
"message": "template_id is not a valid UUID",
} in error["errors"]
ve = ValidationError("phone_number is a required property")
assert ve.message == "Your notification has failed validation"
assert ve.status_code == 400
def test_data_errors(app_for_test):
with app_for_test.test_request_context():
with app_for_test.test_client() as client:
response = client.get(url_for("v2_under_test.raising_data_error"))
assert response.status_code == 404
error = response.json
assert error == {
"status_code": 404,
"errors": [{"error": "DataError", "message": "No result found"}],
}
def test_internal_server_error_handler(app_for_test):
with app_for_test.test_request_context():
with app_for_test.test_client() as client:
response = client.get(url_for("v2_under_test.raising_exception"))
assert response.status_code == 500
error = response.json
assert error == {
"status_code": 500,
"errors": [
{"error": "AssertionError", "message": "Internal server error"}
],
}
def test_bad_method(app_for_test):
with app_for_test.test_request_context():
with app_for_test.test_client() as client:
response = client.post(url_for("v2_under_test.raising_exception"))
assert response.status_code == 405
assert response.get_json(force=True) == {
"result": "error",
"message": "The method is not allowed for the requested URL.",
}