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

Merge branch 'master' into dont-send-message-twice

Browse Source
This commit is contained in:
Rebecca Law
2016-11-28 16:43:17 +00:00
parent be113e031f 1b846f3e74
commit 3267a54ded
11 changed files with 448 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
app/dao/notifications_dao.py
View File

@@ -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(

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):

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

@@ -1,7 +1,8 @@
from flask import jsonify
from flask import jsonify, request, url_for
from app import api_user
from app.dao import notifications_dao
from app.schemas import notifications_filter_schema
from app.v2.notifications import notification_blueprint
@@ -14,9 +15,30 @@ 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 = notifications_filter_schema.load(request.args).data
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, **request.args.to_dict(flat=False)),
}
if len(notifications):
next_query_params = dict(request.args.to_dict(flat=False), 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

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

@@ -57,7 +57,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 +68,40 @@ 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_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",

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

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)

253
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,250 @@ def test_get_notification_by_id_returns_200(
}
assert json_response == expected_response
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_notificatons(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_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_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_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)