From 74ac09ae55a90220da999c7798718deadfb4c6ac Mon Sep 17 00:00:00 2001 From: Cliff Hill Date: Tue, 20 Feb 2024 17:00:54 -0500 Subject: [PATCH] Even more cleanup. Signed-off-by: Cliff Hill --- tests/app/user/test_rest.py | 6 +- tests/app/user/test_rest_verify.py | 36 ++- .../notifications/test_post_notifications.py | 221 +++++++++++------- 3 files changed, 162 insertions(+), 101 deletions(-) diff --git a/tests/app/user/test_rest.py b/tests/app/user/test_rest.py index c24b950d8..801a18f38 100644 --- a/tests/app/user/test_rest.py +++ b/tests/app/user/test_rest.py @@ -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", diff --git a/tests/app/user/test_rest_verify.py b/tests/app/user/test_rest_verify.py index 01c28bb25..8bff45391 100644 --- a/tests/app/user/test_rest_verify.py +++ b/tests/app/user/test_rest_verify.py @@ -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], ) diff --git a/tests/app/v2/notifications/test_post_notifications.py b/tests/app/v2/notifications/test_post_notifications.py index 98289dc62..4a90c5372 100644 --- a/tests/app/v2/notifications/test_post_notifications.py +++ b/tests/app/v2/notifications/test_post_notifications.py @@ -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 ):