darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-07-24 01:49:42 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into redo-queue-visibitlity-timeout

Browse Source 
Conflicts:
	app/notifications/process_notifications.py
	app/v2/notifications/post_notifications.py
This commit is contained in:
Martyn Inglis
2017-05-30 10:18:18 +01:00
parent 29b6a918e8 c59746a4a7
commit 75c5e5e5d5
30 changed files with 1473 additions and 345 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
tests/app/celery/test_scheduled_tasks.py
View File

@@ -6,7 +6,8 @@ from functools import partial
from flask import current_app
from freezegun import freeze_time
from app.celery.scheduled_tasks import s3, timeout_job_statistics, delete_sms_notifications_older_than_seven_days, \
delete_letter_notifications_older_than_seven_days, delete_email_notifications_older_than_seven_days
delete_letter_notifications_older_than_seven_days, delete_email_notifications_older_than_seven_days, \
send_scheduled_notifications
from app.celery import scheduled_tasks
from app.celery.scheduled_tasks import (
delete_verify_codes,
@@ -20,6 +21,7 @@ from app.celery.scheduled_tasks import (
)
from app.clients.performance_platform.performance_platform_client import PerformancePlatformClient
from app.dao.jobs_dao import dao_get_job_by_id
from app.dao.notifications_dao import dao_get_scheduled_notifications
from app.dao.provider_details_dao import (
dao_update_provider_details,
get_current_provider
@@ -30,8 +32,7 @@ from tests.app.db import create_notification, create_service
from tests.app.conftest import (
sample_job as create_sample_job,
sample_notification_history as create_notification_history,
create_custom_template
)
create_custom_template)
from tests.conftest import set_config_values
from unittest.mock import call, patch, PropertyMock
@@ -416,6 +417,24 @@ def test_switch_providers_on_slow_delivery_does_not_switch_based_on_older_notifi
assert starting_provider.identifier == current_provider.identifier
@freeze_time("2017-05-01 14:00:00")
def test_should_send_all_scheduled_notifications_to_deliver_queue(sample_template, mocker):
mocked = mocker.patch('app.celery.provider_tasks.deliver_sms')
message_to_deliver = create_notification(template=sample_template, scheduled_for="2017-05-01 13:15")
create_notification(template=sample_template, scheduled_for="2017-05-01 10:15", status='delivered')
create_notification(template=sample_template)
create_notification(template=sample_template, scheduled_for="2017-05-01 14:15")
scheduled_notifications = dao_get_scheduled_notifications()
assert len(scheduled_notifications) == 1
send_scheduled_notifications()
mocked.apply_async.assert_called_once_with([str(message_to_deliver.id)], queue='send-tasks')
scheduled_notifications = dao_get_scheduled_notifications()
assert not scheduled_notifications
def test_timeout_job_statistics_called_with_notification_timeout(notify_api, mocker):
notify_api.config['SENDING_NOTIFICATIONS_TIMEOUT_PERIOD'] = 999
dao_mock = mocker.patch('app.celery.scheduled_tasks.dao_timeout_job_statistics')

24
tests/app/conftest.py
View File

@@ -24,7 +24,7 @@ from app.models import (
NotificationStatistics,
ServiceWhitelist,
KEY_TYPE_NORMAL, KEY_TYPE_TEST, KEY_TYPE_TEAM,
MOBILE_TYPE, EMAIL_TYPE, LETTER_TYPE, NOTIFICATION_STATUS_TYPES_COMPLETED)
MOBILE_TYPE, EMAIL_TYPE, SMS_TYPE, LETTER_TYPE, NOTIFICATION_STATUS_TYPES_COMPLETED, ScheduledNotification)
from app.dao.users_dao import (create_user_code, create_secret_code)
from app.dao.services_dao import (dao_create_service, dao_add_user_to_service)
from app.dao.templates_dao import dao_create_template
@@ -124,7 +124,8 @@ def sample_service(
restricted=False,
limit=1000,
email_from=None,
can_send_international_sms=False
can_send_international_sms=False,
permissions=[SMS_TYPE, EMAIL_TYPE]
):
if user is None:
user = create_user()
@@ -142,7 +143,7 @@ def sample_service(
service = Service.query.filter_by(name=service_name).first()
if not service:
service = Service(**data)
dao_create_service(service, user)
dao_create_service(service, user, service_permissions=permissions)
else:
if user not in service.users:
dao_add_user_to_service(service, user)
@@ -445,7 +446,9 @@ def sample_notification(
key_type=KEY_TYPE_NORMAL,
sent_by=None,
client_reference=None,
rate_multiplier=1.0
rate_multiplier=1.0,
scheduled_for=None,
normalised_to=None
):
if created_at is None:
created_at = datetime.utcnow()
@@ -483,12 +486,23 @@ def sample_notification(
'sent_by': sent_by,
'updated_at': created_at if status in NOTIFICATION_STATUS_TYPES_COMPLETED else None,
'client_reference': client_reference,
'rate_multiplier': rate_multiplier
'rate_multiplier': rate_multiplier,
'normalised_to': normalised_to
}
if job_row_number is not None:
data['job_row_number'] = job_row_number
notification = Notification(**data)
dao_create_notification(notification)
if scheduled_for:
scheduled_notification = ScheduledNotification(id=uuid.uuid4(),
notification_id=notification.id,
scheduled_for=datetime.strptime(scheduled_for,
"%Y-%m-%d %H:%M"))
if status != 'created':
scheduled_notification.pending = False
db.session.add(scheduled_notification)
db.session.commit()
return notification

210
tests/app/dao/test_notification_dao.py
View File

@@ -12,9 +12,11 @@ from app.models import (
Job,
NotificationStatistics,
TemplateStatistics,
ScheduledNotification,
NOTIFICATION_STATUS_TYPES,
NOTIFICATION_STATUS_TYPES_FAILED,
NOTIFICATION_SENT,
NOTIFICATION_DELIVERED,
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST
@@ -40,7 +42,9 @@ from app.dao.notifications_dao import (
dao_delete_notifications_and_history_by_id,
dao_timeout_notifications,
is_delivery_slow_for_provider,
dao_update_notifications_sent_to_dvla, dao_get_notifications_by_to_field)
dao_update_notifications_sent_to_dvla,
dao_get_notifications_by_to_field,
dao_created_scheduled_notification, dao_get_scheduled_notifications, set_scheduled_notification_to_processed)
from app.dao.services_dao import dao_update_service
from tests.app.db import create_notification
@@ -354,28 +358,45 @@ def test_should_update_status_by_id_if_created(notify_db, notify_db_session):
assert updated.status == 'failed'
def test_should_not_update_status_by_reference_if_in_sent_status(notify_db, notify_db_session):
notification = sample_notification(
notify_db,
notify_db_session,
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'
)
update_notification_status_by_reference('foo', 'failed')
assert Notification.query.get(notification.id).status == NOTIFICATION_SENT
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_in_sent_status(notify_db, notify_db_session):
notification = sample_notification(
notify_db,
notify_db_session,
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
)
update_notification_status_by_id(notification.id, 'failed')
res = update_notification_status_by_id(notification.id, 'delivered')
assert Notification.query.get(notification.id).status == NOTIFICATION_SENT
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):
@@ -714,13 +735,18 @@ def test_save_notification_with_no_job(sample_template, mmg_provider):
assert notification_from_db.status == 'created'
def test_get_notification_by_id(sample_notification):
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_notification.service.id,
sample_notification.id,
sample_template.service.id,
notification.id,
key_type=None
)
assert sample_notification == notification_from_db
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(notify_db, notify_db_session, sample_service):
@@ -1720,30 +1746,146 @@ def test_dao_update_notifications_sent_to_dvla_does_update_history_if_test_key(
def test_dao_get_notifications_by_to_field(sample_template):
notification1 = create_notification(template=sample_template, to_field='+447700900855')
notification2 = create_notification(template=sample_template, to_field='jack@gmail.com')
notification3 = create_notification(template=sample_template, to_field='jane@gmail.com')
notification1 = create_notification(
template=sample_template, to_field='+447700900855', normalised_to='447700900855'
)
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, "+447700900855")
assert len(results) == 1
assert results[0].id == notification1.id
assert notification1.id == results[0].id
def test_dao_get_notifications_by_to_field_search_is_not_case_sensitive(sample_template):
notification1 = create_notification(template=sample_template, to_field='+447700900855')
notification2 = create_notification(template=sample_template, to_field='jack@gmail.com')
notification3 = create_notification(template=sample_template, to_field='jane@gmail.com')
results = dao_get_notifications_by_to_field(notification1.service_id, 'JACK@gmail.com')
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 results[0].id == notification2.id
assert notification.id in notification_ids
def test_dao_get_notifications_by_to_field_search_ignores_spaces(sample_template):
notification1 = create_notification(template=sample_template, to_field='+447700900855')
notification2 = create_notification(template=sample_template, to_field='+44 77 00900 855')
notification3 = create_notification(template=sample_template, to_field=' +4477009 00 855 ')
notification4 = create_notification(template=sample_template, to_field='jack@gmail.com')
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 [r.id for r in results]
assert notification2.id in [r.id for r in results]
assert notification3.id in [r.id for r in results]
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

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

@@ -258,15 +258,15 @@ def test_create_service_returns_service_with_default_permissions(service_factory
service = dao_fetch_service_by_id(service.id)
assert len(service.permissions) == 2
assert all(p.permission in [SMS_TYPE, EMAIL_TYPE] for p in service.permissions)
assert set([SMS_TYPE, EMAIL_TYPE]) == set(p.permission for p in service.permissions)
# This test is only for backward compatibility and will be removed
# when the 'can_use' columns are dropped from the Service data model
# when the deprecated 'can_use' columns are not used in the Service data model
@pytest.mark.parametrize("permission_to_add, can_send_letters, can_send_international_sms",
[(LETTER_TYPE, True, False),
(INTERNATIONAL_SMS_TYPE, False, True)])
def test_create_service_by_id_adding_service_permission_returns_service_with_permissions_set(
def test_create_service_by_id_adding_service_permission_returns_service_with_flags_and_permissions_set(
service_factory, permission_to_add, can_send_letters, can_send_international_sms):
service = service_factory.get('testing', email_from='testing')
@@ -275,18 +275,59 @@ def test_create_service_by_id_adding_service_permission_returns_service_with_per
service = dao_fetch_service_by_id(service.id)
assert len(service.permissions) == 3
assert all(p.permission in [SMS_TYPE, EMAIL_TYPE, permission_to_add] for p in service.permissions)
assert set([SMS_TYPE, EMAIL_TYPE, permission_to_add]) == set(p.permission for p in service.permissions)
assert service.can_send_letters == can_send_letters
assert service.can_send_international_sms == can_send_international_sms
def test_remove_permission_from_service_by_id_returns_service_with_correct_permissions(service_factory):
# This test is only for backward compatibility and will be removed
# when the deprecated 'can_use' columns are not used in the Service data model
@pytest.mark.parametrize("permission_to_remove, can_send_letters, can_send_international_sms",
[(LETTER_TYPE, False, True),
(INTERNATIONAL_SMS_TYPE, True, False)])
def test_create_service_by_id_removing_service_permission_returns_service_with_flags_and_permissions_set(
service_factory, permission_to_remove, can_send_letters, can_send_international_sms):
service = service_factory.get('testing', email_from='testing')
dao_remove_service_permission(service_id=service.id, permission=SMS_TYPE)
dao_add_service_permission(service_id=service.id, permission=LETTER_TYPE)
dao_add_service_permission(service_id=service.id, permission=INTERNATIONAL_SMS_TYPE)
service = dao_fetch_service_by_id(service.id)
service.set_permissions()
assert len(service.permissions) == 4
assert service.can_send_letters
assert service.can_send_international_sms
dao_remove_service_permission(service_id=service.id, permission=permission_to_remove)
service.set_permissions()
service = dao_fetch_service_by_id(service.id)
expected_permissions = [SMS_TYPE, EMAIL_TYPE, LETTER_TYPE, INTERNATIONAL_SMS_TYPE]
expected_permissions.remove(permission_to_remove)
assert len(service.permissions) == 3
assert set(expected_permissions) == set(p.permission for p in service.permissions)
assert service.can_send_letters == can_send_letters
assert service.can_send_international_sms == can_send_international_sms
@pytest.mark.parametrize("permission_to_remove, permission_remaining",
[(SMS_TYPE, EMAIL_TYPE),
(EMAIL_TYPE, SMS_TYPE)])
def test_remove_permission_from_service_by_id_returns_service_with_correct_permissions(
sample_service, permission_to_remove, permission_remaining):
dao_remove_service_permission(service_id=sample_service.id, permission=permission_to_remove)
service = dao_fetch_service_by_id(sample_service.id)
assert len(service.permissions) == 1
assert service.permissions[0].permission == EMAIL_TYPE
assert service.permissions[0].permission == permission_remaining
def test_removing_all_permission_returns_service_with_no_permissions(sample_service):
dao_remove_service_permission(service_id=sample_service.id, permission=SMS_TYPE)
dao_remove_service_permission(service_id=sample_service.id, permission=EMAIL_TYPE)
service = dao_fetch_service_by_id(sample_service.id)
assert len(service.permissions) == 0
def test_remove_service_does_not_remove_service_permission_types(sample_service):
@@ -294,7 +335,7 @@ def test_remove_service_does_not_remove_service_permission_types(sample_service)
services = dao_fetch_all_services()
assert len(services) == 0
assert set([p.name for p in ServicePermissionTypes.query.all()]) & set(SERVICE_PERMISSION_TYPES)
assert set(p.name for p in ServicePermissionTypes.query.all()) == set(SERVICE_PERMISSION_TYPES)
def test_create_service_by_id_adding_and_removing_letter_returns_service_without_letter(service_factory):
@@ -372,6 +413,42 @@ def test_update_service_creates_a_history_record_with_current_data(sample_user):
assert Service.get_history_model().query.filter_by(name='updated_service_name').one().version == 2
def test_update_service_permission_creates_a_history_record_with_current_data(sample_user):
assert Service.query.count() == 0
assert Service.get_history_model().query.count() == 0
service = Service(name="service_name",
email_from="email_from",
message_limit=1000,
restricted=False,
created_by=sample_user)
dao_create_service(service, sample_user)
service.permissions.append(ServicePermission(service_id=service.id, permission='letter'))
dao_update_service(service)
assert Service.query.count() == 1
assert Service.get_history_model().query.count() == 2
service_from_db = Service.query.first()
assert service_from_db.version == 2
assert LETTER_TYPE in [p.permission for p in service_from_db.permissions]
permission = [p for p in service.permissions if p.permission == 'sms'][0]
service.permissions.remove(permission)
dao_update_service(service)
assert Service.query.count() == 1
assert Service.get_history_model().query.count() == 3
service_from_db = Service.query.first()
assert service_from_db.version == 3
assert SMS_TYPE not in [p.permission for p in service_from_db.permissions]
assert len(Service.get_history_model().query.filter_by(name='service_name').all()) == 3
assert Service.get_history_model().query.filter_by(name='service_name').all()[2].version == 3
def test_create_service_and_history_is_transactional(sample_user):
assert Service.query.count() == 0
assert Service.get_history_model().query.count() == 0

32
tests/app/db.py
View File

@@ -1,11 +1,22 @@
from datetime import datetime
import uuid
from app.dao.jobs_dao import dao_create_job
from app.models import (Service, User, Template, Notification, EMAIL_TYPE, LETTER_TYPE,
SMS_TYPE, KEY_TYPE_NORMAL, Job, ServicePermission)
from app.models import (
Service,
User,
Template,
Notification,
ScheduledNotification,
ServicePermission,
Job,
EMAIL_TYPE,
SMS_TYPE,
KEY_TYPE_NORMAL,
)
from app.dao.users_dao import save_model_user
from app.dao.notifications_dao import dao_create_notification
from app.dao.notifications_dao import dao_create_notification, dao_created_scheduled_notification
from app.dao.templates_dao import dao_create_template
from app.dao.services_dao import dao_create_service
from app.dao.service_permissions_dao import dao_add_service_permission
@@ -80,7 +91,9 @@ def create_notification(
client_reference=None,
rate_multiplier=None,
international=False,
phone_prefix=None
phone_prefix=None,
scheduled_for=None,
normalised_to=None
):
if created_at is None:
created_at = datetime.utcnow()
@@ -114,10 +127,19 @@ def create_notification(
'job_row_number': job_row_number,
'rate_multiplier': rate_multiplier,
'international': international,
'phone_prefix': phone_prefix
'phone_prefix': phone_prefix,
'normalised_to': normalised_to
}
notification = Notification(**data)
dao_create_notification(notification)
if scheduled_for:
scheduled_notification = ScheduledNotification(id=uuid.uuid4(),
notification_id=notification.id,
scheduled_for=datetime.strptime(scheduled_for,
"%Y-%m-%d %H:%M"))
if status != 'created':
scheduled_notification.pending = False
dao_created_scheduled_notification(scheduled_notification)
return notification

77
tests/app/notifications/test_process_notification.py
View File

@@ -7,13 +7,14 @@ from sqlalchemy.exc import SQLAlchemyError
from freezegun import freeze_time
from collections import namedtuple
from app.models import Template, Notification, NotificationHistory
from app.models import Template, Notification, NotificationHistory, ScheduledNotification
from app.notifications import SendNotificationToQueueError
from app.notifications.process_notifications import (
create_content_for_notification,
persist_notification,
send_notification_to_queue,
simulated_recipient
simulated_recipient,
persist_scheduled_notification
)
from notifications_utils.recipients import validate_and_format_phone_number, validate_and_format_email_address
from app.utils import cache_key_for_service_template_counter
@@ -339,3 +340,75 @@ def test_persist_notification_with_international_info_does_not_store_for_email(
assert persisted_notification.international is False
assert persisted_notification.phone_prefix is None
assert persisted_notification.rate_multiplier is None
def test_persist_scheduled_notification(sample_notification):
persist_scheduled_notification(sample_notification.id, '2017-05-12 14:15')
scheduled_notification = ScheduledNotification.query.all()
assert len(scheduled_notification) == 1
assert scheduled_notification[0].notification_id == sample_notification.id
assert scheduled_notification[0].scheduled_for == datetime.datetime(2017, 5, 12, 13, 15)
@pytest.mark.parametrize('recipient, expected_recipient_normalised', [
('7900900123', '447900900123'),
('+447900 900 123', '447900900123'),
(' 07700900222', '447700900222'),
('07700900222', '447700900222'),
(' 73122345678', '73122345678'),
('360623400400', '360623400400'),
('-077-00900222-', '447700900222'),
('(360623(400400)', '360623400400')
])
def test_persist_sms_notification_stores_normalised_number(
sample_job,
sample_api_key,
mocker,
recipient,
expected_recipient_normalised
):
persist_notification(
template_id=sample_job.template.id,
template_version=sample_job.template.version,
recipient=recipient,
service=sample_job.service,
personalisation=None,
notification_type='sms',
api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type,
job_id=sample_job.id,
)
persisted_notification = Notification.query.all()[0]
assert persisted_notification.to == recipient
assert persisted_notification.normalised_to == expected_recipient_normalised
@pytest.mark.parametrize('recipient, expected_recipient_normalised', [
('FOO@bar.com', 'foo@bar.com'),
('BAR@foo.com', 'bar@foo.com')
])
def test_persist_email_notification_stores_normalised_email(
sample_job,
sample_api_key,
mocker,
recipient,
expected_recipient_normalised
):
persist_notification(
template_id=sample_job.template.id,
template_version=sample_job.template.version,
recipient=recipient,
service=sample_job.service,
personalisation=None,
notification_type='email',
api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type,
job_id=sample_job.id,
)
persisted_notification = Notification.query.all()[0]
assert persisted_notification.to == recipient
assert persisted_notification.normalised_to == expected_recipient_normalised

445
tests/app/service/test_rest.py
View File

@@ -10,7 +10,7 @@ from freezegun import freeze_time
from app.dao.users_dao import save_model_user
from app.dao.services_dao import dao_remove_user_from_service
from app.models import User, Organisation, DVLA_ORG_LAND_REGISTRY, Rate
from app.models import User, Organisation, DVLA_ORG_LAND_REGISTRY, Rate, ServicePermission
from tests import create_authorization_header
from tests.app.db import create_template
from tests.app.conftest import (
@@ -20,28 +20,30 @@ from tests.app.conftest import (
sample_notification_history as create_notification_history,
sample_notification_with_job
)
from app.models import KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST
from app.models import (
Service, ServicePermission,
KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST,
EMAIL_TYPE, SMS_TYPE, LETTER_TYPE, INTERNATIONAL_SMS_TYPE, INBOUND_SMS_TYPE
)
from tests.app.db import create_user
def test_get_service_list(notify_api, service_factory):
with notify_api.test_request_context():
with notify_api.test_client() as client:
service_factory.get('one')
service_factory.get('two')
service_factory.get('three')
auth_header = create_authorization_header()
response = client.get(
'/service',
headers=[auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 3
assert json_resp['data'][0]['name'] == 'one'
assert json_resp['data'][1]['name'] == 'two'
assert json_resp['data'][2]['name'] == 'three'
def test_get_service_list(client, service_factory):
service_factory.get('one')
service_factory.get('two')
service_factory.get('three')
auth_header = create_authorization_header()
response = client.get(
'/service',
headers=[auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 3
assert json_resp['data'][0]['name'] == 'one'
assert json_resp['data'][1]['name'] == 'two'
assert json_resp['data'][2]['name'] == 'three'
def test_get_service_list_with_only_active_flag(client, service_factory):
@@ -117,17 +119,15 @@ def test_get_service_list_by_user_should_return_empty_list_if_no_services(client
assert len(json_resp['data']) == 0
def test_get_service_list_should_return_empty_list_if_no_services(notify_api, notify_db, notify_db_session):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header()
response = client.get(
'/service',
headers=[auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 0
def test_get_service_list_should_return_empty_list_if_no_services(client):
auth_header = create_authorization_header()
response = client.get(
'/service',
headers=[auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 0
def test_get_service_by_id(client, sample_service):
@@ -147,6 +147,32 @@ def test_get_service_by_id(client, sample_service):
assert json_resp['data']['sms_sender'] == current_app.config['FROM_NUMBER']
def test_get_service_list_has_default_permissions(client, service_factory):
service_factory.get('one')
service_factory.get('two')
service_factory.get('three')
auth_header = create_authorization_header()
response = client.get(
'/service',
headers=[auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 3
assert all([set(json['permissions']) == set([EMAIL_TYPE, SMS_TYPE]) for json in json_resp['data']])
def test_get_service_by_id_has_default_service_permissions(client, sample_service):
auth_header = create_authorization_header()
resp = client.get(
'/service/{}'.format(sample_service.id),
headers=[auth_header]
)
json_resp = json.loads(resp.get_data(as_text=True))
assert set(json_resp['data']['permissions']) == set([EMAIL_TYPE, SMS_TYPE])
def test_get_service_by_id_should_404_if_no_service(notify_api, notify_db):
with notify_api.test_request_context():
with notify_api.test_client() as client:
@@ -216,6 +242,10 @@ def test_create_service(client, sample_user):
assert json_resp['data']['dvla_organisation'] == '001'
assert json_resp['data']['sms_sender'] == current_app.config['FROM_NUMBER']
service_db = Service.query.get(json_resp['data']['id'])
assert service_db.name == 'created service'
assert service_db.sms_sender == current_app.config['FROM_NUMBER']
auth_header_fetch = create_authorization_header()
resp = client.get(
@@ -410,39 +440,194 @@ def test_update_service(client, notify_db, sample_service):
assert result['data']['dvla_organisation'] == DVLA_ORG_LAND_REGISTRY
def test_update_service_flags(notify_api, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header()
resp = client.get(
'/service/{}'.format(sample_service.id),
headers=[auth_header]
)
json_resp = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert json_resp['data']['name'] == sample_service.name
assert json_resp['data']['research_mode'] is False
assert json_resp['data']['can_send_letters'] is False
assert json_resp['data']['can_send_international_sms'] is False
def test_update_service_flags(client, sample_service):
auth_header = create_authorization_header()
resp = client.get(
'/service/{}'.format(sample_service.id),
headers=[auth_header]
)
json_resp = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert json_resp['data']['name'] == sample_service.name
assert json_resp['data']['research_mode'] is False
assert json_resp['data']['can_send_letters'] is False
assert json_resp['data']['can_send_international_sms'] is False
data = {
'research_mode': True,
'can_send_letters': True,
'can_send_international_sms': True,
}
data = {
'research_mode': True,
'can_send_letters': True,
'can_send_international_sms': True,
}
auth_header = create_authorization_header()
auth_header = create_authorization_header()
resp = client.post(
'/service/{}'.format(sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert result['data']['research_mode'] is True
assert result['data']['can_send_letters'] is True
assert result['data']['can_send_international_sms'] is True
resp = client.post(
'/service/{}'.format(sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert result['data']['research_mode'] is True
assert result['data']['can_send_letters'] is True
assert result['data']['can_send_international_sms'] is True
@pytest.fixture(scope='function')
def service_with_no_permissions(notify_db, notify_db_session):
return create_service(notify_db, notify_db_session, permissions=[])
def test_update_service_flags_with_service_without_default_service_permissions(client, service_with_no_permissions):
auth_header = create_authorization_header()
data = {
'can_send_letters': True,
'can_send_international_sms': True,
}
resp = client.post(
'/service/{}'.format(service_with_no_permissions.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert result['data']['can_send_letters'] is True
assert result['data']['can_send_international_sms'] is True
assert set(result['data']['permissions']) == set([LETTER_TYPE, INTERNATIONAL_SMS_TYPE])
def test_update_service_flags_will_remove_service_permissions(client, notify_db, notify_db_session):
auth_header = create_authorization_header()
service = create_service(
notify_db, notify_db_session, permissions=[SMS_TYPE, EMAIL_TYPE, INTERNATIONAL_SMS_TYPE])
assert service.can_send_international_sms is True
data = {
'can_send_international_sms': False
}
resp = client.post(
'/service/{}'.format(service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert result['data']['can_send_international_sms'] is False
permissions = ServicePermission.query.filter_by(service_id=service.id).all()
assert set([p.permission for p in permissions]) == set([SMS_TYPE, EMAIL_TYPE])
def test_update_permissions_will_override_permission_flags(client, service_with_no_permissions):
auth_header = create_authorization_header()
data = {
'permissions': [LETTER_TYPE, INTERNATIONAL_SMS_TYPE]
}
resp = client.post(
'/service/{}'.format(service_with_no_permissions.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert result['data']['can_send_letters'] is True
assert result['data']['can_send_international_sms'] is True
assert set(result['data']['permissions']) == set([LETTER_TYPE, INTERNATIONAL_SMS_TYPE])
def test_update_service_permissions_will_add_service_permissions(client, sample_service):
auth_header = create_authorization_header()
data = {
'permissions': [EMAIL_TYPE, SMS_TYPE, LETTER_TYPE]
}
resp = client.post(
'/service/{}'.format(sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 200
assert set(result['data']['permissions']) == set([SMS_TYPE, EMAIL_TYPE, LETTER_TYPE])
@pytest.mark.parametrize(
'permission_to_add',
[
(EMAIL_TYPE),
(SMS_TYPE),
(INTERNATIONAL_SMS_TYPE),
(LETTER_TYPE),
(INBOUND_SMS_TYPE),
]
)
def test_add_service_permission_will_add_permission(client, service_with_no_permissions, permission_to_add):
auth_header = create_authorization_header()
data = {
'permissions': [permission_to_add]
}
resp = client.post(
'/service/{}'.format(service_with_no_permissions.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
permissions = ServicePermission.query.filter_by(service_id=service_with_no_permissions.id).all()
assert resp.status_code == 200
assert [p.permission for p in permissions] == [permission_to_add]
def test_update_permissions_with_an_invalid_permission_will_raise_error(client, sample_service):
auth_header = create_authorization_header()
invalid_permission = 'invalid_permission'
data = {
'permissions': [EMAIL_TYPE, SMS_TYPE, invalid_permission]
}
resp = client.post(
'/service/{}'.format(sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 400
assert result['result'] == 'error'
assert "Invalid Service Permission: '{}'".format(invalid_permission) in result['message']['permissions']
def test_update_permissions_with_duplicate_permissions_will_raise_error(client, sample_service):
auth_header = create_authorization_header()
data = {
'permissions': [EMAIL_TYPE, SMS_TYPE, LETTER_TYPE, LETTER_TYPE]
}
resp = client.post(
'/service/{}'.format(sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header]
)
result = json.loads(resp.get_data(as_text=True))
assert resp.status_code == 400
assert result['result'] == 'error'
assert "Duplicate Service Permission: ['{}']".format(LETTER_TYPE) in result['message']['permissions']
def test_update_service_research_mode_throws_validation_error(notify_api, sample_service):
@@ -1618,48 +1803,59 @@ def test_get_monthly_billing_usage_returns_empty_list_if_no_notifications(client
def test_search_for_notification_by_to_field(client, notify_db, notify_db_session):
notification1 = create_sample_notification(notify_db, notify_db_session,
to_field="+447700900855")
notification2 = create_sample_notification(notify_db, notify_db_session, to_field="jack@gmail.com")
create_notification = partial(create_sample_notification, notify_db, notify_db_session)
notification1 = create_notification(to_field='+447700900855', normalised_to='447700900855')
notification2 = create_notification(to_field='jack@gmail.com', normalised_to='jack@gmail.com')
response = client.get(
'/service/{}/notifications?to={}'.format(notification1.service_id, 'jack@gmail.com'),
headers=[create_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
response = client.get('/service/{}/notifications?to={}'.format(notification1.service_id, "jack@gmail.com"),
headers=[create_authorization_header()])
assert response.status_code == 200
result = json.loads(response.get_data(as_text=True))
assert len(result["notifications"]) == 1
assert result["notifications"][0]["id"] == str(notification2.id)
assert len(notifications) == 1
assert str(notification2.id) == notifications[0]['id']
def test_search_for_notification_by_to_field_return_empty_list_if_there_is_no_match(
client, notify_db, notify_db_session):
notification1 = create_sample_notification(notify_db, notify_db_session,
to_field="+447700900855")
notification2 = create_sample_notification(notify_db, notify_db_session, to_field="jack@gmail.com")
client, notify_db, notify_db_session
):
create_notification = partial(create_sample_notification, notify_db, notify_db_session)
notification1 = create_notification(to_field='+447700900855')
create_notification(to_field='jack@gmail.com')
response = client.get(
'/service/{}/notifications?to={}'.format(notification1.service_id, '+447700900800'),
headers=[create_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
response = client.get('/service/{}/notifications?to={}'.format(notification1.service_id, "+447700900800"),
headers=[create_authorization_header()])
assert response.status_code == 200
assert len(json.loads(response.get_data(as_text=True))["notifications"]) == 0
assert len(notifications) == 0
def test_search_for_notification_by_to_field_return_multiple_matches(
client, notify_db, notify_db_session):
notification1 = create_sample_notification(notify_db, notify_db_session,
to_field="+447700900855")
notification2 = create_sample_notification(notify_db, notify_db_session,
to_field=" +44 77009 00855 ")
notification3 = create_sample_notification(notify_db, notify_db_session,
to_field="+44770 0900 855")
notification4 = create_sample_notification(notify_db, notify_db_session, to_field="jack@gmail.com")
def test_search_for_notification_by_to_field_return_multiple_matches(client, notify_db, notify_db_session):
create_notification = partial(create_sample_notification, notify_db, notify_db_session)
notification1 = create_notification(to_field='+447700900855', normalised_to='447700900855')
notification2 = create_notification(to_field=' +44 77009 00855 ', normalised_to='447700900855')
notification3 = create_notification(to_field='+44770 0900 855', normalised_to='447700900855')
notification4 = create_notification(to_field='jack@gmail.com', normalised_to='jack@gmail.com')
response = client.get(
'/service/{}/notifications?to={}'.format(notification1.service_id, '+447700900855'),
headers=[create_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
notification_ids = [notification['id'] for notification in notifications]
response = client.get('/service/{}/notifications?to={}'.format(notification1.service_id, "+447700900855"),
headers=[create_authorization_header()])
assert response.status_code == 200
result = json.loads(response.get_data(as_text=True))
assert len(result["notifications"]) == 3
assert str(notification1.id) in [n["id"] for n in result["notifications"]]
assert str(notification2.id) in [n["id"] for n in result["notifications"]]
assert str(notification3.id) in [n["id"] for n in result["notifications"]]
assert len(notifications) == 3
assert str(notification1.id) in notification_ids
assert str(notification2.id) in notification_ids
assert str(notification3.id) in notification_ids
assert str(notification4.id) not in notification_ids
def test_update_service_calls_send_notification_as_service_becomes_live(notify_db, notify_db_session, client, mocker):
@@ -1731,3 +1927,70 @@ def test_update_service_does_not_call_send_notification_when_restricted_not_chan
assert resp.status_code == 200
assert not send_notification_mock.called
def test_update_service_works_when_sms_sender_is_null(sample_service, client, mocker):
sample_service.sms_sender = None
data = {'name': 'new name'}
resp = client.post(
'service/{}'.format(sample_service.id),
data=json.dumps(data),
headers=[create_authorization_header()],
content_type='application/json'
)
assert resp.status_code == 200
# make sure it wasn't changed to not-null under the hood
assert sample_service.sms_sender is None
def test_search_for_notification_by_to_field_filters_by_status(client, notify_db, notify_db_session):
create_notification = partial(
create_sample_notification,
notify_db,
notify_db_session,
to_field='+447700900855',
normalised_to='447700900855'
)
notification1 = create_notification(status='delivered')
create_notification(status='sending')
response = client.get(
'/service/{}/notifications?to={}&status={}'.format(
notification1.service_id, '+447700900855', 'delivered'
),
headers=[create_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
notification_ids = [notification['id'] for notification in notifications]
assert response.status_code == 200
assert len(notifications) == 1
assert str(notification1.id) in notification_ids
def test_search_for_notification_by_to_field_filters_by_statuses(client, notify_db, notify_db_session):
create_notification = partial(
create_sample_notification,
notify_db,
notify_db_session,
to_field='+447700900855',
normalised_to='447700900855'
)
notification1 = create_notification(status='delivered')
notification2 = create_notification(status='sending')
response = client.get(
'/service/{}/notifications?to={}&status={}&status={}'.format(
notification1.service_id, '+447700900855', 'delivered', 'sending'
),
headers=[create_authorization_header()]
)
notifications = json.loads(response.get_data(as_text=True))['notifications']
notification_ids = [notification['id'] for notification in notifications]
assert response.status_code == 200
assert len(notifications) == 2
assert str(notification1.id) in notification_ids
assert str(notification2.id) in notification_ids

14
tests/app/test_utils.py
View File

@@ -4,8 +4,8 @@ import pytest
from app.utils import (
get_london_midnight_in_utc,
get_midnight_for_day_before,
get_utc_time_in_bst
)
convert_utc_time_in_bst,
convert_bst_to_utc)
@pytest.mark.parametrize('date, expected_date', [
@@ -32,7 +32,15 @@ def test_get_midnight_for_day_before_returns_expected_date(date, expected_date):
(datetime(2017, 3, 28, 10, 0), datetime(2017, 3, 28, 11, 0)),
(datetime(2017, 10, 28, 1, 0), datetime(2017, 10, 28, 2, 0)),
(datetime(2017, 10, 29, 1, 0), datetime(2017, 10, 29, 1, 0)),
(datetime(2017, 5, 12, 14), datetime(2017, 5, 12, 15, 0))
])
def test_get_utc_in_bst_returns_expected_date(date, expected_date):
ret_date = get_utc_time_in_bst(date)
ret_date = convert_utc_time_in_bst(date)
assert ret_date == expected_date
def test_convert_bst_to_utc():
bst = "2017-05-12 13:15"
bst_datetime = datetime.strptime(bst, "%Y-%m-%d %H:%M")
utc = convert_bst_to_utc(bst_datetime)
assert utc == datetime(2017, 5, 12, 12, 15)

125
tests/app/v2/notifications/test_get_notifications.py
View File

@@ -4,10 +4,10 @@ from flask import json
from app import DATETIME_FORMAT
from tests import create_authorization_header
from tests.app.conftest import (
sample_notification as create_sample_notification,
sample_template as create_sample_template
)
from tests.app.db import (
create_notification,
create_template,
create_service)
@pytest.mark.parametrize('billable_units, provider', [
@@ -16,12 +16,15 @@ from tests.app.conftest import (
(1, None)
])
def test_get_notification_by_id_returns_200(
client, notify_db, notify_db_session, sample_provider_rate, billable_units, provider
client, billable_units, provider, sample_template
):
sample_notification = create_sample_notification(
notify_db, notify_db_session, billable_units=billable_units, sent_by=provider
)
sample_notification = create_notification(template=sample_template, billable_units=billable_units, sent_by=provider,
scheduled_for="2017-05-12 15:15"
)
another = create_notification(template=sample_template, billable_units=billable_units, sent_by=provider,
scheduled_for="2017-06-12 15:15"
)
auth_header = create_authorization_header(service_id=sample_notification.service_id)
response = client.get(
path='/v2/notifications/{}'.format(sample_notification.id),
@@ -57,18 +60,19 @@ def test_get_notification_by_id_returns_200(
'body': sample_notification.template.content,
"subject": None,
'sent_at': sample_notification.sent_at,
'completed_at': sample_notification.completed_at()
'completed_at': sample_notification.completed_at(),
'scheduled_for': '2017-05-12T14:15:00.000000Z'
}
assert json_response == expected_response
def test_get_notification_by_id_with_placeholders_returns_200(
client, notify_db, notify_db_session, sample_email_template_with_placeholders
client, sample_email_template_with_placeholders
):
sample_notification = create_sample_notification(
notify_db, notify_db_session, template=sample_email_template_with_placeholders, personalisation={"name": "Bob"}
)
sample_notification = create_notification(template=sample_email_template_with_placeholders,
personalisation={"name": "Bob"}
)
auth_header = create_authorization_header(service_id=sample_notification.service_id)
response = client.get(
@@ -105,15 +109,16 @@ def test_get_notification_by_id_with_placeholders_returns_200(
'body': "Hello Bob\nThis is an email from GOV.\u200bUK",
"subject": "Bob",
'sent_at': sample_notification.sent_at,
'completed_at': sample_notification.completed_at()
'completed_at': sample_notification.completed_at(),
'scheduled_for': None
}
assert json_response == expected_response
def test_get_notification_by_reference_returns_200(client, notify_db, notify_db_session):
sample_notification_with_reference = create_sample_notification(
notify_db, notify_db_session, client_reference='some-client-reference')
def test_get_notification_by_reference_returns_200(client, sample_template):
sample_notification_with_reference = create_notification(template=sample_template,
client_reference='some-client-reference')
auth_header = create_authorization_header(service_id=sample_notification_with_reference.service_id)
response = client.get(
@@ -130,6 +135,26 @@ def test_get_notification_by_reference_returns_200(client, notify_db, notify_db_
assert json_response['notifications'][0]['reference'] == "some-client-reference"
def test_get_notifications_returns_scheduled_for(client, sample_template):
sample_notification_with_reference = create_notification(template=sample_template,
client_reference='some-client-reference',
scheduled_for='2017-05-23 17:15')
auth_header = create_authorization_header(service_id=sample_notification_with_reference.service_id)
response = client.get(
path='/v2/notifications?reference={}'.format(sample_notification_with_reference.client_reference),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 200
assert response.headers['Content-type'] == 'application/json'
json_response = json.loads(response.get_data(as_text=True))
assert len(json_response['notifications']) == 1
assert json_response['notifications'][0]['id'] == str(sample_notification_with_reference.id)
assert json_response['notifications'][0]['scheduled_for'] == "2017-05-23T16:15:00.000000Z"
def test_get_notification_by_reference_nonexistent_reference_returns_no_notifications(client, sample_service):
auth_header = create_authorization_header(service_id=sample_service.id)
response = client.get(
@@ -182,8 +207,8 @@ def test_get_notification_by_id_invalid_id(client, sample_notification, id):
}
def test_get_all_notifications_returns_200(client, notify_db, notify_db_session):
notifications = [create_sample_notification(notify_db, notify_db_session) for _ in range(2)]
def test_get_all_notifications_returns_200(client, sample_template):
notifications = [create_notification(template=sample_template) for _ in range(2)]
notification = notifications[-1]
auth_header = create_authorization_header(service_id=notification.service_id)
@@ -208,6 +233,7 @@ def test_get_all_notifications_returns_200(client, notify_db, notify_db_session)
}
assert json_response['notifications'][0]['phone_number'] == "+447700900855"
assert json_response['notifications'][0]['type'] == "sms"
assert not json_response['notifications'][0]['scheduled_for']
def test_get_all_notifications_no_notifications_if_no_notifications(client, sample_service):
@@ -225,13 +251,13 @@ def test_get_all_notifications_no_notifications_if_no_notifications(client, samp
assert len(json_response['notifications']) == 0
def test_get_all_notifications_filter_by_template_type(client, notify_db, notify_db_session):
email_template = create_sample_template(notify_db, notify_db_session, template_type="email")
sms_template = create_sample_template(notify_db, notify_db_session, template_type="sms")
def test_get_all_notifications_filter_by_template_type(client):
service = create_service()
email_template = create_template(service=service, template_type="email")
sms_template = create_template(service=service, template_type="sms")
notification = create_sample_notification(
notify_db, notify_db_session, template=email_template, to_field="don.draper@scdp.biz")
create_sample_notification(notify_db, notify_db_session, template=sms_template)
notification = create_notification(template=email_template, to_field="don.draper@scdp.biz")
create_notification(template=sms_template)
auth_header = create_authorization_header(service_id=notification.service_id)
response = client.get(
@@ -273,9 +299,9 @@ def test_get_all_notifications_filter_by_template_type_invalid_template_type(cli
assert json_response['errors'][0]['message'] == "template_type orange is not one of [sms, email, letter]"
def test_get_all_notifications_filter_by_single_status(client, notify_db, notify_db_session):
notification = create_sample_notification(notify_db, notify_db_session, status="pending")
create_sample_notification(notify_db, notify_db_session)
def test_get_all_notifications_filter_by_single_status(client, sample_template):
notification = create_notification(template=sample_template, status="pending")
create_notification(template=sample_template)
auth_header = create_authorization_header(service_id=notification.service_id)
response = client.get(
@@ -311,12 +337,12 @@ def test_get_all_notifications_filter_by_status_invalid_status(client, sample_no
"delivered, pending, failed, technical-failure, temporary-failure, permanent-failure]"
def test_get_all_notifications_filter_by_multiple_statuses(client, notify_db, notify_db_session):
def test_get_all_notifications_filter_by_multiple_statuses(client, sample_template):
notifications = [
create_sample_notification(notify_db, notify_db_session, status=_status)
create_notification(template=sample_template, status=_status)
for _status in ["created", "pending", "sending"]
]
failed_notification = create_sample_notification(notify_db, notify_db_session, status="permanent-failure")
failed_notification = create_notification(template=sample_template, status="permanent-failure")
auth_header = create_authorization_header(service_id=notifications[0].service_id)
response = client.get(
@@ -338,10 +364,10 @@ def test_get_all_notifications_filter_by_multiple_statuses(client, notify_db, no
assert failed_notification.id not in returned_notification_ids
def test_get_all_notifications_filter_by_failed_status(client, notify_db, notify_db_session):
created_notification = create_sample_notification(notify_db, notify_db_session, status="created")
def test_get_all_notifications_filter_by_failed_status(client, sample_template):
created_notification = create_notification(template=sample_template, status="created")
failed_notifications = [
create_sample_notification(notify_db, notify_db_session, status=_status)
create_notification(template=sample_template, status=_status)
for _status in ["technical-failure", "temporary-failure", "permanent-failure"]
]
@@ -365,9 +391,9 @@ def test_get_all_notifications_filter_by_failed_status(client, notify_db, notify
assert created_notification.id not in returned_notification_ids
def test_get_all_notifications_filter_by_id(client, notify_db, notify_db_session):
older_notification = create_sample_notification(notify_db, notify_db_session)
newer_notification = create_sample_notification(notify_db, notify_db_session)
def test_get_all_notifications_filter_by_id(client, sample_template):
older_notification = create_notification(template=sample_template)
newer_notification = create_notification(template=sample_template)
auth_header = create_authorization_header(service_id=newer_notification.service_id)
response = client.get(
@@ -398,8 +424,8 @@ def test_get_all_notifications_filter_by_id_invalid_id(client, sample_notificati
assert json_response['errors'][0]['message'] == "older_than is not a valid UUID"
def test_get_all_notifications_filter_by_id_no_notifications_if_nonexistent_id(client, notify_db, notify_db_session):
notification = create_sample_notification(notify_db, notify_db_session)
def test_get_all_notifications_filter_by_id_no_notifications_if_nonexistent_id(client, sample_template):
notification = create_notification(template=sample_template)
auth_header = create_authorization_header(service_id=notification.service_id)
response = client.get(
@@ -416,8 +442,8 @@ def test_get_all_notifications_filter_by_id_no_notifications_if_nonexistent_id(c
assert len(json_response['notifications']) == 0
def test_get_all_notifications_filter_by_id_no_notifications_if_last_notification(client, notify_db, notify_db_session):
notification = create_sample_notification(notify_db, notify_db_session)
def test_get_all_notifications_filter_by_id_no_notifications_if_last_notification(client, sample_template):
notification = create_notification(template=sample_template)
auth_header = create_authorization_header(service_id=notification.service_id)
response = client.get(
@@ -433,23 +459,22 @@ def test_get_all_notifications_filter_by_id_no_notifications_if_last_notificatio
assert len(json_response['notifications']) == 0
def test_get_all_notifications_filter_multiple_query_parameters(client, notify_db, notify_db_session):
email_template = create_sample_template(notify_db, notify_db_session, template_type="email")
def test_get_all_notifications_filter_multiple_query_parameters(client, sample_email_template):
# this is the notification we are looking for
older_notification = create_sample_notification(
notify_db, notify_db_session, template=email_template, status="pending")
older_notification = create_notification(
template=sample_email_template, status="pending")
# wrong status
create_sample_notification(notify_db, notify_db_session, template=email_template)
create_notification(template=sample_email_template)
wrong_template = create_template(sample_email_template.service, template_type='sms')
# wrong template
create_sample_notification(notify_db, notify_db_session, status="pending")
create_notification(template=wrong_template, status="pending")
# we only want notifications created before this one
newer_notification = create_sample_notification(notify_db, notify_db_session)
newer_notification = create_notification(template=sample_email_template)
# this notification was created too recently
create_sample_notification(notify_db, notify_db_session, template=email_template, status="pending")
create_notification(template=sample_email_template, status="pending")
auth_header = create_authorization_header(service_id=newer_notification.service_id)
response = client.get(

74
tests/app/v2/notifications/test_notification_schemas.py
View File

@@ -2,6 +2,7 @@ import uuid
import pytest
from flask import json
from freezegun import freeze_time
from jsonschema import ValidationError
from app.v2.notifications.notification_schemas import (
@@ -246,7 +247,8 @@ def valid_email_response():
"id": str(uuid.uuid4()),
"version": 1,
"uri": "http://notify.api/v2/template/id"
}
},
"scheduled_for": ""
}
@@ -262,7 +264,8 @@ def valid_email_response_with_optionals():
"id": str(uuid.uuid4()),
"version": 1,
"uri": "http://notify.api/v2/template/id"
}
},
"schedule_for": "2017-05-12 13:00:00"
}
@@ -346,7 +349,72 @@ def test_get_notifications_response_with_email_and_phone_number():
"subject": "some subject",
"created_at": "2016-01-01",
"sent_at": "2016-01-01",
"completed_at": "2016-01-01"
"completed_at": "2016-01-01",
"schedule_for": ""
}
assert validate(response, get_notification_response) == response
@pytest.mark.parametrize("schema",
[post_email_request_schema, post_sms_request_schema])
@freeze_time("2017-05-12 13:00:00")
def test_post_schema_valid_scheduled_for(schema):
j = {"template_id": str(uuid.uuid4()),
"email_address": "joe@gmail.com",
"scheduled_for": "2017-05-12 13:15"}
if schema == post_email_request_schema:
j.update({"email_address": "joe@gmail.com"})
else:
j.update({"phone_number": "07515111111"})
assert validate(j, schema) == j
@pytest.mark.parametrize("invalid_datetime",
["13:00:00 2017-01-01",
"2017-31-12 13:00:00",
"01-01-2017T14:00:00.0000Z"
])
@pytest.mark.parametrize("schema",
[post_email_request_schema, post_sms_request_schema])
def test_post_email_schema_invalid_scheduled_for(invalid_datetime, schema):
j = {"template_id": str(uuid.uuid4()),
"scheduled_for": invalid_datetime}
if schema == post_email_request_schema:
j.update({"email_address": "joe@gmail.com"})
else:
j.update({"phone_number": "07515111111"})
with pytest.raises(ValidationError) as e:
validate(j, schema)
error = json.loads(str(e.value))
assert error['status_code'] == 400
assert error['errors'] == [{'error': 'ValidationError',
'message': "scheduled_for datetime format is invalid. "
"It must be a valid ISO8601 date time format, "
"https://en.wikipedia.org/wiki/ISO_8601"}]
@freeze_time("2017-05-12 13:00:00")
def test_scheduled_for_raises_validation_error_when_in_the_past():
j = {"phone_number": "07515111111",
"template_id": str(uuid.uuid4()),
"scheduled_for": "2017-05-12 10:00"}
with pytest.raises(ValidationError) as e:
validate(j, post_sms_request_schema)
error = json.loads(str(e.value))
assert error['status_code'] == 400
assert error['errors'] == [{'error': 'ValidationError',
'message': "scheduled_for datetime can not be in the past"}]
@freeze_time("2017-05-12 13:00:00")
def test_scheduled_for_raises_validation_error_when_more_than_24_hours_in_the_future():
j = {"phone_number": "07515111111",
"template_id": str(uuid.uuid4()),
"scheduled_for": "2017-05-13 14:00"}
with pytest.raises(ValidationError) as e:
validate(j, post_sms_request_schema)
error = json.loads(str(e.value))
assert error['status_code'] == 400
assert error['errors'] == [{'error': 'ValidationError',
'message': "scheduled_for datetime can only be 24 hours in the future"}]

157
tests/app/v2/notifications/test_post_notifications.py
View File

@@ -1,6 +1,9 @@
import uuid
import pytest
from freezegun import freeze_time
from app.models import Notification, ScheduledNotification
from flask import json, current_app
from app.models import Notification
@@ -10,39 +13,38 @@ from tests.app.conftest import sample_template as create_sample_template, sample
@pytest.mark.parametrize("reference", [None, "reference_from_client"])
def test_post_sms_notification_returns_201(notify_api, sample_template_with_placeholders, mocker, reference):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocked = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
data = {
'phone_number': '+447700900855',
'template_id': str(sample_template_with_placeholders.id),
'personalisation': {' Name': 'Jo'}
}
if reference:
data.update({"reference": reference})
auth_header = create_authorization_header(service_id=sample_template_with_placeholders.service_id)
def test_post_sms_notification_returns_201(client, sample_template_with_placeholders, mocker, reference):
mocked = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
data = {
'phone_number': '+447700900855',
'template_id': str(sample_template_with_placeholders.id),
'personalisation': {' Name': 'Jo'}
}
if reference:
data.update({"reference": reference})
auth_header = create_authorization_header(service_id=sample_template_with_placeholders.service_id)
response = client.post(
path='/v2/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 201
resp_json = json.loads(response.get_data(as_text=True))
notifications = Notification.query.all()
assert len(notifications) == 1
notification_id = notifications[0].id
assert resp_json['id'] == str(notification_id)
assert resp_json['reference'] == reference
assert resp_json['content']['body'] == sample_template_with_placeholders.content.replace("(( Name))", "Jo")
assert resp_json['content']['from_number'] == current_app.config['FROM_NUMBER']
assert 'v2/notifications/{}'.format(notification_id) in resp_json['uri']
assert resp_json['template']['id'] == str(sample_template_with_placeholders.id)
assert resp_json['template']['version'] == sample_template_with_placeholders.version
assert 'services/{}/templates/{}'.format(sample_template_with_placeholders.service_id,
sample_template_with_placeholders.id) \
in resp_json['template']['uri']
assert mocked.called
response = client.post(
path='/v2/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 201
resp_json = json.loads(response.get_data(as_text=True))
notifications = Notification.query.all()
assert len(notifications) == 1
notification_id = notifications[0].id
assert resp_json['id'] == str(notification_id)
assert resp_json['reference'] == reference
assert resp_json['content']['body'] == sample_template_with_placeholders.content.replace("(( Name))", "Jo")
assert resp_json['content']['from_number'] == current_app.config['FROM_NUMBER']
assert 'v2/notifications/{}'.format(notification_id) in resp_json['uri']
assert resp_json['template']['id'] == str(sample_template_with_placeholders.id)
assert resp_json['template']['version'] == sample_template_with_placeholders.version
assert 'services/{}/templates/{}'.format(sample_template_with_placeholders.service_id,
sample_template_with_placeholders.id) \
in resp_json['template']['uri']
assert not resp_json["scheduled_for"]
assert mocked.called
@pytest.mark.parametrize("notification_type, key_send_to, send_to",
@@ -150,6 +152,7 @@ def test_post_email_notification_returns_201(client, sample_email_template_with_
assert 'services/{}/templates/{}'.format(str(sample_email_template_with_placeholders.service_id),
str(sample_email_template_with_placeholders.id)) \
in resp_json['template']['uri']
assert not resp_json["scheduled_for"]
assert mocked.called
@@ -319,32 +322,76 @@ def test_post_sms_notification_returns_201_if_allowed_to_send_int_sms(notify_db,
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
print(json.loads(response.get_data(as_text=True)))
assert response.status_code == 201
assert response.headers['Content-type'] == 'application/json'
def test_post_sms_should_persist_supplied_sms_number(notify_api, sample_template_with_placeholders, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocked = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
data = {
'phone_number': '+(44) 77009-00855',
'template_id': str(sample_template_with_placeholders.id),
'personalisation': {' Name': 'Jo'}
}
def test_post_sms_should_persist_supplied_sms_number(client, sample_template_with_placeholders, mocker):
mocked = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
data = {
'phone_number': '+(44) 77009-00855',
'template_id': str(sample_template_with_placeholders.id),
'personalisation': {' Name': 'Jo'}
}
auth_header = create_authorization_header(service_id=sample_template_with_placeholders.service_id)
auth_header = create_authorization_header(service_id=sample_template_with_placeholders.service_id)
response = client.post(
path='/v2/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 201
resp_json = json.loads(response.get_data(as_text=True))
notifications = Notification.query.all()
assert len(notifications) == 1
notification_id = notifications[0].id
assert '+(44) 77009-00855' == notifications[0].to
assert resp_json['id'] == str(notification_id)
assert mocked.called
response = client.post(
path='/v2/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 201
resp_json = json.loads(response.get_data(as_text=True))
notifications = Notification.query.all()
assert len(notifications) == 1
notification_id = notifications[0].id
assert '+(44) 77009-00855' == notifications[0].to
assert resp_json['id'] == str(notification_id)
assert mocked.called
@pytest.mark.skip("Once the service can be invited to schedule notifications we can add this test.")
@pytest.mark.parametrize("notification_type, key_send_to, send_to",
[("sms", "phone_number", "07700 900 855"),
("email", "email_address", "sample@email.com")])
@freeze_time("2017-05-14 14:00:00")
def test_post_notification_with_scheduled_for(client, sample_template, sample_email_template,
notification_type, key_send_to, send_to):
data = {
key_send_to: send_to,
'template_id': str(sample_email_template.id) if notification_type == 'email' else str(sample_template.id),
'scheduled_for': '2017-05-14 14:15'
}
auth_header = create_authorization_header(service_id=sample_template.service_id)
response = client.post('/v2/notifications/{}'.format(notification_type),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 201
resp_json = json.loads(response.get_data(as_text=True))
scheduled_notification = ScheduledNotification.query.all()
assert len(scheduled_notification) == 1
assert resp_json["id"] == str(scheduled_notification[0].notification_id)
assert resp_json["scheduled_for"] == '2017-05-14 14:15'
@pytest.mark.parametrize("notification_type, key_send_to, send_to",
[("sms", "phone_number", "07700 900 855"),
("email", "email_address", "sample@email.com")])
@freeze_time("2017-05-14 14:00:00")
def test_post_notification_with_scheduled_for_raises_bad_request(client, sample_template, sample_email_template,
notification_type, key_send_to, send_to):
data = {
key_send_to: send_to,
'template_id': str(sample_email_template.id) if notification_type == 'email' else str(sample_template.id),
'scheduled_for': '2017-05-14 14:15'
}
auth_header = create_authorization_header(service_id=sample_template.service_id)
response = client.post('/v2/notifications/{}'.format(notification_type),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 400
error_json = json.loads(response.get_data(as_text=True))
assert error_json['errors'] == [
{"error": "BadRequestError", "message": 'Your service must be invited to schedule notifications via the API.'}]