darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-01 15:46:07 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into research-mode-csv-file-queues

Browse Source 
Conflicts:
	tests/app/celery/test_tasks.py
	tests/app/notifications/rest/test_send_notification.py
This commit is contained in:
Martyn Inglis
2016-09-30 11:07:32 +01:00
parent 4e03f81fcf c2b57b79ee
commit e3ed8fad1a
14 changed files with 586 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,9 +1,8 @@
import uuid
import pytest
from datetime import datetime
import pytest
from freezegun import freeze_time
from unittest.mock import ANY
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm.exc import NoResultFound
@@ -17,7 +16,7 @@ from app.celery.tasks import (
send_email
)
from app.dao import jobs_dao, services_dao
from app.models import Notification, KEY_TYPE_TEAM, KEY_TYPE_TEST
from app.models import Notification, KEY_TYPE_TEAM, KEY_TYPE_TEST, KEY_TYPE_NORMAL
from tests.app import load_example_csv
from tests.app.conftest import (
sample_service,
@@ -560,7 +559,7 @@ def test_should_send_sms_template_to_and_persist_with_job_id(sample_job, sample_
encryption.encrypt(notification),
datetime.utcnow().strftime(DATETIME_FORMAT),
api_key_id=str(sample_api_key.id),
key_type=KEY_TYPE_TEAM
key_type=KEY_TYPE_NORMAL
)
provider_tasks.deliver_sms.apply_async.assert_called_once_with(
(notification_id),
@@ -577,14 +576,70 @@ def test_should_send_sms_template_to_and_persist_with_job_id(sample_job, sample_
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_TEAM
assert persisted_notification.key_type == KEY_TYPE_NORMAL
assert persisted_notification.notification_type == 'sms'
def test_should_not_send_email_if_team_key_and_recipient_not_in_team(sample_email_template_with_placeholders,
sample_team_api_key,
mocker):
notification = _notification_json(
sample_email_template_with_placeholders,
"my_email@my_email.com",
{"name": "Jo"},
row_number=1)
apply_async = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
notification_id = uuid.uuid4()
team_members = [user.email_address for user in sample_email_template_with_placeholders.service.users]
assert "my_email@my_email.com" not in team_members
with freeze_time("2016-01-01 11:09:00.00000"):
now = datetime.utcnow()
send_email(
sample_email_template_with_placeholders.service_id,
notification_id,
encryption.encrypt(notification),
now.strftime(DATETIME_FORMAT),
api_key_id=str(sample_team_api_key.id),
key_type=KEY_TYPE_TEAM
)
with pytest.raises(NoResultFound):
persisted_notification = Notification.query.filter_by(id=notification_id).one()
print(persisted_notification)
apply_async.not_called()
def test_should_not_send_sms_if_team_key_and_recipient_not_in_team(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)
team_members = [user.mobile_number for user in service.users]
assert "07890 300000" not in team_members
notification = _notification_json(template, "07700 900849")
mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
notification_id = uuid.uuid4()
send_sms(
service.id,
notification_id,
encryption.encrypt(notification),
datetime.utcnow().strftime(DATETIME_FORMAT)
)
provider_tasks.send_sms_to_provider.apply_async.assert_not_called()
with pytest.raises(NoResultFound):
Notification.query.filter_by(id=notification_id).one()
def test_should_use_email_template_and_persist(sample_email_template_with_placeholders, sample_api_key, mocker):
notification = _notification_json(
sample_email_template_with_placeholders,
"my_email@my_email.com",
'my_email@my_email.com',
{"name": "Jo"},
row_number=1)
mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
@@ -601,7 +656,7 @@ def test_should_use_email_template_and_persist(sample_email_template_with_placeh
encryption.encrypt(notification),
now.strftime(DATETIME_FORMAT),
api_key_id=str(sample_api_key.id),
key_type=KEY_TYPE_TEAM
key_type=sample_api_key.key_type
)
persisted_notification = Notification.query.filter_by(id=notification_id).one()
@@ -620,7 +675,7 @@ def test_should_use_email_template_and_persist(sample_email_template_with_placeh
assert persisted_notification.personalisation == {'name': 'Jo'}
assert persisted_notification._personalisation == encryption.encrypt({"name": "Jo"})
assert persisted_notification.api_key_id == sample_api_key.id
assert persisted_notification.key_type == KEY_TYPE_TEAM
assert persisted_notification.key_type == KEY_TYPE_NORMAL
assert persisted_notification.notification_type == 'email'

25
tests/app/conftest.py
View File

@@ -19,7 +19,9 @@ from app.models import (
ProviderStatistics,
ProviderDetails,
NotificationStatistics,
KEY_TYPE_NORMAL, KEY_TYPE_TEST, KEY_TYPE_TEAM)
ServiceWhitelist,
KEY_TYPE_NORMAL, KEY_TYPE_TEST, KEY_TYPE_TEAM,
MOBILE_TYPE, EMAIL_TYPE)
from app.dao.users_dao import (save_model_user, create_user_code, create_secret_code)
from app.dao.services_dao import (dao_create_service, dao_add_user_to_service)
from app.dao.templates_dao import dao_create_template
@@ -129,9 +131,11 @@ def sample_service(notify_db,
user=None,
restricted=False,
limit=1000,
email_from="sample.service"):
email_from=None):
if user is None:
user = sample_user(notify_db, notify_db_session)
if email_from is None:
email_from = service_name.lower().replace(' ', '.')
data = {
'name': service_name,
'message_limit': limit,
@@ -874,3 +878,20 @@ def already_registered_template(notify_db,
template = Template(**data)
db.session.add(template)
return template
@pytest.fixture(scope='function')
def sample_service_whitelist(notify_db, notify_db_session, service=None, email_address=None, mobile_number=None):
if service is None:
service = sample_service(notify_db, notify_db_session)
if email_address:
whitelisted_user = ServiceWhitelist.from_string(service.id, EMAIL_TYPE, email_address)
elif mobile_number:
whitelisted_user = ServiceWhitelist.from_string(service.id, MOBILE_TYPE, mobile_number)
else:
whitelisted_user = ServiceWhitelist.from_string(service.id, EMAIL_TYPE, 'whitelisted_user@digital.gov.uk')
notify_db.session.add(whitelisted_user)
notify_db.session.commit()
return whitelisted_user

56
tests/app/dao/test_service_whitelist_dao.py Normal file
View File

@@ -0,0 +1,56 @@
import uuid
from app.models import (
ServiceWhitelist,
MOBILE_TYPE, EMAIL_TYPE)
from app.dao.service_whitelist_dao import (
dao_fetch_service_whitelist,
dao_add_and_commit_whitelisted_contacts,
dao_remove_service_whitelist
)
from tests.app.conftest import sample_service as create_service
def test_fetch_service_whitelist_gets_whitelists(sample_service_whitelist):
whitelist = dao_fetch_service_whitelist(sample_service_whitelist.service_id)
assert len(whitelist) == 1
assert whitelist[0].id == sample_service_whitelist.id
def test_fetch_service_whitelist_ignores_other_service(sample_service_whitelist):
assert len(dao_fetch_service_whitelist(uuid.uuid4())) == 0
def test_add_and_commit_whitelisted_contacts_saves_data(sample_service):
whitelist = ServiceWhitelist.from_string(sample_service.id, EMAIL_TYPE, 'foo@example.com')
dao_add_and_commit_whitelisted_contacts([whitelist])
db_contents = ServiceWhitelist.query.all()
assert len(db_contents) == 1
assert db_contents[0].id == whitelist.id
def test_remove_service_whitelist_only_removes_for_my_service(notify_db, notify_db_session):
service_1 = create_service(notify_db, notify_db_session, service_name="service 1")
service_2 = create_service(notify_db, notify_db_session, service_name="service 2")
dao_add_and_commit_whitelisted_contacts([
ServiceWhitelist.from_string(service_1.id, EMAIL_TYPE, 'service1@example.com'),
ServiceWhitelist.from_string(service_2.id, EMAIL_TYPE, 'service2@example.com')
])
dao_remove_service_whitelist(service_1.id)
assert service_1.whitelist == []
assert len(service_2.whitelist) == 1
def test_remove_service_whitelist_does_not_commit(notify_db, sample_service_whitelist):
dao_remove_service_whitelist(sample_service_whitelist.service_id)
# since dao_remove_service_whitelist doesn't commit, we can still rollback its changes
notify_db.session.rollback()
assert ServiceWhitelist.query.count() == 1

116
tests/app/notifications/rest/test_send_notification.py
View File

@@ -1,7 +1,7 @@
from datetime import datetime
import random
import string
import pytest
from datetime import datetime
from flask import (json, current_app)
from freezegun import freeze_time
@@ -10,7 +10,7 @@ from notifications_python_client.authentication import create_jwt_token
import app
from app.dao import notifications_dao
from app.models import ApiKey, KEY_TYPE_TEAM, KEY_TYPE_TEST, Notification, NotificationHistory
from app.models import ApiKey, KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST, Notification, NotificationHistory
from app.dao.templates_dao import dao_get_all_templates_for_service, dao_update_template
from app.dao.services_dao import dao_update_service
from app.dao.api_key_dao import save_model_api_key
@@ -20,7 +20,9 @@ from tests.app.conftest import (
sample_service as create_sample_service,
sample_email_template as create_sample_email_template,
sample_template as create_sample_template,
sample_service, sample_template, sample_email_template)
sample_service_whitelist as create_sample_service_whitelist,
sample_api_key as create_sample_api_key
)
def test_create_sms_should_reject_if_missing_required_fields(notify_api, sample_api_key, mocker):
@@ -1030,6 +1032,106 @@ def test_should_not_persist_notification_or_send_sms_if_simulated_number(
assert Notification.query.count() == 0
@pytest.mark.parametrize('notification_type,to, key_type', [
('sms', '07827992635', KEY_TYPE_NORMAL),
('email', 'non_whitelist_recipient@mail.com', KEY_TYPE_NORMAL),
('sms', '07827992635', KEY_TYPE_TEAM),
('email', 'non_whitelist_recipient@mail.com', KEY_TYPE_TEAM)])
def test_should_not_send_notification_to_non_whitelist_recipient_in_trial_mode(client,
notify_db,
notify_db_session,
notification_type,
to,
key_type,
mocker):
service = create_sample_service(notify_db, notify_db_session, limit=2, restricted=True)
service_whitelist = create_sample_service_whitelist(notify_db, notify_db_session, service=service)
if notification_type == 'sms':
apply_async = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
template = create_sample_template(notify_db, notify_db_session, service=service)
elif notification_type == 'email':
apply_async = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
template = create_sample_email_template(notify_db, notify_db_session, service=service)
assert service_whitelist.service_id == service.id
assert to not in [member.recipient for member in service.whitelist]
create_sample_notification(notify_db, notify_db_session, template=template, service=service)
data = {
'to': to,
'template': str(template.id)
}
api_key = create_sample_api_key(notify_db, notify_db_session, service, key_type=key_type)
auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
response = client.post(
path='/notifications/{}'.format(notification_type),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), ('Authorization', 'Bearer {}'.format(auth_header))])
expected_response_message = (
'Can’t send to this recipient when service is in trial mode '
'– see https://www.notifications.service.gov.uk/trial-mode'
) if key_type == KEY_TYPE_NORMAL else ('Can’t send to this recipient using a team-only API key')
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert json_resp['result'] == 'error'
assert expected_response_message in json_resp['message']['to']
apply_async.assert_not_called()
@pytest.mark.parametrize('notification_type,to', [
('sms', '07123123123'),
('email', 'whitelist_recipient@mail.com')])
def test_should_send_notification_to_whitelist_recipient_in_trial_mode_with_live_key(client,
notify_db,
notify_db_session,
notification_type,
to,
mocker):
service = create_sample_service(notify_db, notify_db_session, limit=2, restricted=True)
if notification_type == 'sms':
apply_async = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
template = create_sample_template(notify_db, notify_db_session, service=service)
service_whitelist = create_sample_service_whitelist(notify_db, notify_db_session,
service=service, mobile_number=to)
elif notification_type == 'email':
apply_async = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
template = create_sample_email_template(notify_db, notify_db_session, service=service)
service_whitelist = create_sample_service_whitelist(notify_db, notify_db_session,
service=service, email_address=to)
assert service_whitelist.service_id == service.id
assert to in [member.recipient for member in service.whitelist]
create_sample_notification(notify_db, notify_db_session, template=template, service=service)
data = {
'to': to,
'template': str(template.id)
}
sample_live_key = create_sample_api_key(notify_db, notify_db_session, service)
auth_header = create_jwt_token(secret=sample_live_key.unsigned_secret, client_id=str(sample_live_key.service_id))
response = client.post(
path='/notifications/{}'.format(notification_type),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), ('Authorization', 'Bearer {}'.format(auth_header))])
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 201
assert json_resp['data']['notification']['id']
assert json_resp['data']['body'] == template.content
assert json_resp['data']['template_version'] == template.version
apply_async.called
@pytest.mark.parametrize(
'notification_type, template_type, to', [
('email', 'sms', 'notify@digital.cabinet-office.gov.uk'),
@@ -1067,11 +1169,11 @@ def test_should_send_sms_in_research_mode_queue_if_research_mode_service(
with notify_api.test_client() as client:
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
service = sample_service(notify_db, notify_db_session)
service = create_sample_service(notify_db, notify_db_session)
service.research_mode = True
dao_update_service(service)
template = sample_template(notify_db, notify_db_session, service=service)
template = create_sample_template(notify_db, notify_db_session, service=service)
data = {
'to': service.created_by.mobile_number,
@@ -1099,11 +1201,11 @@ def test_should_send_email_in_research_mode_queue_if_research_mode_service(
with notify_api.test_client() as client:
mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
service = sample_service(notify_db, notify_db_session)
service = create_sample_service(notify_db, notify_db_session)
service.research_mode = True
dao_update_service(service)
template = sample_email_template(notify_db, notify_db_session, service=service)
template = create_sample_email_template(notify_db, notify_db_session, service=service)
data = {
'to': service.created_by.email_address,

95
tests/app/service/test_service_whitelist.py Normal file
View File

@@ -0,0 +1,95 @@
import uuid
import json
from tests import create_authorization_header
from app.models import (
ServiceWhitelist,
MOBILE_TYPE, EMAIL_TYPE)
from app.dao.service_whitelist_dao import dao_add_and_commit_whitelisted_contacts
def test_get_whitelist_returns_data(client, sample_service_whitelist):
service_id = sample_service_whitelist.service_id
response = client.get('service/{}/whitelist'.format(service_id), headers=[create_authorization_header()])
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == {
'email_addresses': [sample_service_whitelist.recipient],
'phone_numbers': []
}
def test_get_whitelist_separates_emails_and_phones(client, sample_service):
dao_add_and_commit_whitelisted_contacts([
ServiceWhitelist.from_string(sample_service.id, EMAIL_TYPE, 'service@example.com'),
ServiceWhitelist.from_string(sample_service.id, MOBILE_TYPE, '07123456789')
])
response = client.get('service/{}/whitelist'.format(sample_service.id), headers=[create_authorization_header()])
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == {
'email_addresses': ['service@example.com'],
'phone_numbers': ['07123456789']
}
def test_get_whitelist_404s_with_unknown_service_id(client):
path = 'service/{}/whitelist'.format(uuid.uuid4())
response = client.get(path, headers=[create_authorization_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'
def test_get_whitelist_returns_no_data(client, sample_service):
path = 'service/{}/whitelist'.format(sample_service.id)
response = client.get(path, headers=[create_authorization_header()])
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True)) == {'email_addresses': [], 'phone_numbers': []}
def test_update_whitelist_replaces_old_whitelist(client, sample_service_whitelist):
data = {
'email_addresses': ['foo@bar.com'],
'phone_numbers': ['07123456789']
}
response = client.put(
'service/{}/whitelist'.format(sample_service_whitelist.service_id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
)
assert response.status_code == 204
whitelist = ServiceWhitelist.query.order_by(ServiceWhitelist.recipient).all()
assert len(whitelist) == 2
assert whitelist[0].recipient == '07123456789'
assert whitelist[1].recipient == 'foo@bar.com'
def test_update_whitelist_doesnt_remove_old_whitelist_if_error(client, sample_service_whitelist):
data = {
'email_addresses': [''],
'phone_numbers': ['07123456789']
}
response = client.put(
'service/{}/whitelist'.format(sample_service_whitelist.service_id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
)
assert response.status_code == 400
assert json.loads(response.get_data(as_text=True)) == {
'result': 'error',
'message': 'Invalid whitelist: "" is not a valid email address or phone number'
}
whitelist = ServiceWhitelist.query.one()
assert whitelist.id == sample_service_whitelist.id

36
tests/app/test_model.py
View File

@@ -1,7 +1,12 @@
from datetime import datetime
import pytest
from app import DATETIME_FORMAT
from app.models import Notification
from app.models import (
Notification,
ServiceWhitelist,
MOBILE_TYPE, EMAIL_TYPE)
def test_should_build_notification_from_minimal_set_of_api_derived_params(notify_api):
@@ -70,3 +75,32 @@ def test_should_build_notification_from_full_set_of_api_derived_params(notify_ap
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'
])
def test_should_build_service_whitelist_from_mobile_number(mobile_number):
service_whitelist = ServiceWhitelist.from_string('service_id', MOBILE_TYPE, mobile_number)
assert service_whitelist.recipient == mobile_number
@pytest.mark.parametrize('email_address', [
'test@example.com'
])
def test_should_build_service_whitelist_from_email_address(email_address):
service_whitelist = ServiceWhitelist.from_string('service_id', EMAIL_TYPE, email_address)
assert service_whitelist.recipient == email_address
@pytest.mark.parametrize('contact, recipient_type', [
('', None),
('07700dsadsad', MOBILE_TYPE),
('gmail.com', EMAIL_TYPE)
])
def test_should_not_build_service_whitelist_from_invalid_contact(recipient_type, contact):
with pytest.raises(ValueError):
ServiceWhitelist.from_string('service_id', recipient_type, contact)