darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-01 15:46:07 -05:00
Code Issues Packages Projects Releases Wiki Activity

DRY up status aggregation tests (move DAO tests up)

Browse Source 
The previous DAO tests were also confusing because they were testing
two functions at the same time, so moving the tests up to the task
level seems very reasonable, and will make it easier to change how
this code works in the next commits.
This commit is contained in:
Ben Thorner
2022-01-11 15:55:24 +00:00
parent 081e0cab88
commit 9fc8b904c6
2 changed files with 85 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,6 +1,7 @@
import itertools
from datetime import date, datetime, timedelta
from datetime import date, datetime, time, timedelta
from decimal import Decimal
from uuid import UUID
import pytest
from freezegun import freeze_time
@@ -15,6 +16,9 @@ from app.config import QueueNames
from app.dao.fact_billing_dao import get_rate
from app.models import (
EMAIL_TYPE,
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
LETTER_TYPE,
SMS_TYPE,
FactBilling,
@@ -24,8 +28,10 @@ from app.models import (
from tests.app.db import (
create_letter_rate,
create_notification,
create_notification_history,
create_rate,
create_service,
create_service_data_retention,
create_template,
)
@@ -495,7 +501,6 @@ def test_create_nightly_billing_for_day_update_when_record_exists(
assert records[0].updated_at
@freeze_time('2019-01-05')
def test_create_nightly_notification_status_for_day(notify_db_session):
first_service = create_service(service_name='First Service')
first_template = create_template(service=first_service)
@@ -504,35 +509,92 @@ def test_create_nightly_notification_status_for_day(notify_db_session):
third_service = create_service(service_name='third Service')
third_template = create_template(service=third_service, template_type='letter')
create_notification(template=first_template, status='delivered')
create_notification(template=first_template, status='delivered', created_at=datetime(2019, 1, 1, 12, 0))
create_service_data_retention(second_service, 'email', days_of_retention=3)
create_notification(template=second_template, status='temporary-failure')
create_notification(template=second_template, status='temporary-failure', created_at=datetime(2019, 1, 1, 12, 0))
process_day = date.today() - timedelta(days=5)
with freeze_time(datetime.combine(process_day, time.min)):
create_notification(template=first_template, status='delivered')
create_notification(template=third_template, status='created')
create_notification(template=third_template, status='created', created_at=datetime(2019, 1, 1, 12, 0))
# 2nd service email has 3 day data retention - data has been moved to history and doesn't exist in notifications
create_notification_history(template=second_template, status='temporary-failure')
# team API key notifications are included
create_notification(template=third_template, status='sending', key_type=KEY_TYPE_TEAM)
# test notifications are ignored
create_notification(template=third_template, status='sending', key_type=KEY_TYPE_TEST)
# these created notifications from a different day get ignored
with freeze_time(datetime.combine(date.today() - timedelta(days=4), time.min)):
create_notification(template=first_template)
create_notification_history(template=second_template)
create_notification(template=third_template)
assert len(FactNotificationStatus.query.all()) == 0
create_nightly_notification_status_for_day('2019-01-01', 'sms')
create_nightly_notification_status_for_day('2019-01-01', 'email')
create_nightly_notification_status_for_day('2019-01-01', 'letter')
create_nightly_notification_status_for_day(str(process_day), 'sms')
create_nightly_notification_status_for_day(str(process_day), 'email')
create_nightly_notification_status_for_day(str(process_day), 'letter')
new_data = FactNotificationStatus.query.order_by(FactNotificationStatus.created_at).all()
new_fact_data = FactNotificationStatus.query.order_by(
FactNotificationStatus.notification_type
).all()
assert len(new_data) == 3
assert new_data[0].bst_date == date(2019, 1, 1)
assert new_data[1].bst_date == date(2019, 1, 1)
assert new_data[2].bst_date == date(2019, 1, 1)
assert len(new_fact_data) == 3
assert new_fact_data[0].bst_date == process_day
assert new_fact_data[0].template_id == second_template.id
assert new_fact_data[0].service_id == second_service.id
assert new_fact_data[0].job_id == UUID('00000000-0000-0000-0000-000000000000')
assert new_fact_data[0].notification_type == 'email'
assert new_fact_data[0].notification_status == 'temporary-failure'
assert new_fact_data[0].notification_count == 1
assert new_fact_data[0].key_type == KEY_TYPE_NORMAL
assert new_data[0].notification_type == 'sms'
assert new_data[1].notification_type == 'email'
assert new_data[2].notification_type == 'letter'
assert new_fact_data[1].bst_date == process_day
assert new_fact_data[1].template_id == third_template.id
assert new_fact_data[1].service_id == third_service.id
assert new_fact_data[1].job_id == UUID('00000000-0000-0000-0000-000000000000')
assert new_fact_data[1].notification_type == 'letter'
assert new_fact_data[1].notification_status == 'sending'
assert new_fact_data[1].notification_count == 1
assert new_fact_data[1].key_type == KEY_TYPE_TEAM
assert new_data[0].notification_status == 'delivered'
assert new_data[1].notification_status == 'temporary-failure'
assert new_data[2].notification_status == 'created'
assert new_fact_data[2].bst_date == process_day
assert new_fact_data[2].template_id == first_template.id
assert new_fact_data[2].service_id == first_service.id
assert new_fact_data[2].job_id == UUID('00000000-0000-0000-0000-000000000000')
assert new_fact_data[2].notification_type == 'sms'
assert new_fact_data[2].notification_status == 'delivered'
assert new_fact_data[2].notification_count == 1
assert new_fact_data[2].key_type == KEY_TYPE_NORMAL
def test_create_nightly_notification_status_for_day_overwrites_old_data(notify_db_session):
first_service = create_service(service_name='First Service')
first_template = create_template(service=first_service)
create_notification(template=first_template, status='delivered')
process_day = date.today()
create_nightly_notification_status_for_day(str(process_day), 'sms')
new_fact_data = FactNotificationStatus.query.order_by(
FactNotificationStatus.bst_date,
FactNotificationStatus.notification_type
).all()
assert len(new_fact_data) == 1
assert new_fact_data[0].notification_count == 1
create_notification(template=first_template, status='delivered')
create_nightly_notification_status_for_day(str(process_day), 'sms')
updated_fact_data = FactNotificationStatus.query.order_by(
FactNotificationStatus.bst_date,
FactNotificationStatus.notification_type
).all()
assert len(updated_fact_data) == 1
assert updated_fact_data[0].notification_count == 2
# the job runs at 12:30am London time. 04/01 is in BST.

101
tests/app/dao/test_fact_notification_status_dao.py
View File

@@ -1,4 +1,4 @@
from datetime import date, datetime, time, timedelta
from datetime import date, datetime, timedelta
from unittest import mock
from uuid import UUID
@@ -8,7 +8,6 @@ from freezegun import freeze_time
from app.dao.fact_notification_status_dao import (
fetch_monthly_notification_statuses_per_service,
fetch_monthly_template_usage_for_service,
fetch_notification_status_for_day,
fetch_notification_status_for_service_by_month,
fetch_notification_status_for_service_for_day,
fetch_notification_status_for_service_for_today_and_7_previous_days,
@@ -16,11 +15,9 @@ from app.dao.fact_notification_status_dao import (
fetch_notification_statuses_for_job,
fetch_stats_for_all_services_by_date_range,
get_total_notifications_for_date_range,
update_fact_notification_status,
)
from app.models import (
EMAIL_TYPE,
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
LETTER_TYPE,
@@ -34,112 +31,16 @@ from app.models import (
NOTIFICATION_TECHNICAL_FAILURE,
NOTIFICATION_TEMPORARY_FAILURE,
SMS_TYPE,
FactNotificationStatus,
)
from tests.app.db import (
create_ft_notification_status,
create_job,
create_notification,
create_notification_history,
create_service,
create_service_data_retention,
create_template,
)
def test_update_fact_notification_status(notify_db_session):
first_service = create_service(service_name='First Service')
first_template = create_template(service=first_service)
second_service = create_service(service_name='second Service')
second_template = create_template(service=second_service, template_type='email')
third_service = create_service(service_name='third Service')
third_template = create_template(service=third_service, template_type='letter')
create_service_data_retention(second_service, 'email', days_of_retention=3)
process_day = date.today() - timedelta(days=5)
with freeze_time(datetime.combine(process_day, time.min)):
create_notification(template=first_template, status='delivered')
# 2nd service email has 3 day data retention - data has been moved to history and doesn't exist in notifications
create_notification_history(template=second_template, status='temporary-failure')
# team API key notifications are included
create_notification(template=third_template, status='sending', key_type=KEY_TYPE_TEAM)
# test notifications are ignored
create_notification(template=third_template, status='sending', key_type=KEY_TYPE_TEST)
# these created notifications from a different day get ignored
with freeze_time(datetime.combine(date.today() - timedelta(days=4), time.min)):
create_notification(template=first_template)
create_notification_history(template=second_template)
create_notification(template=third_template)
for notification_type in ('letter', 'sms', 'email'):
data = fetch_notification_status_for_day(process_day=process_day, notification_type=notification_type)
update_fact_notification_status(data=data, process_day=process_day, notification_type=notification_type)
new_fact_data = FactNotificationStatus.query.order_by(FactNotificationStatus.bst_date,
FactNotificationStatus.notification_type
).all()
assert len(new_fact_data) == 3
assert new_fact_data[0].bst_date == process_day
assert new_fact_data[0].template_id == second_template.id
assert new_fact_data[0].service_id == second_service.id
assert new_fact_data[0].job_id == UUID('00000000-0000-0000-0000-000000000000')
assert new_fact_data[0].notification_type == 'email'
assert new_fact_data[0].notification_status == 'temporary-failure'
assert new_fact_data[0].notification_count == 1
assert new_fact_data[0].key_type == KEY_TYPE_NORMAL
assert new_fact_data[1].bst_date == process_day
assert new_fact_data[1].template_id == third_template.id
assert new_fact_data[1].service_id == third_service.id
assert new_fact_data[1].job_id == UUID('00000000-0000-0000-0000-000000000000')
assert new_fact_data[1].notification_type == 'letter'
assert new_fact_data[1].notification_status == 'sending'
assert new_fact_data[1].notification_count == 1
assert new_fact_data[1].key_type == KEY_TYPE_TEAM
assert new_fact_data[2].bst_date == process_day
assert new_fact_data[2].template_id == first_template.id
assert new_fact_data[2].service_id == first_service.id
assert new_fact_data[2].job_id == UUID('00000000-0000-0000-0000-000000000000')
assert new_fact_data[2].notification_type == 'sms'
assert new_fact_data[2].notification_status == 'delivered'
assert new_fact_data[2].notification_count == 1
assert new_fact_data[2].key_type == KEY_TYPE_NORMAL
def test__update_fact_notification_status_updates_row(notify_db_session):
first_service = create_service(service_name='First Service')
first_template = create_template(service=first_service)
create_notification(template=first_template, status='delivered')
process_day = date.today()
data = fetch_notification_status_for_day(process_day=process_day, notification_type='sms')
update_fact_notification_status(data=data, process_day=process_day, notification_type='sms')
new_fact_data = FactNotificationStatus.query.order_by(FactNotificationStatus.bst_date,
FactNotificationStatus.notification_type
).all()
assert len(new_fact_data) == 1
assert new_fact_data[0].notification_count == 1
create_notification(template=first_template, status='delivered')
data = fetch_notification_status_for_day(process_day=process_day, notification_type='sms')
update_fact_notification_status(data=data, process_day=process_day, notification_type='sms')
updated_fact_data = FactNotificationStatus.query.order_by(FactNotificationStatus.bst_date,
FactNotificationStatus.notification_type
).all()
assert len(updated_fact_data) == 1
assert updated_fact_data[0].notification_count == 2
def test_fetch_notification_status_for_service_by_month(notify_db_session):
service_1 = create_service(service_name='service_1')
service_2 = create_service(service_name='service_2')