convert cloud events to background events (GoogleCloudPlatform#124)

* convert cloud events to background events

* review feedback

Author: matthewrobertson

.gitignore

@@ -5,3 +5,4 @@ __pycache__/
 build/
 dist/
 .coverage
+.vscode/

src/functions_framework/__init__.py

@@ -112,7 +112,11 @@ def view_func(path):
 
 def _event_view_func_wrapper(function, request):
     def view_func(path):
-        if is_binary(request.headers):
+        if event_conversion.is_convertable_cloudevent(request):
+            # Convert this CloudEvent to the equivalent background event data and context.
+            data, context = event_conversion.cloudevent_to_background_event(request)
+            function(data, context)
+        elif is_binary(request.headers):
             # Support CloudEvents in binary content mode, with data being the
             # whole request body and context attributes retrieved from request
             # headers.

src/functions_framework/event_conversion.py

@@ -14,9 +14,10 @@
 import re
 
 from datetime import datetime
-from typing import Optional, Tuple
+from typing import Any, Optional, Tuple
 
-from cloudevents.http import CloudEvent
+from cloudevents.exceptions import MissingRequiredFields
+from cloudevents.http import CloudEvent, from_http, is_binary
 
 from functions_framework.background_event import BackgroundEvent
 from functions_framework.exceptions import EventConversionException
@@ -48,6 +49,19 @@
     "providers/cloud.storage/eventTypes/object.change": "google.cloud.storage.object.v1.finalized",
 }
 
+# _BACKGROUND_TO_CE_TYPE contains duplicate values for some keys. This set contains the duplicates
+# that should be dropped when generating the inverse mapping _CE_TO_BACKGROUND_TYPE
+_NONINVERTALBE_CE_TYPES = {
+    "providers/cloud.pubsub/eventTypes/topic.publish",
+    "providers/cloud.storage/eventTypes/object.change",
+}
+
+# Maps CloudEvent types to the equivalent background/legacy event types (inverse
+# of _BACKGROUND_TO_CE_TYPE)
+_CE_TO_BACKGROUND_TYPE = {
+    v: k for k, v in _BACKGROUND_TO_CE_TYPE.items() if k not in _NONINVERTALBE_CE_TYPES
+}
+
 # CloudEvent service names.
 _FIREBASE_AUTH_CE_SERVICE = "firebaseauth.googleapis.com"
 _FIREBASE_CE_SERVICE = "firebase.googleapis.com"
@@ -93,6 +107,11 @@
     "createdAt": "createTime",
     "lastSignedInAt": "lastSignInTime",
 }
+# Maps Firebase Auth CloudEvent metadata field names to their equivalent
+# background event field names (inverse of _FIREBASE_AUTH_METADATA_FIELDS_BACKGROUND_TO_CE).
+_FIREBASE_AUTH_METADATA_FIELDS_CE_TO_BACKGROUND = {
+    v: k for k, v in _FIREBASE_AUTH_METADATA_FIELDS_BACKGROUND_TO_CE.items()
+}
 
 
 def background_event_to_cloudevent(request) -> CloudEvent:
@@ -143,6 +162,74 @@ def background_event_to_cloudevent(request) -> CloudEvent:
     return CloudEvent(metadata, data)
 
 
+def is_convertable_cloudevent(request) -> bool:
+    """Is the given request a known CloudEvent that can be converted to background event."""
+    if is_binary(request.headers):
+        event_type = request.headers.get("ce-type")
+        event_source = request.headers.get("ce-source")
+        return (
+            event_source is not None
+            and event_type is not None
+            and event_type in _CE_TO_BACKGROUND_TYPE
+        )
+    return False
+
+
+def _split_ce_source(source) -> Tuple[str, str]:
+    """Splits a CloudEvent source string into resource and subject components."""
+    regex = re.compile(r"\/\/([^/]+)\/(.+)")
+    match = regex.fullmatch(source)
+    if not match:
+        raise EventConversionException("Unexpected CloudEvent source.")
+
+    return match.group(1), match.group(2)
+
+
+def cloudevent_to_background_event(request) -> Tuple[Any, Context]:
+    """Converts a background event represented by the given HTTP request into a CloudEvent."""
+    try:
+        event = from_http(request.headers, request.get_data())
+        data = event.data
+        service, name = _split_ce_source(event["source"])
+
+        if event["type"] not in _CE_TO_BACKGROUND_TYPE:
+            raise EventConversionException(
+                f'Unable to find background event equivalent type for "{event["type"]}"'
+            )
+
+        if service == _PUBSUB_CE_SERVICE:
+            resource = {"service": service, "name": name, "type": _PUBSUB_MESSAGE_TYPE}
+            if "message" in data:
+                data = data["message"]
+        elif service == _FIREBASE_AUTH_CE_SERVICE:
+            resource = name
+            if "metadata" in data:
+                for old, new in _FIREBASE_AUTH_METADATA_FIELDS_CE_TO_BACKGROUND.items():
+                    if old in data["metadata"]:
+                        data["metadata"][new] = data["metadata"][old]
+                        del data["metadata"][old]
+        elif service == _STORAGE_CE_SERVICE:
+            resource = {
+                "name": f"{name}/{event['subject']}",
+                "service": service,
+                "type": data["kind"],
+            }
+        else:
+            resource = f"{name}/{event['subject']}"
+
+        context = Context(
+            eventId=event["id"],
+            timestamp=event["time"],
+            eventType=_CE_TO_BACKGROUND_TYPE[event["type"]],
+            resource=resource,
+        )
+        return (data, context)
+    except (AttributeError, KeyError, TypeError, MissingRequiredFields):
+        raise EventConversionException(
+            "Failed to convert CloudEvent to BackgroundEvent."
+        )
+
+
 def _split_resource(context: Context) -> Tuple[str, str, str]:
     """Splits a background event's resource into a CloudEvent service, resource, and subject."""
     service = ""

tests/test_convert.py

@@ -17,7 +17,7 @@
 import flask
 import pytest
 
-from cloudevents.http import from_json
+from cloudevents.http import from_json, to_binary
 
 from functions_framework import event_conversion
 from functions_framework.exceptions import EventConversionException
@@ -138,6 +138,18 @@ def firebase_auth_cloudevent_output():
         return from_json(f.read())
 
 
+@pytest.fixture
+def create_ce_headers():
+    return lambda event_type, source: {
+        "ce-id": "my-id",
+        "ce-type": event_type,
+        "ce-source": source,
+        "ce-specversion": "1.0",
+        "ce-subject": "my/subject",
+        "ce-time": "2020-08-16T13:58:54.471765",
+    }
+
+
 @pytest.mark.parametrize(
     "event", [PUBSUB_BACKGROUND_EVENT, PUBSUB_BACKGROUND_EVENT_WITHOUT_CONTEXT]
 )
@@ -329,3 +341,189 @@ def test_pubsub_emulator_request_with_invalid_message(
     with pytest.raises(EventConversionException) as exc_info:
         cloudevent = event_conversion.background_event_to_cloudevent(req)
     assert "Failed to convert Pub/Sub payload to event" in exc_info.value.args[0]
+
+
+@pytest.mark.parametrize(
+    "ce_event_type, ce_source, expected_type, expected_resource",
+    [
+        (
+            "google.firebase.database.document.v1.written",
+            "//firebasedatabase.googleapis.com/projects/_/instances/my-project-id",
+            "providers/google.firebase.database/eventTypes/ref.write",
+            "projects/_/instances/my-project-id/my/subject",
+        ),
+        (
+            "google.cloud.pubsub.topic.v1.messagePublished",
+            "//pubsub.googleapis.com/projects/sample-project/topics/gcf-test",
+            "google.pubsub.topic.publish",
+            {
+                "service": "pubsub.googleapis.com",
+                "name": "projects/sample-project/topics/gcf-test",
+                "type": "type.googleapis.com/google.pubsub.v1.PubsubMessage",
+            },
+        ),
+        (
+            "google.cloud.storage.object.v1.finalized",
+            "//storage.googleapis.com/projects/_/buckets/some-bucket",
+            "google.storage.object.finalize",
+            {
+                "service": "storage.googleapis.com",
+                "name": "projects/_/buckets/some-bucket/my/subject",
+                "type": "value",
+            },
+        ),
+        (
+            "google.firebase.auth.user.v1.created",
+            "//firebaseauth.googleapis.com/projects/my-project-id",
+            "providers/firebase.auth/eventTypes/user.create",
+            "projects/my-project-id",
+        ),
+    ],
+)
+def test_cloudevent_to_legacy_event(
+    create_ce_headers,
+    ce_event_type,
+    ce_source,
+    expected_type,
+    expected_resource,
+):
+    headers = create_ce_headers(ce_event_type, ce_source)
+    req = flask.Request.from_values(headers=headers, json={"kind": "value"})
+
+    (res_data, res_context) = event_conversion.cloudevent_to_background_event(req)
+
+    assert res_context.event_id == "my-id"
+    assert res_context.timestamp == "2020-08-16T13:58:54.471765"
+    assert res_context.event_type == expected_type
+    assert res_context.resource == expected_resource
+    assert res_data == {"kind": "value"}
+
+
+def test_cloudevent_to_legacy_event_with_pubsub_message_payload(
+    create_ce_headers,
+):
+    headers = create_ce_headers(
+        "google.cloud.pubsub.topic.v1.messagePublished",
+        "//pubsub.googleapis.com/projects/sample-project/topics/gcf-test",
+    )
+    data = {"message": {"data": "fizzbuzz"}}
+    req = flask.Request.from_values(headers=headers, json=data)
+
+    (res_data, res_context) = event_conversion.cloudevent_to_background_event(req)
+
+    assert res_context.event_type == "google.pubsub.topic.publish"
+    assert res_data == {"data": "fizzbuzz"}
+
+
+def test_cloudevent_to_legacy_event_with_firebase_auth_ce(
+    create_ce_headers,
+):
+    headers = create_ce_headers(
+        "google.firebase.auth.user.v1.created",
+        "//firebaseauth.googleapis.com/projects/my-project-id",
+    )
+    data = {
+        "metadata": {
+            "createTime": "2020-05-26T10:42:27Z",
+            "lastSignInTime": "2020-10-24T11:00:00Z",
+        },
+        "uid": "my-id",
+    }
+    req = flask.Request.from_values(headers=headers, json=data)
+
+    (res_data, res_context) = event_conversion.cloudevent_to_background_event(req)
+
+    assert res_context.event_type == "providers/firebase.auth/eventTypes/user.create"
+    assert res_data == {
+        "metadata": {
+            "createdAt": "2020-05-26T10:42:27Z",
+            "lastSignedInAt": "2020-10-24T11:00:00Z",
+        },
+        "uid": "my-id",
+    }
+
+
+def test_cloudevent_to_legacy_event_with_firebase_auth_ce_empty_metadata(
+    create_ce_headers,
+):
+    headers = create_ce_headers(
+        "google.firebase.auth.user.v1.created",
+        "//firebaseauth.googleapis.com/projects/my-project-id",
+    )
+    data = {"metadata": {}, "uid": "my-id"}
+    req = flask.Request.from_values(headers=headers, json=data)
+
+    (res_data, res_context) = event_conversion.cloudevent_to_background_event(req)
+
+    assert res_context.event_type == "providers/firebase.auth/eventTypes/user.create"
+    assert res_data == data
+
+
+@pytest.mark.parametrize(
+    "header_overrides, exception_message",
+    [
+        (
+            {"ce-source": "invalid-source-format"},
+            "Unexpected CloudEvent source",
+        ),
+        (
+            {"ce-source": None},
+            "Failed to convert CloudEvent to BackgroundEvent",
+        ),
+        (
+            {"ce-subject": None},
+            "Failed to convert CloudEvent to BackgroundEvent",
+        ),
+        (
+            {"ce-type": "unknown-type"},
+            "Unable to find background event equivalent type for",
+        ),
+    ],
+)
+def test_cloudevent_to_legacy_event_with_invalid_event(
+    create_ce_headers,
+    header_overrides,
+    exception_message,
+):
+    headers = create_ce_headers(
+        "google.firebase.database.document.v1.written",
+        "//firebasedatabase.googleapis.com/projects/_/instances/my-project-id",
+    )
+    for k, v in header_overrides.items():
+        if v is None:
+            del headers[k]
+        else:
+            headers[k] = v
+
+    req = flask.Request.from_values(headers=headers, json={"some": "val"})
+
+    with pytest.raises(EventConversionException) as exc_info:
+        event_conversion.cloudevent_to_background_event(req)
+
+    assert exception_message in exc_info.value.args[0]
+
+
+@pytest.mark.parametrize(
+    "source,expected_service,expected_name",
+    [
+        (
+            "//firebasedatabase.googleapis.com/projects/_/instances/my-project-id",
+            "firebasedatabase.googleapis.com",
+            "projects/_/instances/my-project-id",
+        ),
+        (
+            "//firebaseauth.googleapis.com/projects/my-project-id",
+            "firebaseauth.googleapis.com",
+            "projects/my-project-id",
+        ),
+        (
+            "//firestore.googleapis.com/projects/project-id/databases/(default)",
+            "firestore.googleapis.com",
+            "projects/project-id/databases/(default)",
+        ),
+    ],
+)
+def test_split_ce_source(source, expected_service, expected_name):
+    service, name = event_conversion._split_ce_source(source)
+    assert service == expected_service
+    assert name == expected_name