Merge pull request #687 from alphagov/optionally-get-test-notifications

Optionally get notifications created with a test key

app/dao/notifications_dao.py

@@ -228,14 +228,17 @@ def get_notifications(filter_dict=None):
 
 
 @statsd(namespace="dao")
-def get_notifications_for_service(service_id,
-                                  filter_dict=None,
-                                  page=1,
-                                  page_size=None,
-                                  limit_days=None,
-                                  key_type=None,
-                                  personalisation=False,
-                                  include_jobs=False):
+def get_notifications_for_service(
+    service_id,
+    filter_dict=None,
+    page=1,
+    page_size=None,
+    limit_days=None,
+    key_type=None,
+    personalisation=False,
+    include_jobs=False,
+    include_from_test_key=False
+):
     if page_size is None:
         page_size = current_app.config['PAGE_SIZE']
 
@@ -250,7 +253,7 @@ def get_notifications_for_service(service_id,
 
     if key_type is not None:
         filters.append(Notification.key_type == key_type)
-    else:
+    elif not include_from_test_key:
         filters.append(Notification.key_type != KEY_TYPE_TEST)
 
     query = Notification.query.filter(*filters)

app/schemas.py

@@ -422,6 +422,7 @@ class NotificationsFilterSchema(ma.Schema):
     page_size = fields.Int(required=False)
     limit_days = fields.Int(required=False)
     include_jobs = fields.Boolean(required=False)
+    include_from_test_key = fields.Boolean(required=False)
 
     @pre_load
     def handle_multidict(self, in_data):

app/service/rest.py

@@ -214,6 +214,7 @@ def get_all_notifications_for_service(service_id):
     page_size = data['page_size'] if 'page_size' in data else current_app.config.get('PAGE_SIZE')
     limit_days = data.get('limit_days')
     include_jobs = data.get('include_jobs', True)
+    include_from_test_key = data.get('include_from_test_key', False)
 
     pagination = notifications_dao.get_notifications_for_service(
         service_id,
@@ -221,7 +222,9 @@ def get_all_notifications_for_service(service_id):
         page=page,
         page_size=page_size,
         limit_days=limit_days,
-        include_jobs=include_jobs)
+        include_jobs=include_jobs,
+        include_from_test_key=include_from_test_key
+    )
     kwargs = request.args.to_dict()
     kwargs['service_id'] = service_id
     return jsonify(