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

merge from main

Browse Source
This commit is contained in:
Kenneth Kehl
2025-02-10 08:57:20 -08:00
parent 0107e0143f 1cbd40f36b
commit 1e77491ecd
145 changed files with 13517 additions and 1749 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -3,8 +3,10 @@ from unittest.mock import ANY, call
import pytest
from freezegun import freeze_time
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
from app import db
from app.celery import nightly_tasks
from app.celery.nightly_tasks import (
_delete_notifications_older_than_retention_by_type,
@@ -230,7 +232,7 @@ def test_save_daily_notification_processing_time(
save_daily_notification_processing_time(date_provided)
persisted_to_db = FactProcessingTime.query.all()
persisted_to_db = db.session.execute(select(FactProcessingTime)).scalars().all()
assert len(persisted_to_db) == 1
assert persisted_to_db[0].local_date == date(2021, 1, 17)
assert persisted_to_db[0].messages_total == 2
@@ -269,7 +271,7 @@ def test_save_daily_notification_processing_time_when_in_est(
save_daily_notification_processing_time(date_provided)
persisted_to_db = FactProcessingTime.query.all()
persisted_to_db = db.session.execute(select(FactProcessingTime)).scalars().all()
assert len(persisted_to_db) == 1
assert persisted_to_db[0].local_date == date(2021, 4, 17)
assert persisted_to_db[0].messages_total == 2

7
tests/app/celery/test_process_ses_receipts_tasks.py
View File

@@ -2,8 +2,9 @@ import json
from unittest.mock import ANY
from freezegun import freeze_time
from sqlalchemy import select
from app import encryption
from app import db, encryption
from app.celery.process_ses_receipts_tasks import (
process_ses_results,
remove_emails_from_bounce,
@@ -168,7 +169,7 @@ def test_process_ses_results_in_complaint(sample_email_template, mocker):
)
process_ses_results(response=ses_complaint_callback())
assert mocked.call_count == 0
complaints = Complaint.query.all()
complaints = db.session.execute(select(Complaint)).scalars().all()
assert len(complaints) == 1
assert complaints[0].notification_id == notification.id
@@ -420,7 +421,7 @@ def test_ses_callback_should_send_on_complaint_to_user_callback_api(
assert send_mock.call_count == 1
assert encryption.decrypt(send_mock.call_args[0][0][0]) == {
"complaint_date": "2018-06-05T13:59:58.000000Z",
"complaint_id": str(Complaint.query.one().id),
"complaint_id": str(db.session.execute(select(Complaint)).scalars().one().id),
"notification_id": str(notification.id),
"reference": None,
"service_callback_api_bearer_token": "some_super_secret",

117
tests/app/celery/test_provider_tasks.py
View File

@@ -1,4 +1,5 @@
import json
from unittest.mock import ANY
import pytest
from botocore.exceptions import ClientError
@@ -6,11 +7,7 @@ from celery.exceptions import MaxRetriesExceededError
import app
from app.celery import provider_tasks
from app.celery.provider_tasks import (
check_sms_delivery_receipt,
deliver_email,
deliver_sms,
)
from app.celery.provider_tasks import deliver_email, deliver_sms
from app.clients.email import EmailClientNonRetryableException
from app.clients.email.aws_ses import (
AwsSesClientException,
@@ -26,110 +23,10 @@ def test_should_have_decorated_tasks_functions():
assert deliver_email.__wrapped__.__name__ == "deliver_email"
def test_should_check_delivery_receipts_success(sample_notification, mocker):
mocker.patch("app.delivery.send_to_providers.send_sms_to_provider")
mocker.patch(
"app.celery.provider_tasks.aws_cloudwatch_client.is_localstack",
return_value=False,
)
mocker.patch(
"app.celery.provider_tasks.aws_cloudwatch_client.check_sms",
return_value=("success", "okay", "AT&T"),
)
mock_sanitize = mocker.patch(
"app.celery.provider_tasks.sanitize_successful_notification_by_id"
)
check_sms_delivery_receipt(
"message_id", sample_notification.id, "2024-10-20 00:00:00+0:00"
)
# This call should be made if the message was successfully delivered
mock_sanitize.assert_called_once()
def test_should_check_delivery_receipts_failure(sample_notification, mocker):
mocker.patch("app.delivery.send_to_providers.send_sms_to_provider")
mocker.patch(
"app.celery.provider_tasks.aws_cloudwatch_client.is_localstack",
return_value=False,
)
mock_update = mocker.patch(
"app.celery.provider_tasks.update_notification_status_by_id"
)
mocker.patch(
"app.celery.provider_tasks.aws_cloudwatch_client.check_sms",
return_value=("failure", "not okay", "AT&T"),
)
mock_sanitize = mocker.patch(
"app.celery.provider_tasks.sanitize_successful_notification_by_id"
)
check_sms_delivery_receipt(
"message_id", sample_notification.id, "2024-10-20 00:00:00+0:00"
)
mock_sanitize.assert_not_called()
mock_update.assert_called_once()
def test_should_check_delivery_receipts_client_error(sample_notification, mocker):
mocker.patch("app.delivery.send_to_providers.send_sms_to_provider")
mocker.patch(
"app.celery.provider_tasks.aws_cloudwatch_client.is_localstack",
return_value=False,
)
mock_update = mocker.patch(
"app.celery.provider_tasks.update_notification_status_by_id"
)
error_response = {"Error": {"Code": "SomeCode", "Message": "Some Message"}}
operation_name = "SomeOperation"
mocker.patch(
"app.celery.provider_tasks.aws_cloudwatch_client.check_sms",
side_effect=ClientError(error_response, operation_name),
)
mock_sanitize = mocker.patch(
"app.celery.provider_tasks.sanitize_successful_notification_by_id"
)
try:
check_sms_delivery_receipt(
"message_id", sample_notification.id, "2024-10-20 00:00:00+0:00"
)
assert 1 == 0
except ClientError:
mock_sanitize.assert_not_called()
mock_update.assert_called_once()
def test_should_check_delivery_receipts_ntfe(sample_notification, mocker):
mocker.patch("app.delivery.send_to_providers.send_sms_to_provider")
mocker.patch(
"app.celery.provider_tasks.aws_cloudwatch_client.is_localstack",
return_value=False,
)
mock_update = mocker.patch(
"app.celery.provider_tasks.update_notification_status_by_id"
)
mocker.patch(
"app.celery.provider_tasks.aws_cloudwatch_client.check_sms",
side_effect=NotificationTechnicalFailureException(),
)
mock_sanitize = mocker.patch(
"app.celery.provider_tasks.sanitize_successful_notification_by_id"
)
try:
check_sms_delivery_receipt(
"message_id", sample_notification.id, "2024-10-20 00:00:00+0:00"
)
assert 1 == 0
except NotificationTechnicalFailureException:
mock_sanitize.assert_not_called()
mock_update.assert_called_once()
def test_should_call_send_sms_to_provider_from_deliver_sms_task(
sample_notification, mocker
):
mocker.patch("app.delivery.send_to_providers.send_sms_to_provider")
mocker.patch("app.celery.provider_tasks.check_sms_delivery_receipt")
deliver_sms(sample_notification.id)
app.delivery.send_to_providers.send_sms_to_provider.assert_called_with(
@@ -148,7 +45,7 @@ def test_should_add_to_retry_queue_if_notification_not_found_in_deliver_sms_task
deliver_sms(notification_id)
app.delivery.send_to_providers.send_sms_to_provider.assert_not_called()
app.celery.provider_tasks.deliver_sms.retry.assert_called_with(
queue="retry-tasks", countdown=0
queue="retry-tasks", countdown=0, expires=ANY
)
@@ -208,7 +105,7 @@ def test_should_go_into_technical_error_if_exceeds_retries_on_deliver_sms_task(
assert str(sample_notification.id) in str(e.value)
provider_tasks.deliver_sms.retry.assert_called_with(
queue="retry-tasks", countdown=0
queue="retry-tasks", countdown=0, expires=ANY
)
assert sample_notification.status == NotificationStatus.TEMPORARY_FAILURE
@@ -240,7 +137,7 @@ def test_should_add_to_retry_queue_if_notification_not_found_in_deliver_email_ta
deliver_email(notification_id)
app.delivery.send_to_providers.send_email_to_provider.assert_not_called()
app.celery.provider_tasks.deliver_email.retry.assert_called_with(
queue="retry-tasks"
queue="retry-tasks", expires=ANY
)
@@ -268,7 +165,9 @@ def test_should_go_into_technical_error_if_exceeds_retries_on_deliver_email_task
deliver_email(sample_notification.id)
assert str(sample_notification.id) in str(e.value)
provider_tasks.deliver_email.retry.assert_called_with(queue="retry-tasks")
provider_tasks.deliver_email.retry.assert_called_with(
queue="retry-tasks", expires=ANY
)
assert sample_notification.status == NotificationStatus.TECHNICAL_FAILURE

103
tests/app/celery/test_reporting_tasks.py
View File

@@ -4,7 +4,9 @@ from uuid import UUID
import pytest
from freezegun import freeze_time
from sqlalchemy import func, select
from app import db
from app.celery.reporting_tasks import (
create_nightly_billing,
create_nightly_billing_for_day,
@@ -101,7 +103,6 @@ def test_create_nightly_notification_status_triggers_relevant_tasks(
mock_celery = mocker.patch(
"app.celery.reporting_tasks.create_nightly_notification_status_for_service_and_day"
).apply_async
for notification_type in NotificationType:
template = create_template(sample_service, template_type=notification_type)
create_notification(template=template, created_at=notification_date)
@@ -132,11 +133,11 @@ def test_create_nightly_billing_for_day_checks_history(
status=NotificationStatus.DELIVERED,
)
records = FactBilling.query.all()
records = _get_fact_billing_records()
assert len(records) == 0
create_nightly_billing_for_day(str(yesterday.date()))
records = FactBilling.query.all()
records = _get_fact_billing_records()
assert len(records) == 1
record = records[0]
@@ -144,6 +145,11 @@ def test_create_nightly_billing_for_day_checks_history(
assert record.notifications_sent == 2
def _get_fact_billing_records():
stmt = select(FactBilling)
return db.session.execute(stmt).scalars().all()
@pytest.mark.parametrize(
"second_rate, records_num, billable_units, multiplier",
[(1.0, 1, 2, [1]), (2.0, 2, 1, [1, 2])],
@@ -181,11 +187,15 @@ def test_create_nightly_billing_for_day_sms_rate_multiplier(
billable_units=1,
)
records = FactBilling.query.all()
records = _get_fact_billing_records()
assert len(records) == 0
create_nightly_billing_for_day(str(yesterday.date()))
records = FactBilling.query.order_by("rate_multiplier").all()
records = (
db.session.execute(select(FactBilling).order_by("rate_multiplier"))
.scalars()
.all()
)
assert len(records) == records_num
for i, record in enumerate(records):
@@ -221,11 +231,15 @@ def test_create_nightly_billing_for_day_different_templates(
billable_units=0,
)
records = FactBilling.query.all()
records = _get_fact_billing_records()
assert len(records) == 0
create_nightly_billing_for_day(str(yesterday.date()))
records = FactBilling.query.order_by("rate_multiplier").all()
records = (
db.session.execute(select(FactBilling).order_by("rate_multiplier"))
.scalars()
.all()
)
assert len(records) == 2
multiplier = [0, 1]
billable_units = [0, 1]
@@ -265,11 +279,15 @@ def test_create_nightly_billing_for_day_same_sent_by(
billable_units=1,
)
records = FactBilling.query.all()
records = _get_fact_billing_records()
assert len(records) == 0
create_nightly_billing_for_day(str(yesterday.date()))
records = FactBilling.query.order_by("rate_multiplier").all()
records = (
db.session.execute(select(FactBilling).order_by("rate_multiplier"))
.scalars()
.all()
)
assert len(records) == 1
for _, record in enumerate(records):
@@ -296,11 +314,11 @@ def test_create_nightly_billing_for_day_null_sent_by_sms(
billable_units=1,
)
records = FactBilling.query.all()
records = _get_fact_billing_records()
assert len(records) == 0
create_nightly_billing_for_day(str(yesterday.date()))
records = FactBilling.query.all()
records = _get_fact_billing_records()
assert len(records) == 1
record = records[0]
@@ -356,12 +374,19 @@ def test_create_nightly_billing_for_day_use_BST(
rate_multiplier=1.0,
billable_units=4,
)
assert Notification.query.count() == 3
assert FactBilling.query.count() == 0
stmt = select(func.count()).select_from(Notification)
count = db.session.execute(stmt).scalar() or 0
assert count == 3
stmt = select(func.count()).select_from(FactBilling)
count = db.session.execute(stmt).scalar() or 0
assert count == 0
create_nightly_billing_for_day("2018-03-25")
records = FactBilling.query.order_by(FactBilling.local_date).all()
records = (
db.session.execute(select(FactBilling).order_by(FactBilling.local_date))
.scalars()
.all()
)
assert len(records) == 1
assert records[0].local_date == date(2018, 3, 25)
@@ -384,11 +409,15 @@ def test_create_nightly_billing_for_day_update_when_record_exists(
billable_units=1,
)
records = FactBilling.query.all()
records = _get_fact_billing_records()
assert len(records) == 0
create_nightly_billing_for_day("2018-01-14")
records = FactBilling.query.order_by(FactBilling.local_date).all()
records = (
db.session.execute(select(FactBilling).order_by(FactBilling.local_date))
.scalars()
.all()
)
assert len(records) == 1
assert records[0].local_date == date(2018, 1, 14)
@@ -454,7 +483,7 @@ def test_create_nightly_notification_status_for_service_and_day(notify_db_sessio
create_notification(template=first_template)
create_notification_history(template=second_template)
assert len(FactNotificationStatus.query.all()) == 0
assert len(db.session.execute(select(FactNotificationStatus)).scalars().all()) == 0
create_nightly_notification_status_for_service_and_day(
str(process_day),
@@ -467,10 +496,16 @@ def test_create_nightly_notification_status_for_service_and_day(notify_db_sessio
NotificationType.EMAIL,
)
new_fact_data = FactNotificationStatus.query.order_by(
FactNotificationStatus.notification_type,
FactNotificationStatus.notification_status,
).all()
new_fact_data = (
db.session.execute(
select(FactNotificationStatus).order_by(
FactNotificationStatus.notification_type,
FactNotificationStatus.notification_status,
)
)
.scalars()
.all()
)
assert len(new_fact_data) == 4
@@ -530,7 +565,7 @@ def test_create_nightly_notification_status_for_service_and_day_overwrites_old_d
NotificationType.SMS,
)
new_fact_data = FactNotificationStatus.query.all()
new_fact_data = db.session.execute(select(FactNotificationStatus)).scalars().all()
assert len(new_fact_data) == 1
assert new_fact_data[0].notification_count == 1
@@ -545,9 +580,15 @@ def test_create_nightly_notification_status_for_service_and_day_overwrites_old_d
NotificationType.SMS,
)
updated_fact_data = FactNotificationStatus.query.order_by(
FactNotificationStatus.notification_status
).all()
updated_fact_data = (
db.session.execute(
select(FactNotificationStatus).order_by(
FactNotificationStatus.notification_status
)
)
.scalars()
.all()
)
assert len(updated_fact_data) == 2
assert updated_fact_data[0].notification_count == 1
@@ -590,9 +631,13 @@ def test_create_nightly_notification_status_for_service_and_day_respects_bst(
NotificationType.SMS,
)
noti_status = FactNotificationStatus.query.order_by(
FactNotificationStatus.local_date
).all()
noti_status = (
db.session.execute(
select(FactNotificationStatus).order_by(FactNotificationStatus.local_date)
)
.scalars()
.all()
)
assert len(noti_status) == 1
assert noti_status[0].local_date == date(2019, 4, 1)

178
tests/app/celery/test_scheduled_tasks.py
View File

@@ -1,17 +1,20 @@
import json
from collections import namedtuple
from datetime import timedelta
from unittest import mock
from unittest.mock import ANY, call
from unittest.mock import ANY, MagicMock, call
import pytest
from app.celery import scheduled_tasks
from app.celery.scheduled_tasks import (
batch_insert_notifications,
check_for_missing_rows_in_completed_jobs,
check_for_services_with_high_failure_rates_or_sending_to_tv_numbers,
check_job_status,
delete_verify_codes,
expire_or_delete_invitations,
process_delivery_receipts,
replay_created_notifications,
run_scheduled_jobs,
)
@@ -23,6 +26,8 @@ from notifications_utils.clients.zendesk.zendesk_client import NotifySupportTick
from tests.app import load_example_csv
from tests.app.db import create_job, create_notification, create_template
CHECK_JOB_STATUS_TOO_OLD_MINUTES = 241
def test_should_call_delete_codes_on_delete_verify_codes_task(
notify_db_session, mocker
@@ -108,8 +113,9 @@ def test_check_job_status_task_calls_process_incomplete_jobs(mocker, sample_temp
job = create_job(
template=sample_template,
notification_count=3,
created_at=utc_now() - timedelta(minutes=31),
processing_started=utc_now() - timedelta(minutes=31),
created_at=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
processing_started=utc_now()
- timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
job_status=JobStatus.IN_PROGRESS,
)
create_notification(template=sample_template, job=job)
@@ -125,9 +131,10 @@ def test_check_job_status_task_calls_process_incomplete_jobs_when_scheduled_job_
job = create_job(
template=sample_template,
notification_count=3,
created_at=utc_now() - timedelta(hours=2),
scheduled_for=utc_now() - timedelta(minutes=31),
processing_started=utc_now() - timedelta(minutes=31),
created_at=utc_now() - timedelta(hours=5),
scheduled_for=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
processing_started=utc_now()
- timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
job_status=JobStatus.IN_PROGRESS,
)
check_job_status()
@@ -142,8 +149,8 @@ def test_check_job_status_task_calls_process_incomplete_jobs_for_pending_schedul
job = create_job(
template=sample_template,
notification_count=3,
created_at=utc_now() - timedelta(hours=2),
scheduled_for=utc_now() - timedelta(minutes=31),
created_at=utc_now() - timedelta(hours=5),
scheduled_for=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
job_status=JobStatus.PENDING,
)
@@ -175,17 +182,19 @@ def test_check_job_status_task_calls_process_incomplete_jobs_for_multiple_jobs(
job = create_job(
template=sample_template,
notification_count=3,
created_at=utc_now() - timedelta(hours=2),
scheduled_for=utc_now() - timedelta(minutes=31),
processing_started=utc_now() - timedelta(minutes=31),
created_at=utc_now() - timedelta(hours=5),
scheduled_for=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
processing_started=utc_now()
- timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
job_status=JobStatus.IN_PROGRESS,
)
job_2 = create_job(
template=sample_template,
notification_count=3,
created_at=utc_now() - timedelta(hours=2),
scheduled_for=utc_now() - timedelta(minutes=31),
processing_started=utc_now() - timedelta(minutes=31),
created_at=utc_now() - timedelta(hours=5),
scheduled_for=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
processing_started=utc_now()
- timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
job_status=JobStatus.IN_PROGRESS,
)
check_job_status()
@@ -200,23 +209,24 @@ def test_check_job_status_task_only_sends_old_tasks(mocker, sample_template):
job = create_job(
template=sample_template,
notification_count=3,
created_at=utc_now() - timedelta(hours=2),
scheduled_for=utc_now() - timedelta(minutes=31),
processing_started=utc_now() - timedelta(minutes=31),
created_at=utc_now() - timedelta(hours=5),
scheduled_for=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
processing_started=utc_now()
- timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
job_status=JobStatus.IN_PROGRESS,
)
create_job(
template=sample_template,
notification_count=3,
created_at=utc_now() - timedelta(minutes=31),
processing_started=utc_now() - timedelta(minutes=29),
created_at=utc_now() - timedelta(minutes=300),
processing_started=utc_now() - timedelta(minutes=239),
job_status=JobStatus.IN_PROGRESS,
)
create_job(
template=sample_template,
notification_count=3,
created_at=utc_now() - timedelta(minutes=50),
scheduled_for=utc_now() - timedelta(minutes=29),
created_at=utc_now() - timedelta(minutes=300),
scheduled_for=utc_now() - timedelta(minutes=239),
job_status=JobStatus.PENDING,
)
check_job_status()
@@ -230,16 +240,17 @@ def test_check_job_status_task_sets_jobs_to_error(mocker, sample_template):
job = create_job(
template=sample_template,
notification_count=3,
created_at=utc_now() - timedelta(hours=2),
scheduled_for=utc_now() - timedelta(minutes=31),
processing_started=utc_now() - timedelta(minutes=31),
created_at=utc_now() - timedelta(hours=5),
scheduled_for=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
processing_started=utc_now()
- timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
job_status=JobStatus.IN_PROGRESS,
)
job_2 = create_job(
template=sample_template,
notification_count=3,
created_at=utc_now() - timedelta(minutes=31),
processing_started=utc_now() - timedelta(minutes=29),
created_at=utc_now() - timedelta(minutes=300),
processing_started=utc_now() - timedelta(minutes=239),
job_status=JobStatus.IN_PROGRESS,
)
check_job_status()
@@ -300,10 +311,10 @@ def test_replay_created_notifications(notify_db_session, sample_service, mocker)
replay_created_notifications()
email_delivery_queue.assert_called_once_with(
[str(old_email.id)], queue="send-email-tasks"
[str(old_email.id)], queue="send-email-tasks", countdown=60
)
sms_delivery_queue.assert_called_once_with(
[str(old_sms.id)], queue="send-sms-tasks"
[str(old_sms.id)], queue="send-sms-tasks", countdown=60
)
@@ -311,16 +322,18 @@ def test_check_job_status_task_does_not_raise_error(sample_template):
create_job(
template=sample_template,
notification_count=3,
created_at=utc_now() - timedelta(hours=2),
scheduled_for=utc_now() - timedelta(minutes=31),
processing_started=utc_now() - timedelta(minutes=31),
created_at=utc_now() - timedelta(hours=5),
scheduled_for=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
processing_started=utc_now()
- timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
job_status=JobStatus.FINISHED,
)
create_job(
template=sample_template,
notification_count=3,
created_at=utc_now() - timedelta(minutes=31),
processing_started=utc_now() - timedelta(minutes=31),
created_at=utc_now() - timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
processing_started=utc_now()
- timedelta(minutes=CHECK_JOB_STATUS_TOO_OLD_MINUTES),
job_status=JobStatus.FINISHED,
)
@@ -415,6 +428,7 @@ def test_check_for_missing_rows_in_completed_jobs_calls_save_email(
),
{},
queue="database-tasks",
expires=ANY,
)
@@ -512,3 +526,101 @@ def test_check_for_services_with_high_failure_rates_or_sending_to_tv_numbers(
technical_ticket=True,
)
mock_send_ticket_to_zendesk.assert_called_once()
def test_batch_insert_with_valid_notifications(mocker):
mocker.patch("app.celery.scheduled_tasks.dao_batch_insert_notifications")
rs = MagicMock()
mocker.patch("app.celery.scheduled_tasks.redis_store", rs)
notifications = [
{"id": 1, "notification_status": "pending"},
{"id": 2, "notification_status": "pending"},
]
serialized_notifications = [json.dumps(n).encode("utf-8") for n in notifications]
pipeline_mock = MagicMock()
rs.pipeline.return_value.__enter__.return_value = pipeline_mock
rs.llen.return_value = len(notifications)
rs.lpop.side_effect = serialized_notifications
batch_insert_notifications()
rs.llen.assert_called_once_with("message_queue")
rs.lpop.assert_called_with("message_queue")
def test_batch_insert_with_expired_notifications(mocker):
expired_time = utc_now() - timedelta(minutes=2)
mocker.patch(
"app.celery.scheduled_tasks.dao_batch_insert_notifications",
side_effect=Exception("DB Error"),
)
rs = MagicMock()
mocker.patch("app.celery.scheduled_tasks.redis_store", rs)
notifications = [
{
"id": 1,
"notification_status": "pending",
"created_at": utc_now().isoformat(),
},
{
"id": 2,
"notification_status": "pending",
"created_at": expired_time.isoformat(),
},
]
serialized_notifications = [json.dumps(n).encode("utf-8") for n in notifications]
pipeline_mock = MagicMock()
rs.pipeline.return_value.__enter__.return_value = pipeline_mock
rs.llen.return_value = len(notifications)
rs.lpop.side_effect = serialized_notifications
batch_insert_notifications()
rs.llen.assert_called_once_with("message_queue")
rs.rpush.assert_called_once()
requeued_notification = json.loads(rs.rpush.call_args[0][1])
assert requeued_notification["id"] == 1
def test_batch_insert_with_malformed_notifications(mocker):
rs = MagicMock()
mocker.patch("app.celery.scheduled_tasks.redis_store", rs)
malformed_data = b"not_a_valid_json"
pipeline_mock = MagicMock()
rs.pipeline.return_value.__enter__.return_value = pipeline_mock
rs.llen.return_value = 1
rs.lpop.side_effect = [malformed_data]
with pytest.raises(json.JSONDecodeError):
batch_insert_notifications()
rs.llen.assert_called_once_with("message_queue")
rs.rpush.assert_not_called()
def test_process_delivery_receipts_success(mocker):
dao_update_mock = mocker.patch(
"app.celery.scheduled_tasks.dao_update_delivery_receipts"
)
cloudwatch_mock = mocker.patch("app.celery.scheduled_tasks.AwsCloudwatchClient")
cloudwatch_mock.return_value.check_delivery_receipts.return_value = (
range(2000),
range(500),
)
current_app_mock = mocker.patch("app.celery.scheduled_tasks.current_app")
current_app_mock.return_value = MagicMock()
processor = MagicMock()
processor.process_delivery_receipts = process_delivery_receipts
processor.retry = MagicMock()
processor.process_delivery_receipts()
assert dao_update_mock.call_count == 3
dao_update_mock.assert_any_call(list(range(1000)), True)
dao_update_mock.assert_any_call(list(range(1000, 2000)), True)
dao_update_mock.assert_any_call(list(range(500)), False)
processor.retry.assert_not_called()

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

@@ -1,16 +1,17 @@
import json
import uuid
from datetime import datetime, timedelta
from unittest.mock import MagicMock, Mock, call
from unittest.mock import ANY, MagicMock, Mock, call
import pytest
import requests_mock
from celery.exceptions import Retry
from freezegun import freeze_time
from requests import RequestException
from sqlalchemy import func, select
from sqlalchemy.exc import SQLAlchemyError
from app import encryption
from app import db, encryption
from app.celery import provider_tasks, tasks
from app.celery.tasks import (
__total_sending_limits_for_job_exceeded,
@@ -115,6 +116,7 @@ def test_should_process_sms_job(sample_job, mocker):
(str(sample_job.service_id), "uuid", "something_encrypted"),
{},
queue="database-tasks",
expires=ANY,
)
job = jobs_dao.dao_get_job_by_id(sample_job.id)
assert job.job_status == JobStatus.FINISHED
@@ -135,6 +137,7 @@ def test_should_process_sms_job_with_sender_id(sample_job, mocker, fake_uuid):
(str(sample_job.service_id), "uuid", "something_encrypted"),
{"sender_id": fake_uuid},
queue="database-tasks",
expires=ANY,
)
@@ -179,6 +182,7 @@ def test_should_process_job_if_send_limits_are_not_exceeded(
),
{},
queue="database-tasks",
expires=ANY,
)
@@ -237,6 +241,7 @@ def test_should_process_email_job(email_job_with_placeholders, mocker):
),
{},
queue="database-tasks",
expires=ANY,
)
job = jobs_dao.dao_get_job_by_id(email_job_with_placeholders.id)
assert job.job_status == JobStatus.FINISHED
@@ -262,6 +267,7 @@ def test_should_process_email_job_with_sender_id(
(str(email_job_with_placeholders.service_id), "uuid", "something_encrypted"),
{"sender_id": fake_uuid},
queue="database-tasks",
expires=ANY,
)
@@ -351,6 +357,7 @@ def test_process_row_sends_letter_task(
),
{},
queue=expected_queue,
expires=ANY,
)
@@ -387,6 +394,7 @@ def test_process_row_when_sender_id_is_provided(mocker, fake_uuid):
),
{"sender_id": fake_uuid},
queue="database-tasks",
expires=ANY,
)
@@ -412,7 +420,7 @@ def test_should_send_template_to_correct_sms_task_and_persist(
encryption.encrypt(notification),
)
persisted_notification = Notification.query.one()
persisted_notification = _get_notification_query_one()
assert persisted_notification.to == "1"
assert persisted_notification.template_id == sample_template_with_placeholders.id
assert (
@@ -427,10 +435,15 @@ def test_should_send_template_to_correct_sms_task_and_persist(
assert persisted_notification.personalisation == {}
assert persisted_notification.notification_type == NotificationType.SMS
mocked_deliver_sms.assert_called_once_with(
[str(persisted_notification.id)], queue="send-sms-tasks"
[str(persisted_notification.id)], queue="send-sms-tasks", countdown=60
)
def _get_notification_query_one():
stmt = select(Notification)
return db.session.execute(stmt).scalars().one()
def test_should_save_sms_if_restricted_service_and_valid_number(
notify_db_session, mocker
):
@@ -451,7 +464,7 @@ def test_should_save_sms_if_restricted_service_and_valid_number(
encrypt_notification,
)
persisted_notification = Notification.query.one()
persisted_notification = _get_notification_query_one()
assert persisted_notification.to == "1"
assert persisted_notification.template_id == template.id
assert persisted_notification.template_version == template.version
@@ -463,7 +476,7 @@ def test_should_save_sms_if_restricted_service_and_valid_number(
assert not persisted_notification.personalisation
assert persisted_notification.notification_type == NotificationType.SMS
provider_tasks.deliver_sms.apply_async.assert_called_once_with(
[str(persisted_notification.id)], queue="send-sms-tasks"
[str(persisted_notification.id)], queue="send-sms-tasks", countdown=60
)
@@ -490,7 +503,7 @@ def test_save_email_should_save_default_email_reply_to_text_on_notification(
encryption.encrypt(notification),
)
persisted_notification = Notification.query.one()
persisted_notification = _get_notification_query_one()
assert persisted_notification.reply_to_text == "reply_to@digital.fake.gov"
@@ -510,7 +523,7 @@ def test_save_sms_should_save_default_sms_sender_notification_reply_to_text_on(
encryption.encrypt(notification),
)
persisted_notification = Notification.query.one()
persisted_notification = _get_notification_query_one()
assert persisted_notification.reply_to_text == "12345"
@@ -531,7 +544,17 @@ def test_should_not_save_sms_if_restricted_service_and_invalid_number(
encryption.encrypt(notification),
)
assert provider_tasks.deliver_sms.apply_async.called is False
assert Notification.query.count() == 0
assert _get_notification_query_count() == 0
def _get_notification_query_all():
stmt = select(Notification)
return db.session.execute(stmt).scalars().all()
def _get_notification_query_count():
stmt = select(func.count()).select_from(Notification)
return db.session.execute(stmt).scalar() or 0
def test_should_not_save_email_if_restricted_service_and_invalid_email_address(
@@ -553,7 +576,7 @@ def test_should_not_save_email_if_restricted_service_and_invalid_email_address(
encryption.encrypt(notification),
)
assert Notification.query.count() == 0
assert _get_notification_query_count() == 0
def test_should_save_sms_template_to_and_persist_with_job_id(sample_job, mocker):
@@ -572,7 +595,7 @@ def test_should_save_sms_template_to_and_persist_with_job_id(sample_job, mocker)
notification_id,
encryption.encrypt(notification),
)
persisted_notification = Notification.query.one()
persisted_notification = _get_notification_query_one()
assert persisted_notification.to == "1"
assert persisted_notification.job_id == sample_job.id
assert persisted_notification.template_id == sample_job.template.id
@@ -586,14 +609,14 @@ def test_should_save_sms_template_to_and_persist_with_job_id(sample_job, mocker)
assert persisted_notification.notification_type == NotificationType.SMS
provider_tasks.deliver_sms.apply_async.assert_called_once_with(
[str(persisted_notification.id)], queue="send-sms-tasks"
[str(persisted_notification.id)], queue="send-sms-tasks", countdown=60
)
def test_should_not_save_sms_if_team_key_and_recipient_not_in_team(
notify_db_session, mocker
):
assert Notification.query.count() == 0
assert _get_notification_query_count() == 0
user = create_user(mobile_number="2028675309")
service = create_service(user=user, restricted=True)
template = create_template(service=service)
@@ -611,7 +634,7 @@ def test_should_not_save_sms_if_team_key_and_recipient_not_in_team(
encryption.encrypt(notification),
)
assert provider_tasks.deliver_sms.apply_async.called is False
assert Notification.query.count() == 0
assert _get_notification_query_count() == 0
def test_should_use_email_template_and_persist(
@@ -637,7 +660,7 @@ def test_should_use_email_template_and_persist(
encryption.encrypt(notification),
)
persisted_notification = Notification.query.one()
persisted_notification = _get_notification_query_one()
assert persisted_notification.to == "1"
assert (
persisted_notification.template_id == sample_email_template_with_placeholders.id
@@ -684,7 +707,7 @@ def test_save_email_should_use_template_version_from_job_not_latest(
encryption.encrypt(notification),
)
persisted_notification = Notification.query.one()
persisted_notification = _get_notification_query_one()
assert persisted_notification.to == "1"
assert persisted_notification.template_id == sample_email_template.id
assert persisted_notification.template_version == version_on_notification
@@ -713,7 +736,7 @@ def test_should_use_email_template_subject_placeholders(
notification_id,
encryption.encrypt(notification),
)
persisted_notification = Notification.query.one()
persisted_notification = _get_notification_query_one()
assert persisted_notification.to == "1"
assert (
persisted_notification.template_id == sample_email_template_with_placeholders.id
@@ -754,7 +777,7 @@ def test_save_email_uses_the_reply_to_text_when_provided(sample_email_template,
encryption.encrypt(notification),
sender_id=other_email_reply_to.id,
)
persisted_notification = Notification.query.one()
persisted_notification = _get_notification_query_one()
assert persisted_notification.notification_type == NotificationType.EMAIL
assert persisted_notification.reply_to_text == "other@example.com"
@@ -779,7 +802,7 @@ def test_save_email_uses_the_default_reply_to_text_if_sender_id_is_none(
encryption.encrypt(notification),
sender_id=None,
)
persisted_notification = Notification.query.one()
persisted_notification = _get_notification_query_one()
assert persisted_notification.notification_type == NotificationType.EMAIL
assert persisted_notification.reply_to_text == "default@example.com"
@@ -798,7 +821,7 @@ def test_should_use_email_template_and_persist_without_personalisation(
notification_id,
encryption.encrypt(notification),
)
persisted_notification = Notification.query.one()
persisted_notification = _get_notification_query_one()
assert persisted_notification.to == "1"
assert persisted_notification.template_id == sample_email_template.id
assert persisted_notification.created_at >= now
@@ -834,9 +857,11 @@ def test_save_sms_should_go_to_retry_queue_if_database_errors(sample_template, m
encryption.encrypt(notification),
)
assert provider_tasks.deliver_sms.apply_async.called is False
tasks.save_sms.retry.assert_called_with(exc=expected_exception, queue="retry-tasks")
tasks.save_sms.retry.assert_called_with(
exc=expected_exception, queue="retry-tasks", expires=ANY
)
assert Notification.query.count() == 0
assert _get_notification_query_count() == 0
def test_save_email_should_go_to_retry_queue_if_database_errors(
@@ -863,10 +888,10 @@ def test_save_email_should_go_to_retry_queue_if_database_errors(
)
assert not provider_tasks.deliver_email.apply_async.called
tasks.save_email.retry.assert_called_with(
exc=expected_exception, queue="retry-tasks"
exc=expected_exception, queue="retry-tasks", expires=ANY
)
assert Notification.query.count() == 0
assert _get_notification_query_count() == 0
def test_save_email_does_not_send_duplicate_and_does_not_put_in_retry_queue(
@@ -888,7 +913,7 @@ def test_save_email_does_not_send_duplicate_and_does_not_put_in_retry_queue(
notification_id,
encryption.encrypt(json),
)
assert Notification.query.count() == 1
assert _get_notification_query_count() == 1
assert not deliver_email.called
assert not retry.called
@@ -912,7 +937,7 @@ def test_save_sms_does_not_send_duplicate_and_does_not_put_in_retry_queue(
notification_id,
encryption.encrypt(json),
)
assert Notification.query.count() == 1
assert _get_notification_query_count() == 1
assert not deliver_sms.called
assert not retry.called
@@ -924,14 +949,14 @@ def test_save_sms_uses_sms_sender_reply_to_text(mocker, notify_db_session):
notification = _notification_json(template, to="2028675301")
mocker.patch("app.celery.provider_tasks.deliver_sms.apply_async")
notification_id = uuid.uuid4()
notification_id = str(uuid.uuid4())
save_sms(
service.id,
notification_id,
encryption.encrypt(notification),
)
persisted_notification = Notification.query.one()
persisted_notification = _get_notification_query_one()
assert persisted_notification.reply_to_text == "+12028675309"
@@ -957,7 +982,7 @@ def test_save_sms_uses_non_default_sms_sender_reply_to_text_if_provided(
sender_id=new_sender.id,
)
persisted_notification = Notification.query.one()
persisted_notification = _get_notification_query_one()
assert persisted_notification.reply_to_text == "new-sender"
@@ -1167,11 +1192,18 @@ def test_process_incomplete_job_sms(mocker, sample_template):
create_notification(sample_template, job, 0)
create_notification(sample_template, job, 1)
assert Notification.query.filter(Notification.job_id == job.id).count() == 2
stmt = (
select(func.count())
.select_from(Notification)
.where(Notification.job_id == job.id)
)
count = db.session.execute(stmt).scalar()
assert count == 2
process_incomplete_job(str(job.id))
completed_job = Job.query.filter(Job.id == job.id).one()
stmt = select(Job).where(Job.id == job.id)
completed_job = db.session.execute(stmt).scalars().one()
assert completed_job.job_status == JobStatus.FINISHED
@@ -1207,11 +1239,17 @@ def test_process_incomplete_job_with_notifications_all_sent(mocker, sample_templ
create_notification(sample_template, job, 8)
create_notification(sample_template, job, 9)
assert Notification.query.filter(Notification.job_id == job.id).count() == 10
stmt = (
select(func.count())
.select_from(Notification)
.where(Notification.job_id == job.id)
)
assert db.session.execute(stmt).scalar() == 10
process_incomplete_job(str(job.id))
completed_job = Job.query.filter(Job.id == job.id).one()
stmt = select(Job).where(Job.id == job.id)
completed_job = db.session.execute(stmt).scalars().one()
assert completed_job.job_status == JobStatus.FINISHED
@@ -1239,7 +1277,12 @@ def test_process_incomplete_jobs_sms(mocker, sample_template):
create_notification(sample_template, job, 1)
create_notification(sample_template, job, 2)
assert Notification.query.filter(Notification.job_id == job.id).count() == 3
stmt = (
select(func.count())
.select_from(Notification)
.where(Notification.job_id == job.id)
)
assert db.session.execute(stmt).scalar() == 3
job2 = create_job(
template=sample_template,
@@ -1256,13 +1299,21 @@ def test_process_incomplete_jobs_sms(mocker, sample_template):
create_notification(sample_template, job2, 3)
create_notification(sample_template, job2, 4)
assert Notification.query.filter(Notification.job_id == job2.id).count() == 5
stmt = (
select(func.count())
.select_from(Notification)
.where(Notification.job_id == job2.id)
)
assert db.session.execute(stmt).scalar() == 5
jobs = [job.id, job2.id]
process_incomplete_jobs(jobs)
completed_job = Job.query.filter(Job.id == job.id).one()
completed_job2 = Job.query.filter(Job.id == job2.id).one()
stmt = select(Job).where(Job.id == job.id)
completed_job = db.session.execute(stmt).scalars().one()
stmt = select(Job).where(Job.id == job2.id)
completed_job2 = db.session.execute(stmt).scalars().one()
assert completed_job.job_status == JobStatus.FINISHED
@@ -1288,12 +1339,16 @@ def test_process_incomplete_jobs_no_notifications_added(mocker, sample_template)
processing_started=utc_now() - timedelta(minutes=31),
job_status=JobStatus.ERROR,
)
assert Notification.query.filter(Notification.job_id == job.id).count() == 0
stmt = (
select(func.count())
.select_from(Notification)
.where(Notification.job_id == job.id)
)
assert db.session.execute(stmt).scalar() == 0
process_incomplete_job(job.id)
completed_job = Job.query.filter(Job.id == job.id).one()
stmt = select(Job).where(Job.id == job.id)
completed_job = db.session.execute(stmt).scalars().one()
assert completed_job.job_status == JobStatus.FINISHED
@@ -1349,11 +1404,17 @@ def test_process_incomplete_job_email(mocker, sample_email_template):
create_notification(sample_email_template, job, 0)
create_notification(sample_email_template, job, 1)
assert Notification.query.filter(Notification.job_id == job.id).count() == 2
stmt = (
select(func.count())
.select_from(Notification)
.where(Notification.job_id == job.id)
)
assert db.session.execute(stmt).scalar() == 2
process_incomplete_job(str(job.id))
completed_job = Job.query.filter(Job.id == job.id).one()
stmt = select(Job).where(Job.id == job.id)
completed_job = db.session.execute(stmt).scalars().one()
assert completed_job.job_status == JobStatus.FINISHED
@@ -1435,12 +1496,12 @@ def test_save_api_email_or_sms(mocker, sample_service, notification_type):
encrypted = encryption.encrypt(data)
assert len(Notification.query.all()) == 0
assert len(_get_notification_query_all()) == 0
if notification_type == NotificationType.EMAIL:
save_api_email(encrypted_notification=encrypted)
else:
save_api_sms(encrypted_notification=encrypted)
notifications = Notification.query.all()
notifications = _get_notification_query_all()
assert len(notifications) == 1
assert str(notifications[0].id) == data["id"]
assert notifications[0].created_at == datetime(2020, 3, 25, 14, 30)
@@ -1488,20 +1549,20 @@ def test_save_api_email_dont_retry_if_notification_already_exists(
expected_queue = QueueNames.SEND_SMS
encrypted = encryption.encrypt(data)
assert len(Notification.query.all()) == 0
assert len(_get_notification_query_all()) == 0
if notification_type == NotificationType.EMAIL:
save_api_email(encrypted_notification=encrypted)
else:
save_api_sms(encrypted_notification=encrypted)
notifications = Notification.query.all()
notifications = _get_notification_query_all()
assert len(notifications) == 1
# call the task again with the same notification
if notification_type == NotificationType.EMAIL:
save_api_email(encrypted_notification=encrypted)
else:
save_api_sms(encrypted_notification=encrypted)
notifications = Notification.query.all()
notifications = _get_notification_query_all()
assert len(notifications) == 1
assert str(notifications[0].id) == data["id"]
assert notifications[0].created_at == datetime(2020, 3, 25, 14, 30)
@@ -1565,7 +1626,7 @@ def test_save_tasks_use_cached_service_and_template(
]
# But we save 2 notifications and enqueue 2 tasks
assert len(Notification.query.all()) == 2
assert len(_get_notification_query_all()) == 2
assert len(delivery_mock.call_args_list) == 2
@@ -1626,14 +1687,14 @@ def test_save_api_tasks_use_cache(
}
)
assert len(Notification.query.all()) == 0
assert len(_get_notification_query_all()) == 0
for _ in range(3):
task_function(encrypted_notification=create_encrypted_notification())
assert service_dict_mock.call_args_list == [call(str(template.service_id))]
assert len(Notification.query.all()) == 3
assert len(_get_notification_query_all()) == 3
assert len(mock_provider_task.call_args_list) == 3