mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-24 01:11:38 -05:00
Create query to sum notificaitons for the message limit check, this is a bit more efficient.
Refactored send_notifications method so that it is more readible. Refectored the test_send_notificaitons so that it uses parametrized test to avoid duplication.
This commit is contained in:
Rebecca Law
@@ -23,21 +23,28 @@ from tests.app.conftest import (
|
||||
sample_api_key as create_sample_api_key
|
||||
)
|
||||
|
||||
from app.models import Template
|
||||
from app.errors import InvalidRequest
|
||||
|
||||
def test_create_sms_should_reject_if_missing_required_fields(notify_api, sample_api_key, mocker):
|
||||
|
||||
@pytest.mark.parametrize('template_type',
|
||||
['sms', 'email'])
|
||||
def test_create_notification_should_reject_if_missing_required_fields(notify_api,
|
||||
sample_api_key, mocker, template_type):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
|
||||
mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
|
||||
|
||||
data = {}
|
||||
auth_header = create_authorization_header(service_id=sample_api_key.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms',
|
||||
path='/notifications/{}'.format(template_type),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
app.celery.provider_tasks.send_email_to_provider.apply_async.assert_not_called()
|
||||
assert json_resp['result'] == 'error'
|
||||
assert 'Missing data for required field.' in json_resp['message']['to'][0]
|
||||
assert 'Missing data for required field.' in json_resp['message']['template'][0]
|
||||
@@ -68,19 +75,22 @@ def test_should_reject_bad_phone_numbers(notify_api, sample_template, mocker):
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_send_notification_invalid_template_id(notify_api, sample_template, mocker, fake_uuid):
|
||||
@pytest.mark.parametrize('template_type, to',
|
||||
[('sms', '+447700900855'),
|
||||
('email', 'ok@ok.com')])
|
||||
def test_send_notification_invalid_template_id(notify_api, sample_template, mocker, fake_uuid, template_type, to):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
|
||||
|
||||
data = {
|
||||
'to': '+447700900855',
|
||||
'to': to,
|
||||
'template': fake_uuid
|
||||
}
|
||||
auth_header = create_authorization_header(service_id=sample_template.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms',
|
||||
path='/notifications/{}'.format(template_type),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
@@ -146,77 +156,30 @@ def test_should_not_send_notification_for_archived_template(notify_api, sample_t
|
||||
assert 'Template has been deleted' in json_resp['message']['template']
|
||||
|
||||
|
||||
def test_send_notification_with_missing_personalisation(notify_api, sample_template_with_placeholders, mocker):
|
||||
@pytest.mark.parametrize('template_type, to',
|
||||
[('sms', '+447700900855'),
|
||||
('email', 'not-someone-we-trust@email-address.com')])
|
||||
def test_should_not_send_notification_if_restricted_and_not_a_service_user(notify_api,
|
||||
sample_template,
|
||||
sample_email_template,
|
||||
mocker,
|
||||
template_type,
|
||||
to):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
|
||||
|
||||
template = sample_template if template_type == 'sms' else sample_email_template
|
||||
template.service.restricted = True
|
||||
dao_update_service(template.service)
|
||||
data = {
|
||||
'to': '+447700900855',
|
||||
'template': sample_template_with_placeholders.id,
|
||||
'personalisation': {
|
||||
'foo': 'bar'
|
||||
}
|
||||
}
|
||||
auth_header = create_authorization_header(service_id=sample_template_with_placeholders.service.id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
app.celery.provider_tasks.send_sms_to_provider.apply_async.assert_not_called()
|
||||
|
||||
assert response.status_code == 400
|
||||
assert 'Missing personalisation: name' in json_resp['message']['template']
|
||||
|
||||
|
||||
def test_send_notification_with_too_much_personalisation_data(
|
||||
notify_api, sample_template_with_placeholders, mocker
|
||||
):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
|
||||
|
||||
data = {
|
||||
'to': '+447700900855',
|
||||
'template': sample_template_with_placeholders.id,
|
||||
'personalisation': {
|
||||
'name': 'Jo', 'foo': 'bar'
|
||||
}
|
||||
}
|
||||
auth_header = create_authorization_header(service_id=sample_template_with_placeholders.service.id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
app.celery.provider_tasks.send_sms_to_provider.apply_async.assert_not_called()
|
||||
|
||||
assert response.status_code == 400
|
||||
assert 'Personalisation not needed for template: foo' in json_resp['message']['template']
|
||||
|
||||
|
||||
def test_should_not_send_sms_if_restricted_and_not_a_service_user(notify_api, sample_template, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
|
||||
|
||||
sample_template.service.restricted = True
|
||||
dao_update_service(sample_template.service)
|
||||
invalid_mob = '+447700900855'
|
||||
data = {
|
||||
'to': invalid_mob,
|
||||
'template': sample_template.id
|
||||
'to': to,
|
||||
'template': template.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(service_id=sample_template.service_id)
|
||||
auth_header = create_authorization_header(service_id=template.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms',
|
||||
path='/notifications/{}'.format(template_type),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
@@ -230,131 +193,84 @@ def test_should_not_send_sms_if_restricted_and_not_a_service_user(notify_api, sa
|
||||
)] == json_resp['message']['to']
|
||||
|
||||
|
||||
def test_should_send_sms_if_restricted_and_a_service_user(notify_api, sample_template, mocker):
|
||||
@pytest.mark.parametrize('template_type',
|
||||
['sms', 'email'])
|
||||
def test_should_send_notification_if_restricted_and_a_service_user(notify_api,
|
||||
sample_template,
|
||||
sample_email_template,
|
||||
template_type,
|
||||
mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
|
||||
if template_type == 'sms':
|
||||
mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
|
||||
else:
|
||||
mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
|
||||
|
||||
sample_template.service.restricted = True
|
||||
dao_update_service(sample_template.service)
|
||||
template = sample_template if template_type == 'sms' else sample_email_template
|
||||
to = template.service.created_by.mobile_number if template_type == 'sms' \
|
||||
else template.service.created_by.email_address
|
||||
template.service.restricted = True
|
||||
dao_update_service(template.service)
|
||||
data = {
|
||||
'to': sample_template.service.created_by.mobile_number,
|
||||
'template': sample_template.id
|
||||
'to': to,
|
||||
'template': template.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(service_id=sample_template.service_id)
|
||||
auth_header = create_authorization_header(service_id=template.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms',
|
||||
path='/notifications/{}'.format(template_type),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
assert app.celery.provider_tasks.send_sms_to_provider.apply_async.called
|
||||
if template_type == 'sms':
|
||||
assert app.celery.provider_tasks.send_sms_to_provider.apply_async.called
|
||||
else:
|
||||
assert app.celery.provider_tasks.send_email_to_provider.apply_async.called
|
||||
assert response.status_code == 201
|
||||
|
||||
|
||||
def test_should_send_email_if_restricted_and_a_service_user(notify_api, sample_email_template, mocker):
|
||||
@pytest.mark.parametrize('template_type',
|
||||
['sms', 'email'])
|
||||
def test_should_not_allow_template_from_another_service(notify_api,
|
||||
service_factory,
|
||||
sample_user,
|
||||
mocker,
|
||||
template_type):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
|
||||
|
||||
sample_email_template.service.restricted = True
|
||||
dao_update_service(sample_email_template.service)
|
||||
data = {
|
||||
'to': sample_email_template.service.created_by.email_address,
|
||||
'template': sample_email_template.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
assert app.celery.provider_tasks.send_email_to_provider.apply_async.called
|
||||
assert response.status_code == 201
|
||||
|
||||
|
||||
def test_should_not_allow_template_from_another_service(notify_api, service_factory, sample_user, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
|
||||
|
||||
if template_type == 'sms':
|
||||
mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
|
||||
else:
|
||||
mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
|
||||
service_1 = service_factory.get('service 1', user=sample_user, email_from='service.1')
|
||||
service_2 = service_factory.get('service 2', user=sample_user, email_from='service.2')
|
||||
|
||||
service_2_templates = dao_get_all_templates_for_service(service_id=service_2.id)
|
||||
to = sample_user.mobile_number if template_type == 'sms' else sample_user.email_address
|
||||
data = {
|
||||
'to': sample_user.mobile_number,
|
||||
'to': to,
|
||||
'template': service_2_templates[0].id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(service_id=service_1.id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms',
|
||||
path='/notifications/{}'.format(template_type),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
app.celery.provider_tasks.send_sms_to_provider.apply_async.assert_not_called()
|
||||
|
||||
if template_type == 'sms':
|
||||
app.celery.provider_tasks.send_sms_to_provider.apply_async.assert_not_called()
|
||||
else:
|
||||
app.celery.provider_tasks.send_email_to_provider.apply_async.assert_not_called()
|
||||
assert response.status_code == 404
|
||||
test_string = 'No result found'
|
||||
assert test_string in json_resp['message']
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'template_type, should_error', [
|
||||
('sms', True),
|
||||
('email', False)
|
||||
]
|
||||
)
|
||||
def test_should_not_allow_template_content_too_large(
|
||||
notify_api,
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
sample_user,
|
||||
template_type,
|
||||
mocker,
|
||||
should_error
|
||||
):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
|
||||
template = create_sample_template(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
content="((long_text))",
|
||||
template_type=template_type
|
||||
)
|
||||
limit = current_app.config.get('SMS_CHAR_COUNT_LIMIT')
|
||||
json_data = json.dumps({
|
||||
'to': sample_user.mobile_number if template_type == 'sms' else sample_user.email_address,
|
||||
'template': template.id,
|
||||
'personalisation': {
|
||||
'long_text': ''.join(
|
||||
random.choice(string.ascii_uppercase + string.digits) for _ in range(limit + 1))
|
||||
}
|
||||
})
|
||||
auth_header = create_authorization_header(service_id=template.service_id)
|
||||
|
||||
resp = client.post(
|
||||
path='/notifications/{}'.format(template_type),
|
||||
data=json_data,
|
||||
headers=[('Content-Type', 'application/json'), auth_header]
|
||||
)
|
||||
if should_error:
|
||||
assert resp.status_code == 400
|
||||
json_resp = json.loads(resp.get_data(as_text=True))
|
||||
assert json_resp['message']['content'][0] == (
|
||||
'Content has a character count greater'
|
||||
' than the limit of {}').format(limit)
|
||||
else:
|
||||
assert resp.status_code == 201
|
||||
|
||||
|
||||
@freeze_time("2016-01-01 11:09:00.061258")
|
||||
def test_should_allow_valid_sms_notification(notify_api, sample_template, mocker):
|
||||
with notify_api.test_request_context():
|
||||
@@ -386,27 +302,6 @@ def test_should_allow_valid_sms_notification(notify_api, sample_template, mocker
|
||||
assert response_data['template_version'] == sample_template.version
|
||||
|
||||
|
||||
def test_create_email_should_reject_if_missing_required_fields(notify_api, sample_api_key, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
|
||||
|
||||
data = {}
|
||||
auth_header = create_authorization_header(service_id=sample_api_key.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
app.celery.provider_tasks.send_email_to_provider.apply_async.assert_not_called()
|
||||
assert json_resp['result'] == 'error'
|
||||
assert 'Missing data for required field.' in json_resp['message']['to'][0]
|
||||
assert 'Missing data for required field.' in json_resp['message']['template'][0]
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_should_reject_email_notification_with_bad_email(notify_api, sample_email_template, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
@@ -430,92 +325,6 @@ def test_should_reject_email_notification_with_bad_email(notify_api, sample_emai
|
||||
assert data['message']['to'][0] == 'Not a valid email address'
|
||||
|
||||
|
||||
def test_should_reject_email_notification_with_template_id_that_cant_be_found(
|
||||
notify_api, sample_email_template, mocker, fake_uuid):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
|
||||
data = {
|
||||
'to': 'ok@ok.com',
|
||||
'template': fake_uuid
|
||||
}
|
||||
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
data = json.loads(response.get_data(as_text=True))
|
||||
app.celery.provider_tasks.send_email_to_provider.apply_async.assert_not_called()
|
||||
assert response.status_code == 404
|
||||
assert data['result'] == 'error'
|
||||
test_string = 'No result found'
|
||||
assert test_string in data['message']
|
||||
|
||||
|
||||
def test_should_not_allow_email_template_from_another_service(notify_api, service_factory, sample_user, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
|
||||
|
||||
service_1 = service_factory.get('service 1', template_type='email', user=sample_user,
|
||||
email_from='service.1')
|
||||
service_2 = service_factory.get('service 2', template_type='email', user=sample_user,
|
||||
email_from='service.2')
|
||||
|
||||
service_2_templates = dao_get_all_templates_for_service(service_id=service_2.id)
|
||||
|
||||
data = {
|
||||
'to': sample_user.email_address,
|
||||
'template': str(service_2_templates[0].id)
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(service_id=service_1.id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
app.celery.provider_tasks.send_email_to_provider.apply_async.assert_not_called()
|
||||
|
||||
assert response.status_code == 404
|
||||
test_string = 'No result found'
|
||||
assert test_string in json_resp['message']
|
||||
|
||||
|
||||
def test_should_not_send_email_if_restricted_and_not_a_service_user(notify_api, sample_email_template, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
|
||||
|
||||
sample_email_template.service.restricted = True
|
||||
dao_update_service(sample_email_template.service)
|
||||
|
||||
data = {
|
||||
'to': "not-someone-we-trust@email-address.com",
|
||||
'template': str(sample_email_template.id)
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
app.celery.provider_tasks.send_email_to_provider.apply_async.assert_not_called()
|
||||
|
||||
assert response.status_code == 400
|
||||
assert [(
|
||||
'Can’t send to this recipient when service is in trial mode – see '
|
||||
'https://www.notifications.service.gov.uk/trial-mode'
|
||||
)] == json_resp['message']['to']
|
||||
|
||||
|
||||
@freeze_time("2016-01-01 11:09:00.061258")
|
||||
def test_should_allow_valid_email_notification(notify_api, sample_email_template, mocker):
|
||||
with notify_api.test_request_context():
|
||||
@@ -808,134 +617,99 @@ def test_should_send_sms_if_team_api_key_and_a_service_user(notify_api, sample_t
|
||||
assert response.status_code == 201
|
||||
|
||||
|
||||
def test_should_persist_sms_notification(notify_api, sample_template, fake_uuid, mocker):
|
||||
@pytest.mark.parametrize('template_type',
|
||||
['sms', 'email'])
|
||||
def test_should_persist_notification(notify_api, sample_template,
|
||||
sample_email_template,
|
||||
template_type,
|
||||
fake_uuid, mocker):
|
||||
with notify_api.test_request_context(), notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
|
||||
if template_type == 'sms':
|
||||
mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
|
||||
else:
|
||||
mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
|
||||
mocker.patch('app.notifications.rest.create_uuid', return_value=fake_uuid)
|
||||
|
||||
template = sample_template if template_type == 'sms' else sample_email_template
|
||||
to = sample_template.service.created_by.mobile_number if template_type == 'sms' \
|
||||
else sample_email_template.service.created_by.email_address
|
||||
data = {
|
||||
'to': sample_template.service.created_by.mobile_number,
|
||||
'template': sample_template.id
|
||||
'to': to,
|
||||
'template': template.id
|
||||
}
|
||||
api_key = ApiKey(
|
||||
service=sample_template.service,
|
||||
service=template.service,
|
||||
name='team_key',
|
||||
created_by=sample_template.created_by,
|
||||
created_by=template.created_by,
|
||||
key_type=KEY_TYPE_TEAM)
|
||||
save_model_api_key(api_key)
|
||||
auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms',
|
||||
path='/notifications/{}'.format(template_type),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), ('Authorization', 'Bearer {}'.format(auth_header))])
|
||||
|
||||
app.celery.provider_tasks.send_sms_to_provider.apply_async.assert_called_once_with(
|
||||
(str(sample_template.service.id), fake_uuid), queue='send-sms')
|
||||
if template_type == 'sms':
|
||||
app.celery.provider_tasks.send_sms_to_provider.apply_async.assert_called_once_with(
|
||||
(str(template.service.id), fake_uuid), queue='send-sms')
|
||||
else:
|
||||
app.celery.provider_tasks.send_email_to_provider.apply_async.assert_called_once_with(
|
||||
(str(template.service.id), fake_uuid), queue='send-email')
|
||||
assert response.status_code == 201
|
||||
|
||||
notification = notifications_dao.get_notification_by_id(fake_uuid)
|
||||
assert notification.to == sample_template.service.created_by.mobile_number
|
||||
assert notification.template_id == sample_template.id
|
||||
assert notification.notification_type == 'sms'
|
||||
assert notification.to == to
|
||||
assert notification.template_id == template.id
|
||||
assert notification.notification_type == template_type
|
||||
|
||||
|
||||
def test_should_persist_email_notification(notify_api, sample_email_template, fake_uuid, mocker):
|
||||
with notify_api.test_request_context(), notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
|
||||
mocker.patch('app.notifications.rest.create_uuid', return_value=fake_uuid)
|
||||
|
||||
data = {
|
||||
'to': sample_email_template.service.created_by.email_address,
|
||||
'template': sample_email_template.id
|
||||
}
|
||||
api_key = ApiKey(
|
||||
service=sample_email_template.service,
|
||||
name='team_key',
|
||||
created_by=sample_email_template.created_by,
|
||||
key_type=KEY_TYPE_TEAM)
|
||||
save_model_api_key(api_key)
|
||||
auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), ('Authorization', 'Bearer {}'.format(auth_header))])
|
||||
|
||||
app.celery.provider_tasks.send_email_to_provider.apply_async.assert_called_once_with(
|
||||
(str(sample_email_template.service.id), fake_uuid), queue='send-email')
|
||||
assert response.status_code == 201
|
||||
|
||||
notification = notifications_dao.get_notification_by_id(fake_uuid)
|
||||
assert notification.to == sample_email_template.service.created_by.email_address
|
||||
assert notification.template_id == sample_email_template.id
|
||||
assert notification.notification_type == 'email'
|
||||
|
||||
|
||||
def test_should_delete_email_notification_and_return_error_if_sqs_fails(
|
||||
@pytest.mark.parametrize('template_type',
|
||||
['sms', 'email'])
|
||||
def test_should_delete_notification_and_return_error_if_sqs_fails(
|
||||
notify_api,
|
||||
sample_email_template,
|
||||
sample_template,
|
||||
fake_uuid,
|
||||
mocker):
|
||||
mocker,
|
||||
template_type):
|
||||
with notify_api.test_request_context(), notify_api.test_client() as client:
|
||||
mocker.patch(
|
||||
'app.celery.provider_tasks.send_email_to_provider.apply_async',
|
||||
side_effect=Exception("failed to talk to SQS")
|
||||
)
|
||||
if template_type == 'email':
|
||||
mocker.patch(
|
||||
'app.celery.provider_tasks.send_email_to_provider.apply_async',
|
||||
side_effect=Exception("failed to talk to SQS")
|
||||
)
|
||||
else:
|
||||
mocker.patch(
|
||||
'app.celery.provider_tasks.send_sms_to_provider.apply_async',
|
||||
side_effect=Exception("failed to talk to SQS")
|
||||
)
|
||||
mocker.patch('app.notifications.rest.create_uuid', return_value=fake_uuid)
|
||||
|
||||
template = sample_template if template_type == 'sms' else sample_email_template
|
||||
to = sample_template.service.created_by.mobile_number if template_type == 'sms' \
|
||||
else sample_email_template.service.created_by.email_address
|
||||
data = {
|
||||
'to': sample_email_template.service.created_by.email_address,
|
||||
'template': sample_email_template.id
|
||||
'to': to,
|
||||
'template': template.id
|
||||
}
|
||||
api_key = ApiKey(
|
||||
service=sample_email_template.service,
|
||||
service=template.service,
|
||||
name='team_key',
|
||||
created_by=sample_email_template.created_by,
|
||||
created_by=template.created_by,
|
||||
key_type=KEY_TYPE_TEAM)
|
||||
save_model_api_key(api_key)
|
||||
auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email',
|
||||
path='/notifications/{}'.format(template_type),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), ('Authorization', 'Bearer {}'.format(auth_header))])
|
||||
|
||||
app.celery.provider_tasks.send_email_to_provider.apply_async.assert_called_once_with(
|
||||
(str(sample_email_template.service.id), fake_uuid), queue='send-email')
|
||||
|
||||
assert response.status_code == 500
|
||||
assert not notifications_dao.get_notification_by_id(fake_uuid)
|
||||
assert not NotificationHistory.query.get(fake_uuid)
|
||||
|
||||
|
||||
def test_should_delete_sms_notification_and_return_error_if_sqs_fails(notify_api, sample_template, fake_uuid, mocker):
|
||||
with notify_api.test_request_context(), notify_api.test_client() as client:
|
||||
mocker.patch(
|
||||
'app.celery.provider_tasks.send_sms_to_provider.apply_async',
|
||||
side_effect=Exception("failed to talk to SQS")
|
||||
)
|
||||
mocker.patch('app.notifications.rest.create_uuid', return_value=fake_uuid)
|
||||
|
||||
data = {
|
||||
'to': sample_template.service.created_by.mobile_number,
|
||||
'template': sample_template.id
|
||||
}
|
||||
api_key = ApiKey(
|
||||
service=sample_template.service,
|
||||
name='team_key',
|
||||
created_by=sample_template.created_by,
|
||||
key_type=KEY_TYPE_TEAM)
|
||||
save_model_api_key(api_key)
|
||||
auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), ('Authorization', 'Bearer {}'.format(auth_header))])
|
||||
|
||||
app.celery.provider_tasks.send_sms_to_provider.apply_async.assert_called_once_with(
|
||||
(str(sample_template.service.id), fake_uuid), queue='send-sms')
|
||||
if template_type == 'email':
|
||||
app.celery.provider_tasks.send_email_to_provider.apply_async.assert_called_once_with(
|
||||
(str(template.service.id), fake_uuid), queue='send-email')
|
||||
else:
|
||||
app.celery.provider_tasks.send_sms_to_provider.apply_async.assert_called_once_with(
|
||||
(str(template.service.id), fake_uuid), queue='send-sms')
|
||||
|
||||
assert response.status_code == 500
|
||||
assert not notifications_dao.get_notification_by_id(fake_uuid)
|
||||
@@ -1128,3 +902,57 @@ def test_should_error_if_notification_type_does_not_match_template_type(
|
||||
assert json_resp['result'] == 'error'
|
||||
assert '{0} template is not suitable for {1} notification'.format(template_type, notification_type) \
|
||||
in json_resp['message']
|
||||
|
||||
|
||||
def test_create_template_raises_invalid_request_exception_with_missing_personalisation(
|
||||
sample_template_with_placeholders):
|
||||
template = Template.query.get(sample_template_with_placeholders.id)
|
||||
from app.notifications.rest import create_template_object_for_notification
|
||||
with pytest.raises(InvalidRequest) as e:
|
||||
create_template_object_for_notification(template, {})
|
||||
assert {'template': ['Missing personalisation: name']} == e.value.message
|
||||
|
||||
|
||||
def test_create_template_raises_invalid_request_exception_with_too_much_personalisation_data(
|
||||
sample_template_with_placeholders
|
||||
):
|
||||
from app.notifications.rest import create_template_object_for_notification
|
||||
template = Template.query.get(sample_template_with_placeholders.id)
|
||||
with pytest.raises(InvalidRequest) as e:
|
||||
create_template_object_for_notification(template, {'name': 'Jo', 'extra': 'stuff'})
|
||||
assert {'template': ['Personalisation not needed for template: foo']} in e.value.message
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'template_type, should_error', [
|
||||
('sms', True),
|
||||
('email', False)
|
||||
]
|
||||
)
|
||||
def test_create_template_raises_invalid_request_when_content_too_large(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
template_type,
|
||||
should_error
|
||||
):
|
||||
sample = create_sample_template(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
content="((long_text))",
|
||||
template_type=template_type
|
||||
)
|
||||
limit = current_app.config.get('SMS_CHAR_COUNT_LIMIT')
|
||||
template = Template.query.get(sample.id)
|
||||
from app.notifications.rest import create_template_object_for_notification
|
||||
try:
|
||||
create_template_object_for_notification(template,
|
||||
{'long_text':
|
||||
''.join(
|
||||
random.choice(string.ascii_uppercase + string.digits) for _ in
|
||||
range(limit + 1))})
|
||||
if should_error:
|
||||
pytest.fail("expected an InvalidRequest")
|
||||
except InvalidRequest as e:
|
||||
if not should_error:
|
||||
pytest.fail("do not expect an InvalidRequest")
|
||||
assert e.message == {'content': ['Content has a character count greater than the limit of {}'.format(limit)]}
|
||||
|
||||
Reference in New Issue
Block a user