Add DVLA callback:

* Process SNS callback, trigger the update notifications celery task
* Put autoconfirm into its own method and use in callbacks

app/notifications/notifications_letter_callback.py

@@ -1,39 +1,57 @@
-from datetime import datetime
+from functools import wraps
 
 from flask import (
     Blueprint,
     jsonify,
     request,
-    current_app,
+    current_app
-    json
 )
 
-from app import statsd_client
+from app.celery.tasks import update_letter_notifications_statuses
-from app.clients.email.aws_ses import get_aws_responses
+from app.v2.errors import register_errors
-from app.dao import (
+from app.notifications.utils import autoconfirm_subscription
-    notifications_dao
+from app.schema_validation import validate
-)
 
-from app.notifications.process_client_response import validate_callback_data
 
 letter_callback_blueprint = Blueprint('notifications_letter_callback', __name__)
-
-from app.errors import (
-    register_errors,
-    InvalidRequest
-)
-
 register_errors(letter_callback_blueprint)
 
 
+dvla_sns_callback_schema = {
+    "$schema": "http://json-schema.org/draft-04/schema#",
+    "description": "sns callback received on s3 update",
+    "type": "object",
+    "title": "dvla internal sns callback",
+    "properties": {
+        "Type": {"enum": ["Notification", "SubscriptionConfirmation"]},
+        "MessageId": {"type": "string"},
+        "Message": {"type": ["string", "object"]}
+    },
+    "required": ["Type", "MessageId", "Message"]
+}
+
+
+def validate_schema(schema):
+    def decorator(f):
+        @wraps(f)
+        def wrapper(*args, **kw):
+            validate(request.json, schema)
+            return f(*args, **kw)
+        return wrapper
+    return decorator
+
+
 @letter_callback_blueprint.route('/notifications/letter/dvla', methods=['POST'])
+@validate_schema(dvla_sns_callback_schema)
 def process_letter_response():
-    try:
+    req_json = request.json
-        dvla_request = json.loads(request.data)
+    if not autoconfirm_subscription(req_json):
-        current_app.logger.info(dvla_request)
+        # The callback should have one record for an S3 Put Event.
-        return jsonify(
+        filename = req_json['Message']['Records'][0]['s3']['object']['key']
-            result="success", message="DVLA callback succeeded"
+        current_app.logger.info('Received file from DVLA: {}'.format(filename))
-        ), 200
+        current_app.logger.info('DVLA callback: Calling task to update letter notifications')
-    except ValueError:
+        update_letter_notifications_statuses.apply_async([filename], queue='notify')
-        error = "DVLA callback failed: invalid json"
+
-        raise InvalidRequest(error, status_code=400)
+    return jsonify(
+        result="success", message="DVLA callback succeeded"
+    ), 200

app/notifications/notifications_ses_callback.py

@@ -13,9 +13,8 @@ from app.clients.email.aws_ses import get_aws_responses
 from app.dao import (
     notifications_dao
 )
-
 from app.notifications.process_client_response import validate_callback_data
-from app.notifications.utils import confirm_subscription
+from app.notifications.utils import autoconfirm_subscription
 
 ses_callback_blueprint = Blueprint('notifications_ses_callback', __name__)
 
@@ -32,14 +31,12 @@ def process_ses_response():
     try:
         ses_request = json.loads(request.data)
 
-        if ses_request.get('Type') == 'SubscriptionConfirmation':
+        subscribed_topic = autoconfirm_subscription(ses_request)
-            current_app.logger.info("SNS subscription confirmation url: {}".format(ses_request['SubscribeURL']))
+        if subscribed_topic:
-            subscribed_topic = confirm_subscription(ses_request)
+            current_app.logger.info("Automatically subscribed to topic: {}".format(subscribed_topic))
-            if subscribed_topic:
+            return jsonify(
-                current_app.logger.info("Automatically subscribed to topic: {}".format(subscribed_topic))
+                result="success", message="SES callback succeeded"
-                return jsonify(
+            ), 200
-                    result="success", message="SES callback succeeded"
-                ), 200
 
         errors = validate_callback_data(data=ses_request, fields=['Message'], client_name=client_name)
         if errors:

app/notifications/utils.py

@@ -16,3 +16,10 @@ def confirm_subscription(confirmation_request):
         raise e
 
     return confirmation_request['TopicArn']
+
+
+def autoconfirm_subscription(req_json):
+    if req_json.get('Type') == 'SubscriptionConfirmation':
+        current_app.logger.info("SNS subscription confirmation url: {}".format(req_json['SubscribeURL']))
+        subscribed_topic = confirm_subscription(req_json)
+        return subscribed_topic

tests/app/celery/test_tasks.py

@@ -12,15 +12,19 @@ from celery.exceptions import Retry
 from app import (encryption, DATETIME_FORMAT)
 from app.celery import provider_tasks
 from app.celery import tasks
-from app.celery.tasks import s3, build_dvla_file, create_dvla_file_contents, update_dvla_job_to_error
 from app.celery.tasks import (
+    s3,
+    build_dvla_file,
+    create_dvla_file_contents,
+    update_dvla_job_to_error,
     process_job,
     process_row,
     send_sms,
     send_email,
     persist_letter,
     get_template_class,
-    update_job_to_sent_to_dvla
+    update_job_to_sent_to_dvla,
+    update_letter_notifications_statuses
 )
 from app.dao import jobs_dao, services_dao
 from app.models import (
@@ -34,6 +38,7 @@ from app.models import (
     Job)
 
 from tests.app import load_example_csv
+from tests.conftest import set_config
 from tests.app.conftest import (
     sample_service,
     sample_template,

tests/app/notifications/rest/test_callbacks.py

@@ -10,17 +10,65 @@ from app.dao.notifications_dao import (
     get_notification_by_id
 )
 from app.models import NotificationStatistics
+from tests.app.notifications.test_notifications_ses_callback import ses_confirmation_callback
 from tests.app.conftest import sample_notification as create_sample_notification
 
 
-def test_dvla_callback_should_not_need_auth(client):
+def test_dvla_callback_returns_400_with_invalid_request(client):
-    data = json.dumps({"somekey": "somevalue"})
+    data = json.dumps({"foo": "bar"})
     response = client.post(
         path='/notifications/letter/dvla',
         data=data,
-        headers=[('Content-Type', 'application/json')])
+        headers=[('Content-Type', 'application/json')]
+    )
+
+    assert response.status_code == 400
+
+
+def test_dvla_callback_autoconfirms_subscription(client, mocker):
+    autoconfirm_mock = mocker.patch('app.notifications.notifications_letter_callback.autoconfirm_subscription')
+
+    data = ses_confirmation_callback()
+    response = client.post(
+        path='/notifications/letter/dvla',
+        data=data,
+        headers=[('Content-Type', 'application/json')]
+    )
 
     assert response.status_code == 200
+    assert autoconfirm_mock.called
+
+
+def test_dvla_callback_autoconfirm_does_not_call_update_letter_notifications_task(client, mocker):
+    autoconfirm_mock = mocker.patch('app.notifications.notifications_letter_callback.autoconfirm_subscription')
+    update_task = \
+        mocker.patch('app.notifications.notifications_letter_callback.update_letter_notifications_statuses.apply_async')
+
+    data = ses_confirmation_callback()
+    response = client.post(
+        path='/notifications/letter/dvla',
+        data=data,
+        headers=[('Content-Type', 'application/json')]
+    )
+
+    assert response.status_code == 200
+    assert autoconfirm_mock.called
+    assert not update_task.called
+
+
+def test_dvla_callback_calls_update_letter_notifications_task(client, mocker):
+    update_task = \
+        mocker.patch('app.notifications.notifications_letter_callback.update_letter_notifications_statuses.apply_async')
+    data = _sample_sns_s3_callback()
+    response = client.post(
+        path='/notifications/letter/dvla',
+        data=data,
+        headers=[('Content-Type', 'application/json')]
+    )
+
+    assert response.status_code == 200
+    assert update_task.called
+    update_task.assert_called_with(['bar.txt'], queue='notify')
 
 
 def test_firetext_callback_should_not_need_auth(client, mocker):
@@ -458,3 +506,46 @@ def test_firetext_callback_should_record_statsd(client, notify_db, notify_db_ses
 
 def get_notification_stats(service_id):
     return NotificationStatistics.query.filter_by(service_id=service_id).one()
+
+
+def _sample_sns_s3_callback():
+    return json.dumps({
+        "SigningCertURL": "foo.pem",
+        "UnsubscribeURL": "bar",
+        "Signature": "some-signature",
+        "Type": "Notification",
+        "Timestamp": "2016-05-03T08:35:12.884Z",
+        "SignatureVersion": "1",
+        "MessageId": "6adbfe0a-d610-509a-9c47-af894e90d32d",
+        "Subject": "Amazon S3 Notification",
+        "TopicArn": "sample-topic-arn",
+        "Message": {
+            "Records": [{
+                "eventVersion": "2.0",
+                "eventSource": "aws:s3",
+                "awsRegion": "eu-west-1",
+                "eventTime": "2017-05-03T08:35:12.826Z",
+                "eventName": "ObjectCreated:Put",
+                "userIdentity": {"principalId": "some-p-id"},
+                "requestParameters": {"sourceIPAddress": "8.8.8.8"},
+                "responseElements": {"x-amz-request-id": "some-req-id", "x-amz-id-2": "some-amz-id"},
+                "s3": {
+                    "s3SchemaVersion": "1.0",
+                    "configurationId": "some-config-id",
+                    "bucket": {
+                        "name": "some-bucket",
+                        "ownerIdentity": {"principalId": "some-p-id"},
+                        "arn": "some-bucket-arn"
+                    },
+                    "object": {
+                        "key": "bar.txt",
+                        "size": 200,
+                        "eTag": "some-etag",
+                        "versionId": "some-v-id",
+                        "sequencer": "some-seq"
+                    }
+                }
+            }
+            ]
+        }
+    })