mirror of
https://github.com/GSA/notifications-api.git
synced 2026-02-01 15:46:07 -05:00
only send to team emails/phones if POST /notificaiton/<type> with team api_key
uses same restriction as a service in trial mode
This commit is contained in:
@@ -8,7 +8,7 @@ from freezegun import freeze_time
|
||||
|
||||
import app
|
||||
from app import encryption
|
||||
from app.models import Service
|
||||
from app.models import KEY_TYPE_TEAM
|
||||
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 tests import create_authorization_header
|
||||
@@ -50,7 +50,7 @@ def test_should_reject_bad_phone_numbers(notify_api, sample_template, mocker):
|
||||
'to': 'invalid',
|
||||
'template': sample_template.id
|
||||
}
|
||||
auth_header = create_authorization_header(service_id=sample_template.service.id)
|
||||
auth_header = create_authorization_header(service_id=sample_template.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms',
|
||||
@@ -74,7 +74,7 @@ def test_send_notification_invalid_template_id(notify_api, sample_template, mock
|
||||
'to': '+447700900855',
|
||||
'template': fake_uuid
|
||||
}
|
||||
auth_header = create_authorization_header(service_id=sample_template.service.id)
|
||||
auth_header = create_authorization_header(service_id=sample_template.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms',
|
||||
@@ -135,7 +135,7 @@ def test_should_not_send_notification_for_archived_template(notify_api, sample_t
|
||||
'to': '+447700900855',
|
||||
'template': sample_template.id
|
||||
})
|
||||
auth_header = create_authorization_header(service_id=sample_template.service.id)
|
||||
auth_header = create_authorization_header(service_id=sample_template.service_id)
|
||||
|
||||
resp = client.post(
|
||||
path='/notifications/sms',
|
||||
@@ -200,23 +200,20 @@ def test_send_notification_with_too_much_personalisation_data(
|
||||
assert 'Personalisation not needed for template: foo' in json_resp['message']['template']
|
||||
|
||||
|
||||
def test_prevents_sending_to_any_mobile_on_restricted_service(notify_api, sample_template, mocker):
|
||||
def test_should_not_send_sms_if_restricted_and_not_a_service_user(notify_api, sample_template, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
||||
|
||||
Service.query.filter_by(
|
||||
id=sample_template.service.id
|
||||
).update(
|
||||
{'restricted': True}
|
||||
)
|
||||
sample_template.service.restricted = True
|
||||
dao_update_service(sample_template.service)
|
||||
invalid_mob = '+447700900855'
|
||||
data = {
|
||||
'to': invalid_mob,
|
||||
'template': sample_template.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(service_id=sample_template.service.id)
|
||||
auth_header = create_authorization_header(service_id=sample_template.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms',
|
||||
@@ -230,6 +227,52 @@ def test_prevents_sending_to_any_mobile_on_restricted_service(notify_api, sample
|
||||
assert 'Invalid phone number for restricted service' in json_resp['message']['to']
|
||||
|
||||
|
||||
def test_should_send_sms_if_restricted_and_a_service_user(notify_api, sample_template, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
||||
|
||||
sample_template.service.restricted = True
|
||||
dao_update_service(sample_template.service)
|
||||
data = {
|
||||
'to': sample_template.service.created_by.mobile_number,
|
||||
'template': sample_template.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(service_id=sample_template.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
assert app.celery.tasks.send_sms.apply_async.called
|
||||
assert response.status_code == 201
|
||||
|
||||
|
||||
def test_should_send_email_if_restricted_and_a_service_user(notify_api, sample_email_template, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_email.apply_async')
|
||||
|
||||
sample_email_template.service.restricted = True
|
||||
dao_update_service(sample_email_template.service)
|
||||
data = {
|
||||
'to': sample_email_template.service.created_by.email_address,
|
||||
'template': sample_email_template.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
assert app.celery.tasks.send_email.apply_async.called
|
||||
assert response.status_code == 201
|
||||
|
||||
|
||||
def test_should_not_allow_template_from_another_service(notify_api, service_factory, sample_user, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
@@ -272,7 +315,7 @@ def test_should_not_allow_template_content_too_large(notify_api, notify_db, noti
|
||||
random.choice(string.ascii_uppercase + string.digits) for _ in range(limit + 1))
|
||||
}
|
||||
})
|
||||
auth_header = create_authorization_header(service_id=template.service.id)
|
||||
auth_header = create_authorization_header(service_id=template.service_id)
|
||||
|
||||
resp = client.post(
|
||||
path='/notifications/sms',
|
||||
@@ -352,9 +395,9 @@ def test_should_reject_email_notification_with_bad_email(notify_api, sample_emai
|
||||
to_address = "bad-email"
|
||||
data = {
|
||||
'to': to_address,
|
||||
'template': str(sample_email_template.service.id)
|
||||
'template': str(sample_email_template.service_id)
|
||||
}
|
||||
auth_header = create_authorization_header(service_id=sample_email_template.service.id)
|
||||
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email',
|
||||
@@ -377,7 +420,7 @@ def test_should_reject_email_notification_with_template_id_that_cant_be_found(
|
||||
'to': 'ok@ok.com',
|
||||
'template': fake_uuid
|
||||
}
|
||||
auth_header = create_authorization_header(service_id=sample_email_template.service.id)
|
||||
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email',
|
||||
@@ -430,46 +473,14 @@ def test_should_not_send_email_if_restricted_and_not_a_service_user(notify_api,
|
||||
mocker.patch('app.celery.tasks.send_email.apply_async')
|
||||
|
||||
sample_email_template.service.restricted = True
|
||||
dao_update_service(sample_email_template)
|
||||
dao_update_service(sample_email_template.service)
|
||||
|
||||
data = {
|
||||
'to': "not-someone-we-trust@email-address.com",
|
||||
'template': str(sample_email_template.id)
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(service_id=sample_email_template.service.id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
app.celery.tasks.send_email.apply_async.assert_not_called()
|
||||
|
||||
assert response.status_code == 400
|
||||
assert 'Invalid email address for restricted service' in json_resp['message']['to']
|
||||
|
||||
|
||||
def test_should_not_send_email_for_job_if_restricted_and_not_a_service_user(
|
||||
notify_api,
|
||||
sample_job,
|
||||
sample_email_template,
|
||||
mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_email.apply_async')
|
||||
|
||||
sample_email_template.service.restricted = True
|
||||
dao_update_service(sample_email_template)
|
||||
|
||||
data = {
|
||||
'to': "not-someone-we-trust@email-address.com",
|
||||
'template': str(sample_job.template.id),
|
||||
'job': (sample_job.id)
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(service_id=sample_job.service.id)
|
||||
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email',
|
||||
@@ -554,6 +565,7 @@ def test_should_block_api_call_if_over_day_limit(notify_db, notify_db_session, n
|
||||
|
||||
def test_no_limit_for_live_service(notify_api,
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
mock_celery_send_email,
|
||||
sample_service,
|
||||
sample_email_template):
|
||||
@@ -679,7 +691,7 @@ def test_should_record_sms_request_in_statsd(notify_api, sample_template, mocker
|
||||
app.statsd_client.incr.assert_called_once_with("notifications.api.sms")
|
||||
|
||||
|
||||
def test_should_not_return_html_in_body(notify_api, notify_db, mocker):
|
||||
def test_should_not_return_html_in_body(notify_api, notify_db, notify_db_session, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_email.apply_async')
|
||||
@@ -698,3 +710,92 @@ def test_should_not_return_html_in_body(notify_api, notify_db, mocker):
|
||||
|
||||
assert response.status_code == 201
|
||||
assert json.loads(response.get_data(as_text=True))['data']['body'] == 'hello\nthere'
|
||||
|
||||
|
||||
def test_should_not_send_email_if_team_api_key_and_not_a_service_user(notify_api, sample_email_template, mocker):
|
||||
with notify_api.test_request_context(), notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_email.apply_async')
|
||||
|
||||
data = {
|
||||
'to': "not-someone-we-trust@email-address.com",
|
||||
'template': str(sample_email_template.id),
|
||||
}
|
||||
|
||||
# import pdb
|
||||
# pdb.set_trace()
|
||||
auth_header = create_authorization_header(service_id=sample_email_template.service_id, key_type=KEY_TYPE_TEAM)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
|
||||
app.celery.tasks.send_email.apply_async.assert_not_called()
|
||||
|
||||
assert response.status_code == 400
|
||||
assert 'Invalid email address for restricted service' in json_resp['message']['to']
|
||||
|
||||
|
||||
def test_should_not_send_sms_if_team_api_key_and_not_a_service_user(notify_api, sample_template, mocker):
|
||||
with notify_api.test_request_context(), notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
||||
|
||||
data = {
|
||||
'to': '07123123123',
|
||||
'template': str(sample_template.id),
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(service_id=sample_template.service_id, key_type=KEY_TYPE_TEAM)
|
||||
|
||||
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 'Invalid phone number for restricted service' in json_resp['message']['to']
|
||||
|
||||
|
||||
def test_should_send_email_if_team_api_key_and_a_service_user(notify_api, sample_email_template, mocker):
|
||||
with notify_api.test_request_context(), notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_email.apply_async')
|
||||
|
||||
data = {
|
||||
'to': sample_email_template.service.created_by.email_address,
|
||||
'template': sample_email_template.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(service_id=sample_email_template.service_id, key_type=KEY_TYPE_TEAM)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
assert app.celery.tasks.send_email.apply_async.called
|
||||
assert response.status_code == 201
|
||||
|
||||
|
||||
def test_should_send_sms_if_team_api_key_and_a_service_user(notify_api, sample_template, mocker):
|
||||
with notify_api.test_request_context(), notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
||||
|
||||
data = {
|
||||
'to': sample_template.service.created_by.mobile_number,
|
||||
'template': sample_template.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(service_id=sample_template.service_id, key_type=KEY_TYPE_TEAM)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms',
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
assert app.celery.tasks.send_sms.apply_async.called
|
||||
assert response.status_code == 201
|
||||
|
||||
Reference in New Issue
Block a user