darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-08-16 20:48:37 -04:00
Code Issues Packages Projects Releases Wiki Activity

Stop calling fixures as functions in the tests

Browse Source
This commit is contained in:
Katie Smith
2019-10-31 15:02:30 +00:00
parent 97b807a99d
commit 04c1c35efb
7 changed files with 338 additions and 515 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
tests/app/invite/test_invite_rest.py
View File

@@ -3,6 +3,7 @@ import pytest
from app.models import Notification, SMS_AUTH_TYPE, EMAIL_AUTH_TYPE
from tests import create_authorization_header
from tests.app.db import create_invited_user
@pytest.mark.parametrize('extra_args, expected_start_of_invite_url', [
@@ -118,16 +119,11 @@ def test_create_invited_user_invalid_email(client, sample_service, mocker, fake_
def test_get_all_invited_users_by_service(client, notify_db, notify_db_session, sample_service):
from tests.app.conftest import sample_invited_user
invites = []
for i in range(0, 5):
email = 'invited_user_{}@service.gov.uk'.format(i)
invited_user = create_invited_user(sample_service, to_email_address=email)
invited_user = sample_invited_user(notify_db,
notify_db_session,
sample_service,
email)
invites.append(invited_user)
url = '/service/{}/invite'.format(sample_service.id)

136
tests/app/notifications/rest/test_callbacks.py
View File

@@ -11,8 +11,7 @@ from app.clients import ClientException
from app.dao.notifications_dao import (
get_notification_by_id
)
from tests.app.conftest import sample_notification as create_sample_notification
from tests.app.db import create_service_callback_api
from tests.app.db import create_notification, create_service_callback_api
def firetext_post(client, data):
@@ -161,15 +160,13 @@ def test_firetext_callback_should_return_400_if_no_status(client, mocker):
def test_firetext_callback_should_set_status_technical_failure_if_status_unknown(
client, notify_db, notify_db_session, mocker):
notification = create_sample_notification(
notify_db, notify_db_session, status='sending', sent_at=datetime.utcnow()
)
mocker.patch('app.statsd_client.incr')
data = 'mobile=441234123123&status=99&time=2016-03-10 14:17:00&reference={}'.format(notification.id)
client, mocker, sample_notification):
sample_notification.status = 'sending'
# mocker.patch('app.statsd_client.incr')
data = 'mobile=441234123123&status=99&time=2016-03-10 14:17:00&reference={}'.format(sample_notification.id)
with pytest.raises(ClientException) as e:
firetext_post(client, data)
assert get_notification_by_id(notification.id).status == 'technical-failure'
assert get_notification_by_id(sample_notification.id).status == 'technical-failure'
assert 'Firetext callback failed: status 99 not found.' in str(e.value)
@@ -201,52 +198,40 @@ def test_callback_should_return_200_if_cannot_find_notification_id(
def test_firetext_callback_should_update_notification_status(
notify_db, notify_db_session, client, sample_email_template, mocker
client, mocker, sample_notification
):
mocker.patch('app.statsd_client.incr')
send_mock = mocker.patch(
'app.celery.service_callback_tasks.send_delivery_status_to_service.apply_async'
)
notification = create_sample_notification(
notify_db,
notify_db_session,
template=sample_email_template,
reference='ref',
status='sending',
sent_at=datetime.utcnow())
sample_notification.status = 'sending'
original = get_notification_by_id(notification.id)
original = get_notification_by_id(sample_notification.id)
assert original.status == 'sending'
data = 'mobile=441234123123&status=0&time=2016-03-10 14:17:00&reference={}'.format(
notification.id)
sample_notification.id)
response = firetext_post(client, data)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert json_resp['result'] == 'success'
assert json_resp['message'] == 'Firetext callback succeeded. reference {} updated'.format(
notification.id
sample_notification.id
)
updated = get_notification_by_id(notification.id)
updated = get_notification_by_id(sample_notification.id)
assert updated.status == 'delivered'
assert get_notification_by_id(notification.id).status == 'delivered'
assert send_mock.called_once_with([notification.id], queue="notify-internal-tasks")
assert get_notification_by_id(sample_notification.id).status == 'delivered'
assert send_mock.called_once_with([sample_notification.id], queue="notify-internal-tasks")
def test_firetext_callback_should_update_notification_status_failed(
notify_db, notify_db_session, client, sample_template, mocker
client, mocker, sample_template
):
mocker.patch('app.statsd_client.incr')
mocker.patch(
'app.celery.service_callback_tasks.send_delivery_status_to_service.apply_async'
)
notification = create_sample_notification(
notify_db,
notify_db_session,
template=sample_template,
reference='ref',
status='sending',
sent_at=datetime.utcnow())
notification = create_notification(template=sample_template, status='sending')
original = get_notification_by_id(notification.id)
assert original.status == 'sending'
@@ -264,14 +249,13 @@ def test_firetext_callback_should_update_notification_status_failed(
assert get_notification_by_id(notification.id).status == 'permanent-failure'
def test_firetext_callback_should_update_notification_status_pending(client, notify_db, notify_db_session, mocker):
def test_firetext_callback_should_update_notification_status_pending(client, sample_template, mocker):
mocker.patch('app.statsd_client.incr')
mocker.patch(
'app.celery.service_callback_tasks.send_delivery_status_to_service.apply_async'
)
notification = create_sample_notification(
notify_db, notify_db_session, status='sending', sent_at=datetime.utcnow()
)
notification = create_notification(template=sample_template, status='sending')
original = get_notification_by_id(notification.id)
assert original.status == 'sending'
data = 'mobile=441234123123&status=2&time=2016-03-10 14:17:00&reference={}'.format(
@@ -299,16 +283,14 @@ def test_process_mmg_response_return_200_when_cid_is_send_sms_code(client):
def test_process_mmg_response_returns_200_when_cid_is_valid_notification_id(
notify_db, notify_db_session, client, mocker
sample_notification, client, mocker
):
mocker.patch(
'app.celery.service_callback_tasks.send_delivery_status_to_service.apply_async'
)
notification = create_sample_notification(
notify_db, notify_db_session, status='sending', sent_at=datetime.utcnow()
)
sample_notification.status = 'sending'
data = json.dumps({"reference": "mmg_reference",
"CID": str(notification.id),
"CID": str(sample_notification.id),
"MSISDN": "447777349060",
"status": "3",
"deliverytime": "2016-04-05 16:01:07"})
@@ -318,22 +300,20 @@ def test_process_mmg_response_returns_200_when_cid_is_valid_notification_id(
assert response.status_code == 200
json_data = json.loads(response.data)
assert json_data['result'] == 'success'
assert json_data['message'] == 'MMG callback succeeded. reference {} updated'.format(notification.id)
assert get_notification_by_id(notification.id).status == 'delivered'
assert json_data['message'] == 'MMG callback succeeded. reference {} updated'.format(sample_notification.id)
assert get_notification_by_id(sample_notification.id).status == 'delivered'
def test_process_mmg_response_status_5_updates_notification_with_permanently_failed(
notify_db, notify_db_session, client, mocker
sample_notification, client, mocker
):
mocker.patch(
'app.celery.service_callback_tasks.send_delivery_status_to_service.apply_async'
)
notification = create_sample_notification(
notify_db, notify_db_session, status='sending', sent_at=datetime.utcnow()
)
sample_notification.status = 'sending'
data = json.dumps({"reference": "mmg_reference",
"CID": str(notification.id),
"CID": str(sample_notification.id),
"MSISDN": "447777349060",
"status": 5})
@@ -341,21 +321,19 @@ def test_process_mmg_response_status_5_updates_notification_with_permanently_fai
assert response.status_code == 200
json_data = json.loads(response.data)
assert json_data['result'] == 'success'
assert json_data['message'] == 'MMG callback succeeded. reference {} updated'.format(notification.id)
assert get_notification_by_id(notification.id).status == 'permanent-failure'
assert json_data['message'] == 'MMG callback succeeded. reference {} updated'.format(sample_notification.id)
assert get_notification_by_id(sample_notification.id).status == 'permanent-failure'
def test_process_mmg_response_status_2_updates_notification_with_permanently_failed(
notify_db, notify_db_session, client, mocker
sample_notification, client, mocker
):
mocker.patch(
'app.celery.service_callback_tasks.send_delivery_status_to_service.apply_async'
)
notification = create_sample_notification(
notify_db, notify_db_session, status='sending', sent_at=datetime.utcnow()
)
sample_notification.status = 'sending'
data = json.dumps({"reference": "mmg_reference",
"CID": str(notification.id),
"CID": str(sample_notification.id),
"MSISDN": "447777349060",
"status": 2})
@@ -363,22 +341,20 @@ def test_process_mmg_response_status_2_updates_notification_with_permanently_fai
assert response.status_code == 200
json_data = json.loads(response.data)
assert json_data['result'] == 'success'
assert json_data['message'] == 'MMG callback succeeded. reference {} updated'.format(notification.id)
assert get_notification_by_id(notification.id).status == 'permanent-failure'
assert json_data['message'] == 'MMG callback succeeded. reference {} updated'.format(sample_notification.id)
assert get_notification_by_id(sample_notification.id).status == 'permanent-failure'
def test_process_mmg_response_status_4_updates_notification_with_temporary_failed(
notify_db, notify_db_session, client, mocker
sample_notification, client, mocker
):
mocker.patch(
'app.celery.service_callback_tasks.send_delivery_status_to_service.apply_async'
)
notification = create_sample_notification(
notify_db, notify_db_session, status='sending', sent_at=datetime.utcnow()
)
sample_notification.status = 'sending'
data = json.dumps({"reference": "mmg_reference",
"CID": str(notification.id),
"CID": str(sample_notification.id),
"MSISDN": "447777349060",
"status": 4})
@@ -386,28 +362,26 @@ def test_process_mmg_response_status_4_updates_notification_with_temporary_faile
assert response.status_code == 200
json_data = json.loads(response.data)
assert json_data['result'] == 'success'
assert json_data['message'] == 'MMG callback succeeded. reference {} updated'.format(notification.id)
assert get_notification_by_id(notification.id).status == 'temporary-failure'
assert json_data['message'] == 'MMG callback succeeded. reference {} updated'.format(sample_notification.id)
assert get_notification_by_id(sample_notification.id).status == 'temporary-failure'
def test_process_mmg_response_unknown_status_updates_notification_with_technical_failure(
notify_db, notify_db_session, client, mocker
sample_notification, client, mocker
):
send_mock = mocker.patch(
'app.celery.service_callback_tasks.send_delivery_status_to_service.apply_async'
)
notification = create_sample_notification(
notify_db, notify_db_session, status='sending', sent_at=datetime.utcnow()
)
sample_notification.status = 'sending'
data = json.dumps({"reference": "mmg_reference",
"CID": str(notification.id),
"CID": str(sample_notification.id),
"MSISDN": "447777349060",
"status": 10})
create_service_callback_api(service=notification.service, url="https://original_url.com")
create_service_callback_api(service=sample_notification.service, url="https://original_url.com")
with pytest.raises(ClientException) as e:
mmg_post(client, data)
assert 'MMG callback failed: status 10 not found.' in str(e.value)
assert get_notification_by_id(notification.id).status == 'technical-failure'
assert get_notification_by_id(sample_notification.id).status == 'technical-failure'
assert send_mock.called
@@ -445,7 +419,7 @@ def test_mmg_callback_returns_400_when_notification_id_is_not_a_valid_uuid(clien
assert json_resp['message'] == 'MMG callback with invalid reference 1234'
def test_process_mmg_response_records_statsd(notify_db, notify_db_session, client, mocker):
def test_process_mmg_response_records_statsd(sample_notification, client, mocker):
with freeze_time('2001-01-01T12:00:00'):
mocker.patch('app.statsd_client.incr')
@@ -453,12 +427,11 @@ def test_process_mmg_response_records_statsd(notify_db, notify_db_session, clien
mocker.patch(
'app.celery.service_callback_tasks.send_delivery_status_to_service.apply_async'
)
notification = create_sample_notification(
notify_db, notify_db_session, status='sending', sent_at=datetime.utcnow()
)
sample_notification.status = 'sending'
sample_notification.sent_at = datetime.now()
data = json.dumps({"reference": "mmg_reference",
"CID": str(notification.id),
"CID": str(sample_notification.id),
"MSISDN": "447777349060",
"status": "3",
"deliverytime": "2016-04-05 16:01:07"})
@@ -467,11 +440,11 @@ def test_process_mmg_response_records_statsd(notify_db, notify_db_session, clien
app.statsd_client.incr.assert_any_call("callback.mmg.delivered")
app.statsd_client.timing_with_dates.assert_any_call(
"callback.mmg.elapsed-time", datetime.utcnow(), notification.sent_at
"callback.mmg.elapsed-time", datetime.utcnow(), sample_notification.sent_at
)
def test_firetext_callback_should_record_statsd(client, notify_db, notify_db_session, mocker):
def test_firetext_callback_should_record_statsd(client, sample_notification, mocker):
with freeze_time('2001-01-01T12:00:00'):
mocker.patch('app.statsd_client.incr')
@@ -479,16 +452,15 @@ def test_firetext_callback_should_record_statsd(client, notify_db, notify_db_ses
mocker.patch(
'app.celery.service_callback_tasks.send_delivery_status_to_service.apply_async'
)
notification = create_sample_notification(
notify_db, notify_db_session, status='sending', sent_at=datetime.utcnow()
)
sample_notification.status = 'sending'
sample_notification.sent_at = datetime.now()
data = 'mobile=441234123123&status=0&time=2016-03-10 14:17:00&code=101&reference={}'.format(
notification.id)
sample_notification.id)
firetext_post(client, data)
app.statsd_client.timing_with_dates.assert_any_call(
"callback.firetext.elapsed-time", datetime.utcnow(), notification.sent_at
"callback.firetext.elapsed-time", datetime.utcnow(), sample_notification.sent_at
)
app.statsd_client.incr.assert_any_call("callback.firetext.delivered")

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

@@ -1,7 +1,6 @@
import random
import string
import pytest
from datetime import datetime
from flask import (json, current_app)
from freezegun import freeze_time
@@ -11,7 +10,7 @@ from notifications_utils import SMS_CHAR_COUNT_LIMIT
import app
from app.dao import notifications_dao
from app.models import (
INTERNATIONAL_SMS_TYPE, SMS_TYPE, EMAIL_TYPE,
SMS_TYPE, EMAIL_TYPE,
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
@@ -22,16 +21,14 @@ from app.models import Template
from app.v2.errors import RateLimitError, TooManyRequestsError
from tests import create_authorization_header
from tests.app.conftest import (
sample_notification as create_sample_notification,
sample_service as create_sample_service,
sample_email_template as create_sample_email_template,
sample_template as create_sample_template,
sample_service_whitelist as create_sample_service_whitelist,
sample_api_key as create_sample_api_key,
sample_service,
from tests.app.db import (
create_api_key,
create_notification,
create_service,
create_service_whitelist,
create_template,
create_reply_to_email,
)
from tests.app.db import create_service, create_reply_to_email
@pytest.mark.parametrize('template_type',
@@ -409,7 +406,6 @@ def test_should_allow_valid_email_notification(notify_api, sample_email_template
@freeze_time("2016-01-01 12:00:00.061258")
def test_should_block_api_call_if_over_day_limit_for_live_service(
notify_db,
notify_db_session,
notify_api,
mocker):
@@ -420,12 +416,9 @@ def test_should_block_api_call_if_over_day_limit_for_live_service(
side_effect=TooManyRequestsError(1)
)
mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
service = create_sample_service(notify_db, notify_db_session, limit=1, restricted=False)
email_template = create_sample_email_template(notify_db, notify_db_session, service=service)
create_sample_notification(
notify_db, notify_db_session, template=email_template, service=service, created_at=datetime.utcnow()
)
service = create_service(message_limit=1)
email_template = create_template(service, template_type=EMAIL_TYPE)
create_notification(template=email_template)
data = {
'to': 'ok@ok.com',
@@ -444,7 +437,6 @@ def test_should_block_api_call_if_over_day_limit_for_live_service(
@freeze_time("2016-01-01 12:00:00.061258")
def test_should_block_api_call_if_over_day_limit_for_restricted_service(
notify_db,
notify_db_session,
notify_api,
mocker):
@@ -455,12 +447,9 @@ def test_should_block_api_call_if_over_day_limit_for_restricted_service(
'app.notifications.validators.check_service_over_daily_message_limit',
side_effect=TooManyRequestsError(1)
)
service = create_sample_service(notify_db, notify_db_session, limit=1, restricted=True)
email_template = create_sample_email_template(notify_db, notify_db_session, service=service)
create_sample_notification(
notify_db, notify_db_session, template=email_template, service=service, created_at=datetime.utcnow()
)
service = create_service(restricted=True, message_limit=1)
email_template = create_template(service, template_type=EMAIL_TYPE)
create_notification(template=email_template)
data = {
'to': 'ok@ok.com',
@@ -481,8 +470,6 @@ def test_should_block_api_call_if_over_day_limit_for_restricted_service(
@pytest.mark.parametrize('restricted', [True, False])
@freeze_time("2016-01-01 12:00:00.061258")
def test_should_allow_api_call_if_under_day_limit_regardless_of_type(
notify_db,
notify_db_session,
notify_api,
sample_user,
mocker,
@@ -491,10 +478,10 @@ def test_should_allow_api_call_if_under_day_limit_regardless_of_type(
with notify_api.test_client() as client:
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
service = create_sample_service(notify_db, notify_db_session, limit=2, restricted=restricted)
email_template = create_sample_email_template(notify_db, notify_db_session, service=service)
sms_template = create_sample_template(notify_db, notify_db_session, service=service)
create_sample_notification(notify_db, notify_db_session, template=email_template, service=service)
service = create_service(restricted=restricted, message_limit=2)
email_template = create_template(service, template_type=EMAIL_TYPE)
sms_template = create_template(service, template_type=SMS_TYPE)
create_notification(template=email_template)
data = {
'to': sample_user.mobile_number,
@@ -511,11 +498,11 @@ def test_should_allow_api_call_if_under_day_limit_regardless_of_type(
assert response.status_code == 201
def test_should_not_return_html_in_body(notify_api, notify_db, notify_db_session, mocker):
def test_should_not_return_html_in_body(notify_api, sample_service, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
email_template = create_sample_email_template(notify_db, notify_db_session, content='hello\nthere')
email_template = create_template(sample_service, template_type=EMAIL_TYPE, content='hello\nthere')
data = {
'to': 'ok@ok.com',
@@ -850,36 +837,35 @@ def test_should_not_persist_notification_or_send_sms_if_simulated_number(
@pytest.mark.parametrize('key_type', [
KEY_TYPE_NORMAL, KEY_TYPE_TEAM
])
@pytest.mark.parametrize('notification_type, to, _create_sample_template', [
(SMS_TYPE, '07827992635', create_sample_template),
(EMAIL_TYPE, 'non_whitelist_recipient@mail.com', create_sample_email_template)]
@pytest.mark.parametrize('notification_type, to', [
(SMS_TYPE, '07827992635'),
(EMAIL_TYPE, 'non_whitelist_recipient@mail.com')]
)
def test_should_not_send_notification_to_non_whitelist_recipient_in_trial_mode(
client,
notify_db,
notify_db_session,
sample_service_whitelist,
notification_type,
to,
_create_sample_template,
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)
service = sample_service_whitelist.service
service.restricted = True
service.message_limit = 2
apply_async = mocker.patch('app.celery.provider_tasks.deliver_{}.apply_async'.format(notification_type))
template = _create_sample_template(notify_db, notify_db_session, service=service)
assert service_whitelist.service_id == service.id
template = create_template(service, template_type=notification_type)
assert sample_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)
create_notification(template=template)
data = {
'to': to,
'template': str(template.id)
}
api_key = create_sample_api_key(notify_db, notify_db_session, service, key_type=key_type)
api_key = create_api_key(service, key_type=key_type)
auth_header = create_jwt_token(secret=api_key.secret, client_id=str(api_key.service_id))
response = client.post(
@@ -905,42 +891,40 @@ def test_should_not_send_notification_to_non_whitelist_recipient_in_trial_mode(
@pytest.mark.parametrize('key_type', [
KEY_TYPE_NORMAL, KEY_TYPE_TEAM
])
@pytest.mark.parametrize('notification_type, to, _create_sample_template', [
(SMS_TYPE, '07123123123', create_sample_template),
(EMAIL_TYPE, 'whitelist_recipient@mail.com', create_sample_email_template)]
@pytest.mark.parametrize('notification_type, to', [
(SMS_TYPE, '07123123123'),
(EMAIL_TYPE, 'whitelist_recipient@mail.com')]
)
def test_should_send_notification_to_whitelist_recipient(
client,
notify_db,
notify_db_session,
sample_service,
notification_type,
to,
_create_sample_template,
key_type,
service_restricted,
mocker
):
service = create_sample_service(notify_db, notify_db_session, limit=2, restricted=service_restricted)
sample_service.message_limit = 2
sample_service.restricted = service_restricted
apply_async = mocker.patch('app.celery.provider_tasks.deliver_{}.apply_async'.format(notification_type))
template = _create_sample_template(notify_db, notify_db_session, service=service)
template = create_template(sample_service, template_type=notification_type)
if notification_type == SMS_TYPE:
service_whitelist = create_sample_service_whitelist(notify_db, notify_db_session,
service=service, mobile_number=to)
service_whitelist = create_service_whitelist(sample_service, mobile_number=to)
elif notification_type == EMAIL_TYPE:
service_whitelist = create_sample_service_whitelist(notify_db, notify_db_session,
service=service, email_address=to)
service_whitelist = create_service_whitelist(sample_service, email_address=to)
assert service_whitelist.service_id == service.id
assert to in [member.recipient for member in service.whitelist]
assert service_whitelist.service_id == sample_service.id
assert to in [member.recipient for member in sample_service.whitelist]
create_sample_notification(notify_db, notify_db_session, template=template, service=service)
create_notification(template=template)
data = {
'to': to,
'template': str(template.id)
}
sample_key = create_sample_api_key(notify_db, notify_db_session, service, key_type=key_type)
sample_key = create_api_key(sample_service, key_type=key_type)
auth_header = create_jwt_token(secret=sample_key.secret, client_id=str(sample_key.service_id))
response = client.post(
@@ -963,13 +947,12 @@ def test_should_send_notification_to_whitelist_recipient(
])
def test_should_error_if_notification_type_does_not_match_template_type(
client,
notify_db,
notify_db_session,
sample_service,
template_type,
notification_type,
to
):
template = create_sample_template(notify_db, notify_db_session, template_type=template_type)
template = create_template(sample_service, template_type=template_type)
data = {
'to': to,
'template': template.id
@@ -1010,17 +993,11 @@ def test_create_template_doesnt_raise_with_too_much_personalisation(
]
)
def test_create_template_raises_invalid_request_when_content_too_large(
notify_db,
notify_db_session,
sample_service,
template_type,
should_error
):
sample = create_sample_template(
notify_db,
notify_db_session,
content="((long_text))",
template_type=template_type
)
sample = create_template(sample_service, template_type=template_type, content="((long_text))")
template = Template.query.get(sample.id)
from app.notifications.rest import create_template_object_for_notification
try:
@@ -1041,15 +1018,14 @@ def test_create_template_raises_invalid_request_when_content_too_large(
@pytest.mark.parametrize("notification_type, send_to",
[("sms", "07700 900 855"),
("email", "sample@email.com")])
def test_send_notification_uses_priority_queue_when_template_is_marked_as_priority(client, notify_db,
notify_db_session, mocker,
notification_type, send_to):
sample = create_sample_template(
notify_db,
notify_db_session,
template_type=notification_type,
process_type='priority'
)
def test_send_notification_uses_priority_queue_when_template_is_marked_as_priority(
client,
sample_service,
mocker,
notification_type,
send_to,
):
sample = create_template(sample_service, template_type=notification_type, process_type='priority')
mocked = mocker.patch('app.celery.provider_tasks.deliver_{}.apply_async'.format(notification_type))
data = {
@@ -1077,17 +1053,12 @@ def test_send_notification_uses_priority_queue_when_template_is_marked_as_priori
)
def test_returns_a_429_limit_exceeded_if_rate_limit_exceeded(
client,
notify_db,
notify_db_session,
sample_service,
mocker,
notification_type,
send_to
):
sample = create_sample_template(
notify_db,
notify_db_session,
template_type=notification_type
)
sample = create_template(sample_service, template_type=notification_type)
persist_mock = mocker.patch('app.notifications.rest.persist_notification')
deliver_mock = mocker.patch('app.notifications.rest.send_notification_to_queue')
@@ -1165,18 +1136,16 @@ def test_should_not_allow_international_number_on_sms_notification(client, sampl
assert error_json['message']['to'][0] == 'Cannot send to international mobile numbers'
def test_should_allow_international_number_on_sms_notification(client, notify_db, notify_db_session, mocker):
def test_should_allow_international_number_on_sms_notification(client, sample_service_full_permissions, mocker):
mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
service = sample_service(notify_db, notify_db_session, permissions=[INTERNATIONAL_SMS_TYPE, SMS_TYPE])
template = create_sample_template(notify_db, notify_db_session, service=service)
template = create_template(sample_service_full_permissions)
data = {
'to': '20-12-1234-1234',
'template': str(template.id)
}
auth_header = create_authorization_header(service_id=service.id)
auth_header = create_authorization_header(service_id=sample_service_full_permissions.id)
response = client.post(
path='/notifications/sms',
@@ -1259,16 +1228,14 @@ def test_should_throw_exception_if_notification_type_is_invalid(client, sample_s
("email", "test@gov.uk")
]
)
def test_post_notification_should_set_reply_to_text(client, notify_db, notify_db_session, mocker, notification_type,
def test_post_notification_should_set_reply_to_text(client, sample_service, mocker, notification_type,
recipient):
mocker.patch('app.celery.provider_tasks.deliver_{}.apply_async'.format(notification_type))
service = create_service()
template = create_sample_template(notify_db=notify_db, notify_db_session=notify_db_session,
service=service, template_type=notification_type)
template = create_template(sample_service, template_type=notification_type)
expected_reply_to = current_app.config['FROM_NUMBER']
if notification_type == EMAIL_TYPE:
expected_reply_to = 'reply_to@gov.uk'
create_reply_to_email(service=service, email_address=expected_reply_to, is_default=True)
create_reply_to_email(service=sample_service, email_address=expected_reply_to, is_default=True)
data = {
'to': recipient,
@@ -1277,7 +1244,7 @@ def test_post_notification_should_set_reply_to_text(client, notify_db, notify_db
response = client.post("/notifications/{}".format(notification_type),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'),
create_authorization_header(service_id=service.id)]
create_authorization_header(service_id=sample_service.id)]
)
assert response.status_code == 201
notifications = Notification.query.all()

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

@@ -11,7 +11,6 @@ from app.dao.templates_dao import dao_update_template
from app.models import ApiKey, KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST
from tests import create_authorization_header
from tests.app.conftest import sample_notification as create_sample_notification
from tests.app.db import create_notification, create_api_key
@@ -158,17 +157,11 @@ def test_get_all_notifications(client, sample_notification):
def test_normal_api_key_returns_notifications_created_from_jobs_and_from_api(
client,
notify_db,
notify_db_session,
sample_template,
sample_api_key,
sample_notification
):
api_notification = create_sample_notification(
notify_db,
notify_db_session,
api_key=sample_api_key
)
api_notification.job = None
api_notification = create_notification(template=sample_template, api_key=sample_api_key)
response = client.get(
path='/notifications',
@@ -184,45 +177,25 @@ def test_normal_api_key_returns_notifications_created_from_jobs_and_from_api(
@pytest.mark.parametrize('key_type', [KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST])
def test_get_all_notifications_only_returns_notifications_of_matching_type(
client,
notify_db,
notify_db_session,
sample_service,
sample_template,
sample_api_key,
sample_test_api_key,
sample_team_api_key,
key_type
):
team_api_key = ApiKey(service=sample_service,
name='team_api_key',
created_by=sample_service.created_by,
key_type=KEY_TYPE_TEAM)
save_model_api_key(team_api_key)
normal_api_key = ApiKey(service=sample_service,
name='normal_api_key',
created_by=sample_service.created_by,
key_type=KEY_TYPE_NORMAL)
save_model_api_key(normal_api_key)
test_api_key = ApiKey(service=sample_service,
name='test_api_key',
created_by=sample_service.created_by,
key_type=KEY_TYPE_TEST)
save_model_api_key(test_api_key)
normal_notification = create_sample_notification(
notify_db,
notify_db_session,
api_key=normal_api_key,
normal_notification = create_notification(
sample_template,
api_key=sample_api_key,
key_type=KEY_TYPE_NORMAL
)
team_notification = create_sample_notification(
notify_db,
notify_db_session,
api_key=team_api_key,
team_notification = create_notification(
sample_template,
api_key=sample_team_api_key,
key_type=KEY_TYPE_TEAM
)
test_notification = create_sample_notification(
notify_db,
notify_db_session,
api_key=test_api_key,
test_notification = create_notification(
sample_template,
api_key=sample_test_api_key,
key_type=KEY_TYPE_TEST
)
@@ -283,51 +256,26 @@ def test_do_not_return_job_notifications_by_default(
])
def test_only_normal_api_keys_can_return_job_notifications(
client,
notify_db,
notify_db_session,
sample_service,
sample_job,
sample_notification_with_job,
sample_template,
sample_api_key,
sample_team_api_key,
sample_test_api_key,
key_type
):
team_api_key = ApiKey(service=sample_service,
name='team_api_key',
created_by=sample_service.created_by,
key_type=KEY_TYPE_TEAM)
save_model_api_key(team_api_key)
normal_api_key = ApiKey(service=sample_service,
name='normal_api_key',
created_by=sample_service.created_by,
key_type=KEY_TYPE_NORMAL)
save_model_api_key(normal_api_key)
test_api_key = ApiKey(service=sample_service,
name='test_api_key',
created_by=sample_service.created_by,
key_type=KEY_TYPE_TEST)
save_model_api_key(test_api_key)
create_sample_notification(
notify_db,
notify_db_session,
job=sample_job
)
normal_notification = create_sample_notification(
notify_db,
notify_db_session,
api_key=normal_api_key,
normal_notification = create_notification(
template=sample_template,
api_key=sample_api_key,
key_type=KEY_TYPE_NORMAL
)
team_notification = create_sample_notification(
notify_db,
notify_db_session,
api_key=team_api_key,
team_notification = create_notification(
template=sample_template,
api_key=sample_team_api_key,
key_type=KEY_TYPE_TEAM
)
test_notification = create_sample_notification(
notify_db,
notify_db_session,
api_key=test_api_key,
test_notification = create_notification(
template=sample_template,
api_key=sample_test_api_key,
key_type=KEY_TYPE_TEST
)
@@ -347,10 +295,10 @@ def test_only_normal_api_keys_can_return_job_notifications(
assert notifications[0]['id'] == str(notification_objs[key_type[0]].id)
def test_get_all_notifications_newest_first(client, notify_db, notify_db_session, sample_email_template):
notification_1 = create_sample_notification(notify_db, notify_db_session, sample_email_template.service)
notification_2 = create_sample_notification(notify_db, notify_db_session, sample_email_template.service)
notification_3 = create_sample_notification(notify_db, notify_db_session, sample_email_template.service)
def test_get_all_notifications_newest_first(client, sample_email_template):
notification_1 = create_notification(template=sample_email_template)
notification_2 = create_notification(template=sample_email_template)
notification_3 = create_notification(template=sample_email_template)
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
@@ -379,10 +327,10 @@ def test_should_reject_invalid_page_param(client, sample_email_template):
assert 'Not a valid integer.' in notifications['message']['page']
def test_valid_page_size_param(notify_api, notify_db, notify_db_session, sample_email_template):
def test_valid_page_size_param(notify_api, sample_email_template):
with notify_api.test_request_context():
create_sample_notification(notify_db, notify_db_session)
create_sample_notification(notify_db, notify_db_session)
create_notification(sample_email_template)
create_notification(sample_email_template)
with notify_api.test_client() as client:
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
@@ -397,10 +345,10 @@ def test_valid_page_size_param(notify_api, notify_db, notify_db_session, sample_
assert notifications['page_size'] == 1
def test_invalid_page_size_param(client, notify_db, notify_db_session, sample_email_template):
def test_invalid_page_size_param(client, sample_email_template):
create_notification(sample_email_template)
create_notification(sample_email_template)
create_sample_notification(notify_db, notify_db_session)
create_sample_notification(notify_db, notify_db_session)
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
response = client.get(
@@ -413,15 +361,15 @@ def test_invalid_page_size_param(client, notify_db, notify_db_session, sample_em
assert 'Not a valid integer.' in notifications['message']['page_size']
def test_should_return_pagination_links(client, notify_db, notify_db_session, sample_email_template):
def test_should_return_pagination_links(client, sample_email_template):
# Effectively mocking page size
original_page_size = current_app.config['API_PAGE_SIZE']
try:
current_app.config['API_PAGE_SIZE'] = 1
create_sample_notification(notify_db, notify_db_session, sample_email_template.service)
notification_2 = create_sample_notification(notify_db, notify_db_session, sample_email_template.service)
create_sample_notification(notify_db, notify_db_session, sample_email_template.service)
create_notification(sample_email_template)
notification_2 = create_notification(sample_email_template)
create_notification(sample_email_template)
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
@@ -453,17 +401,9 @@ def test_get_all_notifications_returns_empty_list(client, sample_api_key):
assert len(notifications['notifications']) == 0
def test_filter_by_template_type(client, notify_db, notify_db_session, sample_template, sample_email_template):
create_sample_notification(
notify_db,
notify_db_session,
service=sample_email_template.service,
template=sample_template)
create_sample_notification(
notify_db,
notify_db_session,
service=sample_email_template.service,
template=sample_email_template)
def test_filter_by_template_type(client, sample_template, sample_email_template):
create_notification(sample_template)
create_notification(sample_email_template)
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
@@ -584,23 +524,12 @@ def test_get_notification_by_id_returns_merged_template_content_for_email(
assert notification['content_char_count'] is None
def test_get_notifications_for_service_returns_merged_template_content(client,
notify_db,
notify_db_session,
sample_template_with_placeholders):
def test_get_notifications_for_service_returns_merged_template_content(client, sample_template_with_placeholders):
with freeze_time('2001-01-01T12:00:00'):
create_sample_notification(notify_db,
notify_db_session,
service=sample_template_with_placeholders.service,
template=sample_template_with_placeholders,
personalisation={"name": "merged with first"})
create_notification(sample_template_with_placeholders, personalisation={"name": "merged with first"})
with freeze_time('2001-01-01T12:00:01'):
create_sample_notification(notify_db,
notify_db_session,
service=sample_template_with_placeholders.service,
template=sample_template_with_placeholders,
personalisation={"name": "merged with second"})
create_notification(sample_template_with_placeholders, personalisation={"name": "merged with second"})
auth_header = create_authorization_header(service_id=sample_template_with_placeholders.service_id)
@@ -617,23 +546,14 @@ def test_get_notifications_for_service_returns_merged_template_content(client,
def test_get_notification_selects_correct_template_for_personalisation(client,
notify_db,
notify_db_session,
sample_template):
create_sample_notification(notify_db,
notify_db_session,
service=sample_template.service,
template=sample_template)
create_notification(sample_template)
original_content = sample_template.content
sample_template.content = '((name))'
dao_update_template(sample_template)
notify_db.session.commit()
create_sample_notification(notify_db,
notify_db_session,
service=sample_template.service,
template=sample_template,
personalisation={"name": "foo"})
create_notification(sample_template, personalisation={"name": "foo"})
auth_header = create_authorization_header(service_id=sample_template.service_id)

118
tests/app/notifications/test_validators.py
View File

@@ -4,7 +4,7 @@ from flask import current_app
from notifications_utils import SMS_CHAR_COUNT_LIMIT
import app
from app.models import INTERNATIONAL_SMS_TYPE, SMS_TYPE, EMAIL_TYPE, LETTER_TYPE
from app.models import SMS_TYPE, EMAIL_TYPE, LETTER_TYPE
from app.notifications.validators import (
check_service_over_daily_message_limit,
check_template_is_for_notification_type,
@@ -25,12 +25,16 @@ from app.v2.errors import (
RateLimitError)
from tests.conftest import set_config
from tests.app.conftest import (
sample_notification as create_notification,
sample_service as create_service,
sample_service_whitelist,
sample_api_key)
from tests.app.db import create_reply_to_email, create_service_sms_sender, create_letter_contact
from tests.app.db import (
create_api_key,
create_letter_contact,
create_notification,
create_reply_to_email,
create_service,
create_service_sms_sender,
create_service_whitelist,
create_template,
)
# all of these tests should have redis enabled (except where we specifically disable it)
@@ -76,14 +80,13 @@ def test_should_not_interact_with_cache_for_test_key(sample_service, mocker):
@pytest.mark.parametrize('key_type', ['team', 'normal'])
def test_should_set_cache_value_as_value_from_database_if_cache_not_set(
key_type,
notify_db,
notify_db_session,
sample_template,
sample_service,
mocker
):
with freeze_time("2016-01-01 12:00:00.000000"):
for x in range(5):
create_notification(notify_db, notify_db_session, service=sample_service)
create_notification(sample_template)
mocker.patch('app.notifications.validators.redis_store.get', return_value=None)
mocker.patch('app.notifications.validators.redis_store.set')
check_service_over_daily_message_limit(key_type, sample_service)
@@ -102,27 +105,29 @@ def test_should_not_access_database_if_redis_disabled(notify_api, sample_service
@pytest.mark.parametrize('key_type', ['team', 'normal'])
def test_check_service_message_limit_over_message_limit_fails(key_type, notify_db, notify_db_session, mocker):
def test_check_service_message_limit_over_message_limit_fails(key_type, sample_service, mocker):
with freeze_time("2016-01-01 12:00:00.000000"):
mocker.patch('app.redis_store.get', return_value=None)
mocker.patch('app.notifications.validators.redis_store.set')
service = create_service(notify_db, notify_db_session, restricted=True, limit=4)
sample_service.restricted = True
sample_service.message_limit = 4
template = create_template(sample_service)
for x in range(5):
create_notification(notify_db, notify_db_session, service=service)
create_notification(template)
with pytest.raises(TooManyRequestsError) as e:
check_service_over_daily_message_limit(key_type, service)
check_service_over_daily_message_limit(key_type, sample_service)
assert e.value.status_code == 429
assert e.value.message == 'Exceeded send limits (4) for today'
assert e.value.fields == []
app.notifications.validators.redis_store.set.assert_called_with(
str(service.id) + "-2016-01-01-count", 5, ex=3600
str(sample_service.id) + "-2016-01-01-count", 5, ex=3600
)
@pytest.mark.parametrize('key_type', ['team', 'normal'])
def test_check_service_message_limit_in_cache_over_message_limit_fails(
notify_db,
notify_db_session,
key_type,
mocker):
@@ -131,7 +136,7 @@ def test_check_service_message_limit_in_cache_over_message_limit_fails(
mocker.patch('app.notifications.validators.redis_store.set')
mocker.patch('app.notifications.validators.services_dao')
service = create_service(notify_db, notify_db_session, restricted=True, limit=4)
service = create_service(restricted=True, message_limit=4)
with pytest.raises(TooManyRequestsError) as e:
check_service_over_daily_message_limit(key_type, service)
assert e.value.status_code == 429
@@ -180,8 +185,8 @@ def test_check_template_is_active_fails(sample_template):
@pytest.mark.parametrize('key_type',
['test', 'normal'])
def test_service_can_send_to_recipient_passes(key_type, notify_db, notify_db_session):
trial_mode_service = create_service(notify_db, notify_db_session, service_name='trial mode', restricted=True)
def test_service_can_send_to_recipient_passes(key_type, notify_db_session):
trial_mode_service = create_service(service_name='trial mode', restricted=True)
assert service_can_send_to_recipient(trial_mode_service.users[0].email_address,
key_type,
trial_mode_service) is None
@@ -192,23 +197,21 @@ def test_service_can_send_to_recipient_passes(key_type, notify_db, notify_db_ses
@pytest.mark.parametrize('key_type',
['test', 'normal'])
def test_service_can_send_to_recipient_passes_for_live_service_non_team_member(key_type, notify_db, notify_db_session):
live_service = create_service(notify_db, notify_db_session, service_name='live', restricted=False)
def test_service_can_send_to_recipient_passes_for_live_service_non_team_member(key_type, sample_service):
assert service_can_send_to_recipient("some_other_email@test.com",
key_type,
live_service) is None
sample_service) is None
assert service_can_send_to_recipient('07513332413',
key_type,
live_service) is None
sample_service) is None
def test_service_can_send_to_recipient_passes_for_whitelisted_recipient_passes(notify_db, notify_db_session,
sample_service):
sample_service_whitelist(notify_db, notify_db_session, email_address="some_other_email@test.com")
def test_service_can_send_to_recipient_passes_for_whitelisted_recipient_passes(sample_service):
create_service_whitelist(sample_service, email_address="some_other_email@test.com")
assert service_can_send_to_recipient("some_other_email@test.com",
'team',
sample_service) is None
sample_service_whitelist(notify_db, notify_db_session, mobile_number='07513332413')
create_service_whitelist(sample_service, mobile_number='07513332413')
assert service_can_send_to_recipient('07513332413',
'team',
sample_service) is None
@@ -224,7 +227,7 @@ def test_service_can_send_to_recipient_fails_when_ignoring_whitelist(
sample_service,
recipient,
):
sample_service_whitelist(notify_db, notify_db_session, **recipient)
create_service_whitelist(sample_service, **recipient)
with pytest.raises(BadRequestError) as exec_info:
service_can_send_to_recipient(
next(iter(recipient.values())),
@@ -242,9 +245,13 @@ def test_service_can_send_to_recipient_fails_when_ignoring_whitelist(
[('team', 'Can’t send to this recipient using a team-only API key'),
('normal',
"Can’t send to this recipient when service is in trial mode – see https://www.notifications.service.gov.uk/trial-mode")]) # noqa
def test_service_can_send_to_recipient_fails_when_recipient_is_not_on_team(recipient, key_type, error_message,
notify_db, notify_db_session):
trial_mode_service = create_service(notify_db, notify_db_session, service_name='trial mode', restricted=True)
def test_service_can_send_to_recipient_fails_when_recipient_is_not_on_team(
recipient,
key_type,
error_message,
notify_db_session,
):
trial_mode_service = create_service(service_name='trial mode', restricted=True)
with pytest.raises(BadRequestError) as exec_info:
service_can_send_to_recipient(recipient,
key_type,
@@ -254,12 +261,11 @@ def test_service_can_send_to_recipient_fails_when_recipient_is_not_on_team(recip
assert exec_info.value.fields == []
def test_service_can_send_to_recipient_fails_when_mobile_number_is_not_on_team(notify_db, notify_db_session):
live_service = create_service(notify_db, notify_db_session, service_name='live mode', restricted=False)
def test_service_can_send_to_recipient_fails_when_mobile_number_is_not_on_team(sample_service):
with pytest.raises(BadRequestError) as e:
service_can_send_to_recipient("0758964221",
'team',
live_service)
sample_service)
assert e.value.status_code == 400
assert e.value.message == 'Can’t send to this recipient using a team-only API key'
assert e.value.fields == []
@@ -282,9 +288,8 @@ def test_check_sms_content_char_count_fails(char_count, notify_api):
@pytest.mark.parametrize('key_type', ['team', 'live', 'test'])
def test_that_when_exceed_rate_limit_request_fails(
notify_db,
notify_db_session,
key_type,
sample_service,
mocker):
with freeze_time("2016-01-01 12:00:00.000000"):
@@ -296,45 +301,44 @@ def test_that_when_exceed_rate_limit_request_fails(
mocker.patch('app.redis_store.exceeded_rate_limit', return_value=True)
mocker.patch('app.notifications.validators.services_dao')
service = create_service(notify_db, notify_db_session, restricted=True)
api_key = sample_api_key(notify_db, notify_db_session, service=service, key_type=api_key_type)
sample_service.restricted = True
api_key = create_api_key(sample_service, key_type=api_key_type)
with pytest.raises(RateLimitError) as e:
check_service_over_api_rate_limit(service, api_key)
check_service_over_api_rate_limit(sample_service, api_key)
assert app.redis_store.exceeded_rate_limit.called_with(
"{}-{}".format(str(service.id), api_key.key_type),
service.rate_limit,
"{}-{}".format(str(sample_service.id), api_key.key_type),
sample_service.rate_limit,
60
)
assert e.value.status_code == 429
assert e.value.message == 'Exceeded rate limit for key type {} of {} requests per {} seconds'.format(
key_type.upper(), service.rate_limit, 60
key_type.upper(), sample_service.rate_limit, 60
)
assert e.value.fields == []
def test_that_when_not_exceeded_rate_limit_request_succeeds(
notify_db,
notify_db_session,
sample_service,
mocker):
with freeze_time("2016-01-01 12:00:00.000000"):
mocker.patch('app.redis_store.exceeded_rate_limit', return_value=False)
mocker.patch('app.notifications.validators.services_dao')
service = create_service(notify_db, notify_db_session, restricted=True)
api_key = sample_api_key(notify_db, notify_db_session, service=service, key_type='normal')
sample_service.restricted = True
api_key = create_api_key(sample_service)
check_service_over_api_rate_limit(service, api_key)
check_service_over_api_rate_limit(sample_service, api_key)
assert app.redis_store.exceeded_rate_limit.called_with(
"{}-{}".format(str(service.id), api_key.key_type),
"{}-{}".format(str(sample_service.id), api_key.key_type),
3000,
60
)
def test_should_not_rate_limit_if_limiting_is_disabled(
notify_db,
notify_db_session,
sample_service,
mocker):
with freeze_time("2016-01-01 12:00:00.000000"):
current_app.config['API_RATE_LIMIT_ENABLED'] = False
@@ -342,20 +346,19 @@ def test_should_not_rate_limit_if_limiting_is_disabled(
mocker.patch('app.redis_store.exceeded_rate_limit', return_value=False)
mocker.patch('app.notifications.validators.services_dao')
service = create_service(notify_db, notify_db_session, restricted=True)
api_key = sample_api_key(notify_db, notify_db_session, service=service)
sample_service.restricted = True
api_key = create_api_key(sample_service)
check_service_over_api_rate_limit(service, api_key)
check_service_over_api_rate_limit(sample_service, api_key)
assert not app.redis_store.exceeded_rate_limit.called
@pytest.mark.parametrize('key_type', ['test', 'normal'])
def test_rejects_api_calls_with_international_numbers_if_service_does_not_allow_int_sms(
key_type,
notify_db,
notify_db_session,
):
service = create_service(notify_db, notify_db_session, permissions=[SMS_TYPE])
service = create_service(service_permissions=[SMS_TYPE])
with pytest.raises(BadRequestError) as e:
validate_and_format_recipient('20-12-1234-1234', key_type, service, SMS_TYPE)
assert e.value.status_code == 400
@@ -365,9 +368,8 @@ def test_rejects_api_calls_with_international_numbers_if_service_does_not_allow_
@pytest.mark.parametrize('key_type', ['test', 'normal'])
def test_allows_api_calls_with_international_numbers_if_service_does_allow_int_sms(
key_type, notify_db, notify_db_session):
service = create_service(notify_db, notify_db_session, permissions=[SMS_TYPE, INTERNATIONAL_SMS_TYPE])
result = validate_and_format_recipient('20-12-1234-1234', key_type, service, SMS_TYPE)
key_type, sample_service_full_permissions):
result = validate_and_format_recipient('20-12-1234-1234', key_type, sample_service_full_permissions, SMS_TYPE)
assert result == '201212341234'

22
tests/app/service/test_api_key_endpoints.py
View File

@@ -5,9 +5,7 @@ from flask import url_for
from app.models import ApiKey, KEY_TYPE_NORMAL
from app.dao.api_key_dao import expire_api_key
from tests import create_authorization_header
from tests.app.conftest import sample_api_key as create_sample_api_key
from tests.app.conftest import sample_service as create_sample_service
from tests.app.db import create_user
from tests.app.db import create_api_key, create_service, create_user
def test_api_key_should_create_new_api_key_for_service(notify_api, sample_service):
@@ -96,26 +94,18 @@ def test_api_key_should_create_multiple_new_api_key_for_service(notify_api, samp
assert ApiKey.query.count() == 2
def test_get_api_keys_should_return_all_keys_for_service(notify_api, notify_db,
notify_db_session,
sample_api_key):
def test_get_api_keys_should_return_all_keys_for_service(notify_api, sample_api_key):
with notify_api.test_request_context():
with notify_api.test_client() as client:
another_user = create_user(email='another@it.gov.uk')
another_service = create_sample_service(
notify_db,
notify_db_session,
service_name='another',
user=another_user,
email_from='another'
)
another_service = create_service(user=another_user, service_name='Another service')
# key for another service
create_sample_api_key(notify_db, notify_db_session, service=another_service)
create_api_key(another_service)
# this service already has one key, add two more, one expired
create_sample_api_key(notify_db, notify_db_session, service=sample_api_key.service)
one_to_expire = create_sample_api_key(notify_db, notify_db_session, service=sample_api_key.service)
create_api_key(sample_api_key.service)
one_to_expire = create_api_key(sample_api_key.service)
expire_api_key(service_id=one_to_expire.service_id, api_key_id=one_to_expire.id)
assert ApiKey.query.count() == 4

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

@@ -1,7 +1,6 @@
import json
import uuid
from datetime import datetime, timedelta, date
from functools import partial
from unittest.mock import ANY
import pytest
@@ -10,7 +9,7 @@ from freezegun import freeze_time
from app.dao.organisation_dao import dao_add_service_to_organisation
from app.dao.service_sms_sender_dao import dao_get_sms_senders_by_service_id
from app.dao.services_dao import dao_remove_user_from_service
from app.dao.services_dao import dao_add_user_to_service, dao_remove_user_from_service
from app.dao.service_user_dao import dao_get_service_user
from app.dao.templates_dao import dao_redact_template
from app.dao.users_dao import save_model_user
@@ -18,6 +17,7 @@ from app.models import (
EmailBranding,
InboundNumber,
Notification,
Permission,
Service,
ServiceEmailReplyTo,
ServiceLetterContact,
@@ -29,11 +29,6 @@ from app.models import (
INTERNATIONAL_SMS_TYPE, INBOUND_SMS_TYPE,
)
from tests import create_authorization_header
from tests.app.conftest import (
sample_user_service_permission as create_user_service_permission,
sample_notification as create_sample_notification,
sample_notification_with_job
)
from tests.app.db import (
create_ft_billing,
create_ft_notification_status,
@@ -1494,18 +1489,22 @@ def test_add_unknown_user_to_service_returns404(notify_api, notify_db, notify_db
def test_remove_user_from_service(
notify_db, notify_db_session, client, sample_user_service_permission
client, sample_user_service_permission
):
second_user = create_user(email="new@digital.cabinet-office.gov.uk")
service = sample_user_service_permission.service
# Simulates successfully adding a user to the service
second_permission = create_user_service_permission(
notify_db,
notify_db_session,
user=second_user)
dao_add_user_to_service(
service,
second_user,
permissions=[Permission(service_id=service.id, user_id=second_user.id, permission='manage_settings')]
)
endpoint = url_for(
'service.remove_user_from_service',
service_id=str(second_permission.service.id),
user_id=str(second_permission.user.id))
service_id=str(service.id),
user_id=str(second_user.id))
auth_header = create_authorization_header()
resp = client.delete(
endpoint,
@@ -1549,10 +1548,7 @@ def test_cannot_remove_only_user_from_service(notify_api,
# This test is just here verify get_service_and_api_key_history that is a temp solution
# until proper ui is sorted out on admin app
def test_get_service_and_api_key_history(notify_api, notify_db, notify_db_session, sample_service):
from tests.app.conftest import sample_api_key as create_sample_api_key
api_key = create_sample_api_key(notify_db, notify_db_session, service=sample_service)
def test_get_service_and_api_key_history(notify_api, sample_service, sample_api_key):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header()
@@ -1564,19 +1560,23 @@ def test_get_service_and_api_key_history(notify_api, notify_db, notify_db_sessio
json_resp = json.loads(response.get_data(as_text=True))
assert json_resp['data']['service_history'][0]['id'] == str(sample_service.id)
assert json_resp['data']['api_key_history'][0]['id'] == str(api_key.id)
assert json_resp['data']['api_key_history'][0]['id'] == str(sample_api_key.id)
def test_get_all_notifications_for_service_in_order(notify_api, notify_db, notify_db_session):
def test_get_all_notifications_for_service_in_order(notify_api, notify_db_session):
with notify_api.test_request_context(), notify_api.test_client() as client:
service_1 = create_service(service_name="1", email_from='1')
service_2 = create_service(service_name="2", email_from='2')
create_sample_notification(notify_db, notify_db_session, service=service_2)
service_1_template = create_template(service_1)
service_2_template = create_template(service_2)
notification_1 = create_sample_notification(notify_db, notify_db_session, service=service_1)
notification_2 = create_sample_notification(notify_db, notify_db_session, service=service_1)
notification_3 = create_sample_notification(notify_db, notify_db_session, service=service_1)
# create notification for service_2
create_notification(service_2_template)
notification_1 = create_notification(service_1_template)
notification_2 = create_notification(service_1_template)
notification_3 = create_notification(service_1_template)
auth_header = create_authorization_header()
@@ -1619,18 +1619,21 @@ def test_get_notification_for_service_without_uuid(client, notify_db, notify_db_
assert response.status_code == 404
def test_get_notification_for_service(client, notify_db, notify_db_session):
def test_get_notification_for_service(client, notify_db_session):
service_1 = create_service(service_name="1", email_from='1')
service_2 = create_service(service_name="2", email_from='2')
service_1_template = create_template(service_1)
service_2_template = create_template(service_2)
service_1_notifications = [
create_sample_notification(notify_db, notify_db_session, service=service_1),
create_sample_notification(notify_db, notify_db_session, service=service_1),
create_sample_notification(notify_db, notify_db_session, service=service_1),
create_notification(service_1_template),
create_notification(service_1_template),
create_notification(service_1_template),
]
create_sample_notification(notify_db, notify_db_session, service=service_2)
create_notification(service_2_template)
for notification in service_1_notifications:
response = client.get(
@@ -1694,18 +1697,15 @@ def test_get_notification_for_service_returns_old_template_version(admin_request
)
def test_get_all_notifications_for_service_including_ones_made_by_jobs(
client,
notify_db,
notify_db_session,
sample_service,
include_from_test_key,
expected_count_of_notifications
expected_count_of_notifications,
sample_notification,
sample_notification_with_job,
sample_template,
):
with_job = sample_notification_with_job(notify_db, notify_db_session, service=sample_service)
without_job = create_sample_notification(notify_db, notify_db_session, service=sample_service)
# from_test_api_key
create_sample_notification(
notify_db, notify_db_session, service=sample_service, key_type=KEY_TYPE_TEST
)
# notification from_test_api_key
create_notification(sample_template, key_type=KEY_TYPE_TEST)
auth_header = create_authorization_header()
@@ -1718,8 +1718,8 @@ def test_get_all_notifications_for_service_including_ones_made_by_jobs(
resp = json.loads(response.get_data(as_text=True))
assert len(resp['notifications']) == expected_count_of_notifications
assert resp['notifications'][0]['to'] == with_job.to
assert resp['notifications'][1]['to'] == without_job.to
assert resp['notifications'][0]['to'] == sample_notification_with_job.to
assert resp['notifications'][1]['to'] == sample_notification.to
assert response.status_code == 200
@@ -1829,11 +1829,11 @@ def test_set_sms_prefixing_for_service_cant_be_none(
('False', {'requested': 2, 'delivered': 1, 'failed': 0}),
('True', {'requested': 1, 'delivered': 0, 'failed': 0})
], ids=['seven_days', 'today'])
def test_get_detailed_service(notify_db, notify_db_session, notify_api, sample_service, today_only, stats):
def test_get_detailed_service(sample_template, notify_api, sample_service, today_only, stats):
with notify_api.test_request_context(), notify_api.test_client() as client:
create_ft_notification_status(date(2000, 1, 1), 'sms', sample_service, count=1)
with freeze_time('2000-01-02T12:00:00'):
create_sample_notification(notify_db, notify_db_session, status='created')
create_notification(template=sample_template, status='created')
resp = client.get(
'/service/{}?detailed=True&today_only={}'.format(sample_service.id, today_only),
headers=[create_authorization_header()]
@@ -1847,11 +1847,11 @@ def test_get_detailed_service(notify_db, notify_db_session, notify_api, sample_s
assert service['statistics'][SMS_TYPE] == stats
def test_get_services_with_detailed_flag(client, notify_db, notify_db_session):
def test_get_services_with_detailed_flag(client, sample_template):
notifications = [
create_sample_notification(notify_db, notify_db_session),
create_sample_notification(notify_db, notify_db_session),
create_sample_notification(notify_db, notify_db_session, key_type=KEY_TYPE_TEST)
create_notification(sample_template),
create_notification(sample_template),
create_notification(sample_template, key_type=KEY_TYPE_TEST),
]
resp = client.get(
'/service?detailed=True',
@@ -1870,12 +1870,12 @@ def test_get_services_with_detailed_flag(client, notify_db, notify_db_session):
}
def test_get_services_with_detailed_flag_excluding_from_test_key(notify_api, notify_db, notify_db_session):
create_sample_notification(notify_db, notify_db_session, key_type=KEY_TYPE_NORMAL),
create_sample_notification(notify_db, notify_db_session, key_type=KEY_TYPE_TEAM),
create_sample_notification(notify_db, notify_db_session, key_type=KEY_TYPE_TEST),
create_sample_notification(notify_db, notify_db_session, key_type=KEY_TYPE_TEST),
create_sample_notification(notify_db, notify_db_session, key_type=KEY_TYPE_TEST)
def test_get_services_with_detailed_flag_excluding_from_test_key(notify_api, sample_template):
create_notification(sample_template, key_type=KEY_TYPE_NORMAL)
create_notification(sample_template, key_type=KEY_TYPE_TEAM)
create_notification(sample_template, key_type=KEY_TYPE_TEST)
create_notification(sample_template, key_type=KEY_TYPE_TEST)
create_notification(sample_template, key_type=KEY_TYPE_TEST)
with notify_api.test_request_context(), notify_api.test_client() as client:
resp = client.get(
@@ -1927,16 +1927,19 @@ def test_get_services_with_detailed_flag_defaults_to_today(client, mocker):
assert resp.status_code == 200
def test_get_detailed_services_groups_by_service(notify_db, notify_db_session):
def test_get_detailed_services_groups_by_service(notify_db_session):
from app.service.rest import get_detailed_services
service_1 = create_service(service_name="1", email_from='1')
service_2 = create_service(service_name="2", email_from='2')
create_sample_notification(notify_db, notify_db_session, service=service_1, status='created')
create_sample_notification(notify_db, notify_db_session, service=service_2, status='created')
create_sample_notification(notify_db, notify_db_session, service=service_1, status='delivered')
create_sample_notification(notify_db, notify_db_session, service=service_1, status='created')
service_1_template = create_template(service_1)
service_2_template = create_template(service_2)
create_notification(service_1_template, status='created')
create_notification(service_2_template, status='created')
create_notification(service_1_template, status='delivered')
create_notification(service_1_template, status='created')
data = get_detailed_services(start_date=datetime.utcnow().date(), end_date=datetime.utcnow().date())
data = sorted(data, key=lambda x: x['name'])
@@ -1956,13 +1959,14 @@ def test_get_detailed_services_groups_by_service(notify_db, notify_db_session):
}
def test_get_detailed_services_includes_services_with_no_notifications(notify_db, notify_db_session):
def test_get_detailed_services_includes_services_with_no_notifications(notify_db_session):
from app.service.rest import get_detailed_services
service_1 = create_service(service_name="1", email_from='1')
service_2 = create_service(service_name="2", email_from='2')
create_sample_notification(notify_db, notify_db_session, service=service_1)
service_1_template = create_template(service_1)
create_notification(service_1_template)
data = get_detailed_services(start_date=datetime.utcnow().date(),
end_date=datetime.utcnow().date())
@@ -1983,13 +1987,13 @@ def test_get_detailed_services_includes_services_with_no_notifications(notify_db
}
def test_get_detailed_services_only_includes_todays_notifications(notify_db, notify_db_session):
def test_get_detailed_services_only_includes_todays_notifications(sample_template):
from app.service.rest import get_detailed_services
create_sample_notification(notify_db, notify_db_session, created_at=datetime(2015, 10, 9, 23, 59))
create_sample_notification(notify_db, notify_db_session, created_at=datetime(2015, 10, 10, 0, 0))
create_sample_notification(notify_db, notify_db_session, created_at=datetime(2015, 10, 10, 12, 0))
create_sample_notification(notify_db, notify_db_session, created_at=datetime(2015, 10, 10, 23, 0))
create_notification(sample_template, created_at=datetime(2015, 10, 9, 23, 59))
create_notification(sample_template, created_at=datetime(2015, 10, 10, 0, 0))
create_notification(sample_template, created_at=datetime(2015, 10, 10, 12, 0))
create_notification(sample_template, created_at=datetime(2015, 10, 10, 23, 0))
with freeze_time('2015-10-10T12:00:00'):
data = get_detailed_services(start_date=datetime.utcnow().date(), end_date=datetime.utcnow().date())
@@ -2055,11 +2059,10 @@ def test_search_for_notification_by_to_field(client, sample_template, sample_ema
def test_search_for_notification_by_to_field_return_empty_list_if_there_is_no_match(
client, notify_db, notify_db_session
client, sample_template, sample_email_template
):
create_notification = partial(create_sample_notification, notify_db, notify_db_session)
notification1 = create_notification(to_field='+447700900855')
create_notification(to_field='jack@gmail.com')
notification1 = create_notification(sample_template, to_field='+447700900855')
create_notification(sample_email_template, to_field='jack@gmail.com')
response = client.get(
'/service/{}/notifications?to={}&template_type={}'.format(notification1.service_id, '+447700900800', 'sms'),
@@ -2071,12 +2074,12 @@ def test_search_for_notification_by_to_field_return_empty_list_if_there_is_no_ma
assert len(notifications) == 0
def test_search_for_notification_by_to_field_return_multiple_matches(client, notify_db, notify_db_session):
create_notification = partial(create_sample_notification, notify_db, notify_db_session)
notification1 = create_notification(to_field='+447700900855', normalised_to='447700900855')
notification2 = create_notification(to_field=' +44 77009 00855 ', normalised_to='447700900855')
notification3 = create_notification(to_field='+44770 0900 855', normalised_to='447700900855')
notification4 = create_notification(to_field='jack@gmail.com', normalised_to='jack@gmail.com')
def test_search_for_notification_by_to_field_return_multiple_matches(client, sample_template, sample_email_template):
notification1 = create_notification(sample_template, to_field='+447700900855', normalised_to='447700900855')
notification2 = create_notification(sample_template, to_field=' +44 77009 00855 ', normalised_to='447700900855')
notification3 = create_notification(sample_template, to_field='+44770 0900 855', normalised_to='447700900855')
notification4 = create_notification(
sample_email_template, to_field='jack@gmail.com', normalised_to='jack@gmail.com')
response = client.get(
'/service/{}/notifications?to={}&template_type={}'.format(notification1.service_id, '+447700900855', 'sms'),
@@ -2173,16 +2176,10 @@ def test_update_service_does_not_call_send_notification_when_restricted_not_chan
assert not send_notification_mock.called
def test_search_for_notification_by_to_field_filters_by_status(client, notify_db, notify_db_session):
create_notification = partial(
create_sample_notification,
notify_db,
notify_db_session,
to_field='+447700900855',
normalised_to='447700900855'
)
notification1 = create_notification(status='delivered')
create_notification(status='sending')
def test_search_for_notification_by_to_field_filters_by_status(client, sample_template):
notification1 = create_notification(
sample_template, to_field='+447700900855', status='delivered', normalised_to='447700900855')
create_notification(sample_template, to_field='+447700900855', status='sending', normalised_to='447700900855')
response = client.get(
'/service/{}/notifications?to={}&status={}&template_type={}'.format(
@@ -2198,16 +2195,17 @@ def test_search_for_notification_by_to_field_filters_by_status(client, notify_db
assert str(notification1.id) in notification_ids
def test_search_for_notification_by_to_field_filters_by_statuses(client, notify_db, notify_db_session):
create_notification = partial(
create_sample_notification,
notify_db,
notify_db_session,
def test_search_for_notification_by_to_field_filters_by_statuses(client, sample_template):
notification1 = create_notification(
sample_template,
to_field='+447700900855',
normalised_to='447700900855'
)
notification1 = create_notification(status='delivered')
notification2 = create_notification(status='sending')
status='delivered',
normalised_to='447700900855')
notification2 = create_notification(
sample_template,
to_field='+447700900855',
status='sending',
normalised_to='447700900855')
response = client.get(
'/service/{}/notifications?to={}&status={}&status={}&template_type={}'.format(
@@ -2226,17 +2224,13 @@ def test_search_for_notification_by_to_field_filters_by_statuses(client, notify_
def test_search_for_notification_by_to_field_returns_content(
client,
notify_db,
notify_db_session,
sample_template_with_placeholders
):
notification = create_sample_notification(
notify_db,
notify_db_session,
notification = create_notification(
sample_template_with_placeholders,
to_field='+447700900855',
personalisation={"name": "Foo"},
normalised_to='447700900855',
template=sample_template_with_placeholders,
personalisation={"name": "Foo"}
)
response = client.get(
@@ -2402,17 +2396,13 @@ def test_get_all_notifications_for_service_includes_template_hidden(admin_reques
def test_search_for_notification_by_to_field_returns_personlisation(
client,
notify_db,
notify_db_session,
sample_template_with_placeholders
):
create_sample_notification(
notify_db,
notify_db_session,
create_notification(
sample_template_with_placeholders,
to_field='+447700900855',
personalisation={"name": "Foo"},
normalised_to='447700900855',
template=sample_template_with_placeholders,
personalisation={"name": "Foo"}
)
response = client.get(
@@ -2431,25 +2421,11 @@ def test_search_for_notification_by_to_field_returns_personlisation(
def test_search_for_notification_by_to_field_returns_notifications_by_type(
client,
notify_db,
notify_db_session,
sample_template,
sample_email_template
):
sms_notification = create_sample_notification(
notify_db,
notify_db_session,
to_field='+447700900855',
normalised_to='447700900855',
template=sample_template
)
create_sample_notification(
notify_db,
notify_db_session,
to_field='44770@gamil.com',
normalised_to='44770@gamil.com',
template=sample_email_template
)
sms_notification = create_notification(sample_template, to_field='+447700900855', normalised_to='447700900855')
create_notification(sample_email_template, to_field='44770@gamil.com', normalised_to='44770@gamil.com')
response = client.get(
'/service/{}/notifications?to={}&template_type={}'.format(