Working functionality for filtering notifications and all tests passing.

This commit is contained in:
Nicholas Staples
2016-04-04 16:34:06 +01:00
parent dc5c552f5f
commit a41dda8884
8 changed files with 125 additions and 11 deletions

View File

@@ -9,6 +9,7 @@ from flask import (
request request
) )
from flask_login import login_required from flask_login import login_required
from werkzeug.datastructures import MultiDict
from utils.template import Template from utils.template import Template
from app import (job_api_client, notification_api_client, service_api_client) from app import (job_api_client, notification_api_client, service_api_client)
@@ -19,6 +20,17 @@ from app.utils import (
user_has_permissions) 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") @main.route("/services/<service_id>/jobs")
@login_required @login_required
@user_has_permissions('view_activity', admin_override=True) @user_has_permissions('view_activity', admin_override=True)
@@ -96,17 +108,31 @@ def view_notifications(service_id):
page = get_page_from_request() page = get_page_from_request()
if page is None: if page is None:
abort(404, "Invalid page argument ({}) reverting to page 1.".format(request.args['page'], 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 prev_page = None
if notifications['links'].get('prev', None): if notifications['links'].get('prev', None):
prev_page = generate_previous_next_dict( prev_page = generate_previous_next_dict(
'main.view_notifications', '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 next_page = None
if notifications['links'].get('next', None): if notifications['links'].get('next', None):
next_page = generate_previous_next_dict( next_page = generate_previous_next_dict(
'main.view_notifications', '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( return render_template(
'views/notifications.html', 'views/notifications.html',
service_id=service_id, service_id=service_id,

View File

@@ -21,10 +21,14 @@ class NotificationApiClient(BaseAPIClient):
params=params 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 = {} params = {}
if page is not None: if page is not None:
params['page'] = page 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: if job_id:
return self.get( return self.get(
url='/service/{}/job/{}/notifications'.format(service_id, job_id), url='/service/{}/job/{}/notifications'.format(service_id, job_id),

View File

@@ -4,7 +4,7 @@
</h2> </h2>
{% if current_user.has_permissions(['view_activity'], admin_override=True) %} {% if current_user.has_permissions(['view_activity'], admin_override=True) %}
<ul> <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> </ul>
{% endif %} {% endif %}
{% if current_user.has_permissions(['view_activity', 'manage_templates', 'manage_api_keys'], admin_override=True, any_=True) %} {% if current_user.has_permissions(['view_activity', 'manage_templates', 'manage_api_keys'], admin_override=True, any_=True) %}

View File

@@ -8,7 +8,17 @@
{% block maincolumn_content %} {% block maincolumn_content %}
<h1 class="heading-large">Notifications activity</h1> <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>&emsp;
<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>&emsp;
<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( {% call(item) list_table(
jobs, jobs,
caption="Recent activity", caption="Recent activity",

View File

@@ -9,7 +9,17 @@
{% block maincolumn_content %} {% block maincolumn_content %}
<h1 class="heading-large">Notifications activity</h1> <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>&emsp;
<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>&emsp;
<a href="{{ url_for(".view_notifications", service_id=service_id, status=['failed', 'complaint', 'bounce'], page=1) }}">Failed messages</a>
</p>
{% call(item) list_table( {% call(item) list_table(
notifications, notifications,
caption="Recent activity", caption="Recent activity",

View File

@@ -95,9 +95,13 @@ def get_page_from_request():
return 1 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 { return {
'url': url_for(view, page=page, **view_dict), 'url': url_for(view, service_id=service_id, page=page, **view_dict),
'title': title, 'title': title,
'label': label 'label': label
} }

View File

@@ -107,6 +107,58 @@ def test_should_show_notifications_for_a_service(app_,
assert '.csv' in content 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_, def test_should_show_notifications_for_a_service_with_next_previous(app_,
service_one, service_one,
api_user_active, api_user_active,

View File

@@ -642,7 +642,11 @@ def mock_get_jobs(mocker):
@pytest.fixture(scope='function') @pytest.fixture(scope='function')
def mock_get_notifications(mocker): 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 notification_json(service_id)
return mocker.patch( return mocker.patch(
'app.notification_api_client.get_notifications_for_service', 'app.notification_api_client.get_notifications_for_service',
@@ -652,7 +656,11 @@ def mock_get_notifications(mocker):
@pytest.fixture(scope='function') @pytest.fixture(scope='function')
def mock_get_notifications_with_previous_next(mocker): 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 notification_json(service_id, with_links=True)
return mocker.patch( return mocker.patch(
'app.notification_api_client.get_notifications_for_service', 'app.notification_api_client.get_notifications_for_service',