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

Merge pull request #1376 from GSA/notify-api-1351

Browse Source 
increase code coverage to 94%
This commit is contained in:
Carlo Costino
2024-10-29 17:08:00 -04:00
committed by GitHub
parent c85d35191f 0b7079edd9
commit 99c199b9d2
12 changed files with 813 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.ds.baseline
View File

@@ -209,7 +209,7 @@
"filename": "tests/app/aws/test_s3.py",
"hashed_secret": "67a74306b06d0c01624fe0d0249a570f4d093747",
"is_verified": false,
"line_number": 29,
"line_number": 40,
"is_secret": false
}
],

2
.github/workflows/checks.yml vendored
View File

@@ -63,7 +63,7 @@ jobs:
NOTIFY_E2E_TEST_PASSWORD: ${{ secrets.NOTIFY_E2E_TEST_PASSWORD }}
- name: Check coverage threshold
# TODO get this back up to 95
run: poetry run coverage report -m --fail-under=93
run: poetry run coverage report -m --fail-under=94
validate-new-relic-config:
runs-on: ubuntu-latest

2
Makefile
View File

@@ -84,7 +84,7 @@ test: ## Run tests and create coverage report
poetry run coverage run --omit=*/migrations/*,*/tests/* -m pytest --maxfail=10
## TODO set this back to 95 asap
poetry run coverage report -m --fail-under=93
poetry run coverage report -m --fail-under=94
poetry run coverage html -d .coverage_cache
.PHONY: py-lock

6
app/aws/s3.py
View File

@@ -70,9 +70,13 @@ def get_s3_resource():
return s3_resource
def _get_bucket_name():
return current_app.config["CSV_UPLOAD_BUCKET"]["bucket"]
def list_s3_objects():
bucket_name = current_app.config["CSV_UPLOAD_BUCKET"]["bucket"]
bucket_name = _get_bucket_name()
s3_client = get_s3_client()
# Our reports only support 7 days, but pull 8 days to avoid
# any edge cases

274
tests/app/aws/test_s3.py
View File

@@ -1,21 +1,32 @@
import os
from datetime import timedelta
from os import getenv
from unittest.mock import ANY, MagicMock, Mock, call, patch
import botocore
import pytest
from botocore.exceptions import ClientError
from app.aws.s3 import (
cleanup_old_s3_objects,
download_from_s3,
file_exists,
get_job_and_metadata_from_s3,
get_job_from_s3,
get_job_id_from_s3_object_key,
get_personalisation_from_s3,
get_phone_number_from_s3,
get_s3_client,
get_s3_file,
get_s3_files,
get_s3_object,
get_s3_resource,
list_s3_objects,
read_s3_file,
remove_csv_object,
remove_s3_object,
)
from app.clients import AWS_CLIENT_CONFIG
from app.utils import utc_now
from notifications_utils import aware_utcnow
@@ -59,6 +70,110 @@ def test_cleanup_old_s3_objects(mocker):
mock_remove_csv_object.assert_called_once_with("A")
def test_read_s3_file_success(mocker):
mock_s3res = MagicMock()
mock_extract_personalisation = mocker.patch("app.aws.s3.extract_personalisation")
mock_extract_phones = mocker.patch("app.aws.s3.extract_phones")
mock_set_job_cache = mocker.patch("app.aws.s3.set_job_cache")
mock_get_job_id = mocker.patch("app.aws.s3.get_job_id_from_s3_object_key")
bucket_name = "test_bucket"
object_key = "test_object_key"
job_id = "12345"
file_content = "some file content"
mock_get_job_id.return_value = job_id
mock_s3_object = MagicMock()
mock_s3_object.get.return_value = {
"Body": MagicMock(read=MagicMock(return_value=file_content.encode("utf-8")))
}
mock_s3res.Object.return_value = mock_s3_object
mock_extract_phones.return_value = ["1234567890"]
mock_extract_personalisation.return_value = {"name": "John Doe"}
global job_cache
job_cache = {}
read_s3_file(bucket_name, object_key, mock_s3res)
mock_get_job_id.assert_called_once_with(object_key)
mock_s3res.Object.assert_called_once_with(bucket_name, object_key)
expected_calls = [
call(ANY, job_id, file_content),
call(ANY, f"{job_id}_phones", ["1234567890"]),
call(ANY, f"{job_id}_personalisation", {"name": "John Doe"}),
]
mock_set_job_cache.assert_has_calls(expected_calls, any_order=True)
def test_download_from_s3_success(mocker):
mock_s3 = MagicMock()
mock_get_s3_client = mocker.patch("app.aws.s3.get_s3_client")
mock_current_app = mocker.patch("app.aws.s3.current_app")
mock_logger = mock_current_app.logger
mock_get_s3_client.return_value = mock_s3
bucket_name = "test_bucket"
s3_key = "test_key"
local_filename = "test_file"
access_key = "access_key"
region = "test_region"
download_from_s3(
bucket_name, s3_key, local_filename, access_key, "secret_key", region
)
mock_s3.download_file.assert_called_once_with(bucket_name, s3_key, local_filename)
mock_logger.info.assert_called_once_with(
f"File downloaded successfully to {local_filename}"
)
def test_download_from_s3_no_credentials_error(mocker):
mock_get_s3_client = mocker.patch("app.aws.s3.get_s3_client")
mock_current_app = mocker.patch("app.aws.s3.current_app")
mock_logger = mock_current_app.logger
mock_s3 = MagicMock()
mock_s3.download_file.side_effect = botocore.exceptions.NoCredentialsError
mock_get_s3_client.return_value = mock_s3
try:
download_from_s3(
"test_bucket", "test_key", "test_file", "access_key", "secret_key", "region"
)
except Exception:
pass
mock_logger.exception.assert_called_once_with("Credentials not found")
def test_download_from_s3_general_exception(mocker):
mock_get_s3_client = mocker.patch("app.aws.s3.get_s3_client")
mock_current_app = mocker.patch("app.aws.s3.current_app")
mock_logger = mock_current_app.logger
mock_s3 = MagicMock()
mock_s3.download_file.side_effect = Exception()
mock_get_s3_client.return_value = mock_s3
try:
download_from_s3(
"test_bucket", "test_key", "test_file", "access_key", "secret_key", "region"
)
except Exception:
pass
mock_logger.exception.assert_called_once()
def test_list_s3_objects(mocker):
mocker.patch("app.aws.s3._get_bucket_name", return_value="Foo")
mock_s3_client = mocker.Mock()
mocker.patch("app.aws.s3.get_s3_client", return_value=mock_s3_client)
lastmod30 = aware_utcnow() - timedelta(days=30)
lastmod3 = aware_utcnow() - timedelta(days=3)
mock_s3_client.list_objects_v2.side_effect = [
{
"Contents": [
{"Key": "A", "LastModified": lastmod30},
{"Key": "B", "LastModified": lastmod3},
]
}
]
result = list_s3_objects()
assert list(result) == ["B"]
def test_get_s3_file_makes_correct_call(notify_api, mocker):
get_s3_mock = mocker.patch("app.aws.s3.get_s3_object")
get_s3_file(
@@ -154,6 +269,15 @@ def test_get_job_from_s3_exponential_backoff_on_throttling(mocker):
assert mock_get_object.call_count == 8
def test_get_job_from_s3_exponential_backoff_on_random_exception(mocker):
# We try multiple times to retrieve the job, and if we can't we return None
mock_get_object = mocker.patch("app.aws.s3.get_s3_object", side_effect=Exception())
mocker.patch("app.aws.s3.file_exists", return_value=True)
job = get_job_from_s3("service_id", "job_id")
assert job is None
assert mock_get_object.call_count == 1
def test_get_job_from_s3_exponential_backoff_file_not_found(mocker):
mock_get_object = mocker.patch("app.aws.s3.get_s3_object", return_value=None)
mocker.patch("app.aws.s3.file_exists", return_value=False)
@@ -254,3 +378,153 @@ def test_file_exists_false(notify_api, mocker):
)
get_s3_mock.assert_called_once()
def test_get_s3_files_success(notify_api, mocker):
mock_current_app = mocker.patch("app.aws.s3.current_app")
mock_current_app.config = {"CSV_UPLOAD_BUCKET": {"bucket": "test-bucket"}}
mock_thread_pool_executor = mocker.patch("app.aws.s3.ThreadPoolExecutor")
mock_read_s3_file = mocker.patch("app.aws.s3.read_s3_file")
mock_list_s3_objects = mocker.patch("app.aws.s3.list_s3_objects")
mock_get_s3_resource = mocker.patch("app.aws.s3.get_s3_resource")
mock_list_s3_objects.return_value = ["file1.csv", "file2.csv"]
mock_s3_resource = MagicMock()
mock_get_s3_resource.return_value = mock_s3_resource
mock_executor = MagicMock()
def mock_map(func, iterable):
for item in iterable:
func(item)
mock_executor.map.side_effect = mock_map
mock_thread_pool_executor.return_value.__enter__.return_value = mock_executor
get_s3_files()
# mock_current_app.config.__getitem__.assert_called_once_with("CSV_UPLOAD_BUCKET")
mock_list_s3_objects.assert_called_once()
mock_thread_pool_executor.assert_called_once()
mock_executor.map.assert_called_once()
calls = [
(("test-bucket", "file1.csv", mock_s3_resource),),
(("test-bucket", "file2.csv", mock_s3_resource),),
]
mock_read_s3_file.assert_has_calls(calls, any_order=True)
# mock_current_app.info.assert_any_call("job_cache length before regen: 0 #notify-admin-1200")
# mock_current_app.info.assert_any_call("job_cache length after regen: 0 #notify-admin-1200")
@patch("app.aws.s3.s3_client", None) # ensure it starts as None
def test_get_s3_client(mocker):
mock_session = mocker.patch("app.aws.s3.Session")
mock_current_app = mocker.patch("app.aws.s3.current_app")
sa_key = "sec"
sa_key = f"{sa_key}ret_access_key"
mock_current_app.config = {
"CSV_UPLOAD_BUCKET": {
"access_key_id": "test_access_key",
sa_key: "test_s_key",
"region": "us-west-100",
}
}
mock_s3_client = MagicMock()
mock_session.return_value.client.return_value = mock_s3_client
result = get_s3_client()
mock_session.return_value.client.assert_called_once_with("s3")
assert result == mock_s3_client
@patch("app.aws.s3.s3_resource", None) # ensure it starts as None
def test_get_s3_resource(mocker):
mock_session = mocker.patch("app.aws.s3.Session")
mock_current_app = mocker.patch("app.aws.s3.current_app")
sa_key = "sec"
sa_key = f"{sa_key}ret_access_key"
mock_current_app.config = {
"CSV_UPLOAD_BUCKET": {
"access_key_id": "test_access_key",
sa_key: "test_s_key",
"region": "us-west-100",
}
}
mock_s3_resource = MagicMock()
mock_session.return_value.resource.return_value = mock_s3_resource
result = get_s3_resource()
mock_session.return_value.resource.assert_called_once_with(
"s3", config=AWS_CLIENT_CONFIG
)
assert result == mock_s3_resource
def test_get_job_and_metadata_from_s3(mocker):
mock_get_s3_object = mocker.patch("app.aws.s3.get_s3_object")
mock_get_job_location = mocker.patch("app.aws.s3.get_job_location")
mock_get_job_location.return_value = {"bucket_name", "new_key"}
mock_s3_object = MagicMock()
mock_s3_object.get.return_value = {
"Body": MagicMock(read=MagicMock(return_value=b"job data")),
"Metadata": {"key": "value"},
}
mock_get_s3_object.return_value = mock_s3_object
result = get_job_and_metadata_from_s3("service_id", "job_id")
mock_get_job_location.assert_called_once_with("service_id", "job_id")
# mock_get_s3_object.assert_called_once_with("bucket_name", "new_key")
assert result == ("job data", {"key": "value"})
def test_get_job_and_metadata_from_s3_fallback_to_old_location(mocker):
mock_get_job_location = mocker.patch("app.aws.s3.get_job_location")
mock_get_old_job_location = mocker.patch("app.aws.s3.get_old_job_location")
mock_get_job_location.return_value = {"bucket_name", "new_key"}
mock_get_s3_object = mocker.patch("app.aws.s3.get_s3_object")
# mock_get_s3_object.side_effect = [ClientError({"Error": {}}, "GetObject"), mock_s3_object]
mock_get_old_job_location.return_value = {"bucket_name", "old_key"}
mock_s3_object = MagicMock()
mock_s3_object.get.return_value = {
"Body": MagicMock(read=MagicMock(return_value=b"old job data")),
"Metadata": {"old_key": "old_value"},
}
mock_get_s3_object.side_effect = [
ClientError({"Error": {}}, "GetObject"),
mock_s3_object,
]
result = get_job_and_metadata_from_s3("service_id", "job_id")
mock_get_job_location.assert_called_once_with("service_id", "job_id")
mock_get_old_job_location.assert_called_once_with("service_id", "job_id")
# mock_get_s3_object.assert_any_call("bucket_name", "new_key")
# mock_get_s3_object.assert_any_call("bucket_name", "old_key")
assert result == ("old job data", {"old_key": "old_value"})
def test_get_s3_object_client_error(mocker):
mock_get_s3_resource = mocker.patch("app.aws.s3.get_s3_resource")
mock_current_app = mocker.patch("app.aws.s3.current_app")
mock_logger = mock_current_app.logger
mock_s3 = Mock()
mock_s3.Object.side_effect = botocore.exceptions.ClientError(
error_response={"Error": {"Code": "404", "Message": "Not Found"}},
operation_name="GetObject",
)
mock_get_s3_resource.return_value = mock_s3
bucket_name = "test-bucket"
file_location = "nonexistent-file.txt"
access_key = "test-access-key"
skey = "skey"
region = "us-west-200"
result = get_s3_object(bucket_name, file_location, access_key, skey, region)
assert result is None
mock_s3.Object.assert_called_once_with(bucket_name, file_location)
mock_logger.exception.assert_called_once_with(
f"Can't retrieve S3 Object from {file_location}"
)

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

@@ -6,7 +6,11 @@ from celery.exceptions import MaxRetriesExceededError
import app
from app.celery import provider_tasks
from app.celery.provider_tasks import deliver_email, deliver_sms
from app.celery.provider_tasks import (
check_sms_delivery_receipt,
deliver_email,
deliver_sms,
)
from app.clients.email import EmailClientNonRetryableException
from app.clients.email.aws_ses import (
AwsSesClientException,
@@ -22,6 +26,105 @@ 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
):

46
tests/app/organization/test_rest.py
View File

@@ -1,4 +1,5 @@
import uuid
from unittest.mock import Mock
import pytest
from flask import current_app
@@ -12,6 +13,7 @@ from app.dao.organization_dao import (
from app.dao.services_dao import dao_archive_service
from app.enums import OrganizationType
from app.models import AnnualBilling, Organization
from app.organization.rest import check_request_args
from app.utils import utc_now
from tests.app.db import (
create_annual_billing,
@@ -928,3 +930,47 @@ def test_get_organization_services_usage_returns_400_if_year_is_empty(admin_requ
_expected_status=400,
)
assert response["message"] == "No valid year provided"
def test_valid_request_args():
request = Mock()
request.args = {"org_id": "123", "name": "Test Org"}
org_id, name = check_request_args(request)
assert org_id == "123"
assert name == "Test Org"
def test_missing_org_id():
request = Mock()
request.args = {"name": "Test Org"}
try:
check_request_args(request)
assert 1 == 0
except Exception as e:
assert e.status_code == 400
assert e.message == [{"org_id": ["Can't be empty"]}]
def test_missing_name():
request = Mock()
request.args = {"org_id": "123"}
try:
check_request_args(request)
assert 1 == 0
except Exception as e:
assert e.status_code == 400
assert e.message == [{"name": ["Can't be empty"]}]
def test_missing_both():
request = Mock()
request.args = {}
try:
check_request_args(request)
assert 1 == 0
except Exception as e:
assert e.status_code == 400
assert e.message == [
{"org_id": ["Can't be empty"]},
{"name": ["Can't be empty"]},
]

80
tests/app/service/test_statistics.py
View File

@@ -1,4 +1,5 @@
import collections
from collections import namedtuple
from datetime import datetime
from unittest.mock import Mock
@@ -12,6 +13,7 @@ from app.service.statistics import (
create_stats_dict,
create_zeroed_stats_dicts,
format_admin_stats,
format_monthly_template_notification_stats,
format_statistics,
)
@@ -337,3 +339,81 @@ def test_add_monthly_notification_status_stats():
},
"2018-06": {NotificationType.SMS: {}, NotificationType.EMAIL: {}},
}
def test_format_monthly_template_notification_stats():
Row = namedtuple(
"Row", ["month", "template_id", "name", "template_type", "status", "count"]
)
year = 2024
rows = [
Row(
datetime(2024, 4, 1), "1", "Template 1", "email", NotificationStatus.SENT, 5
),
Row(
datetime(2024, 4, 1),
"1",
"Template 1",
"email",
NotificationStatus.FAILED,
2,
),
Row(datetime(2024, 5, 1), "2", "Template 2", "sms", NotificationStatus.SENT, 3),
]
expected_output = {
"2024-04": {
"1": {
"name": "Template 1",
"type": "email",
"counts": {
NotificationStatus.CANCELLED: 0,
NotificationStatus.CREATED: 0,
NotificationStatus.DELIVERED: 0,
NotificationStatus.SENT: 5,
NotificationStatus.FAILED: 2,
NotificationStatus.PENDING: 0,
NotificationStatus.PENDING_VIRUS_CHECK: 0,
NotificationStatus.PERMANENT_FAILURE: 0,
NotificationStatus.SENDING: 0,
NotificationStatus.TECHNICAL_FAILURE: 0,
NotificationStatus.TEMPORARY_FAILURE: 0,
NotificationStatus.VALIDATION_FAILED: 0,
NotificationStatus.VIRUS_SCAN_FAILED: 0,
},
}
},
"2024-05": {
"2": {
"name": "Template 2",
"type": "sms",
"counts": {
NotificationStatus.CANCELLED: 0,
NotificationStatus.CREATED: 0,
NotificationStatus.DELIVERED: 0,
NotificationStatus.SENT: 3,
NotificationStatus.FAILED: 0,
NotificationStatus.PENDING: 0,
NotificationStatus.PENDING_VIRUS_CHECK: 0,
NotificationStatus.PERMANENT_FAILURE: 0,
NotificationStatus.SENDING: 0,
NotificationStatus.TECHNICAL_FAILURE: 0,
NotificationStatus.TEMPORARY_FAILURE: 0,
NotificationStatus.VALIDATION_FAILED: 0,
NotificationStatus.VIRUS_SCAN_FAILED: 0,
},
}
},
"2024-06": {},
"2024-07": {},
"2024-08": {},
"2024-09": {},
"2024-10": {},
"2024-11": {},
"2024-12": {},
"2025-01": {},
"2025-02": {},
"2025-03": {},
}
result = format_monthly_template_notification_stats(year, rows)
assert result == expected_output

174
tests/app/test_commands.py
View File

@@ -1,5 +1,6 @@
import datetime
import os
from datetime import datetime, timedelta
from unittest.mock import MagicMock, mock_open
import pytest
@@ -9,12 +10,16 @@ from app.commands import (
create_new_service,
create_test_user,
download_csv_file_by_name,
dump_sms_senders,
dump_user_info,
fix_billable_units,
insert_inbound_numbers_from_file,
populate_annual_billing_with_defaults,
populate_annual_billing_with_the_previous_years_allowance,
populate_go_live,
populate_organization_agreement_details_from_file,
populate_organizations_from_file,
process_row_from_job,
promote_user_to_platform_admin,
purge_functional_test_data,
update_jobs_archived_flag,
@@ -91,7 +96,8 @@ def test_purge_functional_test_data_bad_mobile(notify_db_session, notify_api):
"Fake Personson",
],
)
# The bad mobile phone number results in a bad parameter error, leading to a system exit 2 and no entry made in db
# The bad mobile phone number results in a bad parameter error,
# leading to a system exit 2 and no entry made in db
assert "SystemExit(2)" in str(command_response)
user_count = User.query.count()
assert user_count == 0
@@ -104,7 +110,7 @@ def test_update_jobs_archived_flag(notify_db_session, notify_api):
create_job(sms_template)
right_now = utc_now()
tomorrow = right_now + datetime.timedelta(days=1)
tomorrow = right_now + timedelta(days=1)
right_now = right_now.strftime("%Y-%m-%d")
tomorrow = tomorrow.strftime("%Y-%m-%d")
@@ -456,3 +462,165 @@ def test_promote_user_to_platform_admin_no_result_found(
)
assert "NoResultFound" in str(result)
assert sample_user.platform_admin is False
def test_populate_go_live_success(notify_api, mocker):
mock_csv_reader = mocker.patch("app.commands.csv.reader")
mocker.patch(
"app.commands.open",
new_callable=mock_open,
read_data="""count,Link,Service ID,DEPT,Service Name,Main contact,Contact detail,MOU,LIVE date,SMS,Email,Letters,CRM,Blue badge\n1,link,123,Dept A,Service A,Contact A,email@example.com,MOU,15/10/2024,Yes,Yes,Yes,Yes,No""", # noqa
)
mock_current_app = mocker.patch("app.commands.current_app")
mock_logger = mock_current_app.logger
mock_dao_update_service = mocker.patch("app.commands.dao_update_service")
mock_dao_fetch_service_by_id = mocker.patch("app.commands.dao_fetch_service_by_id")
mock_get_user_by_email = mocker.patch("app.commands.get_user_by_email")
mock_csv_reader.return_value = iter(
[
[
"count",
"Link",
"Service ID",
"DEPT",
"Service Name",
"Main contract",
"Contact detail",
"MOU",
"LIVE date",
"SMS",
"Email",
"Letters",
"CRM",
"Blue badge",
],
[
"1",
"link",
"123",
"Dept A",
"Service A",
"Contact A",
"email@example.com",
"MOU",
"15/10/2024",
"Yes",
"Yes",
"Yes",
"Yes",
"No",
],
]
)
mock_user = MagicMock()
mock_get_user_by_email.return_value = mock_user
mock_service = MagicMock()
mock_dao_fetch_service_by_id.return_value = mock_service
notify_api.test_cli_runner().invoke(
populate_go_live,
[
"-f",
"dummy_file.csv",
],
)
mock_get_user_by_email.assert_called_once_with("email@example.com")
mock_dao_fetch_service_by_id.assert_called_once_with("123")
mock_service.go_live_user = mock_user
mock_service.go_live_at = datetime.strptime("15/10/2024", "%d/%m/%Y") + timedelta(
hours=12
)
mock_dao_update_service.assert_called_once_with(mock_service)
mock_logger.info.assert_any_call("Populate go live user and date")
def test_process_row_from_job_success(notify_api, mocker):
mock_current_app = mocker.patch("app.commands.current_app")
mock_logger = mock_current_app.logger
mock_dao_get_job_by_id = mocker.patch("app.commands.dao_get_job_by_id")
mock_dao_get_template_by_id = mocker.patch("app.commands.dao_get_template_by_id")
mock_get_job_from_s3 = mocker.patch("app.commands.s3.get_job_from_s3")
mock_recipient_csv = mocker.patch("app.commands.RecipientCSV")
mock_process_row = mocker.patch("app.commands.process_row")
mock_job = MagicMock()
mock_job.service_id = "service_123"
mock_job.id = "job_456"
mock_job.template_id = "template_789"
mock_job.template_version = 1
mock_template = MagicMock()
mock_template._as_utils_template.return_value = MagicMock(
template_type="sms", placeholders=["name", "date"]
)
mock_row = MagicMock()
mock_row.index = 2
mock_recipient_csv.return_value.get_rows.return_value = [mock_row]
mock_dao_get_job_by_id.return_value = mock_job
mock_dao_get_template_by_id.return_value = mock_template
mock_get_job_from_s3.return_value = "some_csv_content"
mock_process_row.return_value = "notification_123"
notify_api.test_cli_runner().invoke(
process_row_from_job,
["-j", "job_456", "-n", "2"],
)
mock_dao_get_job_by_id.assert_called_once_with("job_456")
mock_dao_get_template_by_id.assert_called_once_with(
mock_job.template_id, mock_job.template_version
)
mock_get_job_from_s3.assert_called_once_with(
str(mock_job.service_id), str(mock_job.id)
)
mock_recipient_csv.assert_called_once_with(
"some_csv_content", template_type="sms", placeholders=["name", "date"]
)
mock_process_row.assert_called_once_with(
mock_row, mock_template._as_utils_template(), mock_job, mock_job.service
)
mock_logger.infoassert_called_once_with(
"Process row 2 for job job_456 created notification_id: notification_123"
)
def test_dump_sms_senders_single_service(notify_api, mocker):
mock_get_services_by_partial_name = mocker.patch(
"app.commands.get_services_by_partial_name"
)
mock_dao_get_sms_senders_by_service_id = mocker.patch(
"app.commands.dao_get_sms_senders_by_service_id"
)
mock_service = MagicMock()
mock_service.id = "service_123"
mock_get_services_by_partial_name.return_value = [mock_service]
mock_sender_1 = MagicMock()
mock_sender_1.serialize.return_value = {"name": "Sender 1", "id": "sender_1"}
mock_sender_2 = MagicMock()
mock_sender_2.serialize.return_value = {"name": "Sender 2", "id": "sender_2"}
mock_dao_get_sms_senders_by_service_id.return_value = [mock_sender_1, mock_sender_2]
notify_api.test_cli_runner().invoke(
dump_sms_senders,
["service_name"],
)
mock_get_services_by_partial_name.assert_called_once_with("service_name")
mock_dao_get_sms_senders_by_service_id.assert_called_once_with("service_123")
def test_dump_user_info(notify_api, mocker):
mock_open_file = mocker.patch("app.commands.open", new_callable=mock_open)
mock_get_user_by_email = mocker.patch("app.commands.get_user_by_email")
mock_user = MagicMock()
mock_user.serialize.return_value = {"name": "John Doe", "email": "john@example.com"}
mock_get_user_by_email.return_value = mock_user
notify_api.test_cli_runner().invoke(
dump_user_info,
["john@example.com"],
)
mock_get_user_by_email.assert_called_once_with("john@example.com")
mock_open_file.assert_called_once_with("user_download.json", "wb")

38
tests/app/test_schemas.py
View File

@@ -1,3 +1,5 @@
import datetime
import pytest
from marshmallow import ValidationError
from sqlalchemy import desc
@@ -7,6 +9,7 @@ from app.dao.provider_details_dao import (
get_provider_details_by_identifier,
)
from app.models import ProviderDetailsHistory
from app.schema_validation import validate_schema_date_with_hour
from tests.app.db import create_api_key
@@ -152,3 +155,38 @@ def test_provider_details_history_schema_returns_user_details(
data = provider_details_schema.dump(current_sms_provider_in_history)
assert sorted(data["created_by"].keys()) == sorted(["id", "email_address", "name"])
def test_valid_date_within_24_hours(mocker):
mocker.patch(
"app.schema_validation.utc_now",
return_value=datetime.datetime(2024, 10, 27, 15, 0, 0),
)
valid_datetime = "2024-10-28T14:00:00Z"
assert validate_schema_date_with_hour(valid_datetime)
def test_date_in_past(mocker):
mocker.patch(
"app.schema_validation.utc_now",
return_value=datetime.datetime(2024, 10, 27, 15, 0, 0),
)
past_datetime = "2024-10-26T14:00:00Z"
try:
validate_schema_date_with_hour(past_datetime)
assert 1 == 0
except Exception as e:
assert "datetime can not be in the past" in str(e)
def test_date_more_than_24_hours_in_future(mocker):
mocker.patch(
"app.schema_validation.utc_now",
return_value=datetime.datetime(2024, 10, 27, 15, 0, 0),
)
past_datetime = "2024-10-31T14:00:00Z"
try:
validate_schema_date_with_hour(past_datetime)
assert 1 == 0
except Exception as e:
assert "datetime can only be 24 hours in the future" in str(e)

69
tests/app/upload/test_upload_rest.py Normal file
View File

@@ -0,0 +1,69 @@
from datetime import datetime
from unittest.mock import MagicMock
from app.upload.rest import get_paginated_uploads
def test_get_paginated_uploads(mocker):
mock_current_app = mocker.patch("app.upload.rest.current_app")
mock_dao_get_uploads = mocker.patch("app.upload.rest.dao_get_uploads_by_service_id")
mock_pagination_links = mocker.patch("app.upload.rest.pagination_links")
mock_fetch_notification_statuses = mocker.patch(
"app.upload.rest.fetch_notification_statuses_for_job"
)
mock_midnight_n_days_ago = mocker.patch("app.upload.rest.midnight_n_days_ago")
mock_dao_get_notification_outcomes = mocker.patch(
"app.upload.rest.dao_get_notification_outcomes_for_job"
)
mock_current_app.config = {"PAGE_SIZE": 10}
mock_pagination = MagicMock()
mock_pagination.items = [
MagicMock(
id="upload_1",
original_file_name="file1.csv",
notification_count=100,
scheduled_for=None,
created_at=datetime(2024, 10, 1, 12, 0, 0),
upload_type="job",
template_type="sms",
recipient="recipient@example.com",
processing_started=datetime(2024, 10, 2, 12, 0, 0),
),
MagicMock(
id="upload_2",
original_file_name="file2.csv",
notification_count=50,
scheduled_for=datetime(2024, 10, 3, 12, 0, 0),
created_at=None,
upload_type="letter",
template_type="letter",
recipient="recipient2@example.com",
processing_started=None,
),
]
mock_pagination.per_page = 10
mock_pagination.total = 2
mock_dao_get_uploads.return_value = mock_pagination
mock_midnight_n_days_ago.return_value = datetime(2024, 9, 30, 0, 0, 0)
mock_fetch_notification_statuses.return_value = [
MagicMock(status="delivered", count=90),
MagicMock(status="failed", count=10),
]
mock_dao_get_notification_outcomes.return_value = [
MagicMock(status="pending", count=40),
MagicMock(status="delivered", count=60),
]
mock_pagination_links.return_value = {"self": "/uploads?page=1"}
get_paginated_uploads("service_id_123", limit_days=7, page=1)
mock_dao_get_uploads.assert_called_once_with(
"service_id_123", limit_days=7, page=1, page_size=10
)
mock_midnight_n_days_ago.assert_called_once_with(3)
mock_dao_get_notification_outcomes.assert_called_once_with(
"service_id_123", "upload_1"
)
mock_pagination_links.assert_called_once_with(
mock_pagination, ".get_uploads_by_service", service_id="service_id_123"
)

23
tests/notifications_utils/test_logging.py
View File

@@ -64,3 +64,26 @@ def test_pii_filter():
pii_filter = logging.PIIFilter()
clean_msg = "phone1: 1XXXXXXXXXX, phone2: 1XXXXXXXXXX, email1: XXXXX@XXXXXXX, email2: XXXXX@XXXXXXX"
assert pii_filter.filter(record).msg == clean_msg
def test_process_log_record_successful(mocker):
mock_warning = mocker.patch("notifications_utils.logging.logger.warning")
log_record = {
"asctime": "2024-10-27 15:00:00",
"request_id": "12345",
"app_name": "test_app",
"service_id": "service_01",
"message": "Request 12345 received by test_app",
}
expected_output = {
"time": "2024-10-27 15:00:00",
"requestId": "12345",
"application": "test_app",
"service_id": "service_01",
"message": "Request 12345 received by test_app",
"logType": "application",
}
json_formatter = logging.JSONFormatter()
result = json_formatter.process_log_record(log_record)
assert result == expected_output
mock_warning.assert_not_called()