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

Cleaning house a bit.

Browse Source 
Signed-off-by: Cliff Hill <Clifford.hill@gsa.gov>
This commit is contained in:
Cliff Hill
2024-02-13 16:22:26 -05:00
parent 64adb09182
commit cf1db4dbf4
3 changed files with 182 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
tests/app/delivery/test_send_to_providers.py
View File

@@ -225,7 +225,8 @@ def test_should_have_sending_status_if_fake_callback_function_fails(
sample_notification, mocker
):
mocker.patch(
"app.delivery.send_to_providers.send_sms_response", side_effect=HTTPError,
"app.delivery.send_to_providers.send_sms_response",
side_effect=HTTPError,
)
sample_notification.key_type = KeyType.TEST
@@ -238,7 +239,10 @@ def test_should_have_sending_status_if_fake_callback_function_fails(
def test_should_not_send_to_provider_when_status_is_not_created(
sample_template, mocker
):
notification = create_notification(template=sample_template, status=NotificationStatus.SENDING,)
notification = create_notification(
template=sample_template,
status=NotificationStatus.SENDING,
)
mocker.patch("app.aws_sns_client.send_sms")
response_mock = mocker.patch("app.delivery.send_to_providers.send_sms_response")
@@ -307,7 +311,9 @@ def test_send_email_to_provider_should_not_send_to_provider_when_status_is_not_c
mock_redis = mocker.patch("app.delivery.send_to_providers.redis_store")
mock_redis.get.return_value = "test@example.com".encode("utf-8")
notification = create_notification(template=sample_email_template, status=NotificationStatus.SENDING)
notification = create_notification(
template=sample_email_template, status=NotificationStatus.SENDING
)
mocker.patch("app.aws_ses_client.send_email")
mocker.patch("app.delivery.send_to_providers.send_email_response")
@@ -332,7 +338,12 @@ def test_send_email_should_use_service_reply_to_email(
send_to_providers.send_email_to_provider(db_notification)
app.aws_ses_client.send_email.assert_called_once_with(
ANY, ANY, ANY, body=ANY, html_body=ANY, reply_to_address="foo@bar.com",
ANY,
ANY,
ANY,
body=ANY,
html_body=ANY,
reply_to_address="foo@bar.com",
)
@@ -391,7 +402,8 @@ def test_get_html_email_renderer_with_branding_details_and_render_govuk_banner_o
def test_get_html_email_renderer_prepends_logo_path(notify_api):
Service = namedtuple("Service", ["email_branding"])
EmailBranding = namedtuple(
"EmailBranding", ["brand_type", "colour", "name", "logo", "text"],
"EmailBranding",
["brand_type", "colour", "name", "logo", "text"],
)
email_branding = EmailBranding(
@@ -415,7 +427,8 @@ def test_get_html_email_renderer_prepends_logo_path(notify_api):
def test_get_html_email_renderer_handles_email_branding_without_logo(notify_api):
Service = namedtuple("Service", ["email_branding"])
EmailBranding = namedtuple(
"EmailBranding", ["brand_type", "colour", "name", "logo", "text"],
"EmailBranding",
["brand_type", "colour", "name", "logo", "text"],
)
email_branding = EmailBranding(
@@ -508,7 +521,10 @@ def test_should_update_billable_units_and_status_according_to_research_mode_and_
sample_template, mocker, research_mode, key_type, billable_units, expected_status
):
notification = create_notification(
template=sample_template, billable_units=0, status=NotificationStatus.CREATED, key_type=key_type,
template=sample_template,
billable_units=0,
status=NotificationStatus.CREATED,
key_type=key_type,
)
mocker.patch("app.aws_sns_client.send_sms")
mocker.patch(
@@ -623,13 +639,19 @@ def test_send_email_to_provider_uses_reply_to_from_notification(
mocker.patch("app.aws_ses_client.send_email", return_value="reference")
db_notification = create_notification(
template=sample_email_template, reply_to_text="test@test.com",
template=sample_email_template,
reply_to_text="test@test.com",
)
send_to_providers.send_email_to_provider(db_notification)
app.aws_ses_client.send_email.assert_called_once_with(
ANY, ANY, ANY, body=ANY, html_body=ANY, reply_to_address="test@test.com",
ANY,
ANY,
ANY,
body=ANY,
html_body=ANY,
reply_to_address="test@test.com",
)

22
tests/app/inbound_sms/test_rest.py
View File

@@ -132,7 +132,11 @@ def test_post_to_get_most_recent_inbound_sms_for_service_limits_to_a_week(
def test_post_to_get_inbound_sms_for_service_respects_data_retention(
admin_request, sample_service, days_of_retention, too_old_date, returned_date
):
create_service_data_retention(sample_service, NotificationType.SMS, days_of_retention,)
create_service_data_retention(
sample_service,
NotificationType.SMS,
days_of_retention,
)
create_inbound_sms(sample_service, created_at=too_old_date)
returned_inbound = create_inbound_sms(sample_service, created_at=returned_date)
@@ -206,10 +210,15 @@ def test_get_inbound_sms_by_id_with_invalid_service_id_returns_404(
@pytest.mark.parametrize(
"page_given, expected_rows, has_next_link", [(True, 10, False), (False, 50, True)],
"page_given, expected_rows, has_next_link",
[(True, 10, False), (False, 50, True)],
)
def test_get_most_recent_inbound_sms_for_service(
admin_request, page_given, sample_service, expected_rows, has_next_link,
admin_request,
page_given,
sample_service,
expected_rows,
has_next_link,
):
for i in range(60):
create_inbound_sms(
@@ -229,13 +238,16 @@ def test_get_most_recent_inbound_sms_for_service(
@freeze_time("Monday 10th April 2017 12:00")
def test_get_most_recent_inbound_sms_for_service_respects_data_retention(
admin_request, sample_service,
admin_request,
sample_service,
):
create_service_data_retention(sample_service, NotificationType.SMS, 5)
for i in range(10):
created = datetime.utcnow() - timedelta(days=i)
create_inbound_sms(
sample_service, user_number="44770090000{}".format(i), created_at=created,
sample_service,
user_number="44770090000{}".format(i),
created_at=created,
)
response = admin_request.get(

172
tests/app/job/test_rest.py
View File

@@ -9,7 +9,13 @@ from freezegun import freeze_time
import app.celery.tasks
from app.dao.templates_dao import dao_update_template
from app.enums import JobStatus, KeyType, NotificationStatus, NotificationType, TemplateType
from app.enums import (
JobStatus,
KeyType,
NotificationStatus,
NotificationType,
TemplateType,
)
from tests import create_admin_authorization_header
from tests.app.db import (
create_ft_notification_status,
@@ -138,7 +144,9 @@ def test_create_unscheduled_job_with_sender_id_in_metadata(
assert response.status_code == 201
app.celery.tasks.process_job.apply_async.assert_called_once_with(
([str(fake_uuid)]), {"sender_id": fake_uuid}, queue="job-tasks",
([str(fake_uuid)]),
{"sender_id": fake_uuid},
queue="job-tasks",
)
@@ -486,7 +494,11 @@ def test_get_all_notifications_for_job_in_order_of_job_number(
],
)
def test_get_all_notifications_for_job_filtered_by_status(
admin_request, sample_job, expected_notification_count, status_args, mocker,
admin_request,
sample_job,
expected_notification_count,
status_args,
mocker,
):
mock_s3 = mocker.patch("app.job.rest.get_phone_number_from_s3")
mock_s3.return_value = "15555555555"
@@ -590,19 +602,42 @@ def test_get_job_by_id_should_return_summed_statistics(admin_request, sample_job
create_notification(job=sample_job, status=NotificationStatus.TEMPORARY_FAILURE)
resp_json = admin_request.get(
"job.get_job_by_service_and_job_id", service_id=service_id, job_id=job_id,
"job.get_job_by_service_and_job_id",
service_id=service_id,
job_id=job_id,
)
assert resp_json["data"]["id"] == job_id
assert {"status": NotificationStatus.CREATED, "count": 3,} in resp_json["data"]["statistics"]
assert {"status": NotificationStatus.SENDING, "count": 1,} in resp_json["data"]["statistics"]
assert {"status": NotificationStatus.FAILED, "count": 3,} in resp_json["data"]["statistics"]
assert {"status": NotificationStatus.TECHNICAL_FAILURE, "count": 1,} in resp_json["data"][
"statistics"
]
assert {"status": NotificationStatus.TEMPORARY_FAILURE, "count": 2,} in resp_json["data"][
"statistics"
]
assert {
"status": NotificationStatus.CREATED,
"count": 3,
} in resp_json[
"data"
]["statistics"]
assert {
"status": NotificationStatus.SENDING,
"count": 1,
} in resp_json[
"data"
]["statistics"]
assert {
"status": NotificationStatus.FAILED,
"count": 3,
} in resp_json[
"data"
]["statistics"]
assert {
"status": NotificationStatus.TECHNICAL_FAILURE,
"count": 1,
} in resp_json[
"data"
]["statistics"]
assert {
"status": NotificationStatus.TEMPORARY_FAILURE,
"count": 2,
} in resp_json[
"data"
]["statistics"]
assert resp_json["data"]["created_by"]["name"] == "Test User"
@@ -641,15 +676,36 @@ def test_get_job_by_id_with_stats_for_old_job_where_notifications_have_been_purg
)
assert resp_json["data"]["id"] == str(old_job.id)
assert {"status": NotificationStatus.CREATED, "count": 3,} in resp_json["data"]["statistics"]
assert {"status": NotificationStatus.SENDING, "count": 1,} in resp_json["data"]["statistics"]
assert {"status": NotificationStatus.FAILED, "count": 3,} in resp_json["data"]["statistics"]
assert {"status": NotificationStatus.TECHNICAL_FAILURE, "count": 1,} in resp_json["data"][
"statistics"
]
assert {"status": NotificationStatus.TEMPORARY_FAILURE, "count": 2,} in resp_json["data"][
"statistics"
]
assert {
"status": NotificationStatus.CREATED,
"count": 3,
} in resp_json[
"data"
]["statistics"]
assert {
"status": NotificationStatus.SENDING,
"count": 1,
} in resp_json[
"data"
]["statistics"]
assert {
"status": NotificationStatus.FAILED,
"count": 3,
} in resp_json[
"data"
]["statistics"]
assert {
"status": NotificationStatus.TECHNICAL_FAILURE,
"count": 1,
} in resp_json[
"data"
]["statistics"]
assert {
"status": NotificationStatus.TEMPORARY_FAILURE,
"count": 2,
} in resp_json[
"data"
]["statistics"]
assert resp_json["data"]["created_by"]["name"] == "Test User"
@@ -723,13 +779,24 @@ def test_get_jobs_should_return_statistics(admin_request, sample_template):
assert len(resp_json["data"]) == 2
assert resp_json["data"][0]["id"] == str(job_2.id)
assert {"status": NotificationStatus.SENDING, "count": 3,} in resp_json["data"][0]["statistics"]
assert {
"status": NotificationStatus.SENDING,
"count": 3,
} in resp_json["data"][
0
]["statistics"]
assert resp_json["data"][1]["id"] == str(job_1.id)
assert {"status": NotificationStatus.CREATED, "count": 3,} in resp_json["data"][1]["statistics"]
assert {
"status": NotificationStatus.CREATED,
"count": 3,
} in resp_json["data"][
1
]["statistics"]
def test_get_jobs_should_return_no_stats_if_no_rows_in_notifications(
admin_request, sample_template,
admin_request,
sample_template,
):
now = datetime.utcnow()
earlier = datetime.utcnow() - timedelta(days=1)
@@ -876,32 +943,49 @@ def test_get_jobs_should_retrieve_from_ft_notification_status_for_old_jobs(
# some notifications created more than three days ago, some created after the midnight cutoff
create_ft_notification_status(
date(2017, 6, 6), job=job_1, notification_status=NotificationStatus.DELIVERED, count=2,
date(2017, 6, 6),
job=job_1,
notification_status=NotificationStatus.DELIVERED,
count=2,
)
create_ft_notification_status(
date(2017, 6, 7), job=job_1, notification_status=NotificationStatus.DELIVERED, count=4,
date(2017, 6, 7),
job=job_1,
notification_status=NotificationStatus.DELIVERED,
count=4,
)
# job2's new enough
create_notification(
job=job_2, status=NotificationStatus.CREATED, created_at=not_quite_three_days_ago,
job=job_2,
status=NotificationStatus.CREATED,
created_at=not_quite_three_days_ago,
)
# this isn't picked up because the job is too new
create_ft_notification_status(
date(2017, 6, 7), job=job_2, notification_status=NotificationStatus.DELIVERED, count=8,
date(2017, 6, 7),
job=job_2,
notification_status=NotificationStatus.DELIVERED,
count=8,
)
# this isn't picked up - while the job is old, it started in last 3 days so we look at notification table instead
create_ft_notification_status(
date(2017, 6, 7), job=job_3, notification_status=NotificationStatus.DELIVERED, count=16,
date(2017, 6, 7),
job=job_3,
notification_status=NotificationStatus.DELIVERED,
count=16,
)
# this isn't picked up because we're using the ft status table for job_1 as it's old
create_notification(
job=job_1, status=NotificationStatus.CREATED, created_at=not_quite_three_days_ago,
job=job_1,
status=NotificationStatus.CREATED,
created_at=not_quite_three_days_ago,
)
resp_json = admin_request.get(
"job.get_jobs_by_service", service_id=sample_template.service_id,
"job.get_jobs_by_service",
service_id=sample_template.service_id,
)
assert resp_json["data"][0]["id"] == str(job_3.id)
@@ -935,26 +1019,38 @@ def test_get_scheduled_job_stats(admin_request):
# Shouldn’t be counted – wrong status
create_job(
service_1_template, job_status=JobStatus.FINISHED, scheduled_for="2017-07-17 07:00",
service_1_template,
job_status=JobStatus.FINISHED,
scheduled_for="2017-07-17 07:00",
)
create_job(
service_1_template, job_status=JobStatus.IN_PROGRESS, scheduled_for="2017-07-17 08:00",
service_1_template,
job_status=JobStatus.IN_PROGRESS,
scheduled_for="2017-07-17 08:00",
)
# Should be counted – service 1
create_job(
service_1_template, job_status=JobStatus.SCHEDULED, scheduled_for="2017-07-17 09:00",
service_1_template,
job_status=JobStatus.SCHEDULED,
scheduled_for="2017-07-17 09:00",
)
create_job(
service_1_template, job_status=JobStatus.SCHEDULED, scheduled_for="2017-07-17 10:00",
service_1_template,
job_status=JobStatus.SCHEDULED,
scheduled_for="2017-07-17 10:00",
)
create_job(
service_1_template, job_status=JobStatus.SCHEDULED, scheduled_for="2017-07-17 11:00",
service_1_template,
job_status=JobStatus.SCHEDULED,
scheduled_for="2017-07-17 11:00",
)
# Should be counted – service 2
create_job(
service_2_template, job_status=JobStatus.SCHEDULED, scheduled_for="2017-07-17 11:00",
service_2_template,
job_status=JobStatus.SCHEDULED,
scheduled_for="2017-07-17 11:00",
)
assert admin_request.get(