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

Added page_size parameter for notifications api. All tests passing.

Browse Source 
Add page_size and total parameters to all calls for notifications.
This commit is contained in:
Nicholas Staples
2016-04-19 10:52:52 +01:00
parent 58ab831df9
commit 31978bd987
4 changed files with 74 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
app/dao/notifications_dao.py
View File

@@ -198,12 +198,14 @@ 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_for_job(service_id, job_id, filter_dict=None, page=1):
def get_notifications_for_job(service_id, job_id, filter_dict=None, page=1, page_size=None):
if page_size is None:
page_size = current_app.config['PAGE_SIZE']
query = Notification.query.filter_by(service_id=service_id, job_id=job_id)
query = filter_query(query, filter_dict)
pagination = query.order_by(desc(Notification.created_at)).paginate(
page=page,
per_page=current_app.config['PAGE_SIZE']
per_page=page_size
)
return pagination
@@ -216,12 +218,14 @@ def get_notification_by_id(notification_id):
return Notification.query.filter_by(id=notification_id).first()
def get_notifications_for_service(service_id, filter_dict=None, page=1):
def get_notifications_for_service(service_id, filter_dict=None, page=1, page_size=None):
if page_size is None:
page_size = current_app.config['PAGE_SIZE']
query = Notification.query.filter_by(service_id=service_id)
query = filter_query(query, filter_dict)
pagination = query.order_by(desc(Notification.created_at)).paginate(
page=page,
per_page=current_app.config['PAGE_SIZE']
per_page=page_size
)
return pagination

19
app/notifications/rest.py
View File

@@ -190,14 +190,17 @@ def get_all_notifications():
return jsonify(result="error", message=errors), 400
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')
pagination = notifications_dao.get_notifications_for_service(
api_user['client'],
filter_dict=data,
page=page)
page=page,
page_size=page_size)
return jsonify(
notifications=notification_status_schema.dump(pagination.items, many=True).data,
page_size=page_size,
total=pagination.total,
links=pagination_links(
pagination,
'.get_all_notifications',
@@ -214,15 +217,19 @@ def get_all_notifications_for_service(service_id):
return jsonify(result="error", message=errors), 400
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')
pagination = notifications_dao.get_notifications_for_service(
service_id,
filter_dict=data,
page=page)
page=page,
page_size=page_size)
kwargs = request.args.to_dict()
kwargs['service_id'] = service_id
return jsonify(
notifications=notification_status_schema.dump(pagination.items, many=True).data,
page_size=page_size,
total=pagination.total,
links=pagination_links(
pagination,
'.get_all_notifications_for_service',
@@ -239,17 +246,21 @@ def get_all_notifications_for_service_job(service_id, job_id):
return jsonify(result="error", message=errors), 400
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')
pagination = notifications_dao.get_notifications_for_job(
service_id,
job_id,
filter_dict=data,
page=page)
page=page,
page_size=page_size)
kwargs = request.args.to_dict()
kwargs['service_id'] = service_id
kwargs['job_id'] = job_id
return jsonify(
notifications=notification_status_schema.dump(pagination.items, many=True).data,
page_size=page_size,
total=pagination.total,
links=pagination_links(
pagination,
'.get_all_notifications_for_service_job',

12
app/schemas.py
View File

@@ -234,6 +234,7 @@ class NotificationsFilterSchema(ma.Schema):
template_type = fields.Nested(BaseTemplateSchema, only=['template_type'], many=True)
status = fields.Nested(NotificationModelSchema, only=['status'], many=True)
page = fields.Int(required=False)
page_size = fields.Int(required=False)
@pre_load
def handle_multidict(self, in_data):
@@ -254,8 +255,7 @@ class NotificationsFilterSchema(ma.Schema):
in_data['status'] = [x.status for x in in_data['status']]
return in_data
@validates('page')
def validate_page(self, value):
def _validate_positive_number(self, value):
try:
page_int = int(value)
if page_int < 1:
@@ -263,6 +263,14 @@ class NotificationsFilterSchema(ma.Schema):
except:
raise ValidationError("Not a positive integer")
@validates('page')
def validate_page(self, value):
self._validate_positive_number(value)
@validates('page_size')
def validate_page_size(self, value):
self._validate_positive_number(value)
class TemplateStatisticsSchema(BaseSchema):

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

@@ -227,6 +227,47 @@ def test_should_reject_invalid_page_param(notify_api, sample_email_template):
assert 'Not a valid integer.' in notifications['message']['page']
def test_valid_page_size_param(notify_api, notify_db, notify_db_session, sample_email_template):
with notify_api.test_request_context():
n1 = create_sample_notification(notify_db, notify_db_session)
n2 = create_sample_notification(notify_db, notify_db_session)
with notify_api.test_client() as client:
auth_header = create_authorization_header(
service_id=sample_email_template.service_id,
path='/notifications',
method='GET')
response = client.get(
'/notifications?page=1&page_size=1',
headers=[auth_header])
notifications = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert len(notifications['notifications']) == 1
assert notifications['total'] == 2
assert notifications['page_size'] == 1
def test_invalid_page_size_param(notify_api, notify_db, notify_db_session, sample_email_template):
with notify_api.test_request_context():
n1 = create_sample_notification(notify_db, notify_db_session)
n2 = create_sample_notification(notify_db, notify_db_session)
with notify_api.test_client() as client:
auth_header = create_authorization_header(
service_id=sample_email_template.service_id,
path='/notifications',
method='GET')
response = client.get(
'/notifications?page=1&page_size=invalid',
headers=[auth_header])
notifications = json.loads(response.get_data(as_text=True))
assert response.status_code == 400
assert notifications['result'] == 'error'
assert 'Not a valid integer.' in notifications['message']['page_size']
def test_should_return_pagination_links(notify_api, notify_db, notify_db_session, sample_email_template):
with notify_api.test_request_context():
with notify_api.test_client() as client:
@@ -1274,7 +1315,6 @@ def test_firetext_callback_should_update_notification_status_sent(notify_api, no
headers=[('Content-Type', 'application/x-www-form-urlencoded')])
json_resp = json.loads(response.get_data(as_text=True))
print(json_resp)
assert response.status_code == 200
assert json_resp['result'] == 'success'
assert json_resp['message'] == 'Firetext callback succeeded. reference {} updated'.format(