darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-01-30 14:31:57 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into implement-v2-send-email

Browse Source
This commit is contained in:
Rebecca Law
2016-11-16 15:18:03 +00:00
parent b0ee09a9f6 80f3bfcb2c
commit a3eeb1a477
12 changed files with 585 additions and 680 deletions
Download Patch File Download Diff File Expand all files Collapse all files

159
tests/app/celery/test_tasks.py
View File

@@ -1,11 +1,8 @@
import uuid
import pytest
from datetime import datetime
from freezegun import freeze_time
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm.exc import NoResultFound
from app import (encryption, DATETIME_FORMAT)
from app.celery import provider_tasks
from app.celery import tasks
@@ -191,8 +188,6 @@ def test_should_not_process_email_job_if_would_exceed_send_limits_inc_today(noti
mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('email'))
mocker.patch('app.celery.tasks.send_email.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
mocker.patch('app.celery.tasks.create_uuid', return_value="uuid")
process_job(job.id)
@@ -210,8 +205,6 @@ def test_should_not_process_email_job_if_would_exceed_send_limits(notify_db, not
mocker.patch('app.celery.tasks.s3.get_job_from_s3')
mocker.patch('app.celery.tasks.send_email.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
mocker.patch('app.celery.tasks.create_uuid', return_value="uuid")
process_job(job.id)
@@ -337,24 +330,16 @@ def test_should_send_template_to_correct_sms_task_and_persist(sample_template_wi
notification = _notification_json(sample_template_with_placeholders,
to="+447234123123", personalisation={"name": "Jo"})
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
notification_id = uuid.uuid4()
mocked_deliver_sms = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
send_sms(
sample_template_with_placeholders.service_id,
notification_id,
uuid.uuid4(),
encryption.encrypt(notification),
datetime.utcnow().strftime(DATETIME_FORMAT)
)
provider_tasks.deliver_sms.apply_async.assert_called_once_with(
[notification_id],
queue="send-sms"
)
persisted_notification = Notification.query.filter_by(id=notification_id).one()
assert persisted_notification.id == notification_id
persisted_notification = Notification.query.one()
assert persisted_notification.to == '+447234123123'
assert persisted_notification.template_id == sample_template_with_placeholders.id
assert persisted_notification.template_version == sample_template_with_placeholders.version
@@ -366,6 +351,10 @@ def test_should_send_template_to_correct_sms_task_and_persist(sample_template_wi
assert persisted_notification.personalisation == {'name': 'Jo'}
assert persisted_notification._personalisation == encryption.encrypt({"name": "Jo"})
assert persisted_notification.notification_type == 'sms'
mocked_deliver_sms.assert_called_once_with(
[persisted_notification.id],
queue="send-sms"
)
def test_should_put_send_sms_task_in_research_mode_queue_if_research_mode_service(notify_db, notify_db_session, mocker):
@@ -377,7 +366,7 @@ def test_should_put_send_sms_task_in_research_mode_queue_if_research_mode_servic
notification = _notification_json(template, to="+447234123123")
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
mocked_deliver_sms = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
notification_id = uuid.uuid4()
@@ -387,11 +376,12 @@ def test_should_put_send_sms_task_in_research_mode_queue_if_research_mode_servic
encryption.encrypt(notification),
datetime.utcnow().strftime(DATETIME_FORMAT)
)
persisted_notification = Notification.query.one()
provider_tasks.deliver_sms.apply_async.assert_called_once_with(
[notification_id],
[persisted_notification.id],
queue="research-mode"
)
assert mocked_deliver_sms.called
def test_should_send_sms_if_restricted_service_and_valid_number(notify_db, notify_db_session, mocker):
@@ -411,13 +401,7 @@ def test_should_send_sms_if_restricted_service_and_valid_number(notify_db, notif
datetime.utcnow().strftime(DATETIME_FORMAT)
)
provider_tasks.deliver_sms.apply_async.assert_called_once_with(
[notification_id],
queue="send-sms"
)
persisted_notification = Notification.query.filter_by(id=notification_id).one()
assert persisted_notification.id == notification_id
persisted_notification = Notification.query.one()
assert persisted_notification.to == '+447700900890'
assert persisted_notification.template_id == template.id
assert persisted_notification.template_version == template.version
@@ -428,17 +412,21 @@ def test_should_send_sms_if_restricted_service_and_valid_number(notify_db, notif
assert not persisted_notification.job_id
assert not persisted_notification.personalisation
assert persisted_notification.notification_type == 'sms'
provider_tasks.deliver_sms.apply_async.assert_called_once_with(
[persisted_notification.id],
queue="send-sms"
)
def test_should_not_send_sms_if_restricted_service_and_invalid_number_with_test_key(notify_db,
notify_db_session,
mocker):
def test_should_send_sms_if_restricted_service_and_non_team_number_with_test_key(notify_db,
notify_db_session,
mocker):
user = sample_user(notify_db, notify_db_session, mobile_numnber="07700 900205")
service = sample_service(notify_db, notify_db_session, user=user, restricted=True)
template = sample_template(notify_db, notify_db_session, service=service)
notification = _notification_json(template, "07700 900849")
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
mocked_deliver_sms = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
notification_id = uuid.uuid4()
send_sms(
@@ -449,18 +437,16 @@ def test_should_not_send_sms_if_restricted_service_and_invalid_number_with_test_
key_type=KEY_TYPE_TEST
)
provider_tasks.deliver_sms.apply_async.assert_called_once_with(
[notification_id],
persisted_notification = Notification.query.one()
mocked_deliver_sms.assert_called_once_with(
[persisted_notification.id],
queue="send-sms"
)
persisted_notification = Notification.query.filter_by(id=notification_id).one()
assert persisted_notification.id == notification_id
def test_should_not_send_email_if_restricted_service_and_invalid_email_address_with_test_key(notify_db,
notify_db_session,
mocker):
def test_should_send_email_if_restricted_service_and_non_team_email_address_with_test_key(notify_db,
notify_db_session,
mocker):
user = sample_user(notify_db, notify_db_session)
service = sample_service(notify_db, notify_db_session, user=user, restricted=True)
template = sample_template(
@@ -468,7 +454,7 @@ def test_should_not_send_email_if_restricted_service_and_invalid_email_address_w
)
notification = _notification_json(template, to="test@example.com")
mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
mocked_deliver_email = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
notification_id = uuid.uuid4()
send_email(
@@ -479,14 +465,12 @@ def test_should_not_send_email_if_restricted_service_and_invalid_email_address_w
key_type=KEY_TYPE_TEST
)
provider_tasks.deliver_email.apply_async.assert_called_once_with(
[notification_id],
persisted_notification = Notification.query.one()
mocked_deliver_email.assert_called_once_with(
[persisted_notification.id],
queue="send-email"
)
persisted_notification = Notification.query.filter_by(id=notification_id).one()
assert persisted_notification.id == notification_id
def test_should_not_send_sms_if_restricted_service_and_invalid_number(notify_db, notify_db_session, mocker):
user = sample_user(notify_db, notify_db_session, mobile_numnber="07700 900205")
@@ -504,8 +488,7 @@ def test_should_not_send_sms_if_restricted_service_and_invalid_number(notify_db,
datetime.utcnow().strftime(DATETIME_FORMAT)
)
assert provider_tasks.deliver_sms.apply_async.called is False
with pytest.raises(NoResultFound):
Notification.query.filter_by(id=notification_id).one()
assert Notification.query.count() == 0
def test_should_not_send_email_if_restricted_service_and_invalid_email_address(notify_db, notify_db_session, mocker):
@@ -524,8 +507,7 @@ def test_should_not_send_email_if_restricted_service_and_invalid_email_address(n
datetime.utcnow().strftime(DATETIME_FORMAT)
)
with pytest.raises(NoResultFound):
Notification.query.filter_by(id=notification_id).one()
assert Notification.query.count() == 0
def test_should_put_send_email_task_in_research_mode_queue_if_research_mode_service(
@@ -550,8 +532,9 @@ def test_should_put_send_email_task_in_research_mode_queue_if_research_mode_serv
datetime.utcnow().strftime(DATETIME_FORMAT)
)
persisted_notification = Notification.query.one()
provider_tasks.deliver_email.apply_async.assert_called_once_with(
[notification_id],
[persisted_notification.id],
queue="research-mode"
)
@@ -565,32 +548,33 @@ def test_should_send_sms_template_to_and_persist_with_job_id(sample_job, sample_
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
notification_id = uuid.uuid4()
now = datetime.utcnow()
send_sms(
sample_job.service.id,
notification_id,
encryption.encrypt(notification),
datetime.utcnow().strftime(DATETIME_FORMAT),
now.strftime(DATETIME_FORMAT),
api_key_id=str(sample_api_key.id),
key_type=KEY_TYPE_NORMAL
)
provider_tasks.deliver_sms.apply_async.assert_called_once_with(
[notification_id],
queue="send-sms"
)
persisted_notification = Notification.query.filter_by(id=notification_id).one()
assert persisted_notification.id == notification_id
persisted_notification = Notification.query.one()
assert persisted_notification.to == '+447234123123'
assert persisted_notification.job_id == sample_job.id
assert persisted_notification.template_id == sample_job.template.id
assert persisted_notification.status == 'created'
assert not persisted_notification.sent_at
assert persisted_notification.created_at <= datetime.utcnow()
assert persisted_notification.created_at <= now
assert not persisted_notification.sent_by
assert persisted_notification.job_row_number == 2
assert persisted_notification.api_key_id == sample_api_key.id
assert persisted_notification.key_type == KEY_TYPE_NORMAL
assert persisted_notification.notification_type == 'sms'
provider_tasks.deliver_sms.apply_async.assert_called_once_with(
[persisted_notification.id],
queue="send-sms"
)
def test_should_not_send_email_if_team_key_and_recipient_not_in_team(sample_email_template_with_placeholders,
sample_team_api_key,
@@ -618,14 +602,13 @@ def test_should_not_send_email_if_team_key_and_recipient_not_in_team(sample_emai
key_type=KEY_TYPE_TEAM
)
with pytest.raises(NoResultFound):
persisted_notification = Notification.query.filter_by(id=notification_id).one()
print(persisted_notification)
assert Notification.query.count() == 0
apply_async.not_called()
def test_should_not_send_sms_if_team_key_and_recipient_not_in_team(notify_db, notify_db_session, mocker):
assert Notification.query.count() == 0
user = sample_user(notify_db, notify_db_session, mobile_numnber="07700 900205")
service = sample_service(notify_db, notify_db_session, user=user, restricted=True)
template = sample_template(notify_db, notify_db_session, service=service)
@@ -644,8 +627,7 @@ def test_should_not_send_sms_if_team_key_and_recipient_not_in_team(notify_db, no
datetime.utcnow().strftime(DATETIME_FORMAT)
)
assert provider_tasks.deliver_sms.apply_async.called is False
with pytest.raises(NoResultFound):
Notification.query.filter_by(id=notification_id).one()
assert Notification.query.count() == 0
def test_should_use_email_template_and_persist(sample_email_template_with_placeholders, sample_api_key, mocker):
@@ -671,11 +653,7 @@ def test_should_use_email_template_and_persist(sample_email_template_with_placeh
key_type=sample_api_key.key_type
)
persisted_notification = Notification.query.filter_by(id=notification_id).one()
provider_tasks.deliver_email.apply_async.assert_called_once_with(
[notification_id], queue='send-email')
assert persisted_notification.id == notification_id
persisted_notification = Notification.query.one()
assert persisted_notification.to == 'my_email@my_email.com'
assert persisted_notification.template_id == sample_email_template_with_placeholders.id
assert persisted_notification.template_version == sample_email_template_with_placeholders.version
@@ -690,6 +668,9 @@ def test_should_use_email_template_and_persist(sample_email_template_with_placeh
assert persisted_notification.key_type == KEY_TYPE_NORMAL
assert persisted_notification.notification_type == 'email'
provider_tasks.deliver_email.apply_async.assert_called_once_with(
[persisted_notification.id], queue='send-email')
def test_send_email_should_use_template_version_from_job_not_latest(sample_email_template, mocker):
notification = _notification_json(sample_email_template, 'my_email@my_email.com')
@@ -701,19 +682,15 @@ def test_send_email_should_use_template_version_from_job_not_latest(sample_email
dao_update_template(sample_email_template)
t = dao_get_template_by_id(sample_email_template.id)
assert t.version > version_on_notification
notification_id = uuid.uuid4()
now = datetime.utcnow()
send_email(
sample_email_template.service_id,
notification_id,
uuid.uuid4(),
encryption.encrypt(notification),
now.strftime(DATETIME_FORMAT)
)
provider_tasks.deliver_email.apply_async.assert_called_once_with([notification_id], queue='send-email')
persisted_notification = Notification.query.filter_by(id=notification_id).one()
assert persisted_notification.id == notification_id
persisted_notification = Notification.query.one()
assert persisted_notification.to == 'my_email@my_email.com'
assert persisted_notification.template_id == sample_email_template.id
assert persisted_notification.template_version == version_on_notification
@@ -722,6 +699,7 @@ def test_send_email_should_use_template_version_from_job_not_latest(sample_email
assert persisted_notification.status == 'created'
assert not persisted_notification.sent_by
assert persisted_notification.notification_type == 'email'
provider_tasks.deliver_email.apply_async.assert_called_once_with([persisted_notification.id], queue='send-email')
def test_should_use_email_template_subject_placeholders(sample_email_template_with_placeholders, mocker):
@@ -737,18 +715,18 @@ def test_should_use_email_template_subject_placeholders(sample_email_template_wi
encryption.encrypt(notification),
now.strftime(DATETIME_FORMAT)
)
provider_tasks.deliver_email.apply_async.assert_called_once_with(
[notification_id], queue='send-email'
)
persisted_notification = Notification.query.filter_by(id=notification_id).one()
assert persisted_notification.id == notification_id
persisted_notification = Notification.query.one()
assert persisted_notification.to == 'my_email@my_email.com'
assert persisted_notification.template_id == sample_email_template_with_placeholders.id
assert persisted_notification.status == 'created'
assert persisted_notification.created_at == now
assert not persisted_notification.sent_by
assert persisted_notification.personalisation == {"name": "Jo"}
assert not persisted_notification.reference
assert persisted_notification.notification_type == 'email'
provider_tasks.deliver_email.apply_async.assert_called_once_with(
[persisted_notification.id], queue='send-email'
)
def test_should_use_email_template_and_persist_without_personalisation(sample_email_template, mocker):
@@ -764,10 +742,7 @@ def test_should_use_email_template_and_persist_without_personalisation(sample_em
encryption.encrypt(notification),
now.strftime(DATETIME_FORMAT)
)
provider_tasks.deliver_email.apply_async.assert_called_once_with([notification_id], queue='send-email')
persisted_notification = Notification.query.filter_by(id=notification_id).one()
assert persisted_notification.id == notification_id
persisted_notification = Notification.query.one()
assert persisted_notification.to == 'my_email@my_email.com'
assert persisted_notification.template_id == sample_email_template.id
assert persisted_notification.created_at == now
@@ -777,6 +752,8 @@ def test_should_use_email_template_and_persist_without_personalisation(sample_em
assert not persisted_notification.personalisation
assert not persisted_notification.reference
assert persisted_notification.notification_type == 'email'
provider_tasks.deliver_email.apply_async.assert_called_once_with([persisted_notification.id],
queue='send-email')
def test_send_sms_should_go_to_retry_queue_if_database_errors(sample_template, mocker):
@@ -786,7 +763,7 @@ def test_send_sms_should_go_to_retry_queue_if_database_errors(sample_template, m
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
mocker.patch('app.celery.tasks.send_sms.retry', side_effect=Exception())
mocker.patch('app.celery.tasks.dao_create_notification', side_effect=expected_exception)
mocker.patch('app.notifications.process_notifications.dao_create_notification', side_effect=expected_exception)
now = datetime.utcnow()
notification_id = uuid.uuid4()
@@ -801,9 +778,7 @@ def test_send_sms_should_go_to_retry_queue_if_database_errors(sample_template, m
assert provider_tasks.deliver_sms.apply_async.called is False
tasks.send_sms.retry.assert_called_with(exc=expected_exception, queue='retry')
with pytest.raises(NoResultFound) as e:
Notification.query.filter_by(id=notification_id).one()
assert 'No row was found for one' in str(e.value)
assert Notification.query.count() == 0
def test_send_email_should_go_to_retry_queue_if_database_errors(sample_email_template, mocker):
@@ -813,7 +788,7 @@ def test_send_email_should_go_to_retry_queue_if_database_errors(sample_email_tem
mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
mocker.patch('app.celery.tasks.send_email.retry', side_effect=Exception())
mocker.patch('app.celery.tasks.dao_create_notification', side_effect=expected_exception)
mocker.patch('app.notifications.process_notifications.dao_create_notification', side_effect=expected_exception)
now = datetime.utcnow()
notification_id = uuid.uuid4()
@@ -825,9 +800,7 @@ def test_send_email_should_go_to_retry_queue_if_database_errors(sample_email_tem
encryption.encrypt(notification),
now.strftime(DATETIME_FORMAT)
)
assert provider_tasks.deliver_email.apply_async.called is False
assert not provider_tasks.deliver_email.apply_async.called
tasks.send_email.retry.assert_called_with(exc=expected_exception, queue='retry')
with pytest.raises(NoResultFound) as e:
Notification.query.filter_by(id=notification_id).one()
assert 'No row was found for one' in str(e.value)
assert Notification.query.count() == 0

2
tests/app/conftest.py
View File

@@ -180,7 +180,7 @@ def sample_template(notify_db,
'created_by': created_by,
'archived': archived
}
if template_type == 'email':
if template_type in ['email', 'letter']:
data.update({
'subject': subject_line
})

28
tests/app/dao/test_templates_dao.py
View File

@@ -10,31 +10,21 @@ from app.models import Template, TemplateHistory
import pytest
def test_create_template(sample_service, sample_user):
@pytest.mark.parametrize('template_type, subject', [
('sms', None),
('email', 'subject'),
('letter', 'subject'),
])
def test_create_template(sample_service, sample_user, template_type, subject):
data = {
'name': 'Sample Template',
'template_type': "sms",
'content': "Template content",
'service': sample_service,
'created_by': sample_user
}
template = Template(**data)
dao_create_template(template)
assert Template.query.count() == 1
assert len(dao_get_all_templates_for_service(sample_service.id)) == 1
assert dao_get_all_templates_for_service(sample_service.id)[0].name == 'Sample Template'
def test_create_email_template(sample_service, sample_user):
data = {
'name': 'Sample Template',
'template_type': "email",
'subject': "subject",
'template_type': template_type,
'content': "Template content",
'service': sample_service,
'created_by': sample_user
}
if subject:
data.update({'subject': subject})
template = Template(**data)
dao_create_template(template)

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

@@ -1,3 +1,5 @@
import datetime
import pytest
from boto3.exceptions import Boto3Error
from sqlalchemy.exc import SQLAlchemyError
@@ -46,21 +48,44 @@ def test_persist_notification_creates_and_save_to_db(sample_template, sample_api
assert NotificationHistory.query.count() == 1
def test_persist_notification_throws_exception_when_missing_template(sample_template, sample_api_key):
def test_persist_notification_throws_exception_when_missing_template(sample_api_key):
assert Notification.query.count() == 0
assert NotificationHistory.query.count() == 0
with pytest.raises(SQLAlchemyError):
persist_notification(template_id=None,
template_version=None,
recipient='+447111111111',
service_id=sample_template.service.id,
personalisation=None, notification_type='sms',
service_id=sample_api_key.service_id,
personalisation=None,
notification_type='sms',
api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type)
assert Notification.query.count() == 0
assert NotificationHistory.query.count() == 0
def test_persist_notification_with_job_and_created(sample_job, sample_api_key):
assert Notification.query.count() == 0
assert NotificationHistory.query.count() == 0
created_at = datetime.datetime(2016, 11, 11, 16, 8, 18)
persist_notification(template_id=sample_job.template.id,
template_version=sample_job.template.version,
recipient='+447111111111',
service_id=sample_job.service.id,
personalisation=None, notification_type='sms',
api_key_id=sample_api_key.id,
key_type=sample_api_key.key_type,
created_at=created_at,
job_id=sample_job.id,
job_row_number=10)
assert Notification.query.count() == 1
assert NotificationHistory.query.count() == 1
persisted_notification = Notification.query.all()[0]
assert persisted_notification.job_id == sample_job.id
assert persisted_notification.job_row_number == 10
assert persisted_notification.created_at == created_at
@pytest.mark.parametrize('research_mode, queue, notification_type, key_type',
[(True, 'research-mode', 'sms', 'normal'),
(True, 'research-mode', 'email', 'normal'),

38
tests/app/service/test_deactivate.py
View File

@@ -1,11 +1,14 @@
import uuid
from unittest import mock
from datetime import datetime
import pytest
from freezegun import freeze_time
from app import db
from app.models import Service, TemplateHistory, ApiKey
from app.models import Service
from app.dao.services_dao import dao_deactive_service
from app.dao.api_key_dao import expire_api_key
from app.dao.templates_dao import dao_update_template
from tests import create_authorization_header, unwrap_function
from tests.app.conftest import (
@@ -79,6 +82,37 @@ def test_deactivating_service_creates_history(deactivated_service):
assert history.active is False
@pytest.fixture
def deactivated_service_with_deleted_stuff(client, notify_db, notify_db_session, sample_service):
with freeze_time('2001-01-01'):
template = create_template(notify_db, notify_db_session, template_name='a')
api_key = create_api_key(notify_db, notify_db_session)
expire_api_key(sample_service.id, api_key.id)
template.archived = True
dao_update_template(template)
with freeze_time('2002-02-02'):
auth_header = create_authorization_header()
response = client.post('/service/{}/deactivate'.format(sample_service.id), headers=[auth_header])
assert response.status_code == 204
assert response.data == b''
return sample_service
def test_deactivating_service_doesnt_affect_existing_archived_templates(deactivated_service_with_deleted_stuff):
assert deactivated_service_with_deleted_stuff.templates[0].archived is True
assert deactivated_service_with_deleted_stuff.templates[0].updated_at == datetime(2001, 1, 1, 0, 0, 0)
assert deactivated_service_with_deleted_stuff.templates[0].version == 2
def test_deactivating_service_doesnt_affect_existing_revoked_api_keys(deactivated_service_with_deleted_stuff):
assert deactivated_service_with_deleted_stuff.api_keys[0].expiry_date == datetime(2001, 1, 1, 0, 0, 0)
assert deactivated_service_with_deleted_stuff.api_keys[0].version == 2
def test_deactivating_service_rolls_back_everything_on_error(sample_service, sample_api_key, sample_template):
unwrapped_deactive_service = unwrap_function(dao_deactive_service)

758
tests/app/template/test_rest.py
View File

@@ -1,3 +1,4 @@
import pytest
import json
import random
import string
@@ -8,322 +9,287 @@ from tests.app.conftest import sample_template as create_sample_template
from app.dao.templates_dao import dao_get_template_by_id
def test_should_create_a_new_sms_template_for_a_service(notify_api, sample_user, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'name': 'my template',
'template_type': 'sms',
'content': 'template <b>content</b>',
'service': str(sample_service.id),
'created_by': str(sample_user.id)
}
data = json.dumps(data)
auth_header = create_authorization_header()
@pytest.mark.parametrize('template_type, subject', [
('sms', None),
('email', 'subject'),
('letter', 'subject'),
])
def test_should_create_a_new_template_for_a_service(
client, sample_user, sample_service, template_type, subject
):
data = {
'name': 'my template',
'template_type': template_type,
'content': 'template <b>content</b>',
'service': str(sample_service.id),
'created_by': str(sample_user.id)
}
if subject:
data.update({'subject': subject})
data = json.dumps(data)
auth_header = create_authorization_header()
response = client.post(
'/service/{}/template'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
assert response.status_code == 201
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['data']['name'] == 'my template'
assert json_resp['data']['template_type'] == 'sms'
assert json_resp['data']['content'] == 'template content'
assert json_resp['data']['service'] == str(sample_service.id)
assert json_resp['data']['id']
assert json_resp['data']['version'] == 1
assert not json_resp['data']['subject']
response = client.post(
'/service/{}/template'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
assert response.status_code == 201
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['data']['name'] == 'my template'
assert json_resp['data']['template_type'] == template_type
assert json_resp['data']['content'] == 'template content'
assert json_resp['data']['service'] == str(sample_service.id)
assert json_resp['data']['id']
assert json_resp['data']['version'] == 1
if subject:
assert json_resp['data']['subject'] == 'subject'
else:
assert not json_resp['data']['subject']
def test_should_create_a_new_email_template_for_a_service(notify_api, sample_user, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'name': 'my template',
'template_type': 'email',
'subject': 'subject',
'content': 'template <b>content</b>',
'service': str(sample_service.id),
'created_by': str(sample_user.id)
}
data = json.dumps(data)
auth_header = create_authorization_header()
def test_should_be_error_if_service_does_not_exist_on_create(client, sample_user, fake_uuid):
data = {
'name': 'my template',
'template_type': 'sms',
'content': 'template content',
'service': fake_uuid,
'created_by': str(sample_user.id)
}
data = json.dumps(data)
auth_header = create_authorization_header()
response = client.post(
'/service/{}/template'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
assert response.status_code == 201
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['data']['name'] == 'my template'
assert json_resp['data']['template_type'] == 'email'
assert json_resp['data']['content'] == 'template content'
assert json_resp['data']['service'] == str(sample_service.id)
assert json_resp['data']['subject'] == 'subject'
assert json_resp['data']['version'] == 1
assert json_resp['data']['id']
response = client.post(
'/service/{}/template'.format(fake_uuid),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 404
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'No result found'
def test_should_be_error_if_service_does_not_exist_on_create(notify_api, sample_user, fake_uuid):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'name': 'my template',
'template_type': 'sms',
'content': 'template content',
'service': fake_uuid,
'created_by': str(sample_user.id)
}
data = json.dumps(data)
auth_header = create_authorization_header()
def test_should_error_if_created_by_missing(client, sample_user, sample_service):
service_id = str(sample_service.id)
data = {
'name': 'my template',
'template_type': 'sms',
'content': 'template content',
'service': service_id
}
data = json.dumps(data)
auth_header = create_authorization_header()
response = client.post(
'/service/{}/template'.format(fake_uuid),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 404
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'No result found'
response = client.post(
'/service/{}/template'.format(service_id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert json_resp['result'] == 'error'
def test_should_error_if_created_by_missing(notify_api, sample_user, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
service_id = str(sample_service.id)
data = {
'name': 'my template',
'template_type': 'sms',
'content': 'template content',
'service': service_id
}
data = json.dumps(data)
auth_header = create_authorization_header()
def test_should_be_error_if_service_does_not_exist_on_update(client, fake_uuid):
data = {
'name': 'my template'
}
data = json.dumps(data)
auth_header = create_authorization_header()
response = client.post(
'/service/{}/template'.format(service_id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert json_resp['result'] == 'error'
response = client.post(
'/service/{}/template/{}'.format(fake_uuid, fake_uuid),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 404
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'No result found'
def test_should_be_error_if_service_does_not_exist_on_update(notify_api, fake_uuid):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'name': 'my template'
}
data = json.dumps(data)
auth_header = create_authorization_header()
@pytest.mark.parametrize('template_type', ['email', 'letter'])
def test_must_have_a_subject_on_an_email_or_letter_template(client, sample_user, sample_service, template_type):
data = {
'name': 'my template',
'template_type': template_type,
'content': 'template content',
'service': str(sample_service.id),
'created_by': str(sample_user.id)
}
data = json.dumps(data)
auth_header = create_authorization_header()
response = client.post(
'/service/{}/template/{}'.format(fake_uuid, fake_uuid),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 404
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'No result found'
response = client.post(
'/service/{}/template'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert json_resp['result'] == 'error'
assert json_resp['message'] == {'subject': ['Invalid template subject']}
def test_must_have_a_subject_on_an_email_template(notify_api, sample_user, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'name': 'my template',
'template_type': 'email',
'content': 'template content',
'service': str(sample_service.id),
'created_by': str(sample_user.id)
}
data = json.dumps(data)
auth_header = create_authorization_header()
def test_update_should_update_a_template(client, sample_user, sample_template):
data = {
'content': 'my template has new content <script type="text/javascript">alert("foo")</script>',
'created_by': str(sample_user.id)
}
data = json.dumps(data)
auth_header = create_authorization_header()
response = client.post(
'/service/{}/template'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert json_resp['result'] == 'error'
assert json_resp['message'] == {'subject': ['Invalid template subject']}
update_response = client.post(
'/service/{}/template/{}'.format(sample_template.service_id, sample_template.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
assert update_response.status_code == 200
update_json_resp = json.loads(update_response.get_data(as_text=True))
assert update_json_resp['data']['content'] == 'my template has new content alert("foo")'
assert update_json_resp['data']['name'] == sample_template.name
assert update_json_resp['data']['template_type'] == sample_template.template_type
assert update_json_resp['data']['version'] == 2
def test_update_should_update_a_template(notify_api, sample_user, sample_template):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'content': 'my template has new content <script type="text/javascript">alert("foo")</script>',
'created_by': str(sample_user.id)
}
data = json.dumps(data)
auth_header = create_authorization_header()
def test_should_be_able_to_archive_template(client, sample_template):
data = {
'name': sample_template.name,
'template_type': sample_template.template_type,
'content': sample_template.content,
'archived': True,
'service': str(sample_template.service.id),
'created_by': str(sample_template.created_by.id)
}
update_response = client.post(
'/service/{}/template/{}'.format(sample_template.service_id, sample_template.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
json_data = json.dumps(data)
assert update_response.status_code == 200
update_json_resp = json.loads(update_response.get_data(as_text=True))
assert update_json_resp['data']['content'] == 'my template has new content alert("foo")'
assert update_json_resp['data']['name'] == sample_template.name
assert update_json_resp['data']['template_type'] == sample_template.template_type
assert update_json_resp['data']['version'] == 2
auth_header = create_authorization_header()
resp = client.post(
'/service/{}/template/{}'.format(sample_template.service.id, sample_template.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=json_data
)
assert resp.status_code == 200
assert Template.query.first().archived
def test_should_be_able_to_archive_template(notify_api, sample_template):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'name': sample_template.name,
'template_type': sample_template.template_type,
'content': sample_template.content,
'archived': True,
'service': str(sample_template.service.id),
'created_by': str(sample_template.created_by.id)
}
def test_should_be_able_to_get_all_templates_for_a_service(client, sample_user, sample_service):
data = {
'name': 'my template 1',
'template_type': 'email',
'subject': 'subject 1',
'content': 'template content',
'service': str(sample_service.id),
'created_by': str(sample_user.id)
}
data_1 = json.dumps(data)
data = {
'name': 'my template 2',
'template_type': 'email',
'subject': 'subject 2',
'content': 'template content',
'service': str(sample_service.id),
'created_by': str(sample_user.id)
}
data_2 = json.dumps(data)
auth_header = create_authorization_header()
client.post(
'/service/{}/template'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data_1
)
auth_header = create_authorization_header()
json_data = json.dumps(data)
client.post(
'/service/{}/template'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data_2
)
auth_header = create_authorization_header()
auth_header = create_authorization_header()
resp = client.post(
'/service/{}/template/{}'.format(sample_template.service.id, sample_template.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=json_data
)
response = client.get(
'/service/{}/template'.format(sample_service.id),
headers=[auth_header]
)
assert resp.status_code == 200
assert Template.query.first().archived
assert response.status_code == 200
update_json_resp = json.loads(response.get_data(as_text=True))
assert update_json_resp['data'][0]['name'] == 'my template 2'
assert update_json_resp['data'][0]['version'] == 1
assert update_json_resp['data'][0]['created_at']
assert update_json_resp['data'][1]['name'] == 'my template 1'
assert update_json_resp['data'][1]['version'] == 1
assert update_json_resp['data'][1]['created_at']
def test_should_be_able_to_get_all_templates_for_a_service(notify_api, sample_user, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'name': 'my template 1',
'template_type': 'email',
'subject': 'subject 1',
'content': 'template content',
'service': str(sample_service.id),
'created_by': str(sample_user.id)
}
data_1 = json.dumps(data)
data = {
'name': 'my template 2',
'template_type': 'email',
'subject': 'subject 2',
'content': 'template content',
'service': str(sample_service.id),
'created_by': str(sample_user.id)
}
data_2 = json.dumps(data)
auth_header = create_authorization_header()
client.post(
'/service/{}/template'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data_1
)
auth_header = create_authorization_header()
def test_should_get_only_templates_for_that_service(client, sample_user, service_factory):
client.post(
'/service/{}/template'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data_2
)
service_1 = service_factory.get('service 1', email_from='service.1')
service_2 = service_factory.get('service 2', email_from='service.2')
auth_header = create_authorization_header()
auth_header_1 = create_authorization_header()
response = client.get(
'/service/{}/template'.format(sample_service.id),
headers=[auth_header]
)
response_1 = client.get(
'/service/{}/template'.format(service_1.id),
headers=[auth_header_1]
)
assert response.status_code == 200
update_json_resp = json.loads(response.get_data(as_text=True))
assert update_json_resp['data'][0]['name'] == 'my template 2'
assert update_json_resp['data'][0]['version'] == 1
assert update_json_resp['data'][0]['created_at']
assert update_json_resp['data'][1]['name'] == 'my template 1'
assert update_json_resp['data'][1]['version'] == 1
assert update_json_resp['data'][1]['created_at']
auth_header_2 = create_authorization_header()
response_2 = client.get(
'/service/{}/template'.format(service_2.id),
headers=[auth_header_2]
)
def test_should_get_only_templates_for_that_service(notify_api, sample_user, service_factory):
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert response_1.status_code == 200
assert response_2.status_code == 200
service_1 = service_factory.get('service 1', email_from='service.1')
service_2 = service_factory.get('service 2', email_from='service.2')
json_resp_1 = json.loads(response_1.get_data(as_text=True))
json_resp_2 = json.loads(response_2.get_data(as_text=True))
auth_header_1 = create_authorization_header()
assert len(json_resp_1['data']) == 1
assert len(json_resp_2['data']) == 1
response_1 = client.get(
'/service/{}/template'.format(service_1.id),
headers=[auth_header_1]
)
data = {
'name': 'my template 2',
'template_type': 'email',
'subject': 'subject 2',
'content': 'template content',
'service': str(service_1.id),
'created_by': str(sample_user.id)
}
data = json.dumps(data)
create_auth_header = create_authorization_header()
resp = client.post(
'/service/{}/template'.format(service_1.id),
headers=[('Content-Type', 'application/json'), create_auth_header],
data=data
)
auth_header_2 = create_authorization_header()
response_3 = client.get(
'/service/{}/template'.format(service_1.id),
headers=[auth_header_1]
)
response_2 = client.get(
'/service/{}/template'.format(service_2.id),
headers=[auth_header_2]
)
response_4 = client.get(
'/service/{}/template'.format(service_2.id),
headers=[auth_header_2]
)
assert response_1.status_code == 200
assert response_2.status_code == 200
assert response_3.status_code == 200
assert response_4.status_code == 200
json_resp_1 = json.loads(response_1.get_data(as_text=True))
json_resp_2 = json.loads(response_2.get_data(as_text=True))
json_resp_3 = json.loads(response_3.get_data(as_text=True))
json_resp_4 = json.loads(response_4.get_data(as_text=True))
assert len(json_resp_1['data']) == 1
assert len(json_resp_2['data']) == 1
data = {
'name': 'my template 2',
'template_type': 'email',
'subject': 'subject 2',
'content': 'template content',
'service': str(service_1.id),
'created_by': str(sample_user.id)
}
data = json.dumps(data)
create_auth_header = create_authorization_header()
resp = client.post(
'/service/{}/template'.format(service_1.id),
headers=[('Content-Type', 'application/json'), create_auth_header],
data=data
)
response_3 = client.get(
'/service/{}/template'.format(service_1.id),
headers=[auth_header_1]
)
response_4 = client.get(
'/service/{}/template'.format(service_2.id),
headers=[auth_header_2]
)
assert response_3.status_code == 200
assert response_4.status_code == 200
json_resp_3 = json.loads(response_3.get_data(as_text=True))
json_resp_4 = json.loads(response_4.get_data(as_text=True))
assert len(json_resp_3['data']) == 2
assert len(json_resp_4['data']) == 1
assert len(json_resp_3['data']) == 2
assert len(json_resp_4['data']) == 1
@pytest.mark.parametrize(
@@ -337,34 +303,38 @@ def test_should_get_only_templates_for_that_service(notify_api, sample_user, ser
None,
'hello ((name)) we’ve received your ((thing))',
'sms'
),
(
'about your ((thing))',
'hello ((name)) we’ve received your ((thing))',
'letter'
)
]
)
def test_should_get_a_single_template(
notify_db,
notify_api,
client,
sample_user,
service_factory,
subject,
content,
template_type
):
with notify_api.test_request_context(), notify_api.test_client() as client:
template = create_sample_template(
notify_db, notify_db.session, subject_line=subject, content=content, template_type=template_type
)
template = create_sample_template(
notify_db, notify_db.session, subject_line=subject, content=content, template_type=template_type
)
response = client.get(
'/service/{}/template/{}'.format(template.service.id, template.id),
headers=[create_authorization_header()]
)
response = client.get(
'/service/{}/template/{}'.format(template.service.id, template.id),
headers=[create_authorization_header()]
)
data = json.loads(response.get_data(as_text=True))['data']
data = json.loads(response.get_data(as_text=True))['data']
assert response.status_code == 200
assert data['content'] == content
assert data['subject'] == subject
assert response.status_code == 200
assert data['content'] == content
assert data['subject'] == subject
@pytest.mark.parametrize(
@@ -403,7 +373,7 @@ def test_should_get_a_single_template(
)
def test_should_preview_a_single_template(
notify_db,
notify_api,
client,
sample_user,
service_factory,
subject,
@@ -413,144 +383,136 @@ def test_should_preview_a_single_template(
expected_content,
expected_error
):
with notify_api.test_request_context(), notify_api.test_client() as client:
template = create_sample_template(
notify_db, notify_db.session, subject_line=subject, content=content, template_type='email'
)
template = create_sample_template(
notify_db, notify_db.session, subject_line=subject, content=content, template_type='email'
)
response = client.get(
path.format(template.service.id, template.id),
headers=[create_authorization_header()]
)
response = client.get(
path.format(template.service.id, template.id),
headers=[create_authorization_header()]
)
content = json.loads(response.get_data(as_text=True))
content = json.loads(response.get_data(as_text=True))
if expected_error:
assert response.status_code == 400
assert content['message']['template'] == [expected_error]
else:
assert response.status_code == 200
assert content['content'] == expected_content
assert content['subject'] == expected_subject
if expected_error:
assert response.status_code == 400
assert content['message']['template'] == [expected_error]
else:
assert response.status_code == 200
assert content['content'] == expected_content
assert content['subject'] == expected_subject
def test_should_return_empty_array_if_no_templates_for_service(notify_api, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
def test_should_return_empty_array_if_no_templates_for_service(client, sample_service):
auth_header = create_authorization_header()
auth_header = create_authorization_header()
response = client.get(
'/service/{}/template'.format(sample_service.id),
headers=[auth_header]
)
response = client.get(
'/service/{}/template'.format(sample_service.id),
headers=[auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 0
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 0
def test_should_return_404_if_no_templates_for_service_with_id(notify_api, sample_service, fake_uuid):
with notify_api.test_request_context():
with notify_api.test_client() as client:
def test_should_return_404_if_no_templates_for_service_with_id(client, sample_service, fake_uuid):
auth_header = create_authorization_header()
auth_header = create_authorization_header()
response = client.get(
'/service/{}/template/{}'.format(sample_service.id, fake_uuid),
headers=[auth_header]
)
response = client.get(
'/service/{}/template/{}'.format(sample_service.id, fake_uuid),
headers=[auth_header]
)
assert response.status_code == 404
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'No result found'
assert response.status_code == 404
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['result'] == 'error'
assert json_resp['message'] == 'No result found'
def test_create_400_for_over_limit_content(notify_api, sample_user, sample_service, fake_uuid):
with notify_api.test_request_context():
with notify_api.test_client() as client:
limit = notify_api.config.get('SMS_CHAR_COUNT_LIMIT')
content = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(limit + 1))
data = {
'name': 'too big template',
'template_type': 'sms',
'content': content,
'service': str(sample_service.id),
'created_by': str(sample_user.id)
}
data = json.dumps(data)
auth_header = create_authorization_header()
def test_create_400_for_over_limit_content(client, notify_api, sample_user, sample_service, fake_uuid):
response = client.post(
'/service/{}/template'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
assert response.status_code == 400
json_resp = json.loads(response.get_data(as_text=True))
assert (
'Content has a character count greater than the limit of {}'
).format(limit) in json_resp['message']['content']
limit = notify_api.config.get('SMS_CHAR_COUNT_LIMIT')
content = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(limit + 1))
data = {
'name': 'too big template',
'template_type': 'sms',
'content': content,
'service': str(sample_service.id),
'created_by': str(sample_user.id)
}
data = json.dumps(data)
auth_header = create_authorization_header()
response = client.post(
'/service/{}/template'.format(sample_service.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=data
)
assert response.status_code == 400
json_resp = json.loads(response.get_data(as_text=True))
assert (
'Content has a character count greater than the limit of {}'
).format(limit) in json_resp['message']['content']
def test_update_400_for_over_limit_content(notify_api, sample_user, sample_template):
with notify_api.test_request_context():
with notify_api.test_client() as client:
limit = notify_api.config.get('SMS_CHAR_COUNT_LIMIT')
json_data = json.dumps({
'content': ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(limit + 1)),
'created_by': str(sample_user.id)
})
auth_header = create_authorization_header()
resp = client.post(
'/service/{}/template/{}'.format(sample_template.service.id, sample_template.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=json_data
)
assert resp.status_code == 400
json_resp = json.loads(resp.get_data(as_text=True))
assert (
'Content has a character count greater than the limit of {}'
).format(limit) in json_resp['message']['content']
def test_update_400_for_over_limit_content(client, notify_api, sample_user, sample_template):
limit = notify_api.config.get('SMS_CHAR_COUNT_LIMIT')
json_data = json.dumps({
'content': ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(limit + 1)),
'created_by': str(sample_user.id)
})
auth_header = create_authorization_header()
resp = client.post(
'/service/{}/template/{}'.format(sample_template.service.id, sample_template.id),
headers=[('Content-Type', 'application/json'), auth_header],
data=json_data
)
assert resp.status_code == 400
json_resp = json.loads(resp.get_data(as_text=True))
assert (
'Content has a character count greater than the limit of {}'
).format(limit) in json_resp['message']['content']
def test_should_return_all_template_versions_for_service_and_template_id(notify_api, sample_template):
def test_should_return_all_template_versions_for_service_and_template_id(client, sample_template):
original_content = sample_template.content
from app.dao.templates_dao import dao_update_template
sample_template.content = original_content + '1'
dao_update_template(sample_template)
sample_template.content = original_content + '2'
dao_update_template(sample_template)
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header()
resp = client.get('/service/{}/template/{}/versions'.format(sample_template.service_id, sample_template.id),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 200
resp_json = json.loads(resp.get_data(as_text=True))['data']
assert len(resp_json) == 3
for x in resp_json:
if x['version'] == 1:
assert x['content'] == original_content
elif x['version'] == 2:
assert x['content'] == original_content + '1'
else:
assert x['content'] == original_content + '2'
auth_header = create_authorization_header()
resp = client.get('/service/{}/template/{}/versions'.format(sample_template.service_id, sample_template.id),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 200
resp_json = json.loads(resp.get_data(as_text=True))['data']
assert len(resp_json) == 3
for x in resp_json:
if x['version'] == 1:
assert x['content'] == original_content
elif x['version'] == 2:
assert x['content'] == original_content + '1'
else:
assert x['content'] == original_content + '2'
def test_update_does_not_create_new_version_when_there_is_no_change(notify_api, sample_template):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header()
data = {
'template_type': sample_template.template_type,
'content': sample_template.content,
}
resp = client.post('/service/{}/template/{}'.format(sample_template.service_id, sample_template.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 200
def test_update_does_not_create_new_version_when_there_is_no_change(client, sample_template):
auth_header = create_authorization_header()
data = {
'template_type': sample_template.template_type,
'content': sample_template.content,
}
resp = client.post('/service/{}/template/{}'.format(sample_template.service_id, sample_template.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert resp.status_code == 200
template = dao_get_template_by_id(sample_template.id)
assert template.version == 1

69
tests/app/test_model.py
View File

@@ -1,3 +1,4 @@
import uuid
from datetime import datetime
import pytest
@@ -9,74 +10,6 @@ from app.models import (
MOBILE_TYPE, EMAIL_TYPE)
def test_should_build_notification_from_minimal_set_of_api_derived_params(notify_api):
now = datetime.utcnow()
notification = {
'template': 'template',
'template_version': '1',
'to': 'someone',
'personalisation': {}
}
notification = Notification.from_api_request(
created_at=now.strftime(DATETIME_FORMAT),
notification=notification,
notification_id="notification_id",
service_id="service_id",
notification_type='SMS',
api_key_id='api_key_id',
key_type='key_type'
)
assert notification.created_at == now
assert notification.id == "notification_id"
assert notification.template_id == 'template'
assert notification.template_version == '1'
assert not notification.job_row_number
assert not notification.job_id
assert notification.to == 'someone'
assert notification.service_id == 'service_id'
assert notification.status == 'created'
assert not notification.personalisation
assert notification.notification_type == 'SMS'
assert notification.api_key_id == 'api_key_id'
assert notification.key_type == 'key_type'
def test_should_build_notification_from_full_set_of_api_derived_params(notify_api):
now = datetime.utcnow()
notification = {
'template': 'template',
'template_version': '1',
'to': 'someone',
'personalisation': {'key': 'value'},
'job': 'job_id',
'row_number': 100
}
notification = Notification.from_api_request(
created_at=now.strftime(DATETIME_FORMAT),
notification=notification,
notification_id="notification_id",
service_id="service_id",
notification_type='SMS',
api_key_id='api_key_id',
key_type='key_type'
)
assert notification.created_at == now
assert notification.id == "notification_id"
assert notification.template_id == 'template'
assert notification.template_version == '1'
assert notification.job_row_number == 100
assert notification.job_id == 'job_id'
assert notification.to == 'someone'
assert notification.service_id == 'service_id'
assert notification.status == 'created'
assert notification.personalisation == {'key': 'value'}
assert notification.notification_type == 'SMS'
assert notification.api_key_id == 'api_key_id'
assert notification.key_type == 'key_type'
@pytest.mark.parametrize('mobile_number', [
'07700 900678',
'+44 7700 900678'