Files
notifications-api/tests/app/service/test_statistics.py
Cliff Hill b5dac890a2 Debugging.
Signed-off-by: Cliff Hill <Clifford.hill@gsa.gov>
2024-02-28 12:58:23 -05:00

327 lines
9.7 KiB
Python

import collections
from datetime import datetime
from unittest.mock import Mock
import pytest
from freezegun import freeze_time
from app.enums import KeyType, NotificationStatus, NotificationType, StatisticsType
from app.service.statistics import (
add_monthly_notification_status_stats,
create_empty_monthly_notification_status_stats_dict,
create_stats_dict,
create_zeroed_stats_dicts,
format_admin_stats,
format_statistics,
)
StatsRow = collections.namedtuple("row", ("notification_type", "status", "count"))
NewStatsRow = collections.namedtuple(
"row", ("notification_type", "status", "key_type", "count")
)
# email_counts and sms_counts are 3-tuple of requested, delivered, failed
@pytest.mark.idparametrize(
"stats, email_counts, sms_counts",
{
"empty": ([], [0, 0, 0], [0, 0, 0]),
"always_increment_requested": (
[
StatsRow(NotificationType.EMAIL, NotificationStatus.DELIVERED, 1),
StatsRow(NotificationType.EMAIL, NotificationStatus.FAILED, 1),
],
[2, 1, 1],
[0, 0, 0],
),
"dont_mix_template_types": (
[
StatsRow(NotificationType.EMAIL, NotificationStatus.DELIVERED, 1),
StatsRow(NotificationType.SMS, NotificationStatus.DELIVERED, 1),
],
[1, 1, 0],
[1, 1, 0],
),
"convert_fail_statuses_to_failed": (
[
StatsRow(NotificationType.EMAIL, NotificationStatus.FAILED, 1),
StatsRow(
NotificationType.EMAIL, NotificationStatus.TECHNICAL_FAILURE, 1
),
StatsRow(
NotificationType.EMAIL, NotificationStatus.TEMPORARY_FAILURE, 1
),
StatsRow(
NotificationType.EMAIL, NotificationStatus.PERMANENT_FAILURE, 1
),
],
[4, 0, 4],
[0, 0, 0],
),
"convert_sent_to_delivered": (
[
StatsRow(NotificationType.SMS, NotificationStatus.SENDING, 1),
StatsRow(NotificationType.SMS, NotificationStatus.DELIVERED, 1),
StatsRow(NotificationType.SMS, NotificationStatus.SENT, 1),
],
[0, 0, 0],
[3, 2, 0],
),
"handles_none_rows": (
[
StatsRow(NotificationType.SMS, NotificationStatus.SENDING, 1),
StatsRow(None, None, None),
],
[0, 0, 0],
[1, 0, 0],
),
},
)
def test_format_statistics(stats, email_counts, sms_counts):
ret = format_statistics(stats)
assert ret[NotificationType.EMAIL] == {
status: count
for status, count in zip(
[
StatisticsType.REQUESTED,
StatisticsType.DELIVERED,
StatisticsType.FAILURE,
],
email_counts,
)
}
assert ret[NotificationType.SMS] == {
status: count
for status, count in zip(
[
StatisticsType.REQUESTED,
StatisticsType.DELIVERED,
StatisticsType.FAILURE,
],
sms_counts,
)
}
def test_create_zeroed_stats_dicts():
assert create_zeroed_stats_dicts() == {
NotificationType.SMS: {
StatisticsType.REQUESTED: 0,
StatisticsType.DELIVERED: 0,
StatisticsType.FAILURE: 0,
},
NotificationType.EMAIL: {
StatisticsType.REQUESTED: 0,
StatisticsType.DELIVERED: 0,
StatisticsType.FAILURE: 0,
},
}
def test_create_stats_dict():
assert create_stats_dict() == {
NotificationType.SMS: {
"total": 0,
"test-key": 0,
"failures": {
NotificationStatus.TECHNICAL_FAILURE: 0,
NotificationStatus.PERMANENT_FAILURE: 0,
NotificationStatus.TEMPORARY_FAILURE: 0,
NotificationStatus.VIRUS_SCAN_FAILED: 0,
},
},
NotificationType.EMAIL: {
"total": 0,
"test-key": 0,
"failures": {
NotificationStatus.TECHNICAL_FAILURE: 0,
NotificationStatus.PERMANENT_FAILURE: 0,
NotificationStatus.TEMPORARY_FAILURE: 0,
NotificationStatus.VIRUS_SCAN_FAILED: 0,
},
},
}
def test_format_admin_stats_only_includes_test_key_notifications_in_test_key_section():
rows = [
NewStatsRow(
NotificationType.EMAIL,
NotificationStatus.TECHNICAL_FAILURE,
KeyType.TEST,
3,
),
NewStatsRow(
NotificationType.SMS, NotificationStatus.PERMANENT_FAILURE, KeyType.TEST, 4
),
]
stats_dict = format_admin_stats(rows)
assert stats_dict[NotificationType.EMAIL]["total"] == 0
assert (
stats_dict[NotificationType.EMAIL]["failures"][
NotificationStatus.TECHNICAL_FAILURE
]
== 0
)
assert stats_dict[NotificationType.EMAIL]["test-key"] == 3
assert stats_dict[NotificationType.SMS]["total"] == 0
assert (
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():
rows = [
NewStatsRow(
NotificationType.EMAIL,
NotificationStatus.TECHNICAL_FAILURE,
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)
assert stats_dict[NotificationType.EMAIL]["total"] == 4
assert (
stats_dict[NotificationType.EMAIL]["failures"][
NotificationStatus.TECHNICAL_FAILURE
]
== 1
)
assert stats_dict[NotificationType.SMS]["total"] == 8
assert (
stats_dict[NotificationType.SMS]["failures"][
NotificationStatus.PERMANENT_FAILURE
]
== 0
)
def _stats(requested, delivered, failed):
return {
StatisticsType.REQUESTED: requested,
StatisticsType.DELIVERED: delivered,
StatisticsType.FAILURE: failed,
}
@pytest.mark.parametrize(
"year, expected_years",
[
(2018, ["2018-01", "2018-02", "2018-03", "2018-04", "2018-05", "2018-06"]),
(
2017,
[
"2017-01",
"2017-02",
"2017-03",
"2017-04",
"2017-05",
"2017-06",
"2017-07",
"2017-08",
"2017-09",
"2017-10",
"2017-11",
"2017-12",
],
),
],
)
@freeze_time("2018-06-01 04:59:59")
def test_create_empty_monthly_notification_status_stats_dict(year, expected_years):
output = create_empty_monthly_notification_status_stats_dict(year)
assert sorted(output.keys()) == expected_years
for v in output.values():
assert v == {NotificationType.SMS: {}, NotificationType.EMAIL: {}}
@freeze_time("2018-06-01 04:59:59")
def test_add_monthly_notification_status_stats():
row_data = [
{
"month": datetime(2018, 4, 1),
"notification_type": NotificationType.SMS,
"notification_status": NotificationStatus.SENDING,
"count": 1,
},
{
"month": datetime(2018, 4, 1),
"notification_type": NotificationType.SMS,
"notification_status": NotificationStatus.DELIVERED,
"count": 2,
},
{
"month": datetime(2018, 4, 1),
"notification_type": NotificationType.EMAIL,
"notification_status": NotificationStatus.SENDING,
"count": 4,
},
{
"month": datetime(2018, 5, 1),
"notification_type": NotificationType.SMS,
"notification_status": NotificationStatus.SENDING,
"count": 8,
},
]
rows = []
for r in row_data:
m = Mock(spec=[])
for k, v in r.items():
setattr(m, k, v)
rows.append(m)
data = create_empty_monthly_notification_status_stats_dict(2018)
# this data won't be affected
data["2018-05"][NotificationType.EMAIL][NotificationStatus.SENDING] = 32
# this data will get combined with the 8 from row_data
data["2018-05"][NotificationType.SMS][NotificationStatus.SENDING] = 16
add_monthly_notification_status_stats(data, rows)
# first 3 months are empty
assert data == {
"2018-01": {NotificationType.SMS: {}, NotificationType.EMAIL: {}},
"2018-02": {NotificationType.SMS: {}, NotificationType.EMAIL: {}},
"2018-03": {NotificationType.SMS: {}, NotificationType.EMAIL: {}},
"2018-04": {
NotificationType.SMS: {
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: {}},
}