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

Merge branch 'master' into fix-mocks

Browse Source 
Conflicts:
	tests/app/notifications/test_validators.py
This commit is contained in:
Martyn Inglis
2016-11-29 13:32:40 +00:00
parent 2182b5715e 2ae8967a5f
commit c437615661
16 changed files with 656 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
app/dao/notifications_dao.py
View File

@@ -8,7 +8,7 @@ from itertools import groupby
from flask import current_app
from werkzeug.datastructures import MultiDict
from sqlalchemy import (desc, func, or_, and_, asc, cast, Text)
from sqlalchemy import (desc, func, or_, and_, asc)
from sqlalchemy.orm import joinedload
from app import db, create_uuid
@@ -262,7 +262,8 @@ def get_notifications_for_service(
key_type=None,
personalisation=False,
include_jobs=False,
include_from_test_key=False
include_from_test_key=False,
older_than=None
):
if page_size is None:
page_size = current_app.config['PAGE_SIZE']
@@ -273,6 +274,11 @@ def get_notifications_for_service(
days_ago = date.today() - timedelta(days=limit_days)
filters.append(func.date(Notification.created_at) >= days_ago)
if older_than is not None:
older_than_created_at = db.session.query(
Notification.created_at).filter(Notification.id == older_than).as_scalar()
filters.append(Notification.created_at < older_than_created_at)
if not include_jobs or (key_type and key_type != KEY_TYPE_NORMAL):
filters.append(Notification.job_id.is_(None))
@@ -296,15 +302,21 @@ def get_notifications_for_service(
def _filter_query(query, filter_dict=None):
if filter_dict is None:
filter_dict = MultiDict()
else:
filter_dict = MultiDict(filter_dict)
statuses = filter_dict.getlist('status') if 'status' in filter_dict else None
return query
multidict = MultiDict(filter_dict)
# filter by status
statuses = multidict.getlist('status')
if statuses:
statuses = Notification.substitute_status(statuses)
query = query.filter(Notification.status.in_(statuses))
template_types = filter_dict.getlist('template_type') if 'template_type' in filter_dict else None
# filter by template
template_types = multidict.getlist('template_type')
if template_types:
query = query.join(Template).filter(Template.template_type.in_(template_types))
return query

96
app/models.py
View File

@@ -6,7 +6,7 @@ from sqlalchemy.dialects.postgresql import (
UUID,
JSON
)
from sqlalchemy import UniqueConstraint, and_, desc
from sqlalchemy import UniqueConstraint, and_
from sqlalchemy.orm import foreign, remote
from notifications_utils.recipients import (
validate_email_address,
@@ -178,9 +178,8 @@ class ServiceWhitelist(db.Model):
else:
return instance
def __repr__(self):
return 'Recipient {} of type: {}'.format(self.recipient,
self.recipient_type)
def __repr__(self):
return 'Recipient {} of type: {}'.format(self.recipient, self.recipient_type)
class ApiKey(db.Model, Versioned):
@@ -285,7 +284,12 @@ class Template(db.Model):
def get_link(self):
# TODO: use "/v2/" route once available
return url_for("template.get_template_by_id_and_service_id", service_id=self.service_id, template_id=self.id)
return url_for(
"template.get_template_by_id_and_service_id",
service_id=self.service_id,
template_id=self.id,
_external=True
)
class TemplateHistory(db.Model):
@@ -466,13 +470,27 @@ NOTIFICATION_TECHNICAL_FAILURE = 'technical-failure'
NOTIFICATION_TEMPORARY_FAILURE = 'temporary-failure'
NOTIFICATION_PERMANENT_FAILURE = 'permanent-failure'
NOTIFICATION_STATUS_TYPES_FAILED = [
NOTIFICATION_TECHNICAL_FAILURE,
NOTIFICATION_TEMPORARY_FAILURE,
NOTIFICATION_PERMANENT_FAILURE,
]
NOTIFICATION_STATUS_TYPES_COMPLETED = [
NOTIFICATION_DELIVERED,
NOTIFICATION_FAILED,
NOTIFICATION_TECHNICAL_FAILURE,
NOTIFICATION_TEMPORARY_FAILURE,
NOTIFICATION_PERMANENT_FAILURE,
]
NOTIFICATION_STATUS_TYPES_BILLABLE = [
NOTIFICATION_SENDING,
NOTIFICATION_DELIVERED,
NOTIFICATION_FAILED,
NOTIFICATION_TECHNICAL_FAILURE,
NOTIFICATION_TEMPORARY_FAILURE,
NOTIFICATION_PERMANENT_FAILURE
NOTIFICATION_PERMANENT_FAILURE,
]
NOTIFICATION_STATUS_TYPES = [
@@ -483,7 +501,7 @@ NOTIFICATION_STATUS_TYPES = [
NOTIFICATION_FAILED,
NOTIFICATION_TECHNICAL_FAILURE,
NOTIFICATION_TEMPORARY_FAILURE,
NOTIFICATION_PERMANENT_FAILURE
NOTIFICATION_PERMANENT_FAILURE,
]
NOTIFICATION_STATUS_TYPES_ENUM = db.Enum(*NOTIFICATION_STATUS_TYPES, name='notify_status_type')
@@ -544,33 +562,52 @@ class Notification(db.Model):
if personalisation:
self._personalisation = encryption.encrypt(personalisation)
def cost(self):
if not self.sent_by or self.billable_units == 0:
return 0
provider_rate = db.session.query(
ProviderRates
).join(ProviderDetails).filter(
ProviderDetails.identifier == self.sent_by,
ProviderRates.provider_id == ProviderDetails.id
).order_by(
desc(ProviderRates.valid_from)
).limit(1).one()
return float(provider_rate.rate * self.billable_units)
def completed_at(self):
if self.status in [
NOTIFICATION_DELIVERED,
NOTIFICATION_FAILED,
NOTIFICATION_TECHNICAL_FAILURE,
NOTIFICATION_TEMPORARY_FAILURE,
NOTIFICATION_PERMANENT_FAILURE
]:
if self.status in NOTIFICATION_STATUS_TYPES_COMPLETED:
return self.updated_at.strftime(DATETIME_FORMAT)
return None
@staticmethod
def substitute_status(status_or_statuses):
"""
static function that takes a status or list of statuses and substitutes our new failure types if it finds
the deprecated one
> IN
'failed'
< OUT
['technical-failure', 'temporary-failure', 'permanent-failure']
-
> IN
['failed', 'created']
< OUT
['technical-failure', 'temporary-failure', 'permanent-failure', 'created']
:param status_or_statuses: a single status or list of statuses
:return: a single status or list with the current failure statuses substituted for 'failure'
"""
def _substitute_status_str(_status):
return NOTIFICATION_STATUS_TYPES_FAILED if _status == NOTIFICATION_FAILED else _status
def _substitute_status_seq(_statuses):
if NOTIFICATION_FAILED in _statuses:
_statuses = list(set(
NOTIFICATION_STATUS_TYPES_FAILED + [_s for _s in _statuses if _s != NOTIFICATION_FAILED]
))
return _statuses
if isinstance(status_or_statuses, str):
return _substitute_status_str(status_or_statuses)
return _substitute_status_seq(status_or_statuses)
def serialize(self):
template_dict = {
@@ -591,7 +628,6 @@ class Notification(db.Model):
"line_5": None,
"line_6": None,
"postcode": None,
"cost": self.cost(),
"type": self.notification_type,
"status": self.status,
"template": template_dict,

4
app/notifications/rest.py
View File

@@ -170,8 +170,8 @@ def get_notification_by_id(notification_id):
def get_all_notifications():
data = notifications_filter_schema.load(request.args).data
include_jobs = data.get('include_jobs', False)
page = data['page'] if 'page' in data else 1
page_size = data['page_size'] if 'page_size' in data else current_app.config.get('PAGE_SIZE')
page = data.get('page', 1)
page_size = data.get('page_size', current_app.config.get('PAGE_SIZE'))
limit_days = data.get('limit_days')
pagination = notifications_dao.get_notifications_for_service(

4
app/schema_validation/__init__.py
View File

@@ -42,6 +42,4 @@ def build_error_message(errors):
def __format_message(e):
s = e.message.split("'")
msg = "{}{}".format(s[1], s[2])
return msg if not e.cause else "{} {}".format(e.path[0], e.cause.message)
return e.message.replace("'", "") if not e.cause else "{} {}".format(e.path[0], e.cause.message)

1
app/schemas.py
View File

@@ -468,6 +468,7 @@ class NotificationsFilterSchema(ma.Schema):
limit_days = fields.Int(required=False)
include_jobs = fields.Boolean(required=False)
include_from_test_key = fields.Boolean(required=False)
older_than = fields.UUID(required=False)
@pre_load
def handle_multidict(self, in_data):

41
app/v2/notifications/get_notifications.py
View File

@@ -1,8 +1,9 @@
from flask import jsonify
from flask import jsonify, request, url_for
from app import api_user
from app.dao import notifications_dao
from app.schema_validation import validate
from app.v2.notifications import notification_blueprint
from app.v2.notifications.notification_schemas import get_notifications_request
@notification_blueprint.route("/<uuid:id>", methods=['GET'])
@@ -14,9 +15,35 @@ def get_notification_by_id(id):
return jsonify(notification.serialize()), 200
@notification_blueprint.route("/", methods=['GET'])
@notification_blueprint.route("", methods=['GET'])
def get_notifications():
# validate notifications request arguments
# fetch all notifications
# return notifications_response schema
pass
_data = request.args.to_dict(flat=False)
if 'older_than' in _data:
# flat=False makes everything a list, but we only ever allow one value for "older_than"
_data['older_than'] = _data['older_than'][0]
data = validate(_data, get_notifications_request)
paginated_notifications = notifications_dao.get_notifications_for_service(
str(api_user.service_id),
filter_dict=data,
key_type=api_user.key_type,
personalisation=True,
older_than=data.get('older_than')
)
def _build_links(notifications):
_links = {
'current': url_for(".get_notifications", _external=True, **data),
}
if len(notifications):
next_query_params = dict(data, older_than=notifications[-1].id)
_links['next'] = url_for(".get_notifications", _external=True, **next_query_params)
return _links
return jsonify(
notifications=[notification.serialize() for notification in paginated_notifications.items],
links=_build_links(paginated_notifications.items)
), 200

57
app/v2/notifications/notification_schemas.py
View File

@@ -1,3 +1,4 @@
from app.models import NOTIFICATION_STATUS_TYPES, TEMPLATE_TYPES
from app.schema_validation.definitions import (uuid, personalisation)
# this may belong in a templates module
@@ -57,7 +58,6 @@ get_notification_response = {
"line_5": {"type": ["string", "null"]},
"line_6": {"type": ["string", "null"]},
"postcode": {"type": ["string", "null"]},
"cost": {"type": "number"},
"type": {"enum": ["sms", "letter", "email"]},
"status": {"type": "string"},
"template": template,
@@ -69,11 +69,62 @@ get_notification_response = {
# technically, all keys are required since we always have all of them
"id", "reference", "email_address", "phone_number",
"line_1", "line_2", "line_3", "line_4", "line_5", "line_6", "postcode",
"cost", "type", "status", "template",
"created_at", "sent_at", "completed_at"
"type", "status", "template", "created_at", "sent_at", "completed_at"
]
}
get_notifications_request = {
"$schema": "http://json-schema.org/draft-04/schema#",
"description": "schema for query parameters allowed when getting list of notifications",
"type": "object",
"properties": {
"status": {
"type": "array",
"items": {
"enum": NOTIFICATION_STATUS_TYPES
}
},
"template_type": {
"type": "array",
"items": {
"enum": TEMPLATE_TYPES
}
},
"older_than": uuid
},
"additionalProperties": False,
}
get_notifications_response = {
"$schema": "http://json-schema.org/draft-04/schema#",
"description": "GET list of notifications response schema",
"type": "object",
"properties": {
"notifications": {
"type": "array",
"items": {
"type": "object",
"ref": get_notification_response
}
},
"links": {
"type": "object",
"properties": {
"current": {
"type": "string"
},
"next": {
"type": "string"
}
},
"additionalProperties": False,
"required": ["current"]
}
},
"additionalProperties": False,
"required": ["notifications", "links"]
}
post_sms_request = {
"$schema": "http://json-schema.org/draft-04/schema#",
"description": "POST sms notification schema",

4
app/v2/notifications/post_notifications.py
View File

@@ -42,7 +42,7 @@ def post_sms_notification():
send_notification_to_queue(notification, service.research_mode)
resp = create_post_sms_response_from_notification(notification,
template_with_content.content,
template_with_content.replaced,
service.sms_sender,
request.url_root)
return jsonify(resp), 201
@@ -70,7 +70,7 @@ def post_email_notification():
send_notification_to_queue(notification, service.research_mode)
resp = create_post_email_response_from_notification(notification=notification,
content=template_with_content.content,
content=template_with_content.replaced,
subject=template_with_content.subject,
email_from=service.email_from,
url_root=request.url_root)

5
tests/app/conftest.py
View File

@@ -21,7 +21,7 @@ from app.models import (
NotificationStatistics,
ServiceWhitelist,
KEY_TYPE_NORMAL, KEY_TYPE_TEST, KEY_TYPE_TEAM,
MOBILE_TYPE, EMAIL_TYPE)
MOBILE_TYPE, EMAIL_TYPE, NOTIFICATION_STATUS_TYPES_COMPLETED)
from app.dao.users_dao import (save_model_user, create_user_code, create_secret_code)
from app.dao.services_dao import (dao_create_service, dao_add_user_to_service)
from app.dao.templates_dao import dao_create_template
@@ -444,7 +444,8 @@ def sample_notification(notify_db,
'notification_type': template.template_type,
'api_key_id': api_key_id,
'key_type': key_type,
'sent_by': sent_by
'sent_by': sent_by,
'updated_at': created_at if status in NOTIFICATION_STATUS_TYPES_COMPLETED else None
}
if job_row_number:
data['job_row_number'] = job_row_number

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

@@ -8,8 +8,6 @@ import pytest
from freezegun import freeze_time
from sqlalchemy.exc import SQLAlchemyError, IntegrityError
from app import db
from app.models import (
Notification,
NotificationHistory,
@@ -17,6 +15,7 @@ from app.models import (
NotificationStatistics,
TemplateStatistics,
NOTIFICATION_STATUS_TYPES,
NOTIFICATION_STATUS_TYPES_FAILED,
KEY_TYPE_NORMAL,
KEY_TYPE_TEAM,
KEY_TYPE_TEST
@@ -683,7 +682,10 @@ def test_get_all_notifications_for_job_by_status(notify_db, notify_db_session, s
assert len(notifications().items) == len(NOTIFICATION_STATUS_TYPES)
for status in NOTIFICATION_STATUS_TYPES:
assert len(notifications(filter_dict={'status': status}).items) == 1
if status == 'failed':
assert len(notifications(filter_dict={'status': status}).items) == len(NOTIFICATION_STATUS_TYPES_FAILED)
else:
assert len(notifications(filter_dict={'status': status}).items) == 1
assert len(notifications(filter_dict={'status': NOTIFICATION_STATUS_TYPES[:3]}).items) == 3

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

@@ -64,6 +64,7 @@ def test_check_service_message_limit_in_cache_with_unrestricted_service_is_allow
mocker.patch('app.notifications.validators.redis_store.get', return_value=1)
mocker.patch('app.notifications.validators.redis_store.set')
mocker.patch('app.notifications.validators.services_dao')
check_service_message_limit(key_type, sample_service)
app.notifications.validators.redis_store.set.assert_not_called()
assert not app.notifications.validators.services_dao.mock_calls

13
tests/app/public_contracts/test_GET_notification.py
View File

@@ -1,7 +1,7 @@
from . import return_json_from_response, validate_v0, validate
from app.models import ApiKey, KEY_TYPE_NORMAL
from app.dao.api_key_dao import save_model_api_key
from app.v2.notifications.notification_schemas import get_notification_response
from app.v2.notifications.notification_schemas import get_notification_response, get_notifications_response
from tests import create_authorization_header
@@ -16,6 +16,8 @@ def _get_notification(client, notification, url):
return client.get(url, headers=[auth_header])
# v2
def test_get_v2_sms_contract(client, sample_notification):
response_json = return_json_from_response(_get_notification(
client, sample_notification, '/v2/notifications/{}'.format(sample_notification.id)
@@ -30,6 +32,15 @@ def test_get_v2_email_contract(client, sample_email_notification):
validate(response_json, get_notification_response)
def test_get_v2_notifications_contract(client, sample_notification):
response_json = return_json_from_response(_get_notification(
client, sample_notification, '/v2/notifications'
))
validate(response_json, get_notifications_response)
# v0
def test_get_api_sms_contract(client, sample_notification):
response_json = return_json_from_response(_get_notification(
client, sample_notification, '/notifications/{}'.format(sample_notification.id)

66
tests/app/test_model.py
View File

@@ -1,11 +1,16 @@
import pytest
from datetime import datetime
from sqlalchemy.orm.exc import NoResultFound
from tests.app.conftest import sample_notification, sample_provider_rate
from app.models import (
ServiceWhitelist,
MOBILE_TYPE, EMAIL_TYPE)
Notification,
MOBILE_TYPE,
EMAIL_TYPE,
NOTIFICATION_CREATED,
NOTIFICATION_PENDING,
NOTIFICATION_FAILED,
NOTIFICATION_TECHNICAL_FAILURE,
NOTIFICATION_STATUS_TYPES_FAILED
)
@pytest.mark.parametrize('mobile_number', [
@@ -37,35 +42,26 @@ def test_should_not_build_service_whitelist_from_invalid_contact(recipient_type,
ServiceWhitelist.from_string('service_id', recipient_type, contact)
@pytest.mark.parametrize('provider, billable_units, expected_cost', [
('mmg', 1, 3.5),
('firetext', 2, 0.025),
('ses', 0, 0)
@pytest.mark.parametrize('initial_statuses, expected_statuses', [
# passing in single statuses as strings
(NOTIFICATION_FAILED, NOTIFICATION_STATUS_TYPES_FAILED),
(NOTIFICATION_CREATED, NOTIFICATION_CREATED),
(NOTIFICATION_TECHNICAL_FAILURE, NOTIFICATION_TECHNICAL_FAILURE),
# passing in lists containing single statuses
([NOTIFICATION_FAILED], NOTIFICATION_STATUS_TYPES_FAILED),
([NOTIFICATION_CREATED], [NOTIFICATION_CREATED]),
([NOTIFICATION_TECHNICAL_FAILURE], [NOTIFICATION_TECHNICAL_FAILURE]),
# passing in lists containing multiple statuses
([NOTIFICATION_FAILED, NOTIFICATION_CREATED], NOTIFICATION_STATUS_TYPES_FAILED + [NOTIFICATION_CREATED]),
([NOTIFICATION_CREATED, NOTIFICATION_PENDING], [NOTIFICATION_CREATED, NOTIFICATION_PENDING]),
([NOTIFICATION_CREATED, NOTIFICATION_TECHNICAL_FAILURE], [NOTIFICATION_CREATED, NOTIFICATION_TECHNICAL_FAILURE]),
# checking we don't end up with duplicates
(
[NOTIFICATION_FAILED, NOTIFICATION_CREATED, NOTIFICATION_TECHNICAL_FAILURE],
NOTIFICATION_STATUS_TYPES_FAILED + [NOTIFICATION_CREATED]
),
])
def test_calculate_cost_from_notification_billable_units(
notify_db, notify_db_session, provider, billable_units, expected_cost
):
provider_rates = [
('mmg', datetime(2016, 7, 1), 1.5),
('firetext', datetime(2016, 7, 1), 0.0125),
('mmg', datetime.utcnow(), 3.5),
]
for provider_identifier, valid_from, rate in provider_rates:
sample_provider_rate(
notify_db,
notify_db_session,
provider_identifier=provider_identifier,
valid_from=valid_from,
rate=rate
)
notification = sample_notification(notify_db, notify_db_session, billable_units=billable_units, sent_by=provider)
assert notification.cost() == expected_cost
def test_billable_units_without_provider_rates_entry_raises_exception(
notify_db, notify_db_session, sample_provider_rate
):
notification = sample_notification(notify_db, notify_db_session, sent_by='not_a_provider')
with pytest.raises(NoResultFound):
notification.cost()
def test_status_conversion_handles_failed_statuses(initial_statuses, expected_statuses):
converted_statuses = Notification.substitute_status(initial_statuses)
assert len(converted_statuses) == len(expected_statuses)
assert set(converted_statuses) == set(expected_statuses)

337
tests/app/v2/notifications/test_get_notifications.py
View File

@@ -3,7 +3,10 @@ import pytest
from app import DATETIME_FORMAT
from tests import create_authorization_header
from tests.app.conftest import sample_notification as create_sample_notification
from tests.app.conftest import (
sample_notification as create_sample_notification,
sample_template as create_sample_template
)
@pytest.mark.parametrize('billable_units, provider', [
@@ -46,7 +49,6 @@ def test_get_notification_by_id_returns_200(
'line_5': None,
'line_6': None,
'postcode': None,
'cost': sample_notification.cost(),
'type': '{}'.format(sample_notification.notification_type),
'status': '{}'.format(sample_notification.status),
'template': expected_template_response,
@@ -56,3 +58,334 @@ def test_get_notification_by_id_returns_200(
}
assert json_response == expected_response
def test_get_notification_by_id_nonexistent_id(client, sample_notification):
auth_header = create_authorization_header(service_id=sample_notification.service_id)
response = client.get(
path='/v2/notifications/dd4b8b9d-d414-4a83-9256-580046bf18f9',
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 404
assert response.headers['Content-type'] == 'application/json'
json_response = json.loads(response.get_data(as_text=True))
assert json_response == {
"errors": [
{
"error": "NoResultFound",
"message": "No result found"
}
],
"status_code": 404
}
def test_get_notification_by_id_invalid_id(client, sample_notification):
auth_header = create_authorization_header(service_id=sample_notification.service_id)
response = client.get(
path='/v2/notifications/1234-badly-formatted-id-7890',
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 404
assert response.headers['Content-type'] == 'application/json'
json_response = json.loads(response.get_data(as_text=True))
assert json_response == {
"message": "The requested URL was not found on the server. "
"If you entered the URL manually please check your spelling and try again.",
"result": "error"
}
def test_get_all_notifications_returns_200(client, notify_db, notify_db_session):
notifications = [create_sample_notification(notify_db, notify_db_session) for _ in range(2)]
notification = notifications[-1]
auth_header = create_authorization_header(service_id=notification.service_id)
response = client.get(
path='/v2/notifications',
headers=[('Content-Type', 'application/json'), auth_header])
json_response = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert response.headers['Content-type'] == "application/json"
assert json_response['links']['current'].endswith("/v2/notifications")
assert 'next' in json_response['links'].keys()
assert len(json_response['notifications']) == 2
assert json_response['notifications'][0]['id'] == str(notification.id)
assert json_response['notifications'][0]['status'] == "created"
assert json_response['notifications'][0]['template'] == {
'id': str(notification.template.id),
'uri': notification.template.get_link(),
'version': 1
}
assert json_response['notifications'][0]['phone_number'] == "+447700900855"
assert json_response['notifications'][0]['type'] == "sms"
def test_get_all_notifications_no_notifications_if_no_notifications(client, sample_service):
auth_header = create_authorization_header(service_id=sample_service.id)
response = client.get(
path='/v2/notifications',
headers=[('Content-Type', 'application/json'), auth_header])
json_response = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert response.headers['Content-type'] == "application/json"
assert json_response['links']['current'].endswith("/v2/notifications")
assert 'next' not in json_response['links'].keys()
assert len(json_response['notifications']) == 0
def test_get_all_notifications_filter_by_template_type(client, notify_db, notify_db_session):
email_template = create_sample_template(notify_db, notify_db_session, template_type="email")
sms_template = create_sample_template(notify_db, notify_db_session, template_type="sms")
notification = create_sample_notification(
notify_db, notify_db_session, template=email_template, to_field="don.draper@scdp.biz")
create_sample_notification(notify_db, notify_db_session, template=sms_template)
auth_header = create_authorization_header(service_id=notification.service_id)
response = client.get(
path='/v2/notifications?template_type=email',
headers=[('Content-Type', 'application/json'), auth_header])
json_response = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert response.headers['Content-type'] == "application/json"
assert json_response['links']['current'].endswith("/v2/notifications?template_type=email")
assert 'next' in json_response['links'].keys()
assert len(json_response['notifications']) == 1
assert json_response['notifications'][0]['id'] == str(notification.id)
assert json_response['notifications'][0]['status'] == "created"
assert json_response['notifications'][0]['template'] == {
'id': str(email_template.id),
'uri': email_template.get_link(),
'version': 1
}
assert json_response['notifications'][0]['email_address'] == "don.draper@scdp.biz"
assert json_response['notifications'][0]['type'] == "email"
def test_get_all_notifications_filter_by_template_type_invalid_template_type(client, sample_notification):
auth_header = create_authorization_header(service_id=sample_notification.service_id)
response = client.get(
path='/v2/notifications?template_type=orange',
headers=[('Content-Type', 'application/json'), auth_header])
json_response = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert response.headers['Content-type'] == "application/json"
assert json_response['status_code'] == 400
assert len(json_response['errors']) == 1
assert json_response['errors'][0]['message'] == "orange is not one of [sms, email, letter]"
def test_get_all_notifications_filter_by_single_status(client, notify_db, notify_db_session):
notification = create_sample_notification(notify_db, notify_db_session, status="pending")
create_sample_notification(notify_db, notify_db_session)
auth_header = create_authorization_header(service_id=notification.service_id)
response = client.get(
path='/v2/notifications?status=pending',
headers=[('Content-Type', 'application/json'), auth_header])
json_response = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert response.headers['Content-type'] == "application/json"
assert json_response['links']['current'].endswith("/v2/notifications?status=pending")
assert 'next' in json_response['links'].keys()
assert len(json_response['notifications']) == 1
assert json_response['notifications'][0]['id'] == str(notification.id)
assert json_response['notifications'][0]['status'] == "pending"
def test_get_all_notifications_filter_by_status_invalid_status(client, sample_notification):
auth_header = create_authorization_header(service_id=sample_notification.service_id)
response = client.get(
path='/v2/notifications?status=elephant',
headers=[('Content-Type', 'application/json'), auth_header])
json_response = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert response.headers['Content-type'] == "application/json"
assert json_response['status_code'] == 400
assert len(json_response['errors']) == 1
assert json_response['errors'][0]['message'] == "elephant is not one of [created, sending, delivered, " \
"pending, failed, technical-failure, temporary-failure, permanent-failure]"
def test_get_all_notifications_filter_by_multiple_statuses(client, notify_db, notify_db_session):
notifications = [
create_sample_notification(notify_db, notify_db_session, status=_status)
for _status in ["created", "pending", "sending"]
]
failed_notification = create_sample_notification(notify_db, notify_db_session, status="permanent-failure")
auth_header = create_authorization_header(service_id=notifications[0].service_id)
response = client.get(
path='/v2/notifications?status=created&status=pending&status=sending',
headers=[('Content-Type', 'application/json'), auth_header])
json_response = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert response.headers['Content-type'] == "application/json"
assert json_response['links']['current'].endswith("/v2/notifications?status=created&status=pending&status=sending")
assert 'next' in json_response['links'].keys()
assert len(json_response['notifications']) == 3
returned_notification_ids = [_n['id'] for _n in json_response['notifications']]
for _id in [_notification.id for _notification in notifications]:
assert str(_id) in returned_notification_ids
assert failed_notification.id not in returned_notification_ids
def test_get_all_notifications_filter_by_failed_status(client, notify_db, notify_db_session):
created_notification = create_sample_notification(notify_db, notify_db_session, status="created")
failed_notifications = [
create_sample_notification(notify_db, notify_db_session, status=_status)
for _status in ["technical-failure", "temporary-failure", "permanent-failure"]
]
auth_header = create_authorization_header(service_id=created_notification.service_id)
response = client.get(
path='/v2/notifications?status=failed',
headers=[('Content-Type', 'application/json'), auth_header])
json_response = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert response.headers['Content-type'] == "application/json"
assert json_response['links']['current'].endswith("/v2/notifications?status=failed")
assert 'next' in json_response['links'].keys()
assert len(json_response['notifications']) == 3
returned_notification_ids = [n['id'] for n in json_response['notifications']]
for _id in [_notification.id for _notification in failed_notifications]:
assert str(_id) in returned_notification_ids
assert created_notification.id not in returned_notification_ids
def test_get_all_notifications_filter_by_id(client, notify_db, notify_db_session):
older_notification = create_sample_notification(notify_db, notify_db_session)
newer_notification = create_sample_notification(notify_db, notify_db_session)
auth_header = create_authorization_header(service_id=newer_notification.service_id)
response = client.get(
path='/v2/notifications?older_than={}'.format(newer_notification.id),
headers=[('Content-Type', 'application/json'), auth_header])
json_response = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert response.headers['Content-type'] == "application/json"
assert json_response['links']['current'].endswith("/v2/notifications?older_than={}".format(newer_notification.id))
assert 'next' in json_response['links'].keys()
assert len(json_response['notifications']) == 1
assert json_response['notifications'][0]['id'] == str(older_notification.id)
def test_get_all_notifications_filter_by_id_invalid_id(client, sample_notification):
auth_header = create_authorization_header(service_id=sample_notification.service_id)
response = client.get(
path='/v2/notifications?older_than=1234-badly-formatted-id-7890',
headers=[('Content-Type', 'application/json'), auth_header])
json_response = json.loads(response.get_data(as_text=True))
assert json_response['status_code'] == 400
assert len(json_response['errors']) == 1
assert json_response['errors'][0]['message'] == "older_than is not a valid UUID"
def test_get_all_notifications_filter_by_id_no_notifications_if_nonexistent_id(client, notify_db, notify_db_session):
notification = create_sample_notification(notify_db, notify_db_session)
auth_header = create_authorization_header(service_id=notification.service_id)
response = client.get(
path='/v2/notifications?older_than=dd4b8b9d-d414-4a83-9256-580046bf18f9',
headers=[('Content-Type', 'application/json'), auth_header])
json_response = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert response.headers['Content-type'] == "application/json"
assert json_response['links']['current'].endswith(
"/v2/notifications?older_than=dd4b8b9d-d414-4a83-9256-580046bf18f9")
assert 'next' not in json_response['links'].keys()
assert len(json_response['notifications']) == 0
def test_get_all_notifications_filter_by_id_no_notifications_if_last_notification(client, notify_db, notify_db_session):
notification = create_sample_notification(notify_db, notify_db_session)
auth_header = create_authorization_header(service_id=notification.service_id)
response = client.get(
path='/v2/notifications?older_than={}'.format(notification.id),
headers=[('Content-Type', 'application/json'), auth_header])
json_response = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert response.headers['Content-type'] == "application/json"
assert json_response['links']['current'].endswith("/v2/notifications?older_than={}".format(notification.id))
assert 'next' not in json_response['links'].keys()
assert len(json_response['notifications']) == 0
def test_get_all_notifications_filter_multiple_query_parameters(client, notify_db, notify_db_session):
email_template = create_sample_template(notify_db, notify_db_session, template_type="email")
# this is the notification we are looking for
older_notification = create_sample_notification(
notify_db, notify_db_session, template=email_template, status="pending")
# wrong status
create_sample_notification(notify_db, notify_db_session, template=email_template)
# wrong template
create_sample_notification(notify_db, notify_db_session, status="pending")
# we only want notifications created before this one
newer_notification = create_sample_notification(notify_db, notify_db_session)
# this notification was created too recently
create_sample_notification(notify_db, notify_db_session, template=email_template, status="pending")
auth_header = create_authorization_header(service_id=newer_notification.service_id)
response = client.get(
path='/v2/notifications?status=pending&template_type=email&older_than={}'.format(newer_notification.id),
headers=[('Content-Type', 'application/json'), auth_header])
json_response = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert response.headers['Content-type'] == "application/json"
# query parameters aren't returned in order
for url_part in [
"/v2/notifications?",
"template_type=email",
"status=pending",
"older_than={}".format(newer_notification.id)
]:
assert url_part in json_response['links']['current']
assert 'next' in json_response['links'].keys()
assert len(json_response['notifications']) == 1
assert json_response['notifications'][0]['id'] == str(older_notification.id)

70
tests/app/v2/notifications/test_notification_schemas.py
View File

@@ -5,10 +5,80 @@ from flask import json
from jsonschema import ValidationError
from app.v2.notifications.notification_schemas import (
get_notifications_request,
post_sms_request, post_sms_response, post_email_request, post_email_response
)
from app.schema_validation import validate
@pytest.mark.parametrize('invalid_statuses, valid_statuses', [
# one invalid status
(["elephant"], []),
# multiple invalid statuses
(["elephant", "giraffe", "cheetah"], []),
# one bad status and one good status
(["elephant"], ["created"]),
])
def test_get_notifications_request_invalid_statuses(
invalid_statuses, valid_statuses
):
partial_error_status = "is not one of " \
"[created, sending, delivered, pending, failed, technical-failure, temporary-failure, permanent-failure]"
with pytest.raises(ValidationError) as e:
validate({'status': invalid_statuses + valid_statuses}, get_notifications_request)
errors = json.loads(e.value.message).get('errors')
assert len(errors) == len(invalid_statuses)
for index, value in enumerate(invalid_statuses):
assert errors[index]['message'] == "{} {}".format(value, partial_error_status)
@pytest.mark.parametrize('invalid_template_types, valid_template_types', [
# one invalid template_type
(["orange"], []),
# multiple invalid template_types
(["orange", "avocado", "banana"], []),
# one bad template_type and one good template_type
(["orange"], ["sms"]),
])
def test_get_notifications_request_invalid_template_types(
invalid_template_types, valid_template_types
):
partial_error_template_type = "is not one of [sms, email, letter]"
with pytest.raises(ValidationError) as e:
validate({'template_type': invalid_template_types + valid_template_types}, get_notifications_request)
errors = json.loads(e.value.message).get('errors')
assert len(errors) == len(invalid_template_types)
for index, value in enumerate(invalid_template_types):
assert errors[index]['message'] == "{} {}".format(value, partial_error_template_type)
def test_get_notifications_request_invalid_statuses_and_template_types():
with pytest.raises(ValidationError) as e:
validate({
'status': ["created", "elephant", "giraffe"],
'template_type': ["sms", "orange", "avocado"]
}, get_notifications_request)
errors = json.loads(e.value.message).get('errors')
assert len(errors) == 4
error_messages = [error['message'] for error in errors]
for invalid_status in ["elephant", "giraffe"]:
assert "{} is not one of [created, sending, delivered, " \
"pending, failed, technical-failure, temporary-failure, permanent-failure]".format(
invalid_status
) in error_messages
for invalid_template_type in ["orange", "avocado"]:
assert "{} is not one of [sms, email, letter]".format(invalid_template_type) in error_messages
valid_json = {"phone_number": "07515111111",
"template_id": str(uuid.uuid4())
}

39
tests/app/v2/notifications/test_post_notifications.py
View File

@@ -7,23 +7,23 @@ from tests import create_authorization_header
@pytest.mark.parametrize("reference", [None, "reference_from_client"])
def test_post_sms_notification_returns_201(notify_api, sample_template, mocker, reference):
def test_post_sms_notification_returns_201(notify_api, sample_template_with_placeholders, mocker, reference):
with notify_api.test_request_context():
with notify_api.test_client() as client:
mocked = mocker.patch('app.celery.provider_tasks.deliver_sms.apply_async')
data = {
'phone_number': '+447700900855',
'template_id': str(sample_template.id)
'template_id': str(sample_template_with_placeholders.id),
'personalisation': {' Name': 'Jo'}
}
if reference:
data.update({"reference": reference})
auth_header = create_authorization_header(service_id=sample_template.service_id)
auth_header = create_authorization_header(service_id=sample_template_with_placeholders.service_id)
response = client.post(
path='/v2/notifications/sms',
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), auth_header])
assert response.status_code == 201
resp_json = json.loads(response.get_data(as_text=True))
notifications = Notification.query.all()
@@ -31,12 +31,12 @@ def test_post_sms_notification_returns_201(notify_api, sample_template, mocker,
notification_id = notifications[0].id
assert resp_json['id'] == str(notification_id)
assert resp_json['reference'] == reference
assert resp_json['content']['body'] == sample_template.content
assert resp_json['content']['from_number'] == sample_template.service.sms_sender
assert resp_json['content']['body'] == sample_template_with_placeholders.content.replace("(( Name))", "Jo")
assert resp_json['content']['from_number'] == sample_template_with_placeholders.service.sms_sender
assert 'v2/notifications/{}'.format(notification_id) in resp_json['uri']
assert resp_json['template']['id'] == str(sample_template.id)
assert resp_json['template']['version'] == sample_template.version
assert 'v2/templates/{}'.format(sample_template.id) in resp_json['template']['uri']
assert resp_json['template']['id'] == str(sample_template_with_placeholders.id)
assert resp_json['template']['version'] == sample_template_with_placeholders.version
assert 'v2/templates/{}'.format(sample_template_with_placeholders.id) in resp_json['template']['uri']
assert mocked.called
@@ -108,15 +108,16 @@ def test_post_sms_notification_returns_400_and_for_schema_problems(notify_api, s
@pytest.mark.parametrize("reference", [None, "reference_from_client"])
def test_post_email_notification_returns_201(client, sample_email_template, mocker, reference):
def test_post_email_notification_returns_201(client, sample_email_template_with_placeholders, mocker, reference):
mocked = mocker.patch('app.celery.provider_tasks.deliver_email.apply_async')
data = {
"email_address": sample_email_template.service.users[0].email_address,
"template_id": sample_email_template.id,
"email_address": sample_email_template_with_placeholders.service.users[0].email_address,
"template_id": sample_email_template_with_placeholders.id,
"personalisation": {"name": "Bob"}
}
if reference:
data.update({"reference": reference})
auth_header = create_authorization_header(service_id=sample_email_template.service_id)
auth_header = create_authorization_header(service_id=sample_email_template_with_placeholders.service_id)
response = client.post(
path="v2/notifications/email",
data=json.dumps(data),
@@ -127,13 +128,13 @@ def test_post_email_notification_returns_201(client, sample_email_template, mock
assert resp_json['id'] == str(notification.id)
assert resp_json['reference'] == reference
assert notification.reference is None
assert resp_json['content']['body'] == sample_email_template.content
assert resp_json['content']['subject'] == sample_email_template.subject
assert resp_json['content']['from_email'] == sample_email_template.service.email_from
assert resp_json['content']['body'] == sample_email_template_with_placeholders.content.replace('((name))', 'Bob')
assert resp_json['content']['subject'] == sample_email_template_with_placeholders.subject
assert resp_json['content']['from_email'] == sample_email_template_with_placeholders.service.email_from
assert 'v2/notifications/{}'.format(notification.id) in resp_json['uri']
assert resp_json['template']['id'] == str(sample_email_template.id)
assert resp_json['template']['version'] == sample_email_template.version
assert 'v2/templates/{}'.format(sample_email_template.id) in resp_json['template']['uri']
assert resp_json['template']['id'] == str(sample_email_template_with_placeholders.id)
assert resp_json['template']['version'] == sample_email_template_with_placeholders.version
assert 'v2/templates/{}'.format(sample_email_template_with_placeholders.id) in resp_json['template']['uri']
assert mocked.called