mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-02-05 02:42:26 -05:00
Merge pull request #379 from alphagov/filter_notifications_page
Working functionality for filtering notifications and all tests passing.
This commit is contained in:
@@ -9,6 +9,7 @@ from flask import (
|
||||
request
|
||||
)
|
||||
from flask_login import login_required
|
||||
from werkzeug.datastructures import MultiDict
|
||||
from utils.template import Template
|
||||
|
||||
from app import (job_api_client, notification_api_client, service_api_client)
|
||||
@@ -19,6 +20,17 @@ from app.utils import (
|
||||
user_has_permissions)
|
||||
|
||||
|
||||
def _parse_filter_args(filter_dict):
|
||||
if not isinstance(filter_dict, MultiDict):
|
||||
filter_dict = MultiDict(filter_dict)
|
||||
out_dict = MultiDict()
|
||||
if 'type' in filter_dict:
|
||||
out_dict.setlist('template_type', filter_dict.getlist('type'))
|
||||
if 'status' in filter_dict:
|
||||
out_dict.setlist('status', filter_dict.getlist('status'))
|
||||
return out_dict
|
||||
|
||||
|
||||
@main.route("/services/<service_id>/jobs")
|
||||
@login_required
|
||||
@user_has_permissions('view_activity', admin_override=True)
|
||||
@@ -96,17 +108,31 @@ def view_notifications(service_id):
|
||||
page = get_page_from_request()
|
||||
if page is None:
|
||||
abort(404, "Invalid page argument ({}) reverting to page 1.".format(request.args['page'], None))
|
||||
notifications = notification_api_client.get_notifications_for_service(service_id=service_id, page=page)
|
||||
filter_args = _parse_filter_args(request.args)
|
||||
notifications = notification_api_client.get_notifications_for_service(
|
||||
service_id=service_id,
|
||||
page=page,
|
||||
template_type=filter_args.getlist('template_type') if 'template_type' in filter_args else None,
|
||||
status=filter_args.getlist('status') if 'status' in filter_args else None)
|
||||
view_dict = MultiDict(request.args)
|
||||
prev_page = None
|
||||
if notifications['links'].get('prev', None):
|
||||
prev_page = generate_previous_next_dict(
|
||||
'main.view_notifications',
|
||||
{'service_id': service_id}, page - 1, 'Previous page', 'page {}'.format(page - 1))
|
||||
service_id,
|
||||
view_dict,
|
||||
page - 1,
|
||||
'Previous page',
|
||||
'page {}'.format(page - 1))
|
||||
next_page = None
|
||||
if notifications['links'].get('next', None):
|
||||
next_page = generate_previous_next_dict(
|
||||
'main.view_notifications',
|
||||
{'service_id': service_id}, page + 1, 'Next page', 'page {}'.format(page + 1))
|
||||
service_id,
|
||||
view_dict,
|
||||
page + 1,
|
||||
'Next page',
|
||||
'page {}'.format(page + 1))
|
||||
return render_template(
|
||||
'views/notifications.html',
|
||||
service_id=service_id,
|
||||
|
||||
@@ -21,10 +21,14 @@ class NotificationApiClient(BaseAPIClient):
|
||||
params=params
|
||||
)
|
||||
|
||||
def get_notifications_for_service(self, service_id, job_id=None, page=None):
|
||||
def get_notifications_for_service(self, service_id, job_id=None, template_type=None, status=None, page=None):
|
||||
params = {}
|
||||
if page is not None:
|
||||
params['page'] = page
|
||||
if template_type is not None:
|
||||
params['template_type'] = template_type
|
||||
if status is not None:
|
||||
params['status'] = status
|
||||
if job_id:
|
||||
return self.get(
|
||||
url='/service/{}/job/{}/notifications'.format(service_id, job_id),
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
</h2>
|
||||
{% if current_user.has_permissions(['view_activity'], admin_override=True) %}
|
||||
<ul>
|
||||
<li><a href="{{ url_for('.view_notifications', service_id=service_id) }}">History</a></li>
|
||||
<li><a href="{{ url_for('.view_notifications', service_id=service_id, page=1) }}">History</a></li>
|
||||
</ul>
|
||||
{% endif %}
|
||||
{% if current_user.has_permissions(['view_activity', 'manage_templates', 'manage_api_keys'], admin_override=True, any_=True) %}
|
||||
|
||||
@@ -8,7 +8,17 @@
|
||||
{% block maincolumn_content %}
|
||||
|
||||
<h1 class="heading-large">Notifications activity</h1>
|
||||
|
||||
<p>
|
||||
<a href="{{ url_for(".view_notifications", service_id=service_id, job_id=job_id, page=1) }}">All messages</a>
|
||||
</p>
|
||||
<p>
|
||||
<a href="{{ url_for(".view_notifications", service_id=service_id, job_id=job_id, type="sms", page=1) }}">Text messages</a> 
|
||||
<a href="{{ url_for(".view_notifications", service_id=service_id, job_id=job_id, type="email", page=1) }}">Email messages</a>
|
||||
</p>
|
||||
<p>
|
||||
<a href="{{ url_for(".view_notifications", service_id=service_id, job_id=job_id, status=['sent', 'delivered'], page=1) }}">Successful messages</a> 
|
||||
<a href="{{ url_for(".view_notifications", service_id=service_id, job_id=job_id, status=['failed', 'complaint', 'bounce'], page=1) }}">Failed messages</a>
|
||||
</p>
|
||||
{% call(item) list_table(
|
||||
jobs,
|
||||
caption="Recent activity",
|
||||
|
||||
@@ -9,7 +9,17 @@
|
||||
{% block maincolumn_content %}
|
||||
|
||||
<h1 class="heading-large">Notifications activity</h1>
|
||||
|
||||
<p>
|
||||
<a href="{{ url_for(".view_notifications", service_id=service_id, page=1) }}">All messages</a>
|
||||
</p>
|
||||
<p>
|
||||
<a href="{{ url_for(".view_notifications", service_id=service_id, type="sms", page=1) }}">Text messages</a> 
|
||||
<a href="{{ url_for(".view_notifications", service_id=service_id, type="email", page=1) }}">Email messages</a>
|
||||
</p>
|
||||
<p>
|
||||
<a href="{{ url_for(".view_notifications", service_id=service_id, status=['sent', 'delivered'], page=1) }}">Successful messages</a> 
|
||||
<a href="{{ url_for(".view_notifications", service_id=service_id, status=['failed', 'complaint', 'bounce'], page=1) }}">Failed messages</a>
|
||||
</p>
|
||||
{% call(item) list_table(
|
||||
notifications,
|
||||
caption="Recent activity",
|
||||
|
||||
@@ -95,9 +95,13 @@ def get_page_from_request():
|
||||
return 1
|
||||
|
||||
|
||||
def generate_previous_next_dict(view, view_dict, page, title, label):
|
||||
def generate_previous_next_dict(view, service_id, view_dict, page, title, label):
|
||||
if 'page' in view_dict:
|
||||
view_dict.pop('page')
|
||||
if 'service_id' in view_dict:
|
||||
view_dict.pop('service_id')
|
||||
return {
|
||||
'url': url_for(view, page=page, **view_dict),
|
||||
'url': url_for(view, service_id=service_id, page=page, **view_dict),
|
||||
'title': title,
|
||||
'label': label
|
||||
}
|
||||
|
||||
@@ -107,6 +107,58 @@ def test_should_show_notifications_for_a_service(app_,
|
||||
assert '.csv' in content
|
||||
|
||||
|
||||
def test_can_view_only_sms_notifications_for_a_service(app_,
|
||||
service_one,
|
||||
api_user_active,
|
||||
mock_login,
|
||||
mock_get_user,
|
||||
mock_get_user_by_email,
|
||||
mock_get_service,
|
||||
mock_get_notifications,
|
||||
mock_has_permissions):
|
||||
with app_.test_request_context():
|
||||
with app_.test_client() as client:
|
||||
client.login(api_user_active)
|
||||
response = client.get(url_for(
|
||||
'main.view_notifications',
|
||||
service_id=service_one['id'],
|
||||
type='sms'))
|
||||
assert response.status_code == 200
|
||||
content = response.get_data(as_text=True)
|
||||
notifications = mock_get_notifications(service_one['id'])
|
||||
notification = notifications['notifications'][0]
|
||||
assert notification['to'] in content
|
||||
assert notification['status'] in content
|
||||
assert notification['template']['name'] in content
|
||||
assert '.csv' in content
|
||||
|
||||
|
||||
def test_can_view_successful_notifications_for_a_service(app_,
|
||||
service_one,
|
||||
api_user_active,
|
||||
mock_login,
|
||||
mock_get_user,
|
||||
mock_get_user_by_email,
|
||||
mock_get_service,
|
||||
mock_get_notifications,
|
||||
mock_has_permissions):
|
||||
with app_.test_request_context():
|
||||
with app_.test_client() as client:
|
||||
client.login(api_user_active)
|
||||
response = client.get(url_for(
|
||||
'main.view_notifications',
|
||||
service_id=service_one['id'],
|
||||
status=['sent', 'delivered']))
|
||||
assert response.status_code == 200
|
||||
content = response.get_data(as_text=True)
|
||||
notifications = mock_get_notifications(service_one['id'])
|
||||
notification = notifications['notifications'][0]
|
||||
assert notification['to'] in content
|
||||
assert notification['status'] in content
|
||||
assert notification['template']['name'] in content
|
||||
assert '.csv' in content
|
||||
|
||||
|
||||
def test_should_show_notifications_for_a_service_with_next_previous(app_,
|
||||
service_one,
|
||||
api_user_active,
|
||||
|
||||
@@ -642,7 +642,11 @@ def mock_get_jobs(mocker):
|
||||
|
||||
@pytest.fixture(scope='function')
|
||||
def mock_get_notifications(mocker):
|
||||
def _get_notifications(service_id, job_id=None, page=1):
|
||||
def _get_notifications(service_id,
|
||||
job_id=None,
|
||||
page=1,
|
||||
template_type=None,
|
||||
status=None):
|
||||
return notification_json(service_id)
|
||||
return mocker.patch(
|
||||
'app.notification_api_client.get_notifications_for_service',
|
||||
@@ -652,7 +656,11 @@ def mock_get_notifications(mocker):
|
||||
|
||||
@pytest.fixture(scope='function')
|
||||
def mock_get_notifications_with_previous_next(mocker):
|
||||
def _get_notifications(service_id, job_id=None, page=1):
|
||||
def _get_notifications(service_id,
|
||||
job_id=None,
|
||||
page=1,
|
||||
template_type=None,
|
||||
status=None):
|
||||
return notification_json(service_id, with_links=True)
|
||||
return mocker.patch(
|
||||
'app.notification_api_client.get_notifications_for_service',
|
||||
|
||||
Reference in New Issue
Block a user