darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-08-18 21:48:49 -04:00
Code Issues Packages Projects Releases Wiki Activity

Reformatting.

Browse Source 
Signed-off-by: Cliff Hill <Clifford.hill@gsa.gov>
This commit is contained in:
Cliff Hill
2024-02-21 10:24:45 -05:00
parent 6f5f5fb864
commit 15eeac6367
10 changed files with 194 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
tests/app/celery/test_nightly_tasks.py
View File

@@ -104,10 +104,16 @@ def test_will_remove_csv_files_for_jobs_older_than_retention_period(
days_of_retention=30,
)
sms_template_service_1 = create_template(service=service_1)
email_template_service_1 = create_template(service=service_1, template_type=TemplateType.EMAIL,)
email_template_service_1 = create_template(
service=service_1,
template_type=TemplateType.EMAIL,
)
sms_template_service_2 = create_template(service=service_2)
email_template_service_2 = create_template(service=service_2, template_type=TemplateType.EMAIL,)
email_template_service_2 = create_template(
service=service_2,
template_type=TemplateType.EMAIL,
)
four_days_ago = datetime.utcnow() - timedelta(days=4)
eight_days_ago = datetime.utcnow() - timedelta(days=8)
@@ -352,10 +358,12 @@ def test_delete_notifications_task_calls_task_for_services_that_have_sent_notifi
create_template(service_will_delete_1)
create_template(service_will_delete_2)
nothing_to_delete_sms_template = create_template(
service_nothing_to_delete, template_type=TemplateType.SMS,
service_nothing_to_delete,
template_type=TemplateType.SMS,
)
nothing_to_delete_email_template = create_template(
service_nothing_to_delete, template_type=TemplateType.EMAIL,
service_nothing_to_delete,
template_type=TemplateType.EMAIL,
)
# will be deleted as service has no custom retention, but past our default 7 days

29
tests/app/celery/test_service_callback_tasks.py
View File

@@ -62,16 +62,20 @@ def test_send_delivery_status_to_service_post_https_request_to_service_with_encr
assert request_mock.request_history[0].method == "POST"
assert request_mock.request_history[0].text == json.dumps(mock_data)
assert request_mock.request_history[0].headers["Content-type"] == "application/json"
assert request_mock.request_history[0].headers[
"Authorization"
] == f"Bearer {callback_api.bearer_token}"
assert (
request_mock.request_history[0].headers["Authorization"]
== f"Bearer {callback_api.bearer_token}"
)
def test_send_complaint_to_service_posts_https_request_to_service_with_encrypted_data(
notify_db_session,
):
with freeze_time("2001-01-01T12:00:00"):
callback_api, template = _set_up_test_data(NotificationType.EMAIL, CallbackType.COMPLAINT,)
callback_api, template = _set_up_test_data(
NotificationType.EMAIL,
CallbackType.COMPLAINT,
)
notification = create_notification(template=template)
complaint = create_complaint(
@@ -100,9 +104,10 @@ def test_send_complaint_to_service_posts_https_request_to_service_with_encrypted
request_mock.request_history[0].headers["Content-type"]
== "application/json"
)
assert request_mock.request_history[0].headers[
"Authorization"
] == f"Bearer {callback_api.bearer_token}"
assert (
request_mock.request_history[0].headers["Authorization"]
== f"Bearer {callback_api.bearer_token}"
)
@pytest.mark.parametrize(
@@ -113,7 +118,10 @@ def test_send_complaint_to_service_posts_https_request_to_service_with_encrypted
def test__send_data_to_service_callback_api_retries_if_request_returns_error_code_with_encrypted_data(
notify_db_session, mocker, notification_type, status_code
):
callback_api, template = _set_up_test_data(notification_type, CallbackType.DELIVERY_STATUS,)
callback_api, template = _set_up_test_data(
notification_type,
CallbackType.DELIVERY_STATUS,
)
datestr = datetime(2017, 6, 20)
notification = create_notification(
template=template,
@@ -143,7 +151,10 @@ def test__send_data_to_service_callback_api_retries_if_request_returns_error_cod
def test__send_data_to_service_callback_api_does_not_retry_if_request_returns_404_with_encrypted_data(
notify_db_session, mocker, notification_type
):
callback_api, template = _set_up_test_data(notification_type, CallbackType.DELIVERY_STATUS,)
callback_api, template = _set_up_test_data(
notification_type,
CallbackType.DELIVERY_STATUS,
)
datestr = datetime(2017, 6, 20)
notification = create_notification(
template=template,

11
tests/app/celery/test_tasks.py
View File

@@ -310,9 +310,7 @@ def test_process_row_sends_letter_task(
template_type, expected_function, expected_queue, mocker
):
mocker.patch("app.celery.tasks.create_uuid", return_value="noti_uuid")
task_mock = mocker.patch(
f"app.celery.tasks.{expected_function}.apply_async"
)
task_mock = mocker.patch(f"app.celery.tasks.{expected_function}.apply_async")
encrypt_mock = mocker.patch("app.celery.tasks.encryption.encrypt")
template = Mock(id="template_id", template_type=template_type)
job = Mock(id="job_id", template_version="temp_vers")
@@ -1039,9 +1037,10 @@ def test_send_inbound_sms_to_service_post_https_request_to_service(
assert request_mock.request_history[0].method == "POST"
assert request_mock.request_history[0].text == json.dumps(data)
assert request_mock.request_history[0].headers["Content-type"] == "application/json"
assert request_mock.request_history[0].headers[
"Authorization"
] == f"Bearer {inbound_api.bearer_token}"
assert (
request_mock.request_history[0].headers["Authorization"]
== f"Bearer {inbound_api.bearer_token}"
)
def test_send_inbound_sms_to_service_does_not_send_request_when_inbound_sms_does_not_exist(

5
tests/app/conftest.py
View File

@@ -863,7 +863,10 @@ def sample_inbound_numbers(sample_service):
inbound_numbers.append(create_inbound_number(number="1", provider="sns"))
inbound_numbers.append(
create_inbound_number(
number="2", provider="sns", active=False, service_id=service.id,
number="2",
provider="sns",
active=False,
service_id=service.id,
)
)
return inbound_numbers

6
tests/app/dao/test_complaint_dao.py
View File

@@ -139,13 +139,15 @@ def test_fetch_count_of_complaints(sample_email_notification):
)
count_of_complaints = fetch_count_of_complaints(
start_date=datetime(2018, 6, 7), end_date=datetime(2018, 6, 7),
start_date=datetime(2018, 6, 7),
end_date=datetime(2018, 6, 7),
)
assert count_of_complaints == 5
def test_fetch_count_of_complaints_returns_zero(notify_db_session):
count_of_complaints = fetch_count_of_complaints(
start_date=datetime(2018, 6, 7), end_date=datetime(2018, 6, 7),
start_date=datetime(2018, 6, 7),
end_date=datetime(2018, 6, 7),
)
assert count_of_complaints == 0

6
tests/app/dao/test_service_data_retention_dao.py
View File

@@ -57,7 +57,11 @@ def test_fetch_service_data_retention_returns_empty_list_when_no_rows_for_servic
def test_fetch_service_data_retention_by_id(sample_service):
email_data_retention = insert_service_data_retention(sample_service.id, NotificationType.EMAIL, 3,)
email_data_retention = insert_service_data_retention(
sample_service.id,
NotificationType.EMAIL,
3,
)
insert_service_data_retention(sample_service.id, NotificationType.SMS, 13)
result = fetch_service_data_retention_by_id(
sample_service.id, email_data_retention.id

29
tests/app/db.py
View File

@@ -140,9 +140,7 @@ def create_service(
else service_name.lower().replace(" ", "."),
created_by=user
if user
else create_user(
email=f"{uuid.uuid4()}@digital.cabinet-office.gov.uk"
),
else create_user(email=f"{uuid.uuid4()}@digital.cabinet-office.gov.uk"),
prefix_sms=prefix_sms,
organization_type=organization_type,
organization=organization,
@@ -498,7 +496,11 @@ def create_service_callback_api(
def create_email_branding(
id=None, colour="blue", logo="test_x2.png", name="test_org_1", text="DisplayName",
id=None,
colour="blue",
logo="test_x2.png",
name="test_org_1",
text="DisplayName",
):
data = {
"colour": colour,
@@ -644,7 +646,9 @@ def create_organization(
def create_invited_org_user(
organization, invited_by, email_address="invite@example.com",
organization,
invited_by,
email_address="invite@example.com",
):
invited_org_user = InvitedOrganizationUser(
email_address=email_address,
@@ -918,13 +922,22 @@ def set_up_usage_data(start_date):
)
create_ft_billing(
local_date=one_week_earlier, template=sms_template_1, billable_unit=2, rate=0.11,
local_date=one_week_earlier,
template=sms_template_1,
billable_unit=2,
rate=0.11,
)
create_ft_billing(
local_date=start_date, template=sms_template_1, billable_unit=2, rate=0.11,
local_date=start_date,
template=sms_template_1,
billable_unit=2,
rate=0.11,
)
create_ft_billing(
local_date=two_days_later, template=sms_template_1, billable_unit=1, rate=0.11,
local_date=two_days_later,
template=sms_template_1,
billable_unit=1,
rate=0.11,
)
# service with emails only:

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

@@ -527,7 +527,9 @@ def test_check_service_over_api_rate_limit_when_rate_limit_has_not_exceeded_limi
check_service_over_api_rate_limit(serialised_service, serialised_api_key)
assert app.redis_store.exceeded_rate_limit.called_with(
f"{sample_service.id}-{api_key.key_type}", 3000, 60,
f"{sample_service.id}-{api_key.key_type}",
3000,
60,
)
@@ -597,7 +599,10 @@ def test_validate_and_format_recipient_succeeds_with_international_numbers_if_se
def test_validate_and_format_recipient_fails_when_no_recipient():
with pytest.raises(BadRequestError) as e:
validate_and_format_recipient(
None, KeyType.NORMAL, "service", NotificationType.SMS,
None,
KeyType.NORMAL,
"service",
NotificationType.SMS,
)
assert e.value.status_code == 400
assert e.value.message == "Recipient can't be empty"
@@ -630,12 +635,9 @@ def test_check_service_email_reply_to_id_where_service_id_is_not_found(
fake_uuid, reply_to_address.id, NotificationType.EMAIL
)
assert e.value.status_code == 400
assert (
e.value.message
== (
f"email_reply_to_id {reply_to_address.id} does not exist in database for i"
f"service id {fake_uuid}"
)
assert e.value.message == (
f"email_reply_to_id {reply_to_address.id} does not exist in database for i"
f"service id {fake_uuid}"
)
@@ -647,12 +649,9 @@ def test_check_service_email_reply_to_id_where_reply_to_id_is_not_found(
sample_service.id, fake_uuid, NotificationType.EMAIL
)
assert e.value.status_code == 400
assert (
e.value.message
== (
f"email_reply_to_id {fake_uuid} does not exist in database for service "
f"id {sample_service.id}"
)
assert e.value.message == (
f"email_reply_to_id {fake_uuid} does not exist in database for service "
f"id {sample_service.id}"
)
@@ -683,12 +682,9 @@ def test_check_service_sms_sender_id_where_service_id_is_not_found(
with pytest.raises(BadRequestError) as e:
check_service_sms_sender_id(fake_uuid, sms_sender.id, NotificationType.SMS)
assert e.value.status_code == 400
assert (
e.value.message
== (
f"sms_sender_id {sms_sender.id} does not exist in database for service "
f"id {fake_uuid}"
)
assert e.value.message == (
f"sms_sender_id {sms_sender.id} does not exist in database for service "
f"id {fake_uuid}"
)
@@ -698,12 +694,9 @@ def test_check_service_sms_sender_id_where_sms_sender_is_not_found(
with pytest.raises(BadRequestError) as e:
check_service_sms_sender_id(sample_service.id, fake_uuid, NotificationType.SMS)
assert e.value.status_code == 400
assert (
e.value.message
== (
f"sms_sender_id {fake_uuid} does not exist in database for service "
f"id {sample_service.id}"
)
assert e.value.message == (
f"sms_sender_id {fake_uuid} does not exist in database for service "
f"id {sample_service.id}"
)

121
tests/app/service/test_rest.py
View File

@@ -151,9 +151,10 @@ def test_find_services_by_name_handles_no_results(
mock_get_services_by_partial_name = mocker.patch(
"app.service.rest.get_services_by_partial_name", return_value=[]
)
response = admin_request.get("service.find_services_by_name", service_name="ABC",)[
"data"
]
response = admin_request.get(
"service.find_services_by_name",
service_name="ABC",
)["data"]
mock_get_services_by_partial_name.assert_called_once_with("ABC")
assert len(response) == 0
@@ -1221,7 +1222,8 @@ def test_default_permissions_are_added_for_user_service(
assert resp.status_code == 200
header = create_admin_authorization_header()
response = client.get(
url_for("user.get_user", user_id=sample_user.id), headers=[header],
url_for("user.get_user", user_id=sample_user.id),
headers=[header],
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
@@ -1517,7 +1519,14 @@ def test_add_existing_user_to_non_existing_service_returns404(
incorrect_id = uuid.uuid4()
data = {
"permissions": [PermissionType.SEND_EMAILS, PermissionType.SEND_TEXTS, PermissionType.MANAGE_USERS, PermissionType.MANAGE_SETTINGS, PermissionType.MANAGE_TEMPLATES, PermissionType.MANAGE_API_KEYS,]
"permissions": [
PermissionType.SEND_EMAILS,
PermissionType.SEND_TEXTS,
PermissionType.MANAGE_USERS,
PermissionType.MANAGE_SETTINGS,
PermissionType.MANAGE_TEMPLATES,
PermissionType.MANAGE_API_KEYS,
]
}
auth_header = create_admin_authorization_header()
@@ -1543,7 +1552,14 @@ def test_add_existing_user_of_service_to_service_returns400(
existing_user_id = sample_service.users[0].id
data = {
"permissions": [PermissionType.SEND_EMAILS, PermissionType.SEND_TEXTS, PermissionType.MANAGE_USERS, PermissionType.MANAGE_SETTINGS, PermissionType.MANAGE_TEMPLATES, PermissionType.MANAGE_API_KEYS,]
"permissions": [
PermissionType.SEND_EMAILS,
PermissionType.SEND_TEXTS,
PermissionType.MANAGE_USERS,
PermissionType.MANAGE_SETTINGS,
PermissionType.MANAGE_TEMPLATES,
PermissionType.MANAGE_API_KEYS,
]
}
auth_header = create_admin_authorization_header()
@@ -1572,7 +1588,14 @@ def test_add_unknown_user_to_service_returns404(
incorrect_id = 9876
data = {
"permissions": [PermissionType.SEND_EMAILS, PermissionType.SEND_TEXTS, PermissionType.MANAGE_USERS, PermissionType.MANAGE_SETTINGS, PermissionType.MANAGE_TEMPLATES, PermissionType.MANAGE_API_KEYS,]
"permissions": [
PermissionType.SEND_EMAILS,
PermissionType.SEND_TEXTS,
PermissionType.MANAGE_USERS,
PermissionType.MANAGE_SETTINGS,
PermissionType.MANAGE_TEMPLATES,
PermissionType.MANAGE_API_KEYS,
]
}
auth_header = create_admin_authorization_header()
@@ -2206,13 +2229,29 @@ def test_get_detailed_services_groups_by_service(notify_db_session):
assert len(data) == 2
assert data[0]["id"] == str(service_1.id)
assert data[0]["statistics"] == {
NotificationType.EMAIL.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 0,},
NotificationType.SMS.value: {NotificationStatus.DELIVERED: 1, NotificationStatus.FAILED: 0, "requested": 3,},
NotificationType.EMAIL.value: {
NotificationStatus.DELIVERED: 0,
NotificationStatus.FAILED: 0,
"requested": 0,
},
NotificationType.SMS.value: {
NotificationStatus.DELIVERED: 1,
NotificationStatus.FAILED: 0,
"requested": 3,
},
}
assert data[1]["id"] == str(service_2.id)
assert data[1]["statistics"] == {
NotificationType.EMAIL.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 0,},
NotificationType.SMS.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 1,},
NotificationType.EMAIL.value: {
NotificationStatus.DELIVERED: 0,
NotificationStatus.FAILED: 0,
"requested": 0,
},
NotificationType.SMS.value: {
NotificationStatus.DELIVERED: 0,
NotificationStatus.FAILED: 0,
"requested": 1,
},
}
@@ -2235,13 +2274,29 @@ def test_get_detailed_services_includes_services_with_no_notifications(
assert len(data) == 2
assert data[0]["id"] == str(service_1.id)
assert data[0]["statistics"] == {
NotificationType.EMAIL.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 0,},
NotificationType.SMS.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 1,},
NotificationType.EMAIL.value: {
NotificationStatus.DELIVERED: 0,
NotificationStatus.FAILED: 0,
"requested": 0,
},
NotificationType.SMS.value: {
NotificationStatus.DELIVERED: 0,
NotificationStatus.FAILED: 0,
"requested": 1,
},
}
assert data[1]["id"] == str(service_2.id)
assert data[1]["statistics"] == {
NotificationType.EMAIL.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 0,},
NotificationType.SMS.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 0,},
NotificationType.EMAIL.value: {
NotificationStatus.DELIVERED: 0,
NotificationStatus.FAILED: 0,
"requested": 0,
},
NotificationType.SMS.value: {
NotificationStatus.DELIVERED: 0,
NotificationStatus.FAILED: 0,
"requested": 0,
},
}
@@ -2261,8 +2316,16 @@ def test_get_detailed_services_only_includes_todays_notifications(sample_templat
assert len(data) == 1
assert data[0]["statistics"] == {
NotificationType.EMAIL.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 0,},
NotificationType.SMS.value: {NotificationStatus.DELIVERED: 0, NotificationStatus.FAILED: 0, "requested": 3,},
NotificationType.EMAIL.value: {
NotificationStatus.DELIVERED: 0,
NotificationStatus.FAILED: 0,
"requested": 0,
},
NotificationType.SMS.value: {
NotificationStatus.DELIVERED: 0,
NotificationStatus.FAILED: 0,
"requested": 3,
},
}
@@ -2372,16 +2435,24 @@ def test_search_for_notification_by_to_field_return_multiple_matches(
client, sample_template, sample_email_template
):
notification1 = create_notification(
sample_template, to_field="+447700900855", normalised_to="447700900855",
sample_template,
to_field="+447700900855",
normalised_to="447700900855",
)
notification2 = create_notification(
sample_template, to_field=" +44 77009 00855 ", normalised_to="447700900855",
sample_template,
to_field=" +44 77009 00855 ",
normalised_to="447700900855",
)
notification3 = create_notification(
sample_template, to_field="+44770 0900 855", normalised_to="447700900855",
sample_template,
to_field="+44770 0900 855",
normalised_to="447700900855",
)
notification4 = create_notification(
sample_email_template, to_field="jack@gmail.com", normalised_to="jack@gmail.com",
sample_email_template,
to_field="jack@gmail.com",
normalised_to="jack@gmail.com",
)
response = client.get(
@@ -2409,7 +2480,9 @@ def test_search_for_notification_by_to_field_returns_next_link_if_more_than_50(
):
for _ in range(51):
create_notification(
sample_template, to_field="+447700900855", normalised_to="447700900855",
sample_template,
to_field="+447700900855",
normalised_to="447700900855",
)
response = client.get(
@@ -2433,7 +2506,9 @@ def test_search_for_notification_by_to_field_returns_no_next_link_if_50_or_less(
):
for _ in range(50):
create_notification(
sample_template, to_field="+447700900855", normalised_to="447700900855",
sample_template,
to_field="+447700900855",
normalised_to="447700900855",
)
response = client.get(

8
tests/app/test_model.py
View File

@@ -42,7 +42,9 @@ from tests.app.db import (
@pytest.mark.parametrize("mobile_number", ["+447700900855", "+12348675309"])
def test_should_build_service_guest_list_from_mobile_number(mobile_number):
service_guest_list = ServiceGuestList.from_string(
"service_id", RecipientType.MOBILE, mobile_number,
"service_id",
RecipientType.MOBILE,
mobile_number,
)
assert service_guest_list.recipient == mobile_number
@@ -51,7 +53,9 @@ def test_should_build_service_guest_list_from_mobile_number(mobile_number):
@pytest.mark.parametrize("email_address", ["test@example.com"])
def test_should_build_service_guest_list_from_email_address(email_address):
service_guest_list = ServiceGuestList.from_string(
"service_id", RecipientType.EMAIL, email_address,
"service_id",
RecipientType.EMAIL,
email_address,
)
assert service_guest_list.recipient == email_address