darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2025-12-23 17:01:35 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
7974e59eb7e6a46fafe08ed92a0a1598de156655
notifications-api/tests/app/dao/notification_dao/test_notification_dao.py
Richard Chapman 7974e59eb7 Changes after review. Updates to style and import order to be more consistent with other code. 
Added clarification to an error message to give better debugging information.
Removed using dao_get_reply_to_by_service_id in tests to be more consistent with other code and use the test db functions or remove the need for a call altogether making the code less complex.
2017-10-05 16:50:18 +01:00

1998 lines
79 KiB
Python
Raw Blame History

from datetime import datetime, timedelta, date
import uuid
from functools import partial
import pytest
from freezegun import freeze_time
from sqlalchemy.exc import SQLAlchemyError, IntegrityError
from app.dao.service_email_reply_to_dao import dao_get_reply_to_by_service_id
from app.models import (
Notification,
NotificationHistory,
Job,
NotificationEmailReplyTo,
NotificationStatistics,
ScheduledNotification,
NOTIFICATION_STATUS_TYPES,
NOTIFICATION_STATUS_TYPES_FAILED,
NOTIFICATION_SENT,
NOTIFICATION_DELIVERED,
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST,
)
from app.dao.notifications_dao import (
dao_create_notification,
dao_create_notification_email_reply_to_mapping,
dao_created_scheduled_notification,
dao_delete_notifications_and_history_by_id,
dao_get_notifications_by_to_field,
dao_get_last_template_usage,
dao_get_notification_email_reply_for_notification,
dao_get_notification_statistics_for_service_and_day,
dao_get_potential_notification_statistics_for_day,
dao_get_scheduled_notifications,
dao_get_template_usage,
dao_timeout_notifications,
dao_update_notification,
dao_update_notifications_for_job_to_sent_to_dvla,
delete_notifications_created_more_than_a_week_ago_by_type,
get_notification_by_id,
get_notification_for_job,
get_notification_with_personalisation,
get_notifications_for_job,
get_notifications_for_service,
get_total_sent_notifications_in_date_range,
is_delivery_slow_for_provider,
set_scheduled_notification_to_processed,
update_notification_status_by_id,
update_notification_status_by_reference
)
from app.dao.services_dao import dao_update_service
from tests.app.db import create_notification, create_api_key, create_reply_to_email
from tests.app.conftest import (
sample_notification,
sample_template,
sample_email_template,
sample_service,
sample_job,
sample_notification_history as create_notification_history,
sample_letter_template)
def test_should_have_decorated_notifications_dao_functions():
assert dao_get_last_template_usage.__wrapped__.__name__ == 'dao_get_last_template_usage' # noqa
assert dao_get_template_usage.__wrapped__.__name__ == 'dao_get_template_usage' # noqa
assert dao_get_potential_notification_statistics_for_day.__wrapped__.__name__ == 'dao_get_potential_notification_statistics_for_day' # noqa
assert dao_create_notification.__wrapped__.__name__ == 'dao_create_notification' # noqa
assert update_notification_status_by_id.__wrapped__.__name__ == 'update_notification_status_by_id' # noqa
assert dao_update_notification.__wrapped__.__name__ == 'dao_update_notification' # noqa
assert update_notification_status_by_reference.__wrapped__.__name__ == 'update_notification_status_by_reference' # noqa
assert get_notification_for_job.__wrapped__.__name__ == 'get_notification_for_job' # noqa
assert get_notifications_for_job.__wrapped__.__name__ == 'get_notifications_for_job' # noqa
assert get_notification_with_personalisation.__wrapped__.__name__ == 'get_notification_with_personalisation' # noqa
assert get_notifications_for_service.__wrapped__.__name__ == 'get_notifications_for_service' # noqa
assert get_notification_by_id.__wrapped__.__name__ == 'get_notification_by_id' # noqa
assert delete_notifications_created_more_than_a_week_ago_by_type.__wrapped__.__name__ == 'delete_notifications_created_more_than_a_week_ago_by_type' # noqa
assert dao_delete_notifications_and_history_by_id.__wrapped__.__name__ == 'dao_delete_notifications_and_history_by_id' # noqa
def test_should_be_able_to_get_template_usage_history(notify_db, notify_db_session, sample_service):
with freeze_time('2000-01-01 12:00:00'):
sms = sample_template(notify_db, notify_db_session)
notification = sample_notification(notify_db, notify_db_session, service=sample_service, template=sms)
results = dao_get_last_template_usage(sms.id)
assert results.template.name == 'Template Name'
assert results.template.template_type == 'sms'
assert results.created_at == datetime(year=2000, month=1, day=1, hour=12, minute=0, second=0)
assert results.template_id == sms.id
assert results.id == notification.id
def test_should_be_able_to_get_all_template_usage_history_order_by_notification_created_at(
notify_db,
notify_db_session,
sample_service):
sms = sample_template(notify_db, notify_db_session)
sample_notification(notify_db, notify_db_session, service=sample_service, template=sms)
sample_notification(notify_db, notify_db_session, service=sample_service, template=sms)
sample_notification(notify_db, notify_db_session, service=sample_service, template=sms)
most_recent = sample_notification(notify_db, notify_db_session, service=sample_service, template=sms)
results = dao_get_last_template_usage(sms.id)
assert results.id == most_recent.id
def test_template_usage_should_ignore_test_keys(
notify_db,
notify_db_session,
sample_team_api_key,
sample_test_api_key
):
sms = sample_template(notify_db, notify_db_session)
one_minute_ago = datetime.utcnow() - timedelta(minutes=1)
two_minutes_ago = datetime.utcnow() - timedelta(minutes=2)
team_key = sample_notification(
notify_db,
notify_db_session,
created_at=two_minutes_ago,
template=sms,
api_key=sample_team_api_key,
key_type=KEY_TYPE_TEAM)
sample_notification(
notify_db,
notify_db_session,
created_at=one_minute_ago,
template=sms,
api_key=sample_test_api_key,
key_type=KEY_TYPE_TEST)
results = dao_get_last_template_usage(sms.id)
assert results.id == team_key.id
def test_should_be_able_to_get_no_template_usage_history_if_no_notifications_using_template(
notify_db,
notify_db_session):
sms = sample_template(notify_db, notify_db_session)
results = dao_get_last_template_usage(sms.id)
assert not results
def test_should_by_able_to_get_template_count(notify_db, notify_db_session, sample_service):
sms = sample_template(notify_db, notify_db_session)
email = sample_email_template(notify_db, notify_db_session)
sample_notification(notify_db, notify_db_session, service=sample_service, template=sms)
sample_notification(notify_db, notify_db_session, service=sample_service, template=sms)
sample_notification(notify_db, notify_db_session, service=sample_service, template=sms)
sample_notification(notify_db, notify_db_session, service=sample_service, template=email)
sample_notification(notify_db, notify_db_session, service=sample_service, template=email)
results = dao_get_template_usage(sample_service.id)
assert results[0].name == 'Email Template Name'
assert results[0].template_type == 'email'
assert results[0].count == 2
assert results[1].name == 'Template Name'
assert results[1].template_type == 'sms'
assert results[1].count == 3
def test_template_history_should_ignore_test_keys(
notify_db,
notify_db_session,
sample_team_api_key,
sample_test_api_key,
sample_api_key
):
sms = sample_template(notify_db, notify_db_session)
sample_notification(
notify_db, notify_db_session, template=sms, api_key=sample_api_key, key_type=KEY_TYPE_NORMAL)
sample_notification(
notify_db, notify_db_session, template=sms, api_key=sample_team_api_key, key_type=KEY_TYPE_TEAM)
sample_notification(
notify_db, notify_db_session, template=sms, api_key=sample_test_api_key, key_type=KEY_TYPE_TEST)
sample_notification(
notify_db, notify_db_session, template=sms)
results = dao_get_template_usage(sms.service_id)
assert results[0].name == 'Template Name'
assert results[0].template_type == 'sms'
assert results[0].count == 3
def test_should_by_able_to_get_template_count_limited_for_service(
notify_db,
notify_db_session):
service_1 = sample_service(notify_db, notify_db_session, service_name="test1", email_from="test1")
service_2 = sample_service(notify_db, notify_db_session, service_name="test2", email_from="test2")
service_3 = sample_service(notify_db, notify_db_session, service_name="test3", email_from="test3")
sms = sample_template(notify_db, notify_db_session)
sample_notification(notify_db, notify_db_session, service=service_1, template=sms)
sample_notification(notify_db, notify_db_session, service=service_1, template=sms)
sample_notification(notify_db, notify_db_session, service=service_2, template=sms)
assert dao_get_template_usage(service_1.id)[0].count == 2
assert dao_get_template_usage(service_2.id)[0].count == 1
assert len(dao_get_template_usage(service_3.id)) == 0
def test_should_by_able_to_get_zero_count_from_notifications_history_if_no_rows(sample_service):
results = dao_get_template_usage(sample_service.id)
assert len(results) == 0
def test_should_by_able_to_get_zero_count_from_notifications_history_if_no_service():
results = dao_get_template_usage(str(uuid.uuid4()))
assert len(results) == 0
def test_should_by_able_to_get_template_count_across_days(
notify_db,
notify_db_session,
sample_service):
sms = sample_template(notify_db, notify_db_session)
email = sample_email_template(notify_db, notify_db_session)
today = datetime.now()
yesterday = datetime.now() - timedelta(days=1)
one_month_ago = datetime.now() - timedelta(days=30)
sample_notification(notify_db, notify_db_session, created_at=today, service=sample_service, template=email)
sample_notification(notify_db, notify_db_session, created_at=today, service=sample_service, template=email)
sample_notification(notify_db, notify_db_session, created_at=today, service=sample_service, template=sms)
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=email)
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=email)
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=email)
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=sms)
sample_notification(notify_db, notify_db_session, created_at=one_month_ago, service=sample_service, template=sms)
sample_notification(notify_db, notify_db_session, created_at=one_month_ago, service=sample_service, template=sms)
sample_notification(notify_db, notify_db_session, created_at=one_month_ago, service=sample_service, template=sms)
results = dao_get_template_usage(sample_service.id)
assert len(results) == 2
assert [(row.name, row.template_type, row.count) for row in results] == [
('Email Template Name', 'email', 5),
('Template Name', 'sms', 5)
]
def test_should_by_able_to_get_template_count_for_under_seven_days(
notify_db,
notify_db_session,
sample_service,
sample_template):
yesterday = datetime.now() - timedelta(days=1)
six_days_ago = datetime.now() - timedelta(days=6)
seven_days_ago = datetime.now() - timedelta(days=7)
eight_days_ago = datetime.now() - timedelta(days=8)
sample_notification(
notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=sample_template
)
sample_notification(
notify_db, notify_db_session, created_at=six_days_ago, service=sample_service, template=sample_template
)
sample_notification(
notify_db, notify_db_session, created_at=seven_days_ago, service=sample_service, template=sample_template
)
sample_notification(
notify_db, notify_db_session, created_at=eight_days_ago, service=sample_service, template=sample_template
)
results = dao_get_template_usage(sample_service.id, limit_days=6)
assert len(results) == 1
assert [(row.name, row.template_type, row.count) for row in results] == [
('Template Name', 'sms', 2)
]
def test_should_by_able_to_get_template_count_for_whole_of_notifications_table_if_seven_days_exactly(
notify_db,
notify_db_session,
sample_service,
sample_template):
yesterday = datetime.now() - timedelta(days=1)
six_days_ago = datetime.now() - timedelta(days=6)
seven_days_ago = datetime.now() - timedelta(days=7)
eight_days_ago = datetime.now() - timedelta(days=8)
sample_notification(
notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=sample_template
)
sample_notification(
notify_db, notify_db_session, created_at=six_days_ago, service=sample_service, template=sample_template
)
sample_notification(
notify_db, notify_db_session, created_at=seven_days_ago, service=sample_service, template=sample_template
)
sample_notification(
notify_db, notify_db_session, created_at=eight_days_ago, service=sample_service, template=sample_template
)
results = dao_get_template_usage(sample_service.id, limit_days=7)
assert len(results) == 1
# note as we haven't run the delete task they'll ALL be in the notifications table.
assert [(row.name, row.template_type, row.count) for row in results] == [
('Template Name', 'sms', 4)
]
def test_should_by_able_to_get_all_template_count_for_more_than_seven_days(
notify_db,
notify_db_session,
sample_service,
sample_template):
yesterday = datetime.now() - timedelta(days=1)
six_days_ago = datetime.now() - timedelta(days=6)
seven_days_ago = datetime.now() - timedelta(days=7)
eight_days_ago = datetime.now() - timedelta(days=8)
sample_notification(
notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=sample_template
)
sample_notification(
notify_db, notify_db_session, created_at=six_days_ago, service=sample_service, template=sample_template
)
sample_notification(
notify_db, notify_db_session, created_at=seven_days_ago, service=sample_service, template=sample_template
)
sample_notification(
notify_db, notify_db_session, created_at=eight_days_ago, service=sample_service, template=sample_template
)
Notification.query.delete()
# gets all from history table
results = dao_get_template_usage(sample_service.id, limit_days=10)
assert len(results) == 1
assert [(row.name, row.template_type, row.count) for row in results] == [
('Template Name', 'sms', 4)
]
def test_should_by_able_to_get_template_count_from_notifications_history_with_day_limit(
notify_db,
notify_db_session,
sample_service):
sms = sample_template(notify_db, notify_db_session)
email = sample_email_template(notify_db, notify_db_session)
today = datetime.now()
yesterday = datetime.now() - timedelta(days=1)
one_month_ago = datetime.now() - timedelta(days=30)
sample_notification(notify_db, notify_db_session, created_at=today, service=sample_service, template=email)
sample_notification(notify_db, notify_db_session, created_at=today, service=sample_service, template=email)
sample_notification(notify_db, notify_db_session, created_at=today, service=sample_service, template=sms)
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=email)
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=email)
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=email)
sample_notification(notify_db, notify_db_session, created_at=yesterday, service=sample_service, template=sms)
sample_notification(notify_db, notify_db_session, created_at=one_month_ago, service=sample_service, template=sms)
sample_notification(notify_db, notify_db_session, created_at=one_month_ago, service=sample_service, template=sms)
sample_notification(notify_db, notify_db_session, created_at=one_month_ago, service=sample_service, template=sms)
results_day_one = dao_get_template_usage(sample_service.id, limit_days=0)
assert len(results_day_one) == 2
results_day_two = dao_get_template_usage(sample_service.id, limit_days=1)
assert len(results_day_two) == 2
results_day_30 = dao_get_template_usage(sample_service.id, limit_days=31)
assert len(results_day_30) == 2
assert [(row.name, row.template_type, row.count) for row in results_day_one] == [
('Email Template Name', 'email', 2),
('Template Name', 'sms', 1)
]
assert [(row.name, row.template_type, row.count) for row in results_day_two] == [
('Email Template Name', 'email', 5),
('Template Name', 'sms', 2),
]
assert [(row.name, row.template_type, row.count) for row in results_day_30] == [
('Email Template Name', 'email', 5),
('Template Name', 'sms', 5),
]
def test_should_by_able_to_update_status_by_reference(sample_email_template, ses_provider):
data = _notification_json(sample_email_template, status='sending')
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.get(notification.id).status == "sending"
notification.reference = 'reference'
dao_update_notification(notification)
updated = update_notification_status_by_reference('reference', 'delivered')
assert updated.status == 'delivered'
assert Notification.query.get(notification.id).status == 'delivered'
def test_should_by_able_to_update_status_by_id(sample_template, sample_job, mmg_provider):
with freeze_time('2000-01-01 12:00:00'):
data = _notification_json(sample_template, job_id=sample_job.id, status='sending')
notification = Notification(**data)
dao_create_notification(notification)
assert notification.status == 'sending'
assert Notification.query.get(notification.id).status == 'sending'
with freeze_time('2000-01-02 12:00:00'):
updated = update_notification_status_by_id(notification.id, 'delivered')
assert updated.status == 'delivered'
assert updated.updated_at == datetime(2000, 1, 2, 12, 0, 0)
assert Notification.query.get(notification.id).status == 'delivered'
assert notification.updated_at == datetime(2000, 1, 2, 12, 0, 0)
assert notification.status == 'delivered'
def test_should_not_update_status_by_id_if_not_sending_and_does_not_update_job(notify_db, notify_db_session):
job = sample_job(notify_db, notify_db_session)
notification = sample_notification(notify_db, notify_db_session, status='delivered', job=job)
assert Notification.query.get(notification.id).status == 'delivered'
assert not update_notification_status_by_id(notification.id, 'failed')
assert Notification.query.get(notification.id).status == 'delivered'
assert job == Job.query.get(notification.job_id)
def test_should_not_update_status_by_reference_if_not_sending_and_does_not_update_job(notify_db, notify_db_session):
job = sample_job(notify_db, notify_db_session)
notification = sample_notification(notify_db, notify_db_session, status='delivered', reference='reference', job=job)
assert Notification.query.get(notification.id).status == 'delivered'
assert not update_notification_status_by_reference('reference', 'failed')
assert Notification.query.get(notification.id).status == 'delivered'
assert job == Job.query.get(notification.job_id)
def test_should_update_status_by_id_if_created(notify_db, notify_db_session):
notification = sample_notification(notify_db, notify_db_session, status='created')
assert Notification.query.get(notification.id).status == 'created'
updated = update_notification_status_by_id(notification.id, 'failed')
assert Notification.query.get(notification.id).status == 'failed'
assert updated.status == 'failed'
def test_should_not_update_status_by_reference_if_from_country_with_no_delivery_receipts(sample_template):
notification = create_notification(
sample_template,
status=NOTIFICATION_SENT,
reference='foo'
)
res = update_notification_status_by_reference('foo', 'failed')
assert res is None
assert notification.status == NOTIFICATION_SENT
def test_should_not_update_status_by_id_if_sent_to_country_with_no_delivery_receipts(sample_template):
notification = create_notification(
sample_template,
status=NOTIFICATION_SENT,
international=True,
phone_prefix='1' # americans only have carrier delivery receipts
)
res = update_notification_status_by_id(notification.id, 'delivered')
assert res is None
assert notification.status == NOTIFICATION_SENT
def test_should_not_update_status_by_id_if_sent_to_country_with_no_delivery_receipts(sample_template):
notification = create_notification(
sample_template,
status=NOTIFICATION_SENT,
international=True,
phone_prefix='7' # russians have full delivery receipts
)
res = update_notification_status_by_id(notification.id, 'delivered')
assert res == notification
assert notification.status == NOTIFICATION_DELIVERED
def test_should_not_update_status_by_reference_if_not_sending(notify_db, notify_db_session):
notification = sample_notification(notify_db, notify_db_session, status='created', reference='reference')
assert Notification.query.get(notification.id).status == 'created'
updated = update_notification_status_by_reference('reference', 'failed')
assert Notification.query.get(notification.id).status == 'created'
assert not updated
def test_should_by_able_to_update_status_by_id_from_pending_to_delivered(sample_template, sample_job):
data = _notification_json(sample_template, job_id=sample_job.id, status='sending')
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.get(notification.id).status == 'sending'
assert update_notification_status_by_id(notification_id=notification.id, status='pending')
assert Notification.query.get(notification.id).status == 'pending'
assert update_notification_status_by_id(notification.id, 'delivered')
assert Notification.query.get(notification.id).status == 'delivered'
def test_should_by_able_to_update_status_by_id_from_pending_to_temporary_failure(sample_template, sample_job):
data = _notification_json(sample_template, job_id=sample_job.id, status='sending')
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.get(notification.id).status == 'sending'
assert update_notification_status_by_id(notification_id=notification.id, status='pending')
assert Notification.query.get(notification.id).status == 'pending'
assert update_notification_status_by_id(
notification.id,
status='permanent-failure')
assert Notification.query.get(notification.id).status == 'temporary-failure'
def test_should_by_able_to_update_status_by_id_from_sending_to_permanent_failure(sample_template, sample_job):
data = _notification_json(sample_template, job_id=sample_job.id, status='sending')
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.get(notification.id).status == 'sending'
assert update_notification_status_by_id(
notification.id,
status='permanent-failure'
)
assert Notification.query.get(notification.id).status == 'permanent-failure'
def test_should_not_update_status_one_notification_status_is_delivered(notify_db, notify_db_session,
sample_email_template,
ses_provider):
notification = sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
template=sample_email_template,
status='sending')
assert Notification.query.get(notification.id).status == "sending"
notification.reference = 'reference'
dao_update_notification(notification)
update_notification_status_by_reference('reference', 'delivered')
assert Notification.query.get(notification.id).status == 'delivered'
update_notification_status_by_reference('reference', 'failed')
assert Notification.query.get(notification.id).status == 'delivered'
def test_should_return_zero_count_if_no_notification_with_id():
assert not update_notification_status_by_id(str(uuid.uuid4()), 'delivered')
def test_should_return_zero_count_if_no_notification_with_reference():
assert not update_notification_status_by_reference('something', 'delivered')
def test_should_return_none_if_no_statistics_for_a_service_for_a_day(sample_template, mmg_provider):
data = _notification_json(sample_template)
notification = Notification(**data)
dao_create_notification(notification)
assert not dao_get_notification_statistics_for_service_and_day(
sample_template.service.id, (datetime.utcnow() - timedelta(days=1)).date())
def test_should_be_able_to_get_all_statistics_for_a_service(sample_template, mmg_provider):
data = _notification_json(sample_template)
notification_1 = Notification(**data)
notification_2 = Notification(**data)
notification_3 = Notification(**data)
dao_create_notification(notification_1)
dao_create_notification(notification_2)
dao_create_notification(notification_3)
def test_create_notification_creates_notification_with_personalisation(notify_db, notify_db_session,
sample_template_with_placeholders,
sample_job, mmg_provider):
assert Notification.query.count() == 0
assert NotificationStatistics.query.count() == 0
data = sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
template=sample_template_with_placeholders,
job=sample_job,
personalisation={'name': 'Jo'},
status='created')
assert Notification.query.count() == 1
notification_from_db = Notification.query.all()[0]
assert notification_from_db.id
assert data.to == notification_from_db.to
assert data.job_id == notification_from_db.job_id
assert data.service == notification_from_db.service
assert data.template == notification_from_db.template
assert data.template_version == notification_from_db.template_version
assert data.created_at == notification_from_db.created_at
assert notification_from_db.status == 'created'
assert {'name': 'Jo'} == notification_from_db.personalisation
def test_save_notification_creates_sms(sample_template, sample_job, mmg_provider):
assert Notification.query.count() == 0
assert NotificationStatistics.query.count() == 0
data = _notification_json(sample_template, job_id=sample_job.id)
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.count() == 1
notification_from_db = Notification.query.all()[0]
assert notification_from_db.id
assert data['to'] == notification_from_db.to
assert data['job_id'] == notification_from_db.job_id
assert data['service'] == notification_from_db.service
assert data['template'] == notification_from_db.template
assert data['template_version'] == notification_from_db.template_version
assert data['created_at'] == notification_from_db.created_at
assert notification_from_db.status == 'created'
def test_save_notification_and_create_email(sample_email_template, sample_job):
assert Notification.query.count() == 0
assert NotificationStatistics.query.count() == 0
data = _notification_json(sample_email_template, job_id=sample_job.id)
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.count() == 1
notification_from_db = Notification.query.all()[0]
assert notification_from_db.id
assert data['to'] == notification_from_db.to
assert data['job_id'] == notification_from_db.job_id
assert data['service'] == notification_from_db.service
assert data['template'] == notification_from_db.template
assert data['template_version'] == notification_from_db.template_version
assert data['created_at'] == notification_from_db.created_at
assert notification_from_db.status == 'created'
def test_save_notification(sample_email_template, sample_job, ses_provider):
assert Notification.query.count() == 0
data = _notification_json(sample_email_template, job_id=sample_job.id)
notification_1 = Notification(**data)
notification_2 = Notification(**data)
dao_create_notification(notification_1)
assert Notification.query.count() == 1
dao_create_notification(notification_2)
assert Notification.query.count() == 2
def test_save_notification_creates_history(sample_email_template, sample_job):
assert Notification.query.count() == 0
data = _notification_json(sample_email_template, job_id=sample_job.id)
notification_1 = Notification(**data)
dao_create_notification(notification_1)
assert Notification.query.count() == 1
assert NotificationHistory.query.count() == 1
def test_save_notification_with_test_api_key_does_not_create_history(sample_email_template, sample_api_key):
assert Notification.query.count() == 0
data = _notification_json(sample_email_template)
data['key_type'] = KEY_TYPE_TEST
data['api_key_id'] = sample_api_key.id
notification_1 = Notification(**data)
dao_create_notification(notification_1)
assert Notification.query.count() == 1
assert NotificationHistory.query.count() == 0
def test_save_notification_with_research_mode_service_does_not_create_history(
notify_db,
notify_db_session):
service = sample_service(notify_db, notify_db_session)
service.research_mode = True
dao_update_service(service)
template = sample_template(notify_db, notify_db_session, service=service)
assert Notification.query.count() == 0
data = _notification_json(template)
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.count() == 1
assert NotificationHistory.query.count() == 0
def test_update_notification_with_test_api_key_does_not_update_or_create_history(sample_email_template, sample_api_key):
assert Notification.query.count() == 0
data = _notification_json(sample_email_template)
data['key_type'] = KEY_TYPE_TEST
data['api_key_id'] = sample_api_key.id
notification = Notification(**data)
dao_create_notification(notification)
notification.status = 'delivered'
dao_update_notification(notification)
assert Notification.query.one().status == 'delivered'
assert NotificationHistory.query.count() == 0
def test_update_notification_with_research_mode_service_does_not_create_or_update_history(
notify_db,
notify_db_session):
service = sample_service(notify_db, notify_db_session)
service.research_mode = True
dao_update_service(service)
template = sample_template(notify_db, notify_db_session, service=service)
data = _notification_json(template)
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.count() == 1
assert NotificationHistory.query.count() == 0
notification.status = 'delivered'
dao_update_notification(notification)
assert Notification.query.one().status == 'delivered'
assert NotificationHistory.query.count() == 0
def test_not_save_notification_and_not_create_stats_on_commit_error(sample_template, sample_job, mmg_provider):
random_id = str(uuid.uuid4())
assert Notification.query.count() == 0
data = _notification_json(sample_template, job_id=random_id)
notification = Notification(**data)
with pytest.raises(SQLAlchemyError):
dao_create_notification(notification)
assert Notification.query.count() == 0
assert Job.query.get(sample_job.id).notifications_sent == 0
assert NotificationStatistics.query.count() == 0
def test_save_notification_and_increment_job(sample_template, sample_job, mmg_provider):
assert Notification.query.count() == 0
data = _notification_json(sample_template, job_id=sample_job.id)
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.count() == 1
notification_from_db = Notification.query.all()[0]
assert notification_from_db.id
assert data['to'] == notification_from_db.to
assert data['job_id'] == notification_from_db.job_id
assert data['service'] == notification_from_db.service
assert data['template'] == notification_from_db.template
assert data['template_version'] == notification_from_db.template_version
assert data['created_at'] == notification_from_db.created_at
assert notification_from_db.status == 'created'
notification_2 = Notification(**data)
dao_create_notification(notification_2)
assert Notification.query.count() == 2
def test_save_notification_and_increment_correct_job(notify_db, notify_db_session, sample_template, mmg_provider):
from tests.app.conftest import sample_job
job_1 = sample_job(notify_db, notify_db_session, sample_template.service)
job_2 = sample_job(notify_db, notify_db_session, sample_template.service)
assert Notification.query.count() == 0
data = _notification_json(sample_template, job_id=job_1.id)
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.count() == 1
notification_from_db = Notification.query.all()[0]
assert notification_from_db.id
assert data['to'] == notification_from_db.to
assert data['job_id'] == notification_from_db.job_id
assert data['service'] == notification_from_db.service
assert data['template'] == notification_from_db.template
assert data['template_version'] == notification_from_db.template_version
assert data['created_at'] == notification_from_db.created_at
assert notification_from_db.status == 'created'
assert job_1.id != job_2.id
def test_save_notification_with_no_job(sample_template, mmg_provider):
assert Notification.query.count() == 0
data = _notification_json(sample_template)
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.count() == 1
notification_from_db = Notification.query.all()[0]
assert notification_from_db.id
assert data['to'] == notification_from_db.to
assert data['service'] == notification_from_db.service
assert data['template'] == notification_from_db.template
assert data['template_version'] == notification_from_db.template_version
assert data['created_at'] == notification_from_db.created_at
assert notification_from_db.status == 'created'
def test_get_notification_by_id(notify_db, notify_db_session, sample_template):
notification = sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
template=sample_template,
scheduled_for='2017-05-05 14:15',
status='created')
notification_from_db = get_notification_with_personalisation(
sample_template.service.id,
notification.id,
key_type=None
)
assert notification == notification_from_db
assert notification_from_db.scheduled_notification.scheduled_for == datetime(2017, 5, 5, 14, 15)
def test_get_notifications_by_reference(sample_template):
client_reference = 'some-client-ref'
assert len(Notification.query.all()) == 0
create_notification(sample_template, client_reference=client_reference)
create_notification(sample_template, client_reference=client_reference)
create_notification(sample_template, client_reference='other-ref')
all_notifications = get_notifications_for_service(
sample_template.service_id,
client_reference=client_reference
).items
assert len(all_notifications) == 2
def test_save_notification_no_job_id(sample_template, mmg_provider):
assert Notification.query.count() == 0
data = _notification_json(sample_template)
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.count() == 1
notification_from_db = Notification.query.all()[0]
assert notification_from_db.id
assert data['to'] == notification_from_db.to
assert data['service'] == notification_from_db.service
assert data['template'] == notification_from_db.template
assert data['template_version'] == notification_from_db.template_version
assert notification_from_db.status == 'created'
assert data.get('job_id') is None
def test_get_notification_for_job(sample_notification):
notification_from_db = get_notification_for_job(
sample_notification.service.id,
sample_notification.job_id,
sample_notification.id)
assert sample_notification == notification_from_db
def test_get_all_notifications_for_job(notify_db, notify_db_session, sample_job):
for i in range(0, 5):
try:
sample_notification(notify_db,
notify_db_session,
service=sample_job.service,
template=sample_job.template,
job=sample_job)
except IntegrityError:
pass
notifications_from_db = get_notifications_for_job(sample_job.service.id, sample_job.id).items
assert len(notifications_from_db) == 5
def test_get_all_notifications_for_job_by_status(notify_db, notify_db_session, sample_job):
notifications = partial(get_notifications_for_job, sample_job.service.id, sample_job.id)
for status in NOTIFICATION_STATUS_TYPES:
sample_notification(
notify_db,
notify_db_session,
service=sample_job.service,
template=sample_job.template,
job=sample_job,
status=status
)
assert len(notifications().items) == len(NOTIFICATION_STATUS_TYPES)
for status in NOTIFICATION_STATUS_TYPES:
if status == 'failed':
assert len(notifications(filter_dict={'status': status}).items) == len(NOTIFICATION_STATUS_TYPES_FAILED)
else:
assert len(notifications(filter_dict={'status': status}).items) == 1
assert len(notifications(filter_dict={'status': NOTIFICATION_STATUS_TYPES[:3]}).items) == 3
def test_update_notification_sets_status(sample_notification):
assert sample_notification.status == 'created'
sample_notification.status = 'failed'
dao_update_notification(sample_notification)
notification_from_db = Notification.query.get(sample_notification.id)
assert notification_from_db.status == 'failed'
@pytest.mark.parametrize('notification_type, expected_sms_count, expected_email_count, expected_letter_count', [
('sms', 8, 10, 10), ('email', 10, 8, 10), ('letter', 10, 10, 8)
])
@freeze_time("2016-01-10 12:00:00.000000")
def test_should_delete_notifications_by_type_after_seven_days(
notify_db,
notify_db_session,
sample_service,
notification_type,
expected_sms_count,
expected_email_count,
expected_letter_count
):
assert len(Notification.query.all()) == 0
email_template = sample_email_template(notify_db, notify_db_session, service=sample_service)
sms_template = sample_template(notify_db, notify_db_session, service=sample_service)
letter_template = sample_letter_template(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-01-{0:02d} {0:02d}:00:00.000000'.format(i)
with freeze_time(past_date):
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
)
all_notifications = Notification.query.all()
assert len(all_notifications) == 30
# Records from before 3rd should be deleted
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, 1, 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 = 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
# create one notification a day between 1st and 9th
for i in range(1, 11):
past_date = '2016-01-{0:02d}'.format(i)
with freeze_time(past_date):
create_notification(sample_template, created_at=datetime.utcnow(), status="failed")
all_notifications = Notification.query.all()
assert len(all_notifications) == 10
all_notifications = get_notifications_for_service(sample_template.service_id, limit_days=10).items
assert len(all_notifications) == 10
all_notifications = get_notifications_for_service(sample_template.service_id, limit_days=1).items
assert len(all_notifications) == 2
def test_creating_notification_adds_to_notification_history(sample_template):
data = _notification_json(sample_template)
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.count() == 1
hist = NotificationHistory.query.one()
assert hist.id == notification.id
assert hist.created_at == notification.created_at
assert hist.status == notification.status
assert not hasattr(hist, 'to')
assert not hasattr(hist, '_personalisation')
def test_updating_notification_updates_notification_history(sample_notification):
hist = NotificationHistory.query.one()
assert hist.id == sample_notification.id
assert hist.status == 'created'
sample_notification.status = 'sending'
dao_update_notification(sample_notification)
notification = Notification.query.one()
hist1 = NotificationHistory.query.one()
assert notification.id == sample_notification.id
assert notification.status == "sending"
assert hist1.id == sample_notification.id
assert hist1.status == 'sending'
def test_should_delete_notification_and_notification_history_for_id(notify_db, notify_db_session, sample_template):
data = _notification_json(sample_template)
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.count() == 1
assert NotificationHistory.query.count() == 1
dao_delete_notifications_and_history_by_id(notification.id)
assert Notification.query.count() == 0
assert NotificationHistory.query.count() == 0
def test_should_delete_notification_and_ignore_history_for_test_api(
notify_db,
notify_db_session,
sample_email_template,
sample_api_key):
data = _notification_json(sample_email_template)
data['key_type'] = KEY_TYPE_TEST
data['api_key_id'] = sample_api_key.id
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.count() == 1
assert NotificationHistory.query.count() == 0
dao_delete_notifications_and_history_by_id(notification.id)
assert Notification.query.count() == 0
assert NotificationHistory.query.count() == 0
def test_should_delete_notification_and_ignore_history_for_research_mode(notify_db, notify_db_session):
service = sample_service(notify_db, notify_db_session)
service.research_mode = True
dao_update_service(service)
template = sample_template(notify_db, notify_db_session, service=service)
data = _notification_json(template)
notification = Notification(**data)
dao_create_notification(notification)
assert Notification.query.count() == 1
assert NotificationHistory.query.count() == 0
dao_delete_notifications_and_history_by_id(notification.id)
assert Notification.query.count() == 0
assert NotificationHistory.query.count() == 0
def test_should_delete_only_notification_and_notification_history_with_id(notify_db, notify_db_session,
sample_template):
id_1 = uuid.uuid4()
id_2 = uuid.uuid4()
data_1 = _notification_json(sample_template, id=id_1)
data_2 = _notification_json(sample_template, id=id_2)
notification_1 = Notification(**data_1)
notification_2 = Notification(**data_2)
dao_create_notification(notification_1)
dao_create_notification(notification_2)
assert Notification.query.count() == 2
assert NotificationHistory.query.count() == 2
dao_delete_notifications_and_history_by_id(notification_1.id)
assert Notification.query.count() == 1
assert NotificationHistory.query.count() == 1
assert Notification.query.first().id == notification_2.id
assert NotificationHistory.query.first().id == notification_2.id
def test_should_delete_no_notifications_or_notification_historys_if_no_matching_ids(
notify_db,
notify_db_session,
sample_template
):
id_1 = uuid.uuid4()
id_2 = uuid.uuid4()
data_1 = _notification_json(sample_template, id=id_1)
notification_1 = Notification(**data_1)
dao_create_notification(notification_1)
assert Notification.query.count() == 1
assert NotificationHistory.query.count() == 1
dao_delete_notifications_and_history_by_id(id_2)
assert Notification.query.count() == 1
assert NotificationHistory.query.count() == 1
def _notification_json(sample_template, job_id=None, id=None, status=None):
data = {
'to': '+44709123456',
'service': sample_template.service,
'service_id': sample_template.service.id,
'template': sample_template,
'template_id': sample_template.id,
'template_version': sample_template.version,
'created_at': datetime.utcnow(),
'billable_units': 1,
'notification_type': sample_template.template_type,
'key_type': KEY_TYPE_NORMAL
}
if job_id:
data.update({'job_id': job_id})
if id:
data.update({'id': id})
if status:
data.update({'status': status})
return data
def test_dao_timeout_notifications(sample_template):
with freeze_time(datetime.utcnow() - timedelta(minutes=2)):
created = create_notification(sample_template, status='created')
sending = create_notification(sample_template, status='sending')
pending = create_notification(sample_template, status='pending')
delivered = create_notification(sample_template, status='delivered')
assert Notification.query.get(created.id).status == 'created'
assert Notification.query.get(sending.id).status == 'sending'
assert Notification.query.get(pending.id).status == 'pending'
assert Notification.query.get(delivered.id).status == 'delivered'
updated = dao_timeout_notifications(1)
assert Notification.query.get(created.id).status == 'technical-failure'
assert Notification.query.get(sending.id).status == 'temporary-failure'
assert Notification.query.get(pending.id).status == 'temporary-failure'
assert Notification.query.get(delivered.id).status == 'delivered'
assert NotificationHistory.query.get(created.id).status == 'technical-failure'
assert NotificationHistory.query.get(sending.id).status == 'temporary-failure'
assert NotificationHistory.query.get(pending.id).status == 'temporary-failure'
assert NotificationHistory.query.get(delivered.id).status == 'delivered'
assert updated == 3
def test_dao_timeout_notifications_only_updates_for_older_notifications(sample_template):
with freeze_time(datetime.utcnow() + timedelta(minutes=10)):
created = create_notification(sample_template, status='created')
sending = create_notification(sample_template, status='sending')
pending = create_notification(sample_template, status='pending')
delivered = create_notification(sample_template, status='delivered')
assert Notification.query.get(created.id).status == 'created'
assert Notification.query.get(sending.id).status == 'sending'
assert Notification.query.get(pending.id).status == 'pending'
assert Notification.query.get(delivered.id).status == 'delivered'
updated = dao_timeout_notifications(1)
assert NotificationHistory.query.get(created.id).status == 'created'
assert NotificationHistory.query.get(sending.id).status == 'sending'
assert NotificationHistory.query.get(pending.id).status == 'pending'
assert NotificationHistory.query.get(delivered.id).status == 'delivered'
assert updated == 0
def test_dao_timeout_notifications_doesnt_affect_letters(sample_letter_template):
with freeze_time(datetime.utcnow() - timedelta(minutes=2)):
created = create_notification(sample_letter_template, status='created')
sending = create_notification(sample_letter_template, status='sending')
pending = create_notification(sample_letter_template, status='pending')
delivered = create_notification(sample_letter_template, status='delivered')
assert Notification.query.get(created.id).status == 'created'
assert Notification.query.get(sending.id).status == 'sending'
assert Notification.query.get(pending.id).status == 'pending'
assert Notification.query.get(delivered.id).status == 'delivered'
updated = dao_timeout_notifications(1)
assert NotificationHistory.query.get(created.id).status == 'created'
assert NotificationHistory.query.get(sending.id).status == 'sending'
assert NotificationHistory.query.get(pending.id).status == 'pending'
assert NotificationHistory.query.get(delivered.id).status == 'delivered'
assert updated == 0
def test_should_return_notifications_excluding_jobs_by_default(sample_template, sample_job, sample_api_key):
with_job = create_notification(sample_template, job=sample_job)
without_job = create_notification(sample_template, api_key=sample_api_key)
include_jobs = get_notifications_for_service(sample_template.service_id, include_jobs=True).items
assert len(include_jobs) == 2
exclude_jobs_by_default = get_notifications_for_service(sample_template.service_id).items
assert len(exclude_jobs_by_default) == 1
assert exclude_jobs_by_default[0].id == without_job.id
exclude_jobs_manually = get_notifications_for_service(sample_template.service_id, include_jobs=False).items
assert len(exclude_jobs_manually) == 1
assert exclude_jobs_manually[0].id == without_job.id
def test_get_notifications_created_by_api_or_csv_are_returned_correctly_excluding_test_key_notifications(
notify_db,
notify_db_session,
sample_service,
sample_job,
sample_api_key,
sample_team_api_key,
sample_test_api_key
):
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), job=sample_job
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_api_key,
key_type=sample_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_team_api_key,
key_type=sample_team_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_test_api_key,
key_type=sample_test_api_key.key_type
)
all_notifications = Notification.query.all()
assert len(all_notifications) == 4
# returns all real API derived notifications
all_notifications = get_notifications_for_service(sample_service.id).items
assert len(all_notifications) == 2
# returns all API derived notifications, including those created with test key
all_notifications = get_notifications_for_service(sample_service.id, include_from_test_key=True).items
assert len(all_notifications) == 3
# all real notifications including jobs
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, include_jobs=True).items
assert len(all_notifications) == 3
def test_get_notifications_with_a_live_api_key_type(
notify_db,
notify_db_session,
sample_service,
sample_job,
sample_api_key,
sample_team_api_key,
sample_test_api_key
):
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), job=sample_job
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_api_key,
key_type=sample_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_team_api_key,
key_type=sample_team_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_test_api_key,
key_type=sample_test_api_key.key_type
)
all_notifications = Notification.query.all()
assert len(all_notifications) == 4
# only those created with normal API key, no jobs
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, key_type=KEY_TYPE_NORMAL).items
assert len(all_notifications) == 1
# only those created with normal API key, with jobs
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, include_jobs=True,
key_type=KEY_TYPE_NORMAL).items
assert len(all_notifications) == 2
def test_get_notifications_with_a_test_api_key_type(
notify_db,
notify_db_session,
sample_service,
sample_job,
sample_api_key,
sample_team_api_key,
sample_test_api_key
):
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), job=sample_job
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_api_key,
key_type=sample_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_team_api_key,
key_type=sample_team_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_test_api_key,
key_type=sample_test_api_key.key_type
)
# only those created with test API key, no jobs
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, key_type=KEY_TYPE_TEST).items
assert len(all_notifications) == 1
# only those created with test API key, no jobs, even when requested
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, include_jobs=True,
key_type=KEY_TYPE_TEST).items
assert len(all_notifications) == 1
def test_get_notifications_with_a_team_api_key_type(
notify_db,
notify_db_session,
sample_service,
sample_job,
sample_api_key,
sample_team_api_key,
sample_test_api_key
):
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), job=sample_job
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_api_key,
key_type=sample_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_team_api_key,
key_type=sample_team_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_test_api_key,
key_type=sample_test_api_key.key_type
)
# only those created with team API key, no jobs
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, key_type=KEY_TYPE_TEAM).items
assert len(all_notifications) == 1
# only those created with team API key, no jobs, even when requested
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, include_jobs=True,
key_type=KEY_TYPE_TEAM).items
assert len(all_notifications) == 1
def test_should_exclude_test_key_notifications_by_default(
notify_db,
notify_db_session,
sample_service,
sample_job,
sample_api_key,
sample_team_api_key,
sample_test_api_key
):
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), job=sample_job
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_api_key,
key_type=sample_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_team_api_key,
key_type=sample_team_api_key.key_type
)
sample_notification(
notify_db, notify_db_session, created_at=datetime.utcnow(), api_key=sample_test_api_key,
key_type=sample_test_api_key.key_type
)
all_notifications = Notification.query.all()
assert len(all_notifications) == 4
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1).items
assert len(all_notifications) == 2
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, include_jobs=True).items
assert len(all_notifications) == 3
all_notifications = get_notifications_for_service(sample_service.id, limit_days=1, key_type=KEY_TYPE_TEST).items
assert len(all_notifications) == 1
@pytest.mark.parametrize('notification_type', ['sms', 'email'])
def test_get_total_sent_notifications_in_date_range_returns_only_in_date_range(
notify_db,
notify_db_session,
sample_template,
notification_type
):
notification_history = partial(
create_notification_history,
notify_db,
notify_db_session,
sample_template,
notification_type=notification_type,
status='delivered'
)
start_date = datetime(2000, 3, 30, 0, 0, 0, 0)
with freeze_time(start_date):
notification_history(created_at=start_date + timedelta(hours=3))
notification_history(created_at=start_date + timedelta(hours=5, minutes=10))
notification_history(created_at=start_date + timedelta(hours=11, minutes=59))
end_date = datetime(2000, 3, 31, 0, 0, 0, 0)
notification_history(created_at=end_date + timedelta(seconds=1))
notification_history(created_at=end_date + timedelta(minutes=10))
total_count = get_total_sent_notifications_in_date_range(start_date, end_date, notification_type)
assert total_count == 3
@pytest.mark.parametrize('notification_type', ['sms', 'email'])
def test_get_total_sent_notifications_in_date_range_excludes_test_key_notifications(
notify_db,
notify_db_session,
sample_template,
notification_type
):
notification_history = partial(
create_notification_history,
notify_db,
notify_db_session,
sample_template,
notification_type=notification_type,
status='delivered'
)
start_date = datetime(2000, 3, 30, 0, 0, 0, 0)
end_date = datetime(2000, 3, 31, 0, 0, 0, 0)
with freeze_time(start_date):
notification_history(key_type=KEY_TYPE_TEAM)
notification_history(key_type=KEY_TYPE_TEAM)
notification_history(key_type=KEY_TYPE_NORMAL)
notification_history(key_type=KEY_TYPE_TEST)
total_count = get_total_sent_notifications_in_date_range(start_date, end_date, notification_type)
assert total_count == 3
def test_get_total_sent_notifications_for_sms_excludes_email_counts(
notify_db,
notify_db_session,
sample_template
):
notification_history = partial(
create_notification_history,
notify_db,
notify_db_session,
sample_template,
status='delivered'
)
start_date = datetime(2000, 3, 30, 0, 0, 0, 0)
end_date = datetime(2000, 3, 31, 0, 0, 0, 0)
with freeze_time(start_date):
notification_history(notification_type='email')
notification_history(notification_type='email')
notification_history(notification_type='sms')
notification_history(notification_type='sms')
notification_history(notification_type='sms')
total_count = get_total_sent_notifications_in_date_range(start_date, end_date, 'sms')
assert total_count == 3
def test_get_total_sent_notifications_for_email_excludes_sms_counts(
notify_db,
notify_db_session,
sample_template
):
notification_history = partial(
create_notification_history,
notify_db,
notify_db_session,
sample_template,
status='delivered'
)
start_date = datetime(2000, 3, 30, 0, 0, 0, 0)
end_date = datetime(2000, 3, 31, 0, 0, 0, 0)
with freeze_time(start_date):
notification_history(notification_type='email')
notification_history(notification_type='email')
notification_history(notification_type='sms')
notification_history(notification_type='sms')
notification_history(notification_type='sms')
total_count = get_total_sent_notifications_in_date_range(start_date, end_date, 'email')
assert total_count == 2
@freeze_time("2016-01-10 12:00:00.000000")
def test_slow_provider_delivery_returns_for_sent_notifications(
sample_template
):
now = datetime.utcnow()
one_minute_from_now = now + timedelta(minutes=1)
five_minutes_from_now = now + timedelta(minutes=5)
notification_five_minutes_to_deliver = partial(
create_notification,
template=sample_template,
status='delivered',
sent_by='mmg',
updated_at=five_minutes_from_now
)
notification_five_minutes_to_deliver(sent_at=now)
notification_five_minutes_to_deliver(sent_at=one_minute_from_now)
notification_five_minutes_to_deliver(sent_at=one_minute_from_now)
slow_delivery = is_delivery_slow_for_provider(
sent_at=one_minute_from_now,
provider='mmg',
threshold=2,
delivery_time=timedelta(minutes=3),
service_id=sample_template.service.id,
template_id=sample_template.id
)
assert slow_delivery
@freeze_time("2016-01-10 12:00:00.000000")
def test_slow_provider_delivery_observes_threshold(
sample_template
):
now = datetime.utcnow()
five_minutes_from_now = now + timedelta(minutes=5)
notification_five_minutes_to_deliver = partial(
create_notification,
template=sample_template,
status='delivered',
sent_at=now,
sent_by='mmg',
updated_at=five_minutes_from_now
)
notification_five_minutes_to_deliver()
notification_five_minutes_to_deliver()
slow_delivery = is_delivery_slow_for_provider(
sent_at=now,
provider='mmg',
threshold=3,
delivery_time=timedelta(minutes=5),
service_id=sample_template.service.id,
template_id=sample_template.id
)
assert not slow_delivery
@freeze_time("2016-01-10 12:00:00.000000")
def test_slow_provider_delivery_returns_for_delivered_notifications_only(
sample_template
):
now = datetime.utcnow()
five_minutes_from_now = now + timedelta(minutes=5)
notification_five_minutes_to_deliver = partial(
create_notification,
template=sample_template,
sent_at=now,
sent_by='firetext',
created_at=now,
updated_at=five_minutes_from_now
)
notification_five_minutes_to_deliver(status='sending')
notification_five_minutes_to_deliver(status='delivered')
notification_five_minutes_to_deliver(status='delivered')
slow_delivery = is_delivery_slow_for_provider(
sent_at=now,
provider='firetext',
threshold=2,
delivery_time=timedelta(minutes=5),
service_id=sample_template.service.id,
template_id=sample_template.id
)
assert slow_delivery
@freeze_time("2016-01-10 12:00:00.000000")
def test_slow_provider_delivery_does_not_return_for_standard_delivery_time(
sample_template
):
now = datetime.utcnow()
five_minutes_from_now = now + timedelta(minutes=5)
notification = partial(
create_notification,
template=sample_template,
created_at=now,
sent_at=now,
sent_by='mmg',
status='delivered'
)
notification(updated_at=five_minutes_from_now - timedelta(seconds=1))
notification(updated_at=five_minutes_from_now - timedelta(seconds=1))
notification(updated_at=five_minutes_from_now)
slow_delivery = is_delivery_slow_for_provider(
sent_at=now,
provider='mmg',
threshold=2,
delivery_time=timedelta(minutes=5),
service_id=sample_template.service.id,
template_id=sample_template.id
)
assert not slow_delivery
def test_dao_update_notifications_for_job_to_sent_to_dvla(notify_db, notify_db_session, sample_letter_template):
job = sample_job(notify_db=notify_db, notify_db_session=notify_db_session, template=sample_letter_template)
notification = create_notification(template=sample_letter_template, job=job)
updated_count = dao_update_notifications_for_job_to_sent_to_dvla(job_id=job.id, provider='some provider')
assert updated_count == 1
updated_notification = Notification.query.get(notification.id)
assert updated_notification.status == 'sending'
assert updated_notification.sent_by == 'some provider'
assert updated_notification.sent_at
assert updated_notification.updated_at
history = NotificationHistory.query.get(notification.id)
assert history.status == 'sending'
assert history.sent_by == 'some provider'
assert history.sent_at
assert history.updated_at
def test_dao_update_notifications_for_job_to_sent_to_dvla_does_update_history_if_test_key(sample_letter_job):
api_key = create_api_key(sample_letter_job.service, key_type=KEY_TYPE_TEST)
notification = create_notification(
sample_letter_job.template,
job=sample_letter_job,
api_key=api_key
)
updated_count = dao_update_notifications_for_job_to_sent_to_dvla(
job_id=sample_letter_job.id,
provider='some provider'
)
assert updated_count == 1
updated_notification = Notification.query.get(notification.id)
assert updated_notification.status == 'sending'
assert updated_notification.sent_by == 'some provider'
assert updated_notification.sent_at
assert updated_notification.updated_at
assert NotificationHistory.query.count() == 0
def test_dao_get_notifications_by_to_field(sample_template):
recipient_to_search_for = {
'to_field': '+447700900855',
'normalised_to': '447700900855'
}
notification1 = create_notification(
template=sample_template, **recipient_to_search_for
)
create_notification(
template=sample_template, key_type=KEY_TYPE_TEST, **recipient_to_search_for
)
create_notification(
template=sample_template, to_field='jack@gmail.com', normalised_to='jack@gmail.com'
)
create_notification(
template=sample_template, to_field='jane@gmail.com', normalised_to='jane@gmail.com'
)
results = dao_get_notifications_by_to_field(
notification1.service_id,
recipient_to_search_for["to_field"]
)
assert len(results) == 1
assert notification1.id == results[0].id
def test_dao_get_notifications_by_to_field_search_is_not_case_sensitive(sample_template):
notification = create_notification(
template=sample_template, to_field='jack@gmail.com', normalised_to='jack@gmail.com'
)
results = dao_get_notifications_by_to_field(notification.service_id, 'JACK@gmail.com')
notification_ids = [notification.id for notification in results]
assert len(results) == 1
assert notification.id in notification_ids
@pytest.mark.parametrize('to', [
'not@email', '123'
])
def test_dao_get_notifications_by_to_field_accepts_invalid_phone_numbers_and_email_addresses(
sample_template,
to,
):
notification = create_notification(
template=sample_template, to_field='test@example.com', normalised_to='test@example.com'
)
results = dao_get_notifications_by_to_field(notification.service_id, to)
assert len(results) == 0
def test_dao_get_notifications_by_to_field_search_ignores_spaces(sample_template):
notification1 = create_notification(
template=sample_template, to_field='+447700900855', normalised_to='447700900855'
)
notification2 = create_notification(
template=sample_template, to_field='+44 77 00900 855', normalised_to='447700900855'
)
notification3 = create_notification(
template=sample_template, to_field=' +4477009 00 855 ', normalised_to='447700900855'
)
create_notification(
template=sample_template, to_field='jaCK@gmail.com', normalised_to='jack@gmail.com'
)
results = dao_get_notifications_by_to_field(notification1.service_id, '+447700900855')
notification_ids = [notification.id for notification in results]
assert len(results) == 3
assert notification1.id in notification_ids
assert notification2.id in notification_ids
assert notification3.id in notification_ids
def test_dao_created_scheduled_notification(sample_notification):
scheduled_notification = ScheduledNotification(notification_id=sample_notification.id,
scheduled_for=datetime.strptime("2017-01-05 14:15",
"%Y-%m-%d %H:%M"))
dao_created_scheduled_notification(scheduled_notification)
saved_notification = ScheduledNotification.query.all()
assert len(saved_notification) == 1
assert saved_notification[0].notification_id == sample_notification.id
assert saved_notification[0].scheduled_for == datetime(2017, 1, 5, 14, 15)
def test_dao_get_scheduled_notifications(notify_db, notify_db_session, sample_template):
notification_1 = sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
template=sample_template, scheduled_for='2017-05-05 14:15',
status='created')
sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
template=sample_template, scheduled_for='2017-05-04 14:15', status='delivered')
sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
template=sample_template, status='created')
scheduled_notifications = dao_get_scheduled_notifications()
assert len(scheduled_notifications) == 1
assert scheduled_notifications[0].id == notification_1.id
assert scheduled_notifications[0].scheduled_notification.pending
def test_set_scheduled_notification_to_processed(notify_db, notify_db_session, sample_template):
notification_1 = sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
template=sample_template, scheduled_for='2017-05-05 14:15',
status='created')
scheduled_notifications = dao_get_scheduled_notifications()
assert len(scheduled_notifications) == 1
assert scheduled_notifications[0].id == notification_1.id
assert scheduled_notifications[0].scheduled_notification.pending
set_scheduled_notification_to_processed(notification_1.id)
scheduled_notifications = dao_get_scheduled_notifications()
assert not scheduled_notifications
def test_dao_get_notifications_by_to_field_filters_status(sample_template):
notification = create_notification(
template=sample_template, to_field='+447700900855',
normalised_to='447700900855', status='delivered'
)
create_notification(
template=sample_template, to_field='+447700900855',
normalised_to='447700900855', status='temporary-failure'
)
notifications = dao_get_notifications_by_to_field(notification.service_id, "+447700900855", statuses=['delivered'])
assert len(notifications) == 1
assert notification.id == notifications[0].id
def test_dao_get_notifications_by_to_field_filters_multiple_statuses(sample_template):
notification1 = create_notification(
template=sample_template, to_field='+447700900855',
normalised_to='447700900855', status='delivered'
)
notification2 = create_notification(
template=sample_template, to_field='+447700900855',
normalised_to='447700900855', status='sending'
)
notifications = dao_get_notifications_by_to_field(
notification1.service_id, "+447700900855", statuses=['delivered', 'sending']
)
notification_ids = [notification.id for notification in notifications]
assert len(notifications) == 2
assert notification1.id in notification_ids
assert notification2.id in notification_ids
def test_dao_get_notifications_by_to_field_returns_all_if_no_status_filter(sample_template):
notification1 = create_notification(
template=sample_template, to_field='+447700900855',
normalised_to='447700900855', status='delivered'
)
notification2 = create_notification(
template=sample_template, to_field='+447700900855',
normalised_to='447700900855', status='temporary-failure'
)
notifications = dao_get_notifications_by_to_field(
notification1.service_id, "+447700900855"
)
notification_ids = [notification.id for notification in notifications]
assert len(notifications) == 2
assert notification1.id in notification_ids
assert notification2.id in notification_ids
@freeze_time('2016-01-01 11:10:00')
def test_dao_get_notifications_by_to_field_orders_by_created_at_desc(sample_template):
notification = partial(
create_notification,
template=sample_template,
to_field='+447700900855',
normalised_to='447700900855'
)
notification_a_minute_ago = notification(created_at=datetime.utcnow() - timedelta(minutes=1))
notification = notification(created_at=datetime.utcnow())
notifications = dao_get_notifications_by_to_field(
sample_template.service_id, '+447700900855'
)
assert len(notifications) == 2
assert notifications[0].id == notification.id
assert notifications[1].id == notification_a_minute_ago.id
def test_dao_create_notification_email_reply_to_mapping(sample_service, sample_notification):
create_reply_to_email(sample_service, "test@test.com")
reply_to_address = dao_get_reply_to_by_service_id(sample_service.id)
dao_create_notification_email_reply_to_mapping(sample_notification.id, reply_to_address[0].id)
email_reply_to = NotificationEmailReplyTo.query.all()
assert len(email_reply_to) == 1
assert email_reply_to[0].notification_id == sample_notification.id
assert email_reply_to[0].service_email_reply_to_id == reply_to_address[0].id
def test_dao_create_multiple_notification_email_reply_to_mapping(sample_service, sample_notification):
reply_to_address = create_reply_to_email(sample_service, "test@test.com")
dao_create_notification_email_reply_to_mapping(sample_notification.id, reply_to_address.id)
with pytest.raises(IntegrityError) as e:
dao_create_notification_email_reply_to_mapping(sample_notification.id, reply_to_address.id)
assert 'duplicate key value' in str(e.value)
email_reply_to = NotificationEmailReplyTo.query.all()
assert len(email_reply_to) == 1
assert email_reply_to[0].notification_id == sample_notification.id
assert email_reply_to[0].service_email_reply_to_id == reply_to_address.id
def test_dao_get_notification_ememail_reply_toail_reply_for_notification(sample_service, sample_notification):
reply_to_address = create_reply_to_email(sample_service, "test@test.com")
dao_create_notification_email_reply_to_mapping(sample_notification.id, reply_to_address.id)
assert dao_get_notification_email_reply_for_notification(sample_notification.id) == "test@test.com"
def test_dao_get_notification_email_reply_for_notification_where_no_mapping(notify_db_session, fake_uuid):
assert dao_get_notification_email_reply_for_notification(fake_uuid) is None
Reference in New Issue View Git Blame Copy Permalink