darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2025-12-22 16:31:15 -05:00
Code Issues Packages Projects Releases Wiki Activity

Fixing up tests to validate the call to the celery tasks.

Browse Source 
- mocker used to test call or otherwise of the task
- no new tests just a spring clean
This commit is contained in:
Martyn Inglis
2016-02-16 14:06:56 +00:00
parent 655beddba6
commit e42da7dd54
7 changed files with 129 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

223
tests/app/notifications/test_rest.py
View File

@@ -1,16 +1,14 @@
import moto
import uuid
import app.celery.tasks
from tests import create_authorization_header
from flask import url_for, json
from flask import json
from app.models import Service
from app.dao.templates_dao import get_model_templates
from mock import ANY
def test_get_notifications(
notify_api, notify_db, notify_db_session, sample_api_key, sample_notification):
"""
Tests GET endpoint '/' to retrieve entire service list.
"""
def test_get_notification_by_id(notify_api, sample_notification):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(
@@ -30,217 +28,208 @@ def test_get_notifications(
assert response.status_code == 200
def test_get_notifications_empty_result(
notify_api, notify_db, notify_db_session, sample_api_key, mocker):
def test_get_notifications_empty_result(notify_api, sample_api_key):
with notify_api.test_request_context():
with notify_api.test_client() as client:
id = uuid.uuid4()
missing_notification_id = uuid.uuid4()
auth_header = create_authorization_header(
service_id=sample_api_key.service_id,
path=url_for('notifications.get_notifications', notification_id=id),
path='/notifications/{}'.format(missing_notification_id),
method='GET')
response = client.get(
url_for('notifications.get_notifications', notification_id=id),
path='/notifications/{}'.format(missing_notification_id),
headers=[auth_header])
notification = json.loads(response.get_data(as_text=True))
assert notification['result'] == "error"
assert notification['message'] == "not found"
assert response.status_code == 404
def test_create_sms_should_reject_if_no_phone_numbers(
notify_api, notify_db, notify_db_session, sample_api_key, mocker):
"""
Tests GET endpoint '/' to retrieve entire service list.
"""
def test_create_sms_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:
data = {
'template': "my message"
}
mocker.patch('app.celery.tasks.send_sms.apply_async')
data = {}
auth_header = create_authorization_header(
service_id=sample_api_key.service_id,
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
path='/notifications/sms',
method='POST')
response = client.post(
url_for('notifications.create_sms_notification'),
path='/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
app.celery.tasks.send_sms.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_bad_phone_numbers(
notify_api, notify_db, notify_db_session, mocker):
"""
Tests GET endpoint '/' to retrieve entire service list.
"""
def test_should_reject_bad_phone_numbers(notify_api, sample_template, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.tasks.send_sms.apply_async')
data = {
'to': 'invalid',
'template': "my message"
'template': sample_template.id
}
auth_header = create_authorization_header(
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
path='/notifications/sms',
method='POST')
response = client.post(
url_for('notifications.create_sms_notification'),
path='/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
app.celery.tasks.send_sms.apply_async.assert_not_called()
assert json_resp['result'] == 'error'
assert len(json_resp['message'].keys()) == 1
assert 'Invalid phone number, must be of format +441234123123' in json_resp['message']['to']
assert response.status_code == 400
def test_send_notification_restrict_mobile(notify_api,
notify_db,
notify_db_session,
sample_api_key,
sample_template,
sample_user,
mocker):
"""
Test POST endpoint '/sms' with service notification with mobile number
not in restricted list.
"""
def test_send_notification_invalid_template_id(notify_api, sample_template, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.tasks.send_sms.apply_async')
data = {
'to': '+441234123123',
'template': 9999
}
auth_header = create_authorization_header(
service_id=sample_template.service.id,
request_body=json.dumps(data),
path='/notifications/sms',
method='POST')
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.tasks.send_sms.apply_async.assert_not_called()
assert response.status_code == 400
assert len(json_resp['message'].keys()) == 1
assert 'Template not found' in json_resp['message']['template']
def test_prevents_sending_to_any_mobile_on_restricted_service(notify_api, sample_template, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.tasks.send_sms.apply_async')
Service.query.filter_by(
id=sample_template.service.id).update({'restricted': True})
id=sample_template.service.id
).update(
{'restricted': True}
)
invalid_mob = '+449999999999'
data = {
'to': invalid_mob,
'template': sample_template.id
}
assert invalid_mob != sample_user.mobile_number
auth_header = create_authorization_header(
service_id=sample_template.service.id,
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
path='/notifications/sms',
method='POST')
response = client.post(
url_for('notifications.create_sms_notification'),
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.tasks.send_sms.apply_async.assert_not_called()
assert response.status_code == 400
assert 'Invalid phone number for restricted service' in json_resp['message']['restricted']
def test_send_notification_invalid_template_id(notify_api,
notify_db,
notify_db_session,
sample_api_key,
sample_template,
sample_user,
mocker):
"""
Tests POST endpoint '/sms' with notifications-admin notification with invalid template id
"""
def test_should_not_allow_template_from_another_service(notify_api, service_factory, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.tasks.send_sms.apply_async')
Service.query.filter_by(
id=sample_template.service.id).update({'restricted': True})
invalid_mob = '+449999999999'
data = {
'to': invalid_mob,
'template': 9999
}
assert invalid_mob != sample_user.mobile_number
auth_header = create_authorization_header(
service_id=sample_template.service.id,
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
method='POST')
service_1 = service_factory.get('service 1')
service_2 = service_factory.get('service 2')
response = client.post(
url_for('notifications.create_sms_notification'),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
service_2_templates = get_model_templates(service_id=service_2.id)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert 'Template not found' in json_resp['message']['template']
@moto.mock_sqs
def test_should_not_allow_template_from_other_service(notify_api,
notify_db,
notify_db_session,
sample_template,
sample_admin_service_id,
mocker):
"""
Tests POST endpoint '/sms' with notifications.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'to': '+441234123123',
'template': sample_template.id
'template': service_2_templates[0].id
}
auth_header = create_authorization_header(
service_id=sample_admin_service_id,
service_id=service_1.id,
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
path='/notifications/sms',
method='POST')
response = client.post(
url_for('notifications.create_sms_notification'),
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.tasks.send_sms.apply_async.assert_not_called()
assert response.status_code == 400
assert 'Invalid template' in json_resp['message']['restricted']
@moto.mock_sqs
def test_should_allow_valid_message(notify_api,
notify_db,
notify_db_session,
sqs_client_conn,
sample_user,
sample_template,
mocker):
"""
Tests POST endpoint '/sms' with notifications-admin notification.
"""
def test_should_allow_valid_sms_notification(notify_api, sample_template, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.tasks.send_sms.apply_async')
data = {
'to': '+441234123123',
'template': sample_template.id
}
auth_header = create_authorization_header(
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
method='POST')
path='/notifications/sms',
method='POST',
service_id=sample_template.service_id
)
response = client.post(
url_for('notifications.create_sms_notification'),
path='/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
notification_id = json.loads(response.data)['notification_id']
app.celery.tasks.send_sms.apply_async.assert_called_once_with(
(str(sample_template.service_id),
notification_id,
ANY,
notify_api.config['SECRET_KEY'],
notify_api.config['DANGEROUS_SALT'])
)
assert response.status_code == 201
assert json.loads(response.data)['notification_id'] is not None
assert notification_id
@moto.mock_sqs
def test_send_email_valid_data(notify_api,
notify_db,
notify_db_session,
@@ -262,11 +251,11 @@ def test_send_email_valid_data(notify_api,
}
auth_header = create_authorization_header(
request_body=json.dumps(data),
path=url_for('notifications.create_email_notification'),
path='/notifications/email',
method='POST')
response = client.post(
url_for('notifications.create_email_notification'),
path='/notifications/email',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
@@ -274,7 +263,6 @@ def test_send_email_valid_data(notify_api,
assert json.loads(response.data)['notification_id'] is not None
@moto.mock_sqs
def test_valid_message_with_service_id(notify_api,
notify_db,
notify_db_session,
@@ -286,7 +274,7 @@ def test_valid_message_with_service_id(notify_api,
with notify_api.test_client() as client:
job_id = uuid.uuid4()
service_id = sample_template.service.id
url = url_for('notifications.create_sms_for_service', service_id=service_id)
url = '/notifications/sms/service/{}'.format(service_id)
data = {
'to': '+441234123123',
'template': sample_template.id,
@@ -306,7 +294,6 @@ def test_valid_message_with_service_id(notify_api,
assert json.loads(response.data)['notification_id'] is not None
@moto.mock_sqs
def test_message_with_incorrect_service_id_should_fail(notify_api,
notify_db,
notify_db_session,
@@ -319,7 +306,7 @@ def test_message_with_incorrect_service_id_should_fail(notify_api,
job_id = uuid.uuid4()
invalid_service_id = uuid.uuid4()
url = url_for('notifications.create_sms_for_service', service_id=invalid_service_id)
url = '/notifications/sms/service/{}'.format(invalid_service_id)
data = {
'to': '+441234123123',