diff --git a/app/dao/notifications_dao.py b/app/dao/notifications_dao.py index fedf8b46d..86cdda671 100644 --- a/app/dao/notifications_dao.py +++ b/app/dao/notifications_dao.py @@ -309,6 +309,7 @@ def _filter_query(query, filter_dict=None): # filter by status statuses = multidict.getlist('status') if statuses: + statuses = Notification.substitute_status(statuses) query = query.filter(Notification.status.in_(statuses)) # filter by template diff --git a/tests/app/dao/test_notification_dao.py b/tests/app/dao/test_notification_dao.py index ea7779ecb..fe768631f 100644 --- a/tests/app/dao/test_notification_dao.py +++ b/tests/app/dao/test_notification_dao.py @@ -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 diff --git a/tests/app/v2/notifications/test_get_notifications.py b/tests/app/v2/notifications/test_get_notifications.py index af785b6b5..423243e6c 100644 --- a/tests/app/v2/notifications/test_get_notifications.py +++ b/tests/app/v2/notifications/test_get_notifications.py @@ -184,7 +184,30 @@ def test_get_all_notifications_filter_by_multiple_statuses(client, notify_db, no def test_get_all_notifications_filter_by_failed_status(client, notify_db, notify_db_session): - pass + 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'] == "/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):