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

Even more cleanup.

Browse Source 
Signed-off-by: Cliff Hill <Clifford.hill@gsa.gov>
This commit is contained in:
Cliff Hill
2024-02-20 17:00:54 -05:00
parent 12658fe667
commit 74ac09ae55
3 changed files with 162 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
tests/app/user/test_rest.py
View File

@@ -8,7 +8,7 @@ from flask import current_app
from freezegun import freeze_time
from app.dao.service_user_dao import dao_get_service_user, dao_update_service_user
from app.enums import AuthType, PermissionType
from app.enums import AuthType, NotificationType, PermissionType
from app.models import Notification, Permission, User
from tests.app.db import (
create_organization,
@@ -262,7 +262,7 @@ def test_post_user_attribute(admin_request, sample_user, user_attribute, user_va
dict(
api_key_id=None,
key_type="normal",
notification_type="email",
notification_type=NotificationType.EMAIL,
personalisation={
"name": "Test User",
"servicemanagername": "Service Manago",
@@ -281,7 +281,7 @@ def test_post_user_attribute(admin_request, sample_user, user_attribute, user_va
dict(
api_key_id=None,
key_type="normal",
notification_type="sms",
notification_type=NotificationType.SMS,
personalisation={
"name": "Test User",
"servicemanagername": "Service Manago",

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

@@ -209,7 +209,11 @@ def test_send_user_sms_code(client, sample_user, sms_code_template, mocker):
mocker.patch("app.celery.provider_tasks.deliver_sms.apply_async")
resp = client.post(
url_for("user.send_user_2fa_code", code_type="sms", user_id=sample_user.id),
url_for(
"user.send_user_2fa_code",
code_type=CodeType.SMS,
user_id=sample_user.id,
),
data=json.dumps({}),
headers=[("Content-Type", "application/json"), auth_header],
)
@@ -247,7 +251,11 @@ def test_send_user_code_for_sms_with_optional_to_field(
auth_header = create_admin_authorization_header()
resp = client.post(
url_for("user.send_user_2fa_code", code_type="sms", user_id=sample_user.id),
url_for(
"user.send_user_2fa_code",
code_type=CodeType.SMS,
user_id=sample_user.id,
),
data=json.dumps({"to": to_number}),
headers=[("Content-Type", "application/json"), auth_header],
)
@@ -265,7 +273,7 @@ def test_send_sms_code_returns_404_for_bad_input_data(client):
uuid_ = uuid.uuid4()
auth_header = create_admin_authorization_header()
resp = client.post(
url_for("user.send_user_2fa_code", code_type="sms", user_id=uuid_),
url_for("user.send_user_2fa_code", code_type=CodeType.SMS, user_id=uuid_),
data=json.dumps({}),
headers=[("Content-Type", "application/json"), auth_header],
)
@@ -278,7 +286,7 @@ def test_send_sms_code_returns_204_when_too_many_codes_already_created(
):
for _ in range(5):
verify_code = VerifyCode(
code_type="sms",
code_type=CodeType.SMS,
_code=12345,
created_at=datetime.utcnow() - timedelta(minutes=10),
expiry_datetime=datetime.utcnow() + timedelta(minutes=40),
@@ -289,7 +297,11 @@ def test_send_sms_code_returns_204_when_too_many_codes_already_created(
assert VerifyCode.query.count() == 5
auth_header = create_admin_authorization_header()
resp = client.post(
url_for("user.send_user_2fa_code", code_type="sms", user_id=sample_user.id),
url_for(
"user.send_user_2fa_code",
code_type=CodeType.SMS,
user_id=sample_user.id,
),
data=json.dumps({}),
headers=[("Content-Type", "application/json"), auth_header],
)
@@ -463,7 +475,7 @@ def test_send_user_email_code(
admin_request.post(
"user.send_user_2fa_code",
code_type="email",
code_type=CodeType.EMAIL,
user_id=sample_user.id,
_data=data,
_expected_status=204,
@@ -493,7 +505,7 @@ def test_send_user_email_code_with_urlencoded_next_param(
data = {"to": None, "next": "/services"}
admin_request.post(
"user.send_user_2fa_code",
code_type="email",
code_type=CodeType.EMAIL,
user_id=sample_user.id,
_data=data,
_expected_status=204,
@@ -505,7 +517,7 @@ def test_send_user_email_code_with_urlencoded_next_param(
def test_send_email_code_returns_404_for_bad_input_data(admin_request):
resp = admin_request.post(
"user.send_user_2fa_code",
code_type="email",
code_type=CodeType.EMAIL,
user_id=uuid.uuid4(),
_data={},
_expected_status=404,
@@ -523,7 +535,7 @@ def test_user_verify_email_code(admin_request, sample_user, auth_type):
magic_code = str(uuid.uuid4())
verify_code = create_user_code(sample_user, magic_code, CodeType.EMAIL)
data = {"code_type": "email", "code": magic_code}
data = {"code_type": CodeType.EMAIL, "code": magic_code}
admin_request.post(
"user.verify_user_code",
@@ -575,7 +587,11 @@ def test_send_user_2fa_code_sends_from_number_for_international_numbers(
mocker.patch("app.user.rest.send_notification_to_queue")
resp = client.post(
url_for("user.send_user_2fa_code", code_type="sms", user_id=sample_user.id),
url_for(
"user.send_user_2fa_code",
code_type=CodeType.SMS,
user_id=sample_user.id,
),
data=json.dumps({}),
headers=[("Content-Type", "application/json"), auth_header],
)

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

@@ -8,7 +8,13 @@ from flask import current_app, json
from app.dao import templates_dao
from app.dao.service_sms_sender_dao import dao_update_service_sms_sender
from app.enums import NotificationStatus, NotificationType, ServicePermissionType
from app.enums import (
KeyType,
NotificationStatus,
NotificationType,
ServicePermissionType,
TemplateType,
)
from app.models import Notification
from app.schema_validation import validate
from app.v2.errors import RateLimitError
@@ -64,16 +70,13 @@ def test_post_sms_notification_returns_201(
"body"
] == sample_template_with_placeholders.content.replace("(( Name))", "Jo")
assert resp_json["content"]["from_number"] == current_app.config["FROM_NUMBER"]
assert "v2/notifications/{}".format(notification_id) in resp_json["uri"]
assert f"v2/notifications/{notification_id}" in resp_json["uri"]
assert resp_json["template"]["id"] == str(sample_template_with_placeholders.id)
assert resp_json["template"]["version"] == sample_template_with_placeholders.version
assert (
"services/{}/templates/{}".format(
sample_template_with_placeholders.service_id,
sample_template_with_placeholders.id,
)
in resp_json["template"]["uri"]
)
f"services/{sample_template_with_placeholders.service_id}/templates/"
f"{sample_template_with_placeholders.service_id}"
) in resp_json["template"]["uri"]
assert not resp_json["scheduled_for"]
assert mocked.called
@@ -369,8 +372,8 @@ def test_should_return_template_if_found_in_redis(mocker, client, sample_templat
@pytest.mark.parametrize(
"notification_type, key_send_to, send_to",
[
("sms", "phone_number", "+447700900855"),
("email", "email_address", "sample@email.com"),
(NotificationType.SMS, "phone_number", "+447700900855"),
(NotificationType.EMAIL, "email_address", "sample@email.com"),
],
)
def test_post_notification_returns_400_and_missing_template(
@@ -380,7 +383,7 @@ def test_post_notification_returns_400_and_missing_template(
auth_header = create_service_authorization_header(service_id=sample_service.id)
response = client.post(
path="/v2/notifications/{}".format(notification_type),
path=f"/v2/notifications/{notification_type}",
data=json.dumps(data),
headers=[("Content-Type", "application/json"), auth_header],
)
@@ -398,8 +401,8 @@ def test_post_notification_returns_400_and_missing_template(
@pytest.mark.parametrize(
"notification_type, key_send_to, send_to",
[
("sms", "phone_number", "+447700900855"),
("email", "email_address", "sample@email.com"),
(NotificationType.SMS, "phone_number", "+447700900855"),
(NotificationType.EMAIL, "email_address", "sample@email.com"),
],
)
def test_post_notification_returns_401_and_well_formed_auth_error(
@@ -408,7 +411,7 @@ def test_post_notification_returns_401_and_well_formed_auth_error(
data = {key_send_to: send_to, "template_id": str(sample_template.id)}
response = client.post(
path="/v2/notifications/{}".format(notification_type),
path=f"/v2/notifications/{notification_type}",
data=json.dumps(data),
headers=[("Content-Type", "application/json")],
)
@@ -428,8 +431,8 @@ def test_post_notification_returns_401_and_well_formed_auth_error(
@pytest.mark.parametrize(
"notification_type, key_send_to, send_to",
[
("sms", "phone_number", "+447700900855"),
("email", "email_address", "sample@email.com"),
(NotificationType.SMS, "phone_number", "+447700900855"),
(NotificationType.EMAIL, "email_address", "sample@email.com"),
],
)
def test_notification_returns_400_and_for_schema_problems(
@@ -441,7 +444,7 @@ def test_notification_returns_400_and_for_schema_problems(
)
response = client.post(
path="/v2/notifications/{}".format(notification_type),
path=f"/v2/notifications/{notification_type}",
data=json.dumps(data),
headers=[("Content-Type", "application/json"), auth_header],
)
@@ -498,11 +501,11 @@ def test_post_email_notification_returns_201(
assert resp_json["content"][
"subject"
] == sample_email_template_with_placeholders.subject.replace("((name))", "Bob")
assert resp_json["content"]["from_email"] == "{}@{}".format(
sample_email_template_with_placeholders.service.email_from,
current_app.config["NOTIFY_EMAIL_DOMAIN"],
assert resp_json["content"]["from_email"] == (
f"{sample_email_template_with_placeholders.service.email_from}@"
f"{current_app.config['NOTIFY_EMAIL_DOMAIN']}"
)
assert "v2/notifications/{}".format(notification.id) in resp_json["uri"]
assert f"v2/notifications/{notification.id}" in resp_json["uri"]
assert resp_json["template"]["id"] == str(
sample_email_template_with_placeholders.id
)
@@ -511,12 +514,9 @@ def test_post_email_notification_returns_201(
== sample_email_template_with_placeholders.version
)
assert (
"services/{}/templates/{}".format(
str(sample_email_template_with_placeholders.service_id),
str(sample_email_template_with_placeholders.id),
)
in resp_json["template"]["uri"]
)
f"services/{sample_email_template_with_placeholders.service_id}/templates/"
f"{sample_email_template_with_placeholders.id}"
) in resp_json["template"]["uri"]
assert not resp_json["scheduled_for"]
assert mocked.called
@@ -527,8 +527,8 @@ def test_post_email_notification_returns_201(
("simulate-delivered@notifications.service.gov.uk", NotificationType.EMAIL),
("simulate-delivered-2@notifications.service.gov.uk", NotificationType.EMAIL),
("simulate-delivered-3@notifications.service.gov.uk", NotificationType.EMAIL),
("+14254147167", "sms"),
("+14254147755", "sms"),
("+14254147167", NotificationType.SMS),
("+14254147755", NotificationType.SMS),
],
)
def test_should_not_persist_or_send_notification_if_simulated_recipient(
@@ -538,7 +538,7 @@ def test_should_not_persist_or_send_notification_if_simulated_recipient(
"app.celery.provider_tasks.deliver_{}.apply_async".format(notification_type)
)
if notification_type == "sms":
if notification_type == NotificationType.SMS:
data = {"phone_number": recipient, "template_id": str(sample_template.id)}
else:
data = {
@@ -551,7 +551,7 @@ def test_should_not_persist_or_send_notification_if_simulated_recipient(
)
response = client.post(
path="/v2/notifications/{}".format(notification_type),
path=f"/v2/notifications/{notification_type}",
data=json.dumps(data),
headers=[("Content-Type", "application/json"), auth_header],
)
@@ -565,22 +565,25 @@ def test_should_not_persist_or_send_notification_if_simulated_recipient(
@pytest.mark.parametrize(
"notification_type, key_send_to, send_to",
[
("sms", "phone_number", "2028675309"),
("email", "email_address", "sample@email.com"),
(NotificationType.SMS, "phone_number", "2028675309"),
(NotificationType.EMAIL, "email_address", "sample@email.com"),
],
)
def test_send_notification_uses_priority_queue_when_template_is_marked_as_priority(
client, sample_service, mocker, notification_type, key_send_to, send_to
client,
sample_service,
mocker,
notification_type,
key_send_to,
send_to,
):
mocker.patch(
"app.celery.provider_tasks.deliver_{}.apply_async".format(notification_type)
)
mocker.patch(f"app.celery.provider_tasks.deliver_{notification_type}.apply_async")
sample = create_template(
service=sample_service, template_type=notification_type, process_type="priority"
)
mocked = mocker.patch(
"app.celery.provider_tasks.deliver_{}.apply_async".format(notification_type)
f"app.celery.provider_tasks.deliver_{notification_type}.apply_async"
)
data = {key_send_to: send_to, "template_id": str(sample.id)}
@@ -588,7 +591,7 @@ def test_send_notification_uses_priority_queue_when_template_is_marked_as_priori
auth_header = create_service_authorization_header(service_id=sample.service_id)
response = client.post(
path="/v2/notifications/{}".format(notification_type),
path=f"/v2/notifications/{notification_type}",
data=json.dumps(data),
headers=[("Content-Type", "application/json"), auth_header],
)
@@ -602,8 +605,8 @@ def test_send_notification_uses_priority_queue_when_template_is_marked_as_priori
@pytest.mark.parametrize(
"notification_type, key_send_to, send_to",
[
("sms", "phone_number", "2028675309"),
("email", "email_address", "sample@email.com"),
(NotificationType.SMS, "phone_number", "2028675309"),
(NotificationType.EMAIL, "email_address", "sample@email.com"),
],
)
def test_returns_a_429_limit_exceeded_if_rate_limit_exceeded(
@@ -626,7 +629,7 @@ def test_returns_a_429_limit_exceeded_if_rate_limit_exceeded(
auth_header = create_service_authorization_header(service_id=sample.service_id)
response = client.post(
path="/v2/notifications/{}".format(notification_type),
path=f"/v2/notifications/{notification_type}",
data=json.dumps(data),
headers=[("Content-Type", "application/json"), auth_header],
)
@@ -698,19 +701,29 @@ def test_post_sms_notification_with_archived_reply_to_id_returns_400(
assert response.status_code == 400
resp_json = json.loads(response.get_data(as_text=True))
assert (
"sms_sender_id {} does not exist in database for service id {}".format(
archived_sender.id, sample_template.service_id
)
in resp_json["errors"][0]["message"]
)
f"sms_sender_id {archived_sender.id} does not exist in database for "
f"service id {sample_template.service_id}"
) in resp_json["errors"][0]["message"]
assert "BadRequestError" in resp_json["errors"][0]["error"]
@pytest.mark.parametrize(
"recipient,label,permission_type, notification_type,expected_error",
[
("2028675309", "phone_number", "email", "sms", "text messages"),
("someone@test.com", "email_address", "sms", "email", "emails"),
(
"2028675309",
"phone_number",
ServicePermissionType.EMAIL,
NotificationType.SMS,
"text messages",
),
(
"someone@test.com",
"email_address",
ServicePermissionType.SMS,
NotificationType.EMAIL,
"emails",
),
],
)
def test_post_sms_notification_returns_400_if_not_allowed_to_send_notification(
@@ -732,9 +745,7 @@ def test_post_sms_notification_returns_400_if_not_allowed_to_send_notification(
)
response = client.post(
path="/v2/notifications/{}".format(
sample_template_without_permission.template_type
),
path=f"/v2/notifications/{sample_template_without_permission.template_type}",
data=json.dumps(data),
headers=[("Content-Type", "application/json"), auth_header],
)
@@ -747,7 +758,7 @@ def test_post_sms_notification_returns_400_if_not_allowed_to_send_notification(
assert error_json["errors"] == [
{
"error": "BadRequestError",
"message": "Service is not allowed to send {}".format(expected_error),
"message": f"Service is not allowed to send {expected_error}",
}
]
@@ -764,14 +775,15 @@ def test_post_sms_notification_returns_400_if_number_not_in_guest_list(
],
)
template = create_template(service=service)
create_api_key(service=service, key_type="team")
create_api_key(service=service, key_type=KeyType.TEAM)
data = {
"phone_number": "+327700900855",
"template_id": template.id,
}
auth_header = create_service_authorization_header(
service_id=service.id, key_type="team"
service_id=service.id,
key_type=KeyType.TEAM,
)
response = client.post(
@@ -855,7 +867,9 @@ def test_post_notification_raises_bad_request_if_not_valid_notification_type(
assert "The requested URL was not found on the server." in error_json["message"]
@pytest.mark.parametrize("notification_type", ["sms", "email"])
@pytest.mark.parametrize(
"notification_type", [NotificationType.SMS, NotificationType.EMAIL]
)
def test_post_notification_with_wrong_type_of_sender(
client, sample_template, sample_email_template, notification_type, fake_uuid
):
@@ -878,14 +892,14 @@ def test_post_notification_with_wrong_type_of_sender(
auth_header = create_service_authorization_header(service_id=template.service_id)
response = client.post(
path="/v2/notifications/{}".format(notification_type),
path=f"/v2/notifications/{notification_type}",
data=json.dumps(data),
headers=[("Content-Type", "application/json"), auth_header],
)
assert response.status_code == 400
resp_json = json.loads(response.get_data(as_text=True))
assert (
"Additional properties are not allowed ({} was unexpected)".format(form_label)
f"Additional properties are not allowed ({form_label} was unexpected)"
in resp_json["errors"][0]["message"]
)
assert "ValidationError" in resp_json["errors"][0]["error"]
@@ -942,11 +956,9 @@ def test_post_email_notification_with_invalid_reply_to_id_returns_400(
assert response.status_code == 400
resp_json = json.loads(response.get_data(as_text=True))
assert (
"email_reply_to_id {} does not exist in database for service id {}".format(
fake_uuid, sample_email_template.service_id
)
in resp_json["errors"][0]["message"]
)
f"email_reply_to_id {fake_uuid} does not exist in database for service id "
f"{sample_email_template.service_id}"
) in resp_json["errors"][0]["message"]
assert "BadRequestError" in resp_json["errors"][0]["error"]
@@ -976,11 +988,9 @@ def test_post_email_notification_with_archived_reply_to_id_returns_400(
assert response.status_code == 400
resp_json = json.loads(response.get_data(as_text=True))
assert (
"email_reply_to_id {} does not exist in database for service id {}".format(
archived_reply_to.id, sample_email_template.service_id
)
in resp_json["errors"][0]["message"]
)
f"email_reply_to_id {archived_reply_to.id} does not exist in database for "
f"service id {sample_email_template.service_id}"
) in resp_json["errors"][0]["message"]
assert "BadRequestError" in resp_json["errors"][0]["error"]
@@ -1000,7 +1010,7 @@ def test_post_notification_with_document_upload(
service.contact_link = "contact.me@gov.uk"
template = create_template(
service=service,
template_type="email",
template_type=TemplateType.EMAIL,
content="Document 1: ((first_link)). Document 2: ((second_link))",
)
@@ -1057,7 +1067,9 @@ def test_post_notification_with_document_upload_simulated(
service = create_service(service_permissions=[ServicePermissionType.EMAIL])
service.contact_link = "contact.me@gov.uk"
template = create_template(
service=service, template_type="email", content="Document: ((document))"
service=service,
template_type=TemplateType.EMAIL,
content="Document: ((document))",
)
mocker.patch("app.celery.provider_tasks.deliver_email.apply_async")
@@ -1093,7 +1105,9 @@ def test_post_notification_without_document_upload_permission(
):
service = create_service(service_permissions=[ServicePermissionType.EMAIL])
template = create_template(
service=service, template_type="email", content="Document: ((document))"
service=service,
template_type=TemplateType.EMAIL,
content="Document: ((document))",
)
mocker.patch("app.celery.provider_tasks.deliver_email.apply_async")
@@ -1135,21 +1149,24 @@ def test_post_notification_returns_400_when_get_json_throws_exception(
@pytest.mark.parametrize(
"notification_type, content_type",
[
("email", "application/json"),
("email", "application/text"),
("sms", "application/json"),
("sms", "application/text"),
(NotificationType.EMAIL, "application/json"),
(NotificationType.EMAIL, "application/text"),
(NotificationType.SMS, "application/json"),
(NotificationType.SMS, "application/text"),
],
)
def test_post_notification_when_payload_is_invalid_json_returns_400(
client, sample_service, notification_type, content_type
client,
sample_service,
notification_type,
content_type,
):
auth_header = create_service_authorization_header(service_id=sample_service.id)
payload_not_json = {
"template_id": "dont-convert-to-json",
}
response = client.post(
path="/v2/notifications/{}".format(notification_type),
path=f"/v2/notifications/{notification_type}",
data=payload_not_json,
headers=[("Content-Type", content_type), auth_header],
)
@@ -1160,14 +1177,18 @@ def test_post_notification_when_payload_is_invalid_json_returns_400(
assert error_msg == "Invalid JSON supplied in POST data"
@pytest.mark.parametrize("notification_type", ["email", "sms"])
@pytest.mark.parametrize(
"notification_type",
[
NotificationType.EMAIL,
NotificationType.SMS,
],
)
def test_post_notification_returns_201_when_content_type_is_missing_but_payload_is_valid_json(
client, sample_service, notification_type, mocker
):
template = create_template(service=sample_service, template_type=notification_type)
mocker.patch(
"app.celery.provider_tasks.deliver_{}.apply_async".format(notification_type)
)
mocker.patch(f"app.celery.provider_tasks.deliver_{notification_type}.apply_async")
auth_header = create_service_authorization_header(service_id=sample_service.id)
valid_json = {
@@ -1178,33 +1199,45 @@ def test_post_notification_returns_201_when_content_type_is_missing_but_payload_
else:
valid_json.update({"phone_number": "+447700900855"})
response = client.post(
path="/v2/notifications/{}".format(notification_type),
path=f"/v2/notifications/{notification_type}",
data=json.dumps(valid_json),
headers=[auth_header],
)
assert response.status_code == 201
@pytest.mark.parametrize("notification_type", ["email", "sms"])
@pytest.mark.parametrize(
"notification_type",
[
NotificationType.EMAIL,
NotificationType.SMS,
],
)
def test_post_email_notification_when_data_is_empty_returns_400(
client, sample_service, notification_type
):
auth_header = create_service_authorization_header(service_id=sample_service.id)
data = None
response = client.post(
path="/v2/notifications/{}".format(notification_type),
path=f"/v2/notifications/{notification_type}",
data=json.dumps(data),
headers=[("Content-Type", "application/json"), auth_header],
)
error_msg = json.loads(response.get_data(as_text=True))["errors"][0]["message"]
assert response.status_code == 400
if notification_type == "sms":
if notification_type == NotificationType.SMS:
assert error_msg == "phone_number is a required property"
else:
assert error_msg == "email_address is a required property"
@pytest.mark.parametrize("notification_type", ("email", "sms"))
@pytest.mark.parametrize(
"notification_type",
(
NotificationType.EMAIL,
NotificationType.SMS,
),
)
def test_post_notifications_saves_email_or_sms_to_queue(
client, notify_db_session, mocker, notification_type
):
@@ -1266,7 +1299,13 @@ def test_post_notifications_saves_email_or_sms_to_queue(
botocore.parsers.ResponseParserError("exceeded max HTTP body length"),
],
)
@pytest.mark.parametrize("notification_type", ("email", "sms"))
@pytest.mark.parametrize(
"notification_type",
(
NotificationType.EMAIL,
NotificationType.SMS,
),
)
def test_post_notifications_saves_email_or_sms_normally_if_saving_to_queue_fails(
client, notify_db_session, mocker, notification_type, exception
):
@@ -1324,7 +1363,13 @@ def test_post_notifications_saves_email_or_sms_normally_if_saving_to_queue_fails
assert Notification.query.count() == 1
@pytest.mark.parametrize("notification_type", ("email", "sms"))
@pytest.mark.parametrize(
"notification_type",
(
NotificationType.EMAIL,
NotificationType.SMS,
),
)
def test_post_notifications_doesnt_use_save_queue_for_test_notifications(
client, notify_db_session, mocker, notification_type
):