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

Merge pull request #78 from alphagov/move-sms-notifications-into-celery

Browse Source 
Move sms notifications into celery
This commit is contained in:
Rebecca Law
2016-02-17 13:29:45 +00:00
parent 5f4b6a7db9 58ae041b64
commit c7cebc4d00
25 changed files with 497 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
.travis.yml
View File

@@ -39,6 +39,16 @@ deploy:
deployment_group: notifications_api_deployment_group
region: eu-west-1
on: *2
- provider: codedeploy
access_key_id: AKIAJQPPNM6P6V53SWKA
secret_access_key: *1
bucket: notifications-api-codedeploy
key: notifications-api-$TRAVIS_BRANCH-$TRAVIS_BUILD_NUMBER-$TRAVIS_COMMIT.zip
bundle_type: zip
application: notifications-api
deployment_group: notifications-api-celery
region: eu-west-1
on: *2
before_deploy:
- ./scripts/update_version_file.sh
- zip -r notifications-api *

14
app/__init__.py
View File

@@ -1,18 +1,20 @@
import os
import re
import ast
from flask import request, url_for
from flask._compat import string_types
from flask import Flask, _request_ctx_stack
from flask.ext.sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from werkzeug.local import LocalProxy
from utils import logging
from app.celery.celery import NotifyCelery
from app.clients.sms.twilio import TwilioClient
from app.encryption import Encryption
db = SQLAlchemy()
ma = Marshmallow()
notify_celery = NotifyCelery()
twilio_client = TwilioClient()
encryption = Encryption()
api_user = LocalProxy(lambda: _request_ctx_stack.top.api_user)
@@ -22,10 +24,14 @@ def create_app():
application.config.from_object(os.environ['NOTIFY_API_ENVIRONMENT'])
init_app(application)
db.init_app(application)
ma.init_app(application)
init_app(application)
logging.init_app(application)
twilio_client.init_app(application)
notify_celery.init_app(application)
encryption.init_app(application)
from app.service.rest import service as service_blueprint
from app.user.rest import user as user_blueprint

0
app/celery/__init__.py Normal file
View File

17
app/celery/celery.py Normal file
View File

@@ -0,0 +1,17 @@
from celery import Celery
class NotifyCelery(Celery):
def init_app(self, app):
super().__init__(app.import_name, broker=app.config['BROKER_URL'])
self.conf.update(app.config)
TaskBase = self.Task
class ContextTask(TaskBase):
abstract = True
def __call__(self, *args, **kwargs):
with app.app_context():
return TaskBase.__call__(self, *args, **kwargs)
self.Task = ContextTask

32
app/celery/tasks.py Normal file
View File

@@ -0,0 +1,32 @@
from app import notify_celery, twilio_client, encryption
from app.clients.sms.twilio import TwilioClientException
from app.dao.templates_dao import get_model_templates
from app.dao.notifications_dao import save_notification
from app.models import Notification
from flask import current_app
from sqlalchemy.exc import SQLAlchemyError
@notify_celery.task(name="send-sms")
def send_sms(service_id, notification_id, encrypted_notification):
notification = encryption.decrypt(encrypted_notification)
template = get_model_templates(notification['template'])
try:
notification_db_object = Notification(
id=notification_id,
template_id=notification['template'],
to=notification['to'],
service_id=service_id,
status='sent'
)
save_notification(notification_db_object)
try:
twilio_client.send_sms(notification['to'], template.content)
except TwilioClientException as e:
current_app.logger.debug(e)
save_notification(notification_db_object, {"status": "failed"})
except SQLAlchemyError as e:
current_app.logger.debug(e)

13
app/clients/__init__.py Normal file
View File

@@ -0,0 +1,13 @@
class ClientException(Exception):
'''
Base Exceptions for sending notifications that fail
'''
pass
class Client(object):
'''
Base client for sending notifications.
'''
pass

17
app/clients/sms/__init__.py Normal file
View File

@@ -0,0 +1,17 @@
from app.clients import (Client, ClientException)
class SmsClientException(ClientException):
'''
Base Exception for SmsClients
'''
pass
class SmsClient(Client):
'''
Base Sms client for sending smss.
'''
def send_sms(self, *args, **kwargs):
raise NotImplemented('TODO Need to implement.')

46
app/clients/sms/twilio.py Normal file
View File

@@ -0,0 +1,46 @@
import logging
from app.clients.sms import (
SmsClient, SmsClientException)
from twilio.rest import TwilioRestClient
from twilio import TwilioRestException
logger = logging.getLogger(__name__)
class TwilioClientException(SmsClientException):
pass
class TwilioClient(SmsClient):
'''
Twilio sms client.
'''
def init_app(self, config, *args, **kwargs):
super(TwilioClient, self).__init__(*args, **kwargs)
self.client = TwilioRestClient(
config.config.get('TWILIO_ACCOUNT_SID'),
config.config.get('TWILIO_AUTH_TOKEN'))
self.from_number = config.config.get('TWILIO_NUMBER')
def send_sms(self, to, content):
try:
response = self.client.messages.create(
body=content,
to=to,
from_=self.from_number
)
return response.sid
except TwilioRestException as e:
logger.exception(e)
raise TwilioClientException(e)
def status(self, message_id):
try:
response = self.client.messages.get(message_id)
if response.status in ('delivered', 'undelivered', 'failed'):
return response.status
return None
except TwilioRestException as e:
logger.exception(e)
raise TwilioClientException(e)

8
app/dao/notifications_dao.py
View File

@@ -14,9 +14,13 @@ def save_notification(notification, update_dict={}):
db.session.commit()
def get_notification(service_id, job_id, notification_id):
def get_notification_for_job(service_id, job_id, notification_id):
return Notification.query.filter_by(service_id=service_id, job_id=job_id, id=notification_id).one()
def get_notifications(service_id, job_id):
def get_notifications_for_job(service_id, job_id):
return Notification.query.filter_by(service_id=service_id, job_id=job_id).all()
def get_notification(service_id, notification_id):
return Notification.query.filter_by(service_id=service_id, id=notification_id).one()

14
app/encryption.py
View File

@@ -1,5 +1,19 @@
from flask.ext.bcrypt import generate_password_hash, check_password_hash
from itsdangerous import URLSafeSerializer
class Encryption:
def init_app(self, app):
self.serializer = URLSafeSerializer(app.config.get('SECRET_KEY'))
self.salt = app.config.get('DANGEROUS_SALT')
def encrypt(self, thing_to_encrypt):
return self.serializer.dumps(thing_to_encrypt, salt=self.salt)
def decrypt(self, thing_to_decrypt):
return self.serializer.loads(thing_to_decrypt, salt=self.salt)
def hashpw(password):
return generate_password_hash(password.encode('UTF-8'), 10)

16
app/job/rest.py
View File

@@ -17,11 +17,7 @@ from app.dao.jobs_dao import (
get_jobs_by_service
)
from app.dao.notifications_dao import (
save_notification,
get_notification,
get_notifications
)
from app.dao import notifications_dao
from app.schemas import (
job_schema,
@@ -89,7 +85,7 @@ def create_notification_for_job(service_id, job_id):
if errors:
return jsonify(result="error", message=errors), 400
try:
save_notification(notification)
notifications_dao.save_notification(notification)
except Exception as e:
return jsonify(result="error", message=str(e)), 500
return jsonify(data=notification_status_schema.dump(notification).data), 201
@@ -100,7 +96,7 @@ def create_notification_for_job(service_id, job_id):
def get_notification_for_job(service_id, job_id, notification_id=None):
if notification_id:
try:
notification = get_notification(service_id, job_id, notification_id)
notification = notifications_dao.get_notification_for_job(service_id, job_id, notification_id)
data, errors = notification_status_schema.dump(notification)
return jsonify(data=data)
except DataError:
@@ -108,7 +104,7 @@ def get_notification_for_job(service_id, job_id, notification_id=None):
except NoResultFound:
return jsonify(result="error", message="Notification not found"), 404
else:
notifications = get_notifications(service_id, job_id)
notifications = notifications_dao.get_notifications_for_job(service_id, job_id)
data, errors = notifications_status_schema.dump(notifications)
return jsonify(data=data)
@@ -116,13 +112,13 @@ def get_notification_for_job(service_id, job_id, notification_id=None):
@job.route('/<job_id>/notification/<notification_id>', methods=['PUT'])
def update_notification_for_job(service_id, job_id, notification_id):
notification = get_notification(service_id, job_id, notification_id)
notification = notifications_dao.get_notification_for_job(service_id, job_id, notification_id)
update_dict, errors = notification_status_schema_load_json.load(request.get_json())
if errors:
return jsonify(result="error", message=errors), 400
try:
save_notification(notification, update_dict=update_dict)
notifications_dao.save_notification(notification, update_dict=update_dict)
except Exception as e:
return jsonify(result="error", message=str(e)), 400

41
app/notifications/rest.py
View File

@@ -6,30 +6,51 @@ from flask import (
request
)
from app import api_user
from app import api_user, encryption
from app.aws_sqs import add_notification_to_queue
from app.dao import (templates_dao)
from app.dao import (templates_dao, notifications_dao)
from app.schemas import (
email_notification_schema, sms_template_notification_schema)
email_notification_schema,
sms_template_notification_schema,
notification_status_schema
)
from app.celery.tasks import send_sms
from sqlalchemy.orm.exc import NoResultFound
notifications = Blueprint('notifications', __name__)
@notifications.route('/<notification_id>', methods=['GET'])
def create_notification_id():
return str(uuid.uuid4())
@notifications.route('/<string:notification_id>', methods=['GET'])
def get_notifications(notification_id):
# TODO return notification id details
return jsonify({'id': notification_id}), 200
try:
notification = notifications_dao.get_notification(api_user['client'], notification_id)
return jsonify({'notification': notification_status_schema.dump(notification).data}), 200
except NoResultFound:
return jsonify(result="error", message="not found"), 404
@notifications.route('/sms', methods=['POST'])
def create_sms_notification():
resp_json = request.get_json()
notification, errors = sms_template_notification_schema.load(resp_json)
notification, errors = sms_template_notification_schema.load(request.get_json())
if errors:
return jsonify(result="error", message=errors), 400
notification_id = add_notification_to_queue(api_user['client'], notification['template'], 'sms', notification)
try:
templates_dao.get_model_templates(template_id=notification['template'], service_id=api_user['client'])
except NoResultFound:
return jsonify(result="error", message={'template': ['Template not found']}), 400
notification_id = create_notification_id()
send_sms.apply_async((
api_user['client'],
notification_id,
encryption.encrypt(notification)),
queue='sms')
return jsonify({'notification_id': notification_id}), 201

7
app/schemas.py
View File

@@ -93,13 +93,6 @@ class SmsTemplateNotificationSchema(SmsNotificationSchema):
template = fields.Int(required=True)
job = fields.String()
@validates('template')
def validate_template(self, value):
if not models.Template.query.filter_by(id=value).first():
# TODO is this message consistent with what marshmallow
# would normally produce.
raise ValidationError('Template not found')
@validates_schema
def validate_schema(self, data):
"""

12
aws_run_celery.py Normal file
View File

@@ -0,0 +1,12 @@
#!/usr/bin/env python
from app import notify_celery, create_app
from credstash import getAllSecrets
import os
# on aws get secrets and export to env
secrets = getAllSecrets(region="eu-west-1")
for key, val in secrets.items():
os.environ[key] = val
application = create_app()
application.app_context().push()

22
config.py
View File

@@ -20,6 +20,28 @@ class Config(object):
SQLALCHEMY_RECORD_QUERIES = True
VERIFY_CODE_FROM_EMAIL_ADDRESS = os.environ['VERIFY_CODE_FROM_EMAIL_ADDRESS']
BROKER_URL = 'sqs://'
BROKER_TRANSPORT_OPTIONS = {
'region': 'eu-west-1',
'polling_interval': 1, # 1 second
'visibility_timeout': 60, # 60 seconds
'queue_name_prefix': os.environ['NOTIFICATION_QUEUE_PREFIX']
}
CELERY_ENABLE_UTC = True,
CELERY_TIMEZONE = 'Europe/London'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
# CELERYBEAT_SCHEDULE = {
# 'refresh-queues': {
# 'task': 'refresh-services',
# 'schedule': timedelta(seconds=5)
# }
# }
CELERY_IMPORTS = ('app.celery.tasks',)
TWILIO_ACCOUNT_SID = os.getenv('TWILIO_ACCOUNT_SID')
TWILIO_AUTH_TOKEN = os.getenv('TWILIO_AUTH_TOKEN')
TWILIO_NUMBER = os.getenv('TWILIO_NUMBER')
class Development(Config):
DEBUG = True

3
environment_test.sh
View File

@@ -10,3 +10,6 @@ export NOTIFICATION_QUEUE_PREFIX='notification_development-test'
export SECRET_KEY='secret-key'
export SQLALCHEMY_DATABASE_URI='postgresql://localhost/test_notification_api'
export VERIFY_CODE_FROM_EMAIL_ADDRESS='no-reply@notify.works'
export TWILIO_ACCOUNT_SID="test"
export TWILIO_AUTH_TOKEN="test"
export TWILIO_NUMBER="test"

3
requirements.txt
View File

@@ -13,6 +13,9 @@ itsdangerous==0.24
Flask-Bcrypt==0.6.2
credstash==1.8.0
boto3==1.2.3
celery==3.1.20
twilio==4.6.0
git+https://github.com/alphagov/notifications-python-client.git@0.2.6#egg=notifications-python-client==0.2.6

5
run_celery.py Normal file
View File

@@ -0,0 +1,5 @@
#!/usr/bin/env python
from app import notify_celery, create_app
application = create_app()
application.app_context().push()

6
scripts/run_celery.sh Executable file
View File

@@ -0,0 +1,6 @@
#!/bin/bash
set -e
source environment.sh
celery -A run_celery.notify_celery worker --loglevel=INFO --logfile=/var/log/notify/application.log --concurrency=4 -Q sms

76
tests/app/celery/test_tasks.py Normal file
View File

@@ -0,0 +1,76 @@
import uuid
import pytest
from app.celery.tasks import send_sms
from app import twilio_client
from app.clients.sms.twilio import TwilioClientException
from app.dao import notifications_dao
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm.exc import NoResultFound
def test_should_send_template_to_correct_sms_provider_and_persist(sample_template, mocker):
notification = {
"template": sample_template.id,
"to": "+441234123123"
}
mocker.patch('app.encryption.decrypt', return_value=notification)
mocker.patch('app.twilio_client.send_sms')
notification_id = uuid.uuid4()
send_sms(
sample_template.service_id,
notification_id,
"encrypted-in-reality")
twilio_client.send_sms.assert_called_once_with("+441234123123", sample_template.content)
persisted_notification = notifications_dao.get_notification(sample_template.service_id, notification_id)
assert persisted_notification.id == notification_id
assert persisted_notification.to == '+441234123123'
assert persisted_notification.template_id == sample_template.id
assert persisted_notification.status == 'sent'
def test_should_persist_notification_as_failed_if_sms_client_fails(sample_template, mocker):
notification = {
"template": sample_template.id,
"to": "+441234123123"
}
mocker.patch('app.encryption.decrypt', return_value=notification)
mocker.patch('app.twilio_client.send_sms', side_effect=TwilioClientException())
notification_id = uuid.uuid4()
send_sms(
sample_template.service_id,
notification_id,
"encrypted-in-reality")
twilio_client.send_sms.assert_called_once_with("+441234123123", sample_template.content)
persisted_notification = notifications_dao.get_notification(sample_template.service_id, notification_id)
assert persisted_notification.id == notification_id
assert persisted_notification.to == '+441234123123'
assert persisted_notification.template_id == sample_template.id
assert persisted_notification.status == 'failed'
def test_should_not_send_sms_if_db_peristance_failed(sample_template, mocker):
notification = {
"template": sample_template.id,
"to": "+441234123123"
}
mocker.patch('app.encryption.decrypt', return_value=notification)
mocker.patch('app.twilio_client.send_sms', side_effect=TwilioClientException())
mocker.patch('app.db.session.add', side_effect=SQLAlchemyError())
notification_id = uuid.uuid4()
send_sms(
sample_template.service_id,
notification_id,
"encrypted-in-reality")
twilio_client.send_sms.assert_not_called()
with pytest.raises(NoResultFound) as e:
notifications_dao.get_notification(sample_template.service_id, notification_id)
assert 'No row was found for one' in str(e.value)

15
tests/app/conftest.py
View File

@@ -1,7 +1,6 @@
import pytest
from flask import jsonify
from app.models import (User, Service, Template, ApiKey, Job, VerifyCode, Notification)
from app.models import (User, Service, Template, ApiKey, Job, Notification)
from app.dao.users_dao import (save_model_user, create_user_code, create_secret_code)
from app.dao.services_dao import save_model_service
from app.dao.templates_dao import save_model_template
@@ -11,6 +10,18 @@ from app.dao.notifications_dao import save_notification
import uuid
@pytest.fixture(scope='function')
def service_factory(notify_db, notify_db_session):
class ServiceFactory(object):
def get(self, service_name):
user = sample_user(notify_db, notify_db_session)
service = sample_service(notify_db, notify_db_session, service_name, user)
sample_template(notify_db, notify_db_session, service=service)
return service
return ServiceFactory()
@pytest.fixture(scope='function')
def sample_user(notify_db,
notify_db_session,

19
tests/app/dao/test_notification_dao.py
View File

@@ -3,7 +3,8 @@ from app.models import Notification
from app.dao.notifications_dao import (
save_notification,
get_notification,
get_notifications
get_notification_for_job,
get_notifications_for_job
)
@@ -31,6 +32,13 @@ def test_save_notification(notify_db, notify_db_session, sample_template, sample
assert 'sent' == notification_from_db.status
def test_get_notification(notify_db, notify_db_session, sample_notification):
notifcation_from_db = get_notification(
sample_notification.service.id,
sample_notification.id)
assert sample_notification == notifcation_from_db
def test_save_notification_no_job_id(notify_db, notify_db_session, sample_template):
assert Notification.query.count() == 0
@@ -54,9 +62,10 @@ def test_save_notification_no_job_id(notify_db, notify_db_session, sample_templa
def test_get_notification_for_job(notify_db, notify_db_session, sample_notification):
notifcation_from_db = get_notification(sample_notification.service.id,
sample_notification.job_id,
sample_notification.id)
notifcation_from_db = get_notification_for_job(
sample_notification.service.id,
sample_notification.job_id,
sample_notification.id)
assert sample_notification == notifcation_from_db
@@ -70,7 +79,7 @@ def test_get_all_notifications_for_job(notify_db, notify_db_session, sample_job)
template=sample_job.template,
job=sample_job)
notifcations_from_db = get_notifications(sample_job.service.id, sample_job.id)
notifcations_from_db = get_notifications_for_job(sample_job.service.id, sample_job.id)
assert len(notifcations_from_db) == 5

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

@@ -1,240 +1,231 @@
import moto
import uuid
import app.celery.tasks
import moto
from tests import create_authorization_header
from flask import url_for, json
from flask import json
from app.models import Service
from tests.app.conftest import sample_service as create_sample_service
from tests.app.conftest import sample_template as create_sample_template
from app.dao.templates_dao import get_model_templates
def test_get_notifications(
notify_api, notify_db, notify_db_session, sample_api_key, mocker):
"""
Tests GET endpoint '/' to retrieve entire service list.
"""
def test_get_notification_by_id(notify_api, sample_notification):
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(
service_id=sample_api_key.service_id,
path=url_for('notifications.get_notifications', notification_id=123),
service_id=sample_notification.service_id,
path='/notifications/{}'.format(sample_notification.id),
method='GET')
response = client.get(
url_for('notifications.get_notifications', notification_id=123),
'/notifications/{}'.format(sample_notification.id),
headers=[auth_header])
notification = json.loads(response.get_data(as_text=True))['notification']
assert notification['status'] == 'sent'
assert notification['template'] == sample_notification.template.id
assert notification['to'] == '+44709123456'
assert notification['service'] == str(sample_notification.service_id)
assert response.status_code == 200
def test_get_notifications_empty_result(
notify_api, notify_db, notify_db_session, sample_api_key, mocker):
"""
Tests GET endpoint '/' to retrieve entire service list.
"""
def test_get_notifications_empty_result(notify_api, sample_api_key):
with notify_api.test_request_context():
with notify_api.test_client() as client:
missing_notification_id = uuid.uuid4()
auth_header = create_authorization_header(
service_id=sample_api_key.service_id,
path=url_for('notifications.get_notifications', notification_id=123),
path='/notifications/{}'.format(missing_notification_id),
method='GET')
response = client.get(
url_for('notifications.get_notifications', notification_id=123),
path='/notifications/{}'.format(missing_notification_id),
headers=[auth_header])
assert response.status_code == 200
notification = json.loads(response.get_data(as_text=True))
assert notification['result'] == "error"
assert notification['message'] == "not found"
assert response.status_code == 404
def test_create_sms_should_reject_if_no_phone_numbers(
notify_api, notify_db, notify_db_session, sample_api_key, mocker):
"""
Tests GET endpoint '/' to retrieve entire service list.
"""
def test_create_sms_should_reject_if_missing_required_fields(notify_api, sample_api_key, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'template': "my message"
}
mocker.patch('app.celery.tasks.send_sms.apply_async')
data = {}
auth_header = create_authorization_header(
service_id=sample_api_key.service_id,
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
path='/notifications/sms',
method='POST')
response = client.post(
url_for('notifications.create_sms_notification'),
path='/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
app.celery.tasks.send_sms.apply_async.assert_not_called()
assert json_resp['result'] == 'error'
assert 'Missing data for required field.' in json_resp['message']['to'][0]
assert 'Missing data for required field.' in json_resp['message']['template'][0]
assert response.status_code == 400
def test_should_reject_bad_phone_numbers(
notify_api, notify_db, notify_db_session, mocker):
"""
Tests GET endpoint '/' to retrieve entire service list.
"""
def test_should_reject_bad_phone_numbers(notify_api, sample_template, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.tasks.send_sms.apply_async')
data = {
'to': 'invalid',
'template': "my message"
'template': sample_template.id
}
auth_header = create_authorization_header(
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
path='/notifications/sms',
method='POST')
response = client.post(
url_for('notifications.create_sms_notification'),
path='/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
app.celery.tasks.send_sms.apply_async.assert_not_called()
assert json_resp['result'] == 'error'
assert len(json_resp['message'].keys()) == 1
assert 'Invalid phone number, must be of format +441234123123' in json_resp['message']['to']
assert response.status_code == 400
def test_send_notification_restrict_mobile(notify_api,
notify_db,
notify_db_session,
sample_api_key,
sample_template,
sample_user,
mocker):
"""
Test POST endpoint '/sms' with service notification with mobile number
not in restricted list.
"""
def test_send_notification_invalid_template_id(notify_api, sample_template, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.tasks.send_sms.apply_async')
data = {
'to': '+441234123123',
'template': 9999
}
auth_header = create_authorization_header(
service_id=sample_template.service.id,
request_body=json.dumps(data),
path='/notifications/sms',
method='POST')
response = client.post(
path='/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
json_resp = json.loads(response.get_data(as_text=True))
app.celery.tasks.send_sms.apply_async.assert_not_called()
assert response.status_code == 400
assert len(json_resp['message'].keys()) == 1
assert 'Template not found' in json_resp['message']['template']
def test_prevents_sending_to_any_mobile_on_restricted_service(notify_api, sample_template, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.tasks.send_sms.apply_async')
Service.query.filter_by(
id=sample_template.service.id).update({'restricted': True})
id=sample_template.service.id
).update(
{'restricted': True}
)
invalid_mob = '+449999999999'
data = {
'to': invalid_mob,
'template': sample_template.id
}
assert invalid_mob != sample_user.mobile_number
auth_header = create_authorization_header(
service_id=sample_template.service.id,
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
path='/notifications/sms',
method='POST')
response = client.post(
url_for('notifications.create_sms_notification'),
path='/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
json_resp = json.loads(response.get_data(as_text=True))
app.celery.tasks.send_sms.apply_async.assert_not_called()
assert response.status_code == 400
assert 'Invalid phone number for restricted service' in json_resp['message']['restricted']
def test_send_notification_invalid_template_id(notify_api,
notify_db,
notify_db_session,
sample_api_key,
sample_template,
sample_user,
mocker):
"""
Tests POST endpoint '/sms' with notifications-admin notification with invalid template id
"""
def test_should_not_allow_template_from_another_service(notify_api, service_factory, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.tasks.send_sms.apply_async')
Service.query.filter_by(
id=sample_template.service.id).update({'restricted': True})
invalid_mob = '+449999999999'
data = {
'to': invalid_mob,
'template': 9999
}
assert invalid_mob != sample_user.mobile_number
auth_header = create_authorization_header(
service_id=sample_template.service.id,
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
method='POST')
service_1 = service_factory.get('service 1')
service_2 = service_factory.get('service 2')
response = client.post(
url_for('notifications.create_sms_notification'),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
service_2_templates = get_model_templates(service_id=service_2.id)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert 'Template not found' in json_resp['message']['template']
@moto.mock_sqs
def test_should_not_allow_template_from_other_service(notify_api,
notify_db,
notify_db_session,
sample_template,
sample_admin_service_id,
mocker):
"""
Tests POST endpoint '/sms' with notifications.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'to': '+441234123123',
'template': sample_template.id
'template': service_2_templates[0].id
}
auth_header = create_authorization_header(
service_id=sample_admin_service_id,
service_id=service_1.id,
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
path='/notifications/sms',
method='POST')
response = client.post(
url_for('notifications.create_sms_notification'),
path='/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
json_resp = json.loads(response.get_data(as_text=True))
app.celery.tasks.send_sms.apply_async.assert_not_called()
assert response.status_code == 400
assert 'Invalid template' in json_resp['message']['restricted']
@moto.mock_sqs
def test_should_allow_valid_message(notify_api,
notify_db,
notify_db_session,
sqs_client_conn,
sample_user,
sample_template,
mocker):
"""
Tests POST endpoint '/sms' with notifications-admin notification.
"""
def test_should_allow_valid_sms_notification(notify_api, sample_template, mocker):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocker.patch('app.celery.tasks.send_sms.apply_async')
mocker.patch('app.encryption.encrypt', return_value="something_encrypted")
data = {
'to': '+441234123123',
'template': sample_template.id
}
auth_header = create_authorization_header(
request_body=json.dumps(data),
path=url_for('notifications.create_sms_notification'),
method='POST')
path='/notifications/sms',
method='POST',
service_id=sample_template.service_id
)
response = client.post(
url_for('notifications.create_sms_notification'),
path='/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
notification_id = json.loads(response.data)['notification_id']
app.celery.tasks.send_sms.apply_async.assert_called_once_with(
(str(sample_template.service_id),
notification_id,
"something_encrypted"),
queue="sms"
)
assert response.status_code == 201
assert json.loads(response.data)['notification_id'] is not None
assert notification_id
@moto.mock_sqs
@@ -259,11 +250,11 @@ def test_send_email_valid_data(notify_api,
}
auth_header = create_authorization_header(
request_body=json.dumps(data),
path=url_for('notifications.create_email_notification'),
path='/notifications/email',
method='POST')
response = client.post(
url_for('notifications.create_email_notification'),
path='/notifications/email',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
@@ -283,7 +274,7 @@ def test_valid_message_with_service_id(notify_api,
with notify_api.test_client() as client:
job_id = uuid.uuid4()
service_id = sample_template.service.id
url = url_for('notifications.create_sms_for_service', service_id=service_id)
url = '/notifications/sms/service/{}'.format(service_id)
data = {
'to': '+441234123123',
'template': sample_template.id,
@@ -316,7 +307,7 @@ def test_message_with_incorrect_service_id_should_fail(notify_api,
job_id = uuid.uuid4()
invalid_service_id = uuid.uuid4()
url = url_for('notifications.create_sms_for_service', service_id=invalid_service_id)
url = '/notifications/sms/service/{}'.format(invalid_service_id)
data = {
'to': '+441234123123',

20
tests/app/test_encryption.py Normal file
View File

@@ -0,0 +1,20 @@
from app.encryption import Encryption
encryption = Encryption()
def test_should_encrypt_content(notify_api):
encryption.init_app(notify_api)
assert encryption.encrypt("this") != "this"
def test_should_decrypt_content(notify_api):
encryption.init_app(notify_api)
encrypted = encryption.encrypt("this")
assert encryption.decrypt(encrypted) == "this"
def test_should_encrypt_json(notify_api):
encryption.init_app(notify_api)
encrypted = encryption.encrypt({"this": "that"})
assert encryption.decrypt(encrypted) == {"this": "that"}

7
tests/conftest.py
View File

@@ -59,6 +59,13 @@ def notify_db_session(request):
request.addfinalizer(teardown)
@pytest.fixture(scope='function')
def notify_config(notify_api):
notify_api.config['NOTIFY_API_ENVIRONMENT'] = 'test'
notify_api.config.from_object(configs['test'])
return notify_api.config
@pytest.fixture(scope='function')
def os_environ(request):
env_patch = mock.patch('os.environ', {})