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

Revert "Purge notifications for configured days of retention"

Browse Source
This commit is contained in:
Rebecca Law
2018-08-03 14:35:36 +01:00
committed by GitHub
parent 691f24f94e
commit 019c6a4e3a
11 changed files with 148 additions and 337 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
tests/app/dao/notification_dao/test_notification_dao.py
View File

@@ -54,9 +54,11 @@ from app.models import (
from tests.app.conftest import (
sample_notification,
sample_template as create_sample_template,
sample_email_template,
sample_service,
sample_job,
sample_notification_history as create_notification_history,
sample_letter_template
)
from tests.app.db import (
create_job,
@@ -605,6 +607,104 @@ def test_update_notification_sets_status(sample_notification):
assert notification_from_db.status == 'failed'
@pytest.mark.parametrize('month, delete_run_time',
[(4, '2016-04-10 23:40'), (1, '2016-01-11 00:40')])
@pytest.mark.parametrize(
'notification_type, expected_sms_count, expected_email_count, expected_letter_count',
[('sms', 7, 10, 10),
('email', 10, 7, 10),
('letter', 10, 10, 7)]
)
def test_should_delete_notifications_by_type_after_seven_days(
sample_service,
month,
delete_run_time,
notification_type,
expected_sms_count,
expected_email_count,
expected_letter_count
):
assert len(Notification.query.all()) == 0
email_template = create_template(service=sample_service, template_type='email')
sms_template = create_template(service=sample_service)
letter_template = create_template(service=sample_service, template_type='letter')
# create one notification a day between 1st and 10th from 11:00 to 19:00 of each type
for i in range(1, 11):
past_date = '2016-0{0}-{1:02d} {1:02d}:00:00.000000'.format(month, i)
with freeze_time(past_date):
create_notification(template=email_template, created_at=datetime.utcnow(), status="permanent-failure")
create_notification(template=sms_template, created_at=datetime.utcnow(), status="delivered")
create_notification(template=letter_template, created_at=datetime.utcnow(), status="temporary-failure")
all_notifications = Notification.query.all()
assert len(all_notifications) == 30
# Records from before 3rd should be deleted
with freeze_time(delete_run_time):
delete_notifications_created_more_than_a_week_ago_by_type(notification_type)
remaining_sms_notifications = Notification.query.filter_by(notification_type='sms').all()
remaining_letter_notifications = Notification.query.filter_by(notification_type='letter').all()
remaining_email_notifications = Notification.query.filter_by(notification_type='email').all()
assert len(remaining_sms_notifications) == expected_sms_count
assert len(remaining_email_notifications) == expected_email_count
assert len(remaining_letter_notifications) == expected_letter_count
if notification_type == 'sms':
notifications_to_check = remaining_sms_notifications
if notification_type == 'email':
notifications_to_check = remaining_email_notifications
if notification_type == 'letter':
notifications_to_check = remaining_letter_notifications
for notification in notifications_to_check:
assert notification.created_at.date() >= date(2016, month, 3)
@pytest.mark.parametrize('notification_type', ['sms', 'email', 'letter'])
@freeze_time("2016-01-10 12:00:00.000000")
def test_should_not_delete_notification_history(notify_db, notify_db_session, sample_service, notification_type):
with freeze_time('2016-01-01 12:00'):
email_template = sample_email_template(notify_db, notify_db_session, service=sample_service)
sms_template = create_sample_template(notify_db, notify_db_session, service=sample_service)
letter_template = sample_letter_template(sample_service)
sample_notification(
notify_db,
notify_db_session,
created_at=datetime.utcnow(),
status="failed",
service=sample_service,
template=email_template
)
sample_notification(
notify_db,
notify_db_session,
created_at=datetime.utcnow(),
status="failed",
service=sample_service,
template=sms_template
)
sample_notification(
notify_db,
notify_db_session,
created_at=datetime.utcnow(),
status="failed",
service=sample_service,
template=letter_template
)
assert Notification.query.count() == 3
assert NotificationHistory.query.count() == 3
delete_notifications_created_more_than_a_week_ago_by_type(notification_type)
assert Notification.query.count() == 2
assert NotificationHistory.query.count() == 3
@freeze_time("2016-01-10")
def test_should_limit_notifications_return_by_day_limit_plus_one(sample_template):
assert len(Notification.query.all()) == 0

187
tests/app/dao/notification_dao/test_notification_dao_delete_notifications.py
View File

@@ -1,187 +0,0 @@
from datetime import (
datetime,
date,
timedelta
)
import pytest
from freezegun import freeze_time
from app.dao.notifications_dao import delete_notifications_created_more_than_a_week_ago_by_type
from app.models import Notification, NotificationHistory
from tests.app.db import (
create_template,
create_notification,
create_service_data_retention,
create_service
)
@pytest.mark.parametrize('month, delete_run_time',
[(4, '2016-04-10 23:40'), (1, '2016-01-11 00:40')])
@pytest.mark.parametrize(
'notification_type, expected_sms_count, expected_email_count, expected_letter_count',
[('sms', 7, 10, 10),
('email', 10, 7, 10),
('letter', 10, 10, 7)]
)
def test_should_delete_notifications_by_type_after_seven_days(
sample_service,
month,
delete_run_time,
notification_type,
expected_sms_count,
expected_email_count,
expected_letter_count
):
assert len(Notification.query.all()) == 0
email_template, letter_template, sms_template = _create_templates(sample_service)
# create one notification a day between 1st and 10th from 11:00 to 19:00 of each type
for i in range(1, 11):
past_date = '2016-0{0}-{1:02d} {1:02d}:00:00.000000'.format(month, i)
with freeze_time(past_date):
create_notification(template=email_template, created_at=datetime.utcnow(), status="permanent-failure")
create_notification(template=sms_template, created_at=datetime.utcnow(), status="delivered")
create_notification(template=letter_template, created_at=datetime.utcnow(), status="temporary-failure")
all_notifications = Notification.query.all()
assert len(all_notifications) == 30
# Records from before 3rd should be deleted
with freeze_time(delete_run_time):
delete_notifications_created_more_than_a_week_ago_by_type(notification_type)
remaining_sms_notifications = Notification.query.filter_by(notification_type='sms').all()
remaining_letter_notifications = Notification.query.filter_by(notification_type='letter').all()
remaining_email_notifications = Notification.query.filter_by(notification_type='email').all()
assert len(remaining_sms_notifications) == expected_sms_count
assert len(remaining_email_notifications) == expected_email_count
assert len(remaining_letter_notifications) == expected_letter_count
if notification_type == 'sms':
notifications_to_check = remaining_sms_notifications
if notification_type == 'email':
notifications_to_check = remaining_email_notifications
if notification_type == 'letter':
notifications_to_check = remaining_letter_notifications
for notification in notifications_to_check:
assert notification.created_at.date() >= date(2016, month, 3)
@pytest.mark.parametrize('notification_type', ['sms', 'email', 'letter'])
@freeze_time("2016-01-10 12:00:00.000000")
def test_should_not_delete_notification_history(sample_service, notification_type):
with freeze_time('2016-01-01 12:00'):
email_template, letter_template, sms_template = _create_templates(sample_service)
create_notification(template=email_template, status='permanent-failure')
create_notification(template=sms_template, status='permanent-failure')
create_notification(template=letter_template, status='permanent-failure')
assert Notification.query.count() == 3
assert NotificationHistory.query.count() == 3
delete_notifications_created_more_than_a_week_ago_by_type(notification_type)
assert Notification.query.count() == 2
assert NotificationHistory.query.count() == 3
@pytest.mark.parametrize('notification_type', ['sms', 'email', 'letter'])
def test_delete_notifications_for_days_of_retention(sample_service, notification_type):
service_with_default_data_retention = create_service(service_name='default data retention')
email_template, letter_template, sms_template = _create_templates(sample_service)
default_email_template, default_letter_template, default_sms_template = _create_templates(
service_with_default_data_retention)
create_notification(template=email_template, status='delivered')
create_notification(template=sms_template, status='permanent-failure')
create_notification(template=letter_template, status='temporary-failure')
create_notification(template=email_template, status='delivered',
created_at=datetime.utcnow() - timedelta(days=4))
create_notification(template=sms_template, status='permanent-failure',
created_at=datetime.utcnow() - timedelta(days=4))
create_notification(template=letter_template, status='temporary-failure',
created_at=datetime.utcnow() - timedelta(days=4))
create_notification(template=default_email_template, status='delivered',
created_at=datetime.utcnow() - timedelta(days=8))
create_notification(template=default_sms_template, status='permanent-failure',
created_at=datetime.utcnow() - timedelta(days=8))
create_notification(template=default_letter_template, status='temporary-failure',
created_at=datetime.utcnow() - timedelta(days=8))
create_service_data_retention(service_id=sample_service.id, notification_type=notification_type)
assert len(Notification.query.all()) == 9
delete_notifications_created_more_than_a_week_ago_by_type(notification_type)
assert len(Notification.query.all()) == 7
assert len(Notification.query.filter_by(notification_type=notification_type).all()) == 1
@pytest.mark.parametrize('notification_type', ['sms', 'email', 'letter'])
def test_delete_notifications_keep_data_for_days_of_retention_is_longer(sample_service, notification_type):
create_service_data_retention(service_id=sample_service.id, notification_type=notification_type,
days_of_retention=15)
email_template, letter_template, sms_template = _create_templates(sample_service)
service_with_default_data_retention = create_service(service_name='default data retention')
default_email_template, default_letter_template, default_sms_template = _create_templates(
service_with_default_data_retention)
create_notification(template=email_template, status='delivered')
create_notification(template=sms_template, status='permanent-failure')
create_notification(template=letter_template, status='temporary-failure')
create_notification(template=email_template, status='delivered',
created_at=datetime.utcnow() - timedelta(days=14))
create_notification(template=sms_template, status='permanent-failure',
created_at=datetime.utcnow() - timedelta(days=14))
create_notification(template=letter_template, status='temporary-failure',
created_at=datetime.utcnow() - timedelta(days=14))
create_notification(template=default_email_template, status='delivered',
created_at=datetime.utcnow() - timedelta(days=8))
create_notification(template=default_sms_template, status='permanent-failure',
created_at=datetime.utcnow() - timedelta(days=8))
create_notification(template=default_letter_template, status='temporary-failure',
created_at=datetime.utcnow() - timedelta(days=8))
assert len(Notification.query.all()) == 9
delete_notifications_created_more_than_a_week_ago_by_type(notification_type)
assert len(Notification.query.filter_by().all()) == 8
assert len(Notification.query.filter_by(notification_type=notification_type).all()) == 2
def test_delete_notifications_delete_notification_type_for_default_time_if_no_days_of_retention_for_type(
sample_service
):
create_service_data_retention(service_id=sample_service.id, notification_type='sms',
days_of_retention=15)
email_template, letter_template, sms_template = _create_templates(sample_service)
create_notification(template=email_template, status='delivered')
create_notification(template=sms_template, status='permanent-failure')
create_notification(template=letter_template, status='temporary-failure')
create_notification(template=email_template, status='delivered',
created_at=datetime.utcnow() - timedelta(days=14))
create_notification(template=sms_template, status='permanent-failure',
created_at=datetime.utcnow() - timedelta(days=14))
create_notification(template=letter_template, status='temporary-failure',
created_at=datetime.utcnow() - timedelta(days=14))
assert len(Notification.query.all()) == 6
delete_notifications_created_more_than_a_week_ago_by_type('email')
assert len(Notification.query.filter_by().all()) == 5
assert len(Notification.query.filter_by(notification_type='email').all()) == 1
def _create_templates(sample_service):
email_template = create_template(service=sample_service, template_type='email')
sms_template = create_template(service=sample_service)
letter_template = create_template(service=sample_service, template_type='letter')
return email_template, letter_template, sms_template

19
tests/app/dao/test_service_data_retention_dao.py
View File

@@ -8,11 +8,10 @@ from app.dao.service_data_retention_dao import (
fetch_service_data_retention,
insert_service_data_retention,
update_service_data_retention,
fetch_service_data_retention_by_id,
fetch_service_data_retention_by_notification_type
fetch_service_data_retention_by_id
)
from app.models import ServiceDataRetention
from tests.app.db import create_service, create_service_data_retention
from tests.app.db import create_service
def test_fetch_service_data_retention(sample_service):
@@ -132,17 +131,3 @@ def test_update_service_data_retention_does_not_update_row_if_data_retention_is_
service_id=uuid.uuid4(),
days_of_retention=5)
assert updated_count == 0
@pytest.mark.parametrize('notification_type, alternate',
[('sms', 'email'),
('email', 'sms'), ('letter', 'email')])
def test_fetch_service_data_retention_by_notification_type(sample_service, notification_type, alternate):
data_retention = create_service_data_retention(service_id=sample_service.id, notification_type=notification_type)
create_service_data_retention(service_id=sample_service.id, notification_type=alternate)
result = fetch_service_data_retention_by_notification_type(sample_service.id, notification_type)
assert result == data_retention
def test_fetch_service_data_retention_by_notification_type_returns_none_when_no_rows(sample_service):
assert not fetch_service_data_retention_by_notification_type(sample_service.id, 'email')

33
tests/app/dao/test_services_dao.py
View File

@@ -534,7 +534,7 @@ def test_fetch_stats_filters_on_service(sample_notification):
message_limit=1000)
dao_create_service(service_two, sample_notification.service.created_by)
stats = dao_fetch_stats_for_service(service_two.id, 7)
stats = dao_fetch_stats_for_service(service_two.id)
assert len(stats) == 0
@@ -546,7 +546,7 @@ def test_fetch_stats_ignores_historical_notification_data(sample_notification):
assert Notification.query.count() == 0
assert NotificationHistory.query.count() == 1
stats = dao_fetch_stats_for_service(service_id, 7)
stats = dao_fetch_stats_for_service(service_id)
assert len(stats) == 0
@@ -557,7 +557,7 @@ def test_fetch_stats_counts_correctly(notify_db, notify_db_session, sample_templ
create_notification(notify_db, notify_db_session, template=sample_email_template, status='technical-failure')
create_notification(notify_db, notify_db_session, template=sample_template, status='created')
stats = dao_fetch_stats_for_service(sample_template.service_id, 7)
stats = dao_fetch_stats_for_service(sample_template.service_id)
stats = sorted(stats, key=lambda x: (x.notification_type, x.status))
assert len(stats) == 3
@@ -591,7 +591,7 @@ def test_fetch_stats_counts_should_ignore_team_key(
create_notification(
notify_db, notify_db_session)
stats = dao_fetch_stats_for_service(sample_template.service_id, 7)
stats = dao_fetch_stats_for_service(sample_template.service_id)
assert len(stats) == 1
assert stats[0].notification_type == 'sms'
assert stats[0].status == 'created'
@@ -634,30 +634,7 @@ def test_fetch_stats_should_not_gather_notifications_older_than_7_days(sample_te
create_notification_db(sample_template,)
with freeze_time('Monday 16th July 2018 12:00'):
stats = dao_fetch_stats_for_service(sample_template.service_id, 7)
assert len(stats) == rows_returned
@pytest.mark.parametrize('created_at, rows_returned', [
('Saturday 7th July 2018 12:00', 0),
('Saturday 7th July 2018 22:59', 0),
('Sunday 8th July 2018 12:00', 1),
('Sunday 8th July 2018 22:59', 1),
('Sunday 8th July 2018 23:00', 1),
('Monday 9th July 2018 09:00', 1),
('Monday 9th July 2018 15:00', 1),
('Monday 16th July 2018 12:00', 1),
])
def test_fetch_stats_should_not_gather_notifications_older_than_the_days_given(
sample_template, created_at, rows_returned
):
# It's monday today. Things made last monday should still show
with freeze_time(created_at):
create_notification_db(sample_template,)
with freeze_time('Monday 16th July 2018 12:00'):
stats = dao_fetch_stats_for_service(sample_template.service_id, 8)
stats = dao_fetch_stats_for_service(sample_template.service_id)
assert len(stats) == rows_returned