darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-08-23 15:56:45 -04:00
Code Issues Packages Projects Releases Wiki Activity

More fixes.

Browse Source 
Signed-off-by: Cliff Hill <Clifford.hill@gsa.gov>
This commit is contained in:
Cliff Hill
2024-02-16 16:30:17 -05:00
parent 287dcd6a46
commit 8e8749a40e
3 changed files with 197 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
tests/app/service/test_service_data_retention_rest.py
View File

@@ -1,6 +1,7 @@
import json import json
import uuid import uuid
from app.enums import NotificationType
from app.models import ServiceDataRetention from app.models import ServiceDataRetention
from tests import create_admin_authorization_header from tests import create_admin_authorization_header
from tests.app.db import create_service_data_retention from tests.app.db import create_service_data_retention
@@ -13,7 +14,7 @@ def test_get_service_data_retention(client, sample_service):
) )
response = client.get( response = client.get(
"/service/{}/data-retention".format(str(sample_service.id)), f"/service/{sample_service.id!s}/data-retention",
headers=[ headers=[
("Content-Type", "application/json"), ("Content-Type", "application/json"),
create_admin_authorization_header(), create_admin_authorization_header(),
@@ -29,7 +30,7 @@ def test_get_service_data_retention(client, sample_service):
def test_get_service_data_retention_returns_empty_list(client, sample_service): def test_get_service_data_retention_returns_empty_list(client, sample_service):
response = client.get( response = client.get(
"/service/{}/data-retention".format(str(sample_service.id)), f"/service/{sample_service.id!s}/data-retention",
headers=[ headers=[
("Content-Type", "application/json"), ("Content-Type", "application/json"),
create_admin_authorization_header(), create_admin_authorization_header(),
@@ -42,9 +43,8 @@ def test_get_service_data_retention_returns_empty_list(client, sample_service):
def test_get_data_retention_for_service_notification_type(client, sample_service): def test_get_data_retention_for_service_notification_type(client, sample_service):
data_retention = create_service_data_retention(service=sample_service) data_retention = create_service_data_retention(service=sample_service)
response = client.get( response = client.get(
"/service/{}/data-retention/notification-type/{}".format( f"/service/{sample_service.id}/data-retention/"
sample_service.id, "sms" f"notification-type/{NotificationType.SMS}",
),
headers=[ headers=[
("Content-Type", "application/json"), ("Content-Type", "application/json"),
create_admin_authorization_header(), create_admin_authorization_header(),
@@ -57,15 +57,17 @@ def test_get_data_retention_for_service_notification_type(client, sample_service
def test_get_service_data_retention_by_id(client, sample_service): def test_get_service_data_retention_by_id(client, sample_service):
sms_data_retention = create_service_data_retention(service=sample_service) sms_data_retention = create_service_data_retention(service=sample_service)
create_service_data_retention( create_service_data_retention(
service=sample_service, notification_type="email", days_of_retention=10 service=sample_service,
notification_type=NotificationType.EMAIL,
days_of_retention=10,
) )
create_service_data_retention( create_service_data_retention(
service=sample_service, notification_type="letter", days_of_retention=30 service=sample_service,
notification_type=NotificationType.LETTER,
days_of_retention=30,
) )
response = client.get( response = client.get(
"/service/{}/data-retention/{}".format( f"/service/{sample_service.id!s}/data-retention/{sms_data_retention.id}",
str(sample_service.id), sms_data_retention.id
),
headers=[ headers=[
("Content-Type", "application/json"), ("Content-Type", "application/json"),
create_admin_authorization_header(), create_admin_authorization_header(),
@@ -79,7 +81,7 @@ def test_get_service_data_retention_by_id_returns_none_when_no_data_retention_ex
client, sample_service client, sample_service
): ):
response = client.get( response = client.get(
"/service/{}/data-retention/{}".format(str(sample_service.id), uuid.uuid4()), f"/service/{sample_service.id!s}/data-retention/{uuid.uuid4()}",
headers=[ headers=[
("Content-Type", "application/json"), ("Content-Type", "application/json"),
create_admin_authorization_header(), create_admin_authorization_header(),
@@ -90,9 +92,9 @@ def test_get_service_data_retention_by_id_returns_none_when_no_data_retention_ex
def test_create_service_data_retention(client, sample_service): def test_create_service_data_retention(client, sample_service):
data = {"notification_type": "sms", "days_of_retention": 3} data = {"notification_type": NotificationType.SMS, "days_of_retention": 3}
response = client.post( response = client.post(
"/service/{}/data-retention".format(str(sample_service.id)), f"/service/{sample_service.id!s}/data-retention",
headers=[ headers=[
("Content-Type", "application/json"), ("Content-Type", "application/json"),
create_admin_authorization_header(), create_admin_authorization_header(),
@@ -113,7 +115,7 @@ def test_create_service_data_retention_returns_400_when_notification_type_is_inv
): ):
data = {"notification_type": "unknown", "days_of_retention": 3} data = {"notification_type": "unknown", "days_of_retention": 3}
response = client.post( response = client.post(
"/service/{}/data-retention".format(str(uuid.uuid4())), f"/service/{uuid.uuid4()!s}/data-retention",
headers=[ headers=[
("Content-Type", "application/json"), ("Content-Type", "application/json"),
create_admin_authorization_header(), create_admin_authorization_header(),
@@ -133,9 +135,9 @@ def test_create_service_data_retention_returns_400_when_data_retention_for_notif
client, sample_service client, sample_service
): ):
create_service_data_retention(service=sample_service) create_service_data_retention(service=sample_service)
data = {"notification_type": "sms", "days_of_retention": 3} data = {"notification_type": NotificationType.SMS, "days_of_retention": 3}
response = client.post( response = client.post(
"/service/{}/data-retention".format(str(uuid.uuid4())), f"/service/{uuid.uuid4()!s}/data-retention",
headers=[ headers=[
("Content-Type", "application/json"), ("Content-Type", "application/json"),
create_admin_authorization_header(), create_admin_authorization_header(),
@@ -156,7 +158,7 @@ def test_modify_service_data_retention(client, sample_service):
data_retention = create_service_data_retention(service=sample_service) data_retention = create_service_data_retention(service=sample_service)
data = {"days_of_retention": 3} data = {"days_of_retention": 3}
response = client.post( response = client.post(
"/service/{}/data-retention/{}".format(sample_service.id, data_retention.id), f"/service/{sample_service.id}/data-retention/{data_retention.id}",
headers=[ headers=[
("Content-Type", "application/json"), ("Content-Type", "application/json"),
create_admin_authorization_header(), create_admin_authorization_header(),
@@ -172,7 +174,7 @@ def test_modify_service_data_retention_returns_400_when_data_retention_does_not_
): ):
data = {"days_of_retention": 3} data = {"days_of_retention": 3}
response = client.post( response = client.post(
"/service/{}/data-retention/{}".format(sample_service.id, uuid.uuid4()), f"/service/{sample_service.id}/data-retention/{uuid.uuid4()}",
headers=[ headers=[
("Content-Type", "application/json"), ("Content-Type", "application/json"),
create_admin_authorization_header(), create_admin_authorization_header(),
@@ -186,7 +188,7 @@ def test_modify_service_data_retention_returns_400_when_data_retention_does_not_
def test_modify_service_data_retention_returns_400_when_data_is_invalid(client): def test_modify_service_data_retention_returns_400_when_data_is_invalid(client):
data = {"bad_key": 3} data = {"bad_key": 3}
response = client.post( response = client.post(
"/service/{}/data-retention/{}".format(uuid.uuid4(), uuid.uuid4()), f"/service/{uuid.uuid4()}/data-retention/{uuid.uuid4()}",
headers=[ headers=[
("Content-Type", "application/json"), ("Content-Type", "application/json"),
create_admin_authorization_header(), create_admin_authorization_header(),

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

@@ -5,6 +5,7 @@ from unittest.mock import Mock
import pytest import pytest
from freezegun import freeze_time from freezegun import freeze_time
from app.enums import KeyType, NotificationStatus, NotificationType
from app.service.statistics import ( from app.service.statistics import (
add_monthly_notification_status_stats, add_monthly_notification_status_stats,
create_empty_monthly_notification_status_stats_dict, create_empty_monthly_notification_status_stats_dict,
@@ -26,39 +27,51 @@ NewStatsRow = collections.namedtuple(
{ {
"empty": ([], [0, 0, 0], [0, 0, 0]), "empty": ([], [0, 0, 0], [0, 0, 0]),
"always_increment_requested": ( "always_increment_requested": (
[StatsRow("email", "delivered", 1), StatsRow("email", "failed", 1)], [
StatsRow(NotificationType.EMAIL, NotificationStatus.DELIVERED, 1),
StatsRow(NotificationType.EMAIL, NotificationStatus.FAILED, 1),
],
[2, 1, 1], [2, 1, 1],
[0, 0, 0], [0, 0, 0],
), ),
"dont_mix_template_types": ( "dont_mix_template_types": (
[ [
StatsRow("email", "delivered", 1), StatsRow(NotificationType.EMAIL, NotificationStatus.DELIVERED, 1),
StatsRow("sms", "delivered", 1), StatsRow(NotificationType.SMS, NotificationStatus.DELIVERED, 1),
], ],
[1, 1, 0], [1, 1, 0],
[1, 1, 0], [1, 1, 0],
), ),
"convert_fail_statuses_to_failed": ( "convert_fail_statuses_to_failed": (
[ [
StatsRow("email", "failed", 1), StatsRow(NotificationType.EMAIL, NotificationStatus.FAILED, 1),
StatsRow("email", "technical-failure", 1), StatsRow(
StatsRow("email", "temporary-failure", 1), NotificationType.EMAIL, NotificationStatus.TECHNICAL_FAILURE, 1
StatsRow("email", "permanent-failure", 1), ),
StatsRow(
NotificationType.EMAIL, NotificationStatus.TEMPORARY_FAILURE, 1
),
StatsRow(
NotificationType.EMAIL, NotificationStatus.PERMANENT_FAILURE, 1
),
], ],
[4, 0, 4], [4, 0, 4],
[0, 0, 0], [0, 0, 0],
), ),
"convert_sent_to_delivered": ( "convert_sent_to_delivered": (
[ [
StatsRow("sms", "sending", 1), StatsRow(NotificationType.SMS, NotificationStatus.SENDING, 1),
StatsRow("sms", "delivered", 1), StatsRow(NotificationType.SMS, NotificationStatus.DELIVERED, 1),
StatsRow("sms", "sent", 1), StatsRow(NotificationType.SMS, NotificationStatus.SENT, 1),
], ],
[0, 0, 0], [0, 0, 0],
[3, 2, 0], [3, 2, 0],
), ),
"handles_none_rows": ( "handles_none_rows": (
[StatsRow("sms", "sending", 1), StatsRow(None, None, None)], [
StatsRow(NotificationType.SMS, NotificationStatus.SENDING, 1),
StatsRow(None, None, None),
],
[0, 0, 0], [0, 0, 0],
[1, 0, 0], [1, 0, 0],
), ),
@@ -67,44 +80,66 @@ NewStatsRow = collections.namedtuple(
def test_format_statistics(stats, email_counts, sms_counts): def test_format_statistics(stats, email_counts, sms_counts):
ret = format_statistics(stats) ret = format_statistics(stats)
assert ret["email"] == { assert ret[NotificationType.EMAIL] == {
status: count status: count
for status, count in zip(["requested", "delivered", "failed"], email_counts) for status, count in zip(
[
NotificationStatus.REQUESTED,
NotificationStatus.DELIVERED,
NotificationStatus.FAILED,
],
email_counts,
)
} }
assert ret["sms"] == { assert ret[NotificationType.SMS] == {
status: count status: count
for status, count in zip(["requested", "delivered", "failed"], sms_counts) for status, count in zip(
[
NotificationStatus.REQUESTED,
NotificationStatus.DELIVERED,
NotificationStatus.FAILED,
],
sms_counts,
)
} }
def test_create_zeroed_stats_dicts(): def test_create_zeroed_stats_dicts():
assert create_zeroed_stats_dicts() == { assert create_zeroed_stats_dicts() == {
"sms": {"requested": 0, "delivered": 0, "failed": 0}, NotificationType.SMS: {
"email": {"requested": 0, "delivered": 0, "failed": 0}, NotificationStatus.REQUESTED: 0,
NotificationStatus.DELIVERED: 0,
NotificationStatus.FAILED: 0,
},
NotificationType.EMAIL: {
NotificationStatus.REQUESTED: 0,
NotificationStatus.DELIVERED: 0,
NotificationStatus.FAILED: 0,
},
} }
def test_create_stats_dict(): def test_create_stats_dict():
assert create_stats_dict() == { assert create_stats_dict() == {
"sms": { NotificationType.SMS: {
"total": 0, "total": 0,
"test-key": 0, "test-key": 0,
"failures": { "failures": {
"technical-failure": 0, NotificationStatus.TECHNICAL_FAILURE: 0,
"permanent-failure": 0, NotificationStatus.PERMANENT_FAILURE: 0,
"temporary-failure": 0, NotificationStatus.TEMPORARY_FAILURE: 0,
"virus-scan-failed": 0, NotificationStatus.VIRUS_SCAN_FAILED: 0,
}, },
}, },
"email": { NotificationType.EMAIL: {
"total": 0, "total": 0,
"test-key": 0, "test-key": 0,
"failures": { "failures": {
"technical-failure": 0, NotificationStatus.TECHNICAL_FAILURE: 0,
"permanent-failure": 0, NotificationStatus.PERMANENT_FAILURE: 0,
"temporary-failure": 0, NotificationStatus.TEMPORARY_FAILURE: 0,
"virus-scan-failed": 0, NotificationStatus.VIRUS_SCAN_FAILED: 0,
}, },
}, },
} }
@@ -112,38 +147,89 @@ def test_create_stats_dict():
def test_format_admin_stats_only_includes_test_key_notifications_in_test_key_section(): def test_format_admin_stats_only_includes_test_key_notifications_in_test_key_section():
rows = [ rows = [
NewStatsRow("email", "technical-failure", "test", 3), NewStatsRow(
NewStatsRow("sms", "permanent-failure", "test", 4), NotificationType.EMAIL,
NotificationStatus.TECHNICAL_FAILURE,
KeyType.TEST,
3,
),
NewStatsRow(
NotificationType.SMS, NotificationType.PERMANENT_FAILURE, KeyType.TEST, 4
),
] ]
stats_dict = format_admin_stats(rows) stats_dict = format_admin_stats(rows)
assert stats_dict["email"]["total"] == 0 assert stats_dict[NotificationType.EMAIL]["total"] == 0
assert stats_dict["email"]["failures"]["technical-failure"] == 0 assert (
assert stats_dict["email"]["test-key"] == 3 stats_dict[NotificationType.EMAIL]["failures"][
NotificationStatus.TECHNICAL_FAILURE
]
== 0
)
assert stats_dict[NotificationType.EMAIL]["test-key"] == 3
assert stats_dict["sms"]["total"] == 0 assert stats_dict[NotificationType.SMS]["total"] == 0
assert stats_dict["sms"]["failures"]["permanent-failure"] == 0 assert (
assert stats_dict["sms"]["test-key"] == 4 stats_dict[NotificationType.SMS]["failures"][
NotificationStatus.PERMANENT_FAILURE
]
== 0
)
assert stats_dict[NotificationType.SMS]["test-key"] == 4
def test_format_admin_stats_counts_non_test_key_notifications_correctly(): def test_format_admin_stats_counts_non_test_key_notifications_correctly():
rows = [ rows = [
NewStatsRow("email", "technical-failure", "normal", 1), NewStatsRow(
NewStatsRow("email", "created", "team", 3), NotificationType.EMAIL,
NewStatsRow("sms", "temporary-failure", "normal", 6), NotificationStatus.TECHNICAL_FAILURE,
NewStatsRow("sms", "sent", "normal", 2), KeyType.NORMAL,
1,
),
NewStatsRow(
NotificationType.EMAIL,
NotificationStatus.CREATED,
KeyType.TEAM,
3,
),
NewStatsRow(
NotificationType.SMS,
NotificationStatus.TEMPORARY_FAILURE,
KeyType.NORMAL,
6,
),
NewStatsRow(
NotificationType.SMS,
NotificationStatus.SENT,
KeyType.NORMAL,
2,
),
] ]
stats_dict = format_admin_stats(rows) stats_dict = format_admin_stats(rows)
assert stats_dict["email"]["total"] == 4 assert stats_dict[NotificationType.EMAIL]["total"] == 4
assert stats_dict["email"]["failures"]["technical-failure"] == 1 assert (
stats_dict[NotificationType.EMAIL]["failures"][
NotificationStatus.TECHNICAL_FAILURE
]
== 1
)
assert stats_dict["sms"]["total"] == 8 assert stats_dict[NotificationType.SMS]["total"] == 8
assert stats_dict["sms"]["failures"]["permanent-failure"] == 0 assert (
stats_dict[NotificationType.SMS]["failures"][
NotificationStatus.PERMANENT_FAILURE
]
== 0
)
def _stats(requested, delivered, failed): def _stats(requested, delivered, failed):
return {"requested": requested, "delivered": delivered, "failed": failed} return {
NotificationStatus.REQUESTED: requested,
NotificationStatus.DELIVERED: delivered,
NotificationStatus.FAILED: failed,
}
@pytest.mark.parametrize( @pytest.mark.parametrize(
@@ -174,7 +260,7 @@ def test_create_empty_monthly_notification_status_stats_dict(year, expected_year
output = create_empty_monthly_notification_status_stats_dict(year) output = create_empty_monthly_notification_status_stats_dict(year)
assert sorted(output.keys()) == expected_years assert sorted(output.keys()) == expected_years
for v in output.values(): for v in output.values():
assert v == {"sms": {}, "email": {}} assert v == {NotificationType.SMS: {}, NotificationType.EMAIL: {}}
@freeze_time("2018-06-01 04:59:59") @freeze_time("2018-06-01 04:59:59")
@@ -182,26 +268,26 @@ def test_add_monthly_notification_status_stats():
row_data = [ row_data = [
{ {
"month": datetime(2018, 4, 1), "month": datetime(2018, 4, 1),
"notification_type": "sms", "notification_type": NotificationType.SMS,
"notification_status": "sending", "notification_status": NotificationStatus.SENDING,
"count": 1, "count": 1,
}, },
{ {
"month": datetime(2018, 4, 1), "month": datetime(2018, 4, 1),
"notification_type": "sms", "notification_type": NotificationType.SMS,
"notification_status": "delivered", "notification_status": NotificationStatus.DELIVERED,
"count": 2, "count": 2,
}, },
{ {
"month": datetime(2018, 4, 1), "month": datetime(2018, 4, 1),
"notification_type": "email", "notification_type": NotificationType.EMAIL,
"notification_status": "sending", "notification_status": NotificationStatus.SENDING,
"count": 4, "count": 4,
}, },
{ {
"month": datetime(2018, 5, 1), "month": datetime(2018, 5, 1),
"notification_type": "sms", "notification_type": NotificationType.SMS,
"notification_status": "sending", "notification_status": NotificationStatus.SENDING,
"count": 8, "count": 8,
}, },
] ]
@@ -214,18 +300,27 @@ def test_add_monthly_notification_status_stats():
data = create_empty_monthly_notification_status_stats_dict(2018) data = create_empty_monthly_notification_status_stats_dict(2018)
# this data won't be affected # this data won't be affected
data["2018-05"]["email"]["sending"] = 32 data["2018-05"][NotificationType.EMAIL][NotificationStatus.SENDING] = 32
# this data will get combined with the 8 from row_data # this data will get combined with the 8 from row_data
data["2018-05"]["sms"]["sending"] = 16 data["2018-05"][NotificationType.SMS][NotificationStatus.SENDING] = 16
add_monthly_notification_status_stats(data, rows) add_monthly_notification_status_stats(data, rows)
# first 3 months are empty # first 3 months are empty
assert data == { assert data == {
"2018-01": {"sms": {}, "email": {}}, "2018-01": {NotificationType.SMS: {}, NotificationType.EMAIL: {}},
"2018-02": {"sms": {}, "email": {}}, "2018-02": {NotificationType.SMS: {}, NotificationType.EMAIL: {}},
"2018-03": {"sms": {}, "email": {}}, "2018-03": {NotificationType.SMS: {}, NotificationType.EMAIL: {}},
"2018-04": {"sms": {"sending": 1, "delivered": 2}, "email": {"sending": 4}}, "2018-04": {
"2018-05": {"sms": {"sending": 24}, "email": {"sending": 32}}, NotificationType.SMS: {
"2018-06": {"sms": {}, "email": {}}, NotificationStatus.SENDING: 1,
NotificationStatus.DELIVERED: 2,
},
NotificationType.EMAIL: {NotificationStatus.SENDING: 4},
},
"2018-05": {
NotificationType.SMS: {NotificationStatus.SENDING: 24},
NotificationType.EMAIL: {NotificationStatus.SENDING: 32},
},
"2018-06": {NotificationType.SMS: {}, NotificationType.EMAIL: {}},
} }

32
tests/app/service/test_statistics_rest.py
View File

@@ -4,7 +4,7 @@ from datetime import date, datetime
import pytest import pytest
from freezegun import freeze_time from freezegun import freeze_time
from app.enums import KeyType, NotificationType, TemplateType from app.enums import KeyType, NotificationStatus, NotificationType, TemplateType
from tests.app.db import ( from tests.app.db import (
create_ft_notification_status, create_ft_notification_status,
create_notification, create_notification,
@@ -112,7 +112,7 @@ def test_get_service_notification_statistics(
date(2000, 1, 1), NotificationType.SMS, sample_service, count=1 date(2000, 1, 1), NotificationType.SMS, sample_service, count=1
) )
with freeze_time("2000-01-02T12:00:00"): with freeze_time("2000-01-02T12:00:00"):
create_notification(sample_template, status="created") create_notification(sample_template, status=NotificationStatus.CREATED)
resp = admin_request.get( resp = admin_request.get(
"service.get_service_notification_statistics", "service.get_service_notification_statistics",
service_id=sample_template.service_id, service_id=sample_template.service_id,
@@ -120,10 +120,10 @@ def test_get_service_notification_statistics(
) )
assert set(resp["data"].keys()) == { assert set(resp["data"].keys()) == {
NotificationType.SMS.value, NotificationType.SMS,
NotificationType.EMAIL.value, NotificationType.EMAIL,
} }
assert resp["data"][NotificationType.SMS.value] == stats assert resp["data"][NotificationType.SMS] == stats
def test_get_service_notification_statistics_with_unknown_service(admin_request): def test_get_service_notification_statistics_with_unknown_service(admin_request):
@@ -132,8 +132,8 @@ def test_get_service_notification_statistics_with_unknown_service(admin_request)
) )
assert resp["data"] == { assert resp["data"] == {
NotificationType.SMS.value: {"requested": 0, "delivered": 0, "failed": 0}, NotificationType.SMS: {"requested": 0, "delivered": 0, "failed": 0},
NotificationType.EMAIL.value: {"requested": 0, "delivered": 0, "failed": 0}, NotificationType.EMAIL: {"requested": 0, "delivered": 0, "failed": 0},
} }
@@ -238,25 +238,33 @@ def test_get_monthly_notification_stats_combines_todays_data_and_historic_stats(
admin_request, sample_template admin_request, sample_template
): ):
create_ft_notification_status( create_ft_notification_status(
datetime(2016, 5, 1, 12), template=sample_template, count=1 datetime(2016, 5, 1, 12),
template=sample_template,
count=1,
) )
create_ft_notification_status( create_ft_notification_status(
datetime(2016, 6, 1, 12), datetime(2016, 6, 1, 12),
template=sample_template, template=sample_template,
notification_status="created", notification_status=NotificationStatus.CREATED,
count=2, count=2,
) # noqa ) # noqa
create_notification( create_notification(
sample_template, created_at=datetime(2016, 6, 5, 12), status="created" sample_template,
created_at=datetime(2016, 6, 5, 12),
status=NotificationStatus.CREATED,
) )
create_notification( create_notification(
sample_template, created_at=datetime(2016, 6, 5, 12), status="delivered" sample_template,
created_at=datetime(2016, 6, 5, 12),
status=NotificationStatus.DELIVERED,
) )
# this doesn't get returned in the stats because it is old - it should be in ft_notification_status by now # this doesn't get returned in the stats because it is old - it should be in ft_notification_status by now
create_notification( create_notification(
sample_template, created_at=datetime(2016, 6, 4, 12), status="sending" sample_template,
created_at=datetime(2016, 6, 4, 12),
status=NotificationStatus.SENDING,
) )
response = admin_request.get( response = admin_request.get(