mirror of
https://github.com/GSA/notifications-api.git
synced 2026-04-11 20:59:39 -04:00
Removed now unused notification for job endpoints
- this is now handled in the tasks
This commit is contained in:
Martyn Inglis
@@ -11,15 +11,12 @@ from app import api_user, encryption, create_uuid
|
||||
from app.dao import (
|
||||
templates_dao,
|
||||
services_dao,
|
||||
notifications_dao,
|
||||
jobs_dao
|
||||
notifications_dao
|
||||
)
|
||||
from app.schemas import (
|
||||
email_notification_schema,
|
||||
sms_template_notification_schema,
|
||||
notification_status_schema,
|
||||
job_sms_template_notification_schema,
|
||||
job_email_template_notification_schema
|
||||
notification_status_schema
|
||||
)
|
||||
from app.celery.tasks import send_sms, send_email
|
||||
from sqlalchemy.orm.exc import NoResultFound
|
||||
@@ -45,36 +42,20 @@ def get_notifications(notification_id):
|
||||
|
||||
@notifications.route('/notifications/sms', methods=['POST'])
|
||||
def create_sms_notification():
|
||||
return send_notification(notification_type=SMS_NOTIFICATION, expects_job=False)
|
||||
|
||||
|
||||
@notifications.route('/notifications/sms/service/<service_id>', methods=['POST'])
|
||||
def create_sms_for_job(service_id):
|
||||
return send_notification(service_id=service_id, notification_type=SMS_NOTIFICATION, expects_job=True)
|
||||
return send_notification(notification_type=SMS_NOTIFICATION)
|
||||
|
||||
|
||||
@notifications.route('/notifications/email', methods=['POST'])
|
||||
def create_email_notification():
|
||||
return send_notification(notification_type=EMAIL_NOTIFICATION, expects_job=False)
|
||||
return send_notification(notification_type=EMAIL_NOTIFICATION)
|
||||
|
||||
|
||||
@notifications.route('/notifications/email/service/<service_id>', methods=['POST'])
|
||||
def create_email_notification_for_job(service_id):
|
||||
return send_notification(service_id=service_id, notification_type=EMAIL_NOTIFICATION, expects_job=True)
|
||||
|
||||
|
||||
def send_notification(notification_type, service_id=None, expects_job=False):
|
||||
def send_notification(notification_type):
|
||||
assert notification_type
|
||||
|
||||
if not service_id:
|
||||
service_id = api_user['client']
|
||||
service_id = api_user['client']
|
||||
|
||||
if expects_job:
|
||||
schema = job_sms_template_notification_schema if notification_type is SMS_NOTIFICATION else \
|
||||
job_email_template_notification_schema
|
||||
else:
|
||||
schema = sms_template_notification_schema if notification_type is SMS_NOTIFICATION else \
|
||||
email_notification_schema
|
||||
schema = sms_template_notification_schema if notification_type is SMS_NOTIFICATION else email_notification_schema
|
||||
|
||||
notification, errors = schema.load(request.get_json())
|
||||
if errors:
|
||||
@@ -91,13 +72,7 @@ def send_notification(notification_type, service_id=None, expects_job=False):
|
||||
message={
|
||||
'template': ['Template {} not found for service {}'.format(notification['template'], service_id)]
|
||||
}
|
||||
), 400
|
||||
|
||||
if expects_job:
|
||||
job = jobs_dao.dao_get_job_by_service_id_and_job_id(service_id, notification['job'])
|
||||
|
||||
if not job:
|
||||
return jsonify(result="error", message={'job': ['Job {} not found'.format(notification['job'])]}), 400
|
||||
), 404
|
||||
|
||||
service = services_dao.dao_fetch_service_by_id(api_user['client'])
|
||||
|
||||
|
||||
@@ -143,7 +143,6 @@ api_keys_schema = ApiKeySchema(many=True)
|
||||
job_schema = JobSchema()
|
||||
job_schema_load_json = JobSchema(load_json=True)
|
||||
jobs_schema = JobSchema(many=True)
|
||||
# TODO: Remove this schema once the admin app has stopped using the /user/<user_id>code endpoint
|
||||
old_request_verify_code_schema = OldRequestVerifyCodeSchema()
|
||||
request_verify_code_schema = RequestVerifyCodeSchema()
|
||||
sms_admin_notification_schema = SmsAdminNotificationSchema()
|
||||
|
||||
@@ -5,7 +5,6 @@ from flask import json
|
||||
from app.models import Service
|
||||
from app.dao.templates_dao import dao_get_all_templates_for_service
|
||||
from app.dao.services_dao import dao_update_service
|
||||
from tests.app.conftest import sample_job
|
||||
from freezegun import freeze_time
|
||||
|
||||
|
||||
@@ -73,32 +72,6 @@ def test_create_sms_should_reject_if_missing_required_fields(notify_api, sample_
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_create_sms_should_reject_if_missing_job_id_on_job_sms(notify_api, sample_api_key, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
||||
|
||||
data = {}
|
||||
auth_header = create_authorization_header(
|
||||
service_id=sample_api_key.service_id,
|
||||
request_body=json.dumps(data),
|
||||
path='/notifications/sms/service/{}'.format(sample_api_key.service_id),
|
||||
method='POST')
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms/service/{}'.format(sample_api_key.service_id),
|
||||
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 json_resp['result'] == 'error'
|
||||
assert 'Missing data for required field.' in json_resp['message']['to'][0]
|
||||
assert 'Missing data for required field.' in json_resp['message']['template'][0]
|
||||
assert 'Missing data for required field.' in json_resp['message']['job'][0]
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_should_reject_bad_phone_numbers(notify_api, sample_template, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
@@ -127,35 +100,6 @@ def test_should_reject_bad_phone_numbers(notify_api, sample_template, mocker):
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_should_reject_bad_phone_numbers_on_job_sms(notify_api, sample_job, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
||||
|
||||
data = {
|
||||
'to': 'invalid',
|
||||
'template': sample_job.template.id,
|
||||
'job': sample_job.id
|
||||
}
|
||||
auth_header = create_authorization_header(
|
||||
request_body=json.dumps(data),
|
||||
path='/notifications/sms/service/{}'.format(sample_job.service_id),
|
||||
method='POST')
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms/service/{}'.format(sample_job.service_id),
|
||||
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 json_resp['result'] == 'error'
|
||||
assert len(json_resp['message'].keys()) == 1
|
||||
assert 'Invalid phone number, must be of format +441234123123' in json_resp['message']['to']
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_send_notification_invalid_template_id(notify_api, sample_template, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
@@ -179,108 +123,12 @@ def test_send_notification_invalid_template_id(notify_api, sample_template, mock
|
||||
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 response.status_code == 404
|
||||
assert len(json_resp['message'].keys()) == 1
|
||||
test_string = 'Template {} not found for service {}'.format(9999, sample_template.service.id)
|
||||
assert test_string in json_resp['message']['template']
|
||||
|
||||
|
||||
def test_send_notification_invalid_service_id_on_job_sms(notify_api, sample_job, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
||||
|
||||
service_id = uuid.uuid4()
|
||||
|
||||
data = {
|
||||
'to': '+441234123123',
|
||||
'template': sample_job.template.id,
|
||||
'job': sample_job.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(
|
||||
service_id=sample_job.service.id,
|
||||
request_body=json.dumps(data),
|
||||
path='/notifications/sms/service/{}'.format(service_id),
|
||||
method='POST')
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms/service/{}'.format(service_id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
app.celery.tasks.send_sms.apply_async.assert_not_called()
|
||||
|
||||
assert response.status_code == 400
|
||||
assert len(json_resp['message'].keys()) == 1
|
||||
test_string = 'Template {} not found for service {}'.format(sample_job.template.id, service_id)
|
||||
assert test_string in json_resp['message']['template']
|
||||
|
||||
|
||||
def test_send_notification_invalid_template_id_on_job_sms(notify_api, sample_job, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
||||
|
||||
data = {
|
||||
'to': '+441234123123',
|
||||
'template': 9999,
|
||||
'job': sample_job.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(
|
||||
service_id=sample_job.service.id,
|
||||
request_body=json.dumps(data),
|
||||
path='/notifications/sms/service/{}'.format(sample_job.service_id),
|
||||
method='POST')
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms/service/{}'.format(sample_job.service_id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
app.celery.tasks.send_sms.apply_async.assert_not_called()
|
||||
|
||||
assert response.status_code == 400
|
||||
assert len(json_resp['message'].keys()) == 1
|
||||
test_string = 'Template {} not found for service {}'.format(9999, sample_job.service.id)
|
||||
assert test_string in json_resp['message']['template']
|
||||
|
||||
|
||||
def test_send_notification_invalid_job_id_on_job_sms(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')
|
||||
job_id = uuid.uuid4()
|
||||
data = {
|
||||
'to': '+441234123123',
|
||||
'template': sample_template.id,
|
||||
'job': job_id
|
||||
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(
|
||||
service_id=sample_template.service.id,
|
||||
request_body=json.dumps(data),
|
||||
path='/notifications/sms/service/{}'.format(sample_template.service_id),
|
||||
method='POST')
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms/service/{}'.format(sample_template.service_id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
app.celery.tasks.send_sms.apply_async.assert_not_called()
|
||||
|
||||
assert response.status_code == 400
|
||||
assert len(json_resp['message'].keys()) == 1
|
||||
test_string = 'Job {} not found'.format(job_id)
|
||||
assert test_string in json_resp['message']['job']
|
||||
|
||||
|
||||
def test_prevents_sending_to_any_mobile_on_restricted_service(notify_api, sample_template, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
@@ -315,41 +163,6 @@ 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_prevents_sending_to_any_mobile_on_restricted_service_on_job_sms(notify_api, sample_job, 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_job.service.id
|
||||
).update(
|
||||
{'restricted': True}
|
||||
)
|
||||
invalid_mob = '+449999999999'
|
||||
data = {
|
||||
'to': invalid_mob,
|
||||
'template': sample_job.template.id,
|
||||
'job': sample_job.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(
|
||||
service_id=sample_job.service.id,
|
||||
request_body=json.dumps(data),
|
||||
path='/notifications/sms/service/{}'.format(sample_job.service_id),
|
||||
method='POST')
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms/service/{}'.format(sample_job.service_id),
|
||||
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_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:
|
||||
@@ -378,52 +191,7 @@ def test_should_not_allow_template_from_another_service(notify_api, service_fact
|
||||
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
|
||||
test_string = 'Template {} not found for service {}'.format(service_2_templates[0].id, service_1.id)
|
||||
assert test_string in json_resp['message']['template']
|
||||
|
||||
|
||||
def test_should_not_allow_template_from_another_service_on_job_sms(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
notify_api,
|
||||
service_factory,
|
||||
sample_user,
|
||||
mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
||||
|
||||
service_1 = service_factory.get('service 1', user=sample_user)
|
||||
service_2 = service_factory.get('service 2', user=sample_user)
|
||||
|
||||
service_1_templates = dao_get_all_templates_for_service(service_id=service_2.id)
|
||||
service_2_templates = dao_get_all_templates_for_service(service_id=service_2.id)
|
||||
|
||||
job_1 = sample_job(notify_db, notify_db_session, service_1, service_1_templates[0])
|
||||
sample_job(notify_db, notify_db_session, service_2, service_2_templates[0])
|
||||
|
||||
data = {
|
||||
'to': sample_user.mobile_number,
|
||||
'template': service_2_templates[0].id,
|
||||
'job': job_1.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(
|
||||
service_id=service_1.id,
|
||||
request_body=json.dumps(data),
|
||||
path='/notifications/sms/service/{}'.format(service_1.id),
|
||||
method='POST')
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms/service/{}'.format(service_1.id),
|
||||
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 response.status_code == 404
|
||||
test_string = 'Template {} not found for service {}'.format(service_2_templates[0].id, service_1.id)
|
||||
assert test_string in json_resp['message']['template']
|
||||
|
||||
@@ -464,43 +232,6 @@ def test_should_allow_valid_sms_notification(notify_api, sample_template, mocker
|
||||
assert notification_id
|
||||
|
||||
|
||||
@freeze_time("2016-01-01 11:09:00.061258")
|
||||
def test_should_allow_valid_sms_notification_for_job(notify_api, sample_job, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_sms.apply_async')
|
||||
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
|
||||
|
||||
data = {
|
||||
'to': '+441234123123',
|
||||
'template': sample_job.template.id,
|
||||
'job': sample_job.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(
|
||||
request_body=json.dumps(data),
|
||||
path='/notifications/sms/service/{}'.format(sample_job.service_id),
|
||||
method='POST',
|
||||
service_id=sample_job.service_id
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/sms/service/{}'.format(sample_job.service_id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
notification_id = json.loads(response.data)['notification_id']
|
||||
app.celery.tasks.send_sms.apply_async.assert_called_once_with(
|
||||
(str(sample_job.service_id),
|
||||
notification_id,
|
||||
"something_encrypted",
|
||||
"2016-01-01 11:09:00.061258"),
|
||||
queue="sms"
|
||||
)
|
||||
assert response.status_code == 201
|
||||
assert notification_id
|
||||
|
||||
|
||||
def test_create_email_should_reject_if_missing_required_fields(notify_api, sample_api_key, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
@@ -526,32 +257,6 @@ def test_create_email_should_reject_if_missing_required_fields(notify_api, sampl
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_create_email_job_should_reject_if_missing_required_fields(notify_api, sample_api_key, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_email.apply_async')
|
||||
|
||||
data = {}
|
||||
auth_header = create_authorization_header(
|
||||
service_id=sample_api_key.service_id,
|
||||
request_body=json.dumps(data),
|
||||
path='/notifications/email/service/{}'.format(sample_api_key.service_id),
|
||||
method='POST')
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email/service/{}'.format(sample_api_key.service_id),
|
||||
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 json_resp['result'] == 'error'
|
||||
assert 'Missing data for required field.' in json_resp['message']['to'][0]
|
||||
assert 'Missing data for required field.' in json_resp['message']['template'][0]
|
||||
assert 'Missing data for required field.' in json_resp['message']['job'][0]
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_should_reject_email_notification_with_bad_email(notify_api, sample_email_template, mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
@@ -579,34 +284,6 @@ def test_should_reject_email_notification_with_bad_email(notify_api, sample_emai
|
||||
assert data['message']['to'][0] == 'Invalid email'
|
||||
|
||||
|
||||
def test_should_reject_email_job_notification_with_bad_email(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')
|
||||
to_address = "bad-email"
|
||||
data = {
|
||||
'to': to_address,
|
||||
'template': sample_email_template.service.id,
|
||||
'job': sample_job.id
|
||||
}
|
||||
auth_header = create_authorization_header(
|
||||
service_id=sample_email_template.service.id,
|
||||
request_body=json.dumps(data),
|
||||
path='/notifications/email/service/{}'.format(sample_email_template.service_id),
|
||||
method='POST')
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email/service/{}'.format(sample_email_template.service_id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
data = json.loads(response.get_data(as_text=True))
|
||||
app.celery.tasks.send_email.apply_async.assert_not_called()
|
||||
assert response.status_code == 400
|
||||
assert data['result'] == 'error'
|
||||
assert data['message']['to'][0] == 'Invalid email'
|
||||
|
||||
|
||||
def test_should_reject_email_notification_with_template_id_that_cant_be_found(
|
||||
notify_api, sample_email_template, mocker):
|
||||
with notify_api.test_request_context():
|
||||
@@ -629,39 +306,7 @@ def test_should_reject_email_notification_with_template_id_that_cant_be_found(
|
||||
|
||||
data = json.loads(response.get_data(as_text=True))
|
||||
app.celery.tasks.send_email.apply_async.assert_not_called()
|
||||
assert response.status_code == 400
|
||||
assert data['result'] == 'error'
|
||||
test_string = 'Template {} not found for service {}'.format(
|
||||
1234,
|
||||
sample_email_template.service.id
|
||||
)
|
||||
assert test_string in data['message']['template']
|
||||
|
||||
|
||||
def test_should_reject_email_job_notification_with_template_id_that_cant_be_found(
|
||||
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')
|
||||
data = {
|
||||
'to': 'ok@ok.com',
|
||||
'template': 1234,
|
||||
'job': sample_job.id
|
||||
}
|
||||
auth_header = create_authorization_header(
|
||||
service_id=sample_email_template.service.id,
|
||||
request_body=json.dumps(data),
|
||||
path='/notifications/email/service/{}'.format(sample_job.service.id),
|
||||
method='POST')
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email/service/{}'.format(sample_job.service.id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
data = json.loads(response.get_data(as_text=True))
|
||||
app.celery.tasks.send_email.apply_async.assert_not_called()
|
||||
assert response.status_code == 400
|
||||
assert response.status_code == 404
|
||||
assert data['result'] == 'error'
|
||||
test_string = 'Template {} not found for service {}'.format(
|
||||
1234,
|
||||
@@ -699,53 +344,7 @@ def test_should_not_allow_email_template_from_another_service(notify_api, servic
|
||||
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
|
||||
test_string = 'Template {} not found for service {}'.format(service_2_templates[0].id, service_1.id)
|
||||
assert test_string in json_resp['message']['template']
|
||||
|
||||
|
||||
def test_should_not_allow_template_from_another_service_on_job_email(
|
||||
notify_db,
|
||||
notify_db_session,
|
||||
notify_api,
|
||||
service_factory,
|
||||
sample_user,
|
||||
mocker):
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
mocker.patch('app.celery.tasks.send_email.apply_async')
|
||||
|
||||
service_1 = service_factory.get('service 1', user=sample_user, template_type='email')
|
||||
service_2 = service_factory.get('service 2', user=sample_user, template_type='email')
|
||||
|
||||
service_1_templates = dao_get_all_templates_for_service(service_id=service_2.id)
|
||||
service_2_templates = dao_get_all_templates_for_service(service_id=service_2.id)
|
||||
|
||||
job_1 = sample_job(notify_db, notify_db_session, service_1, service_1_templates[0])
|
||||
sample_job(notify_db, notify_db_session, service_2, service_2_templates[0])
|
||||
|
||||
data = {
|
||||
'to': sample_user.email_address,
|
||||
'template': service_2_templates[0].id,
|
||||
'job': job_1.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(
|
||||
service_id=service_1.id,
|
||||
request_body=json.dumps(data),
|
||||
path='/notifications/email/service/{}'.format(service_1.id),
|
||||
method='POST')
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email/service/{}'.format(service_1.id),
|
||||
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
|
||||
print(json_resp)
|
||||
assert response.status_code == 404
|
||||
test_string = 'Template {} not found for service {}'.format(service_2_templates[0].id, service_1.id)
|
||||
assert test_string in json_resp['message']['template']
|
||||
|
||||
@@ -853,74 +452,3 @@ def test_should_allow_valid_email_notification(notify_api, sample_email_template
|
||||
)
|
||||
assert response.status_code == 201
|
||||
assert notification_id
|
||||
|
||||
|
||||
def test_send_notification_invalid_job_id_on_job_email(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')
|
||||
job_id = uuid.uuid4()
|
||||
data = {
|
||||
'to': 'test@test.com',
|
||||
'template': sample_email_template.id,
|
||||
'job': job_id
|
||||
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(
|
||||
service_id=sample_email_template.service.id,
|
||||
request_body=json.dumps(data),
|
||||
path='/notifications/email/service/{}'.format(sample_email_template.service_id),
|
||||
method='POST')
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email/service/{}'.format(sample_email_template.service_id),
|
||||
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 len(json_resp['message'].keys()) == 1
|
||||
test_string = 'Job {} not found'.format(job_id)
|
||||
assert test_string in json_resp['message']['job']
|
||||
|
||||
|
||||
@freeze_time("2016-01-01 11:09:00.061258")
|
||||
def test_should_allow_valid_email_notification_for_job(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')
|
||||
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
|
||||
|
||||
data = {
|
||||
'to': 'ok@ok.com',
|
||||
'template': sample_email_template.id,
|
||||
'job': sample_job.id
|
||||
}
|
||||
|
||||
auth_header = create_authorization_header(
|
||||
request_body=json.dumps(data),
|
||||
path='/notifications/email/service/{}'.format(sample_job.service_id),
|
||||
method='POST',
|
||||
service_id=sample_job.service_id
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
path='/notifications/email/service/{}'.format(sample_job.service_id),
|
||||
data=json.dumps(data),
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
notification_id = json.loads(response.data)['notification_id']
|
||||
app.celery.tasks.send_email.apply_async.assert_called_once_with(
|
||||
(str(sample_job.service_id),
|
||||
notification_id,
|
||||
"Email Subject",
|
||||
"sample.service@test.notify.com",
|
||||
"something_encrypted",
|
||||
"2016-01-01 11:09:00.061258"),
|
||||
queue="email"
|
||||
)
|
||||
assert response.status_code == 201
|
||||
assert notification_id
|
||||
|
||||
Reference in New Issue
Block a user