Add Cloud Events support for #55

setup.py

@@ -52,6 +52,7 @@
         "click>=7.0,<8.0",
         "watchdog>=0.10.0",
         "gunicorn>=19.2.0,<21.0; platform_system!='Windows'",
+        "cloudevents<=1.0",
     ],
     extras_require={"test": ["pytest", "tox"]},
     entry_points={

src/functions_framework/__init__.py

@@ -12,8 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import functools
+from enum import Enum
 import importlib.util
+import io
+import json
 import os.path
 import pathlib
 import sys
@@ -31,9 +33,18 @@
 )
 from google.cloud.functions.context import Context
 
+
+from cloudevents.sdk.event import v1
+from cloudevents.sdk import marshaller
+
 DEFAULT_SOURCE = os.path.realpath("./main.py")
 DEFAULT_SIGNATURE_TYPE = "http"
 
+class _EventType(Enum):
+    LEGACY = 1
+    CLOUD_EVENT_BINARY = 2
+    CLOUD_EVENT_TEXT = 3
+
 
 class _Event(object):
     """Event passed to background functions."""
@@ -59,51 +70,100 @@ def __init__(
             }
         self.data = data
 
-
 def _http_view_func_wrapper(function, request):
     def view_func(path):
         return function(request._get_current_object())
 
     return view_func
 
 
-def _is_binary_cloud_event(request):
-    return (
-        request.headers.get("ce-type")
-        and request.headers.get("ce-specversion")
-        and request.headers.get("ce-source")
-        and request.headers.get("ce-id")
-    )
+def _get_cloud_event_version():
+    return v1.Event()
+
+
+def _run_legacy_event(function, request):
+    event_data = request.get_json()
+    if not event_data:
+        flask.abort(400)
+    event_object = _Event(**event_data)
+    data = event_object.data
+    context = Context(**event_object.context)
+    function(data, context)
+
+
+def _run_binary_cloud_event(function, request, cloud_event_def):
+    data = io.BytesIO(request.get_data())
+    http_marshaller = marshaller.NewDefaultHTTPMarshaller()
+    event = http_marshaller.FromRequest(
+        cloud_event_def, request.headers, data, json.load)
+
+    function(event)
+
+
+def _run_text_cloud_event(function, request, cloud_event_def):
+    data = io.StringIO(request.get_data(as_text=True))
+    m = marshaller.NewDefaultHTTPMarshaller()
+    event = m.FromRequest(cloud_event_def, request.headers, data, json.loads)
+    function(event)
+
+
+def _get_event_type(request):
+    if request.headers.get("ce-type") \
+            and request.headers.get("ce-specversion") \
+            and request.headers.get("ce-source") \
+            and request.headers.get("ce-id"):
+        return _EventType.CLOUD_EVENT_BINARY
+    elif request.headers.get("Content-Type") == "application/cloudevents+json":
+        return _EventType.CLOUD_EVENT_TEXT
+    else:
+        return _EventType.LEGACY
 
 
 def _event_view_func_wrapper(function, request):
     def view_func(path):
-        if _is_binary_cloud_event(request):
-            # Support CloudEvents in binary content mode, with data being the
-            # whole request body and context attributes retrieved from request
-            # headers.
-            data = request.get_data()
-            context = Context(
-                eventId=request.headers.get("ce-eventId"),
-                timestamp=request.headers.get("ce-timestamp"),
-                eventType=request.headers.get("ce-eventType"),
-                resource=request.headers.get("ce-resource"),
-            )
-            function(data, context)
+        if _get_event_type(request) == _EventType.LEGACY:
+            _run_legacy_event(function, request)
+        else:
+            # here for defensive backwards compatibility in case we make a mistake in rollout.
+            raise InvalidConfigurationException("The FUNCTION_SIGNATURE_TYPE for this function is set to event "
+                                                "but no legacy event was given. If you are using CloudEvents set "
+                                                "FUNCTION_SIGNATURE_TYPE=cloudevent")
+
+
+        return "OK"
+
+    return view_func
+
+
+def _cloudevent_view_func_wrapper(function, request):
+    def view_func(path):
+        cloud_event_def = _get_cloud_event_version()
+        event_type = _get_event_type(request)
+        if event_type == _EventType.CLOUD_EVENT_TEXT:
+            _run_text_cloud_event(function, request, cloud_event_def)
+        elif event_type == _EventType.CLOUD_EVENT_BINARY:
+            _run_binary_cloud_event(function, request, cloud_event_def)
         else:
-            # This is a regular CloudEvent
-            event_data = request.get_json()
-            if not event_data:
-                flask.abort(400)
-            event_object = _Event(**event_data)
-            data = event_object.data
-            context = Context(**event_object.context)
-            function(data, context)
+            raise InvalidConfigurationException("Function was defined with FUNCTION_SIGNATURE_TYPE=cloudevent "
+                                                " but it did not receive a cloudevent as a request.")
 
         return "OK"
 
     return view_func
 
+def _setup_event_routes(app):
+    app.url_map.add(
+        werkzeug.routing.Rule(
+            "/", defaults={"path": ""}, endpoint="run", methods=["POST"]
+        )
+    )
+    app.url_map.add(
+        werkzeug.routing.Rule("/<path:path>", endpoint="run", methods=["POST"])
+    )
+
+    # Add a dummy endpoint for GET /
+    app.url_map.add(werkzeug.routing.Rule("/", endpoint="get", methods=["GET"]))
+    app.view_functions["get"] = lambda: ""
 
 def create_app(target=None, source=None, signature_type=None):
     # Get the configured function target
@@ -180,18 +240,11 @@ def create_app(target=None, source=None, signature_type=None):
         app.view_functions["run"] = _http_view_func_wrapper(function, flask.request)
         app.view_functions["error"] = lambda: flask.abort(404, description="Not Found")
     elif signature_type == "event":
-        app.url_map.add(
-            werkzeug.routing.Rule(
-                "/", defaults={"path": ""}, endpoint="run", methods=["POST"]
-            )
-        )
-        app.url_map.add(
-            werkzeug.routing.Rule("/<path:path>", endpoint="run", methods=["POST"])
-        )
+        _setup_event_routes(app)
         app.view_functions["run"] = _event_view_func_wrapper(function, flask.request)
-        # Add a dummy endpoint for GET /
-        app.url_map.add(werkzeug.routing.Rule("/", endpoint="get", methods=["GET"]))
-        app.view_functions["get"] = lambda: ""
+    elif signature_type == "cloudevent":
+        _setup_event_routes(app)
+        app.view_functions["run"] = _cloudevent_view_func_wrapper(function, flask.request)
     else:
         raise FunctionsFrameworkException(
             "Invalid signature type: {signature_type}".format(

tests/test_cloudevent_functions.py

@@ -0,0 +1,109 @@
+# Copyright 2020 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import json
+import pathlib
+import pytest
+from cloudevents.sdk import marshaller
+from cloudevents.sdk.event import v1
+from cloudevents.sdk.event import v03
+from cloudevents.sdk import converters
+
+from functions_framework import LazyWSGIApp, create_app, exceptions
+
+TEST_FUNCTIONS_DIR = pathlib.Path(__file__).resolve().parent / "test_functions"
+
+# Python 3.5: ModuleNotFoundError does not exist
+try:
+    _ModuleNotFoundError = ModuleNotFoundError
+except:
+    _ModuleNotFoundError = ImportError
+
+
+@pytest.fixture
+def event_1_10():
+    event = (
+        v1.Event()
+            .SetContentType("application/json")
+            .SetData('{"name":"john"}')
+            .SetEventID("my-id")
+            .SetSource("from-galaxy-far-far-away")
+            .SetEventTime("tomorrow")
+            .SetEventType("cloudevent.greet.you")
+    )
+    return event
+
+
+@pytest.fixture
+def event_0_3():
+    event = (
+        v03.Event()
+            .SetContentType("application/json")
+            .SetData('{"name":"john"}')
+            .SetEventID("my-id")
+            .SetSource("from-galaxy-far-far-away")
+            .SetEventTime("tomorrow")
+            .SetEventType("cloudevent.greet.you")
+    )
+    return event
+
+
+def test_event_1_0(event_1_10):
+    source = TEST_FUNCTIONS_DIR / "cloudevents" / "main.py"
+    target = "function"
+
+    client = create_app(target, source, "cloudevent").test_client()
+
+    m = marshaller.NewDefaultHTTPMarshaller()
+    structured_headers, structured_data = m.ToRequest(
+        event_1_10, converters.TypeStructured, json.dumps
+    )
+
+    resp = client.post("/", headers=structured_headers,
+                        data=structured_data.getvalue())
+    assert resp.status_code == 200
+    assert resp.data == b"OK"
+
+def test_binary_event_1_0(event_1_10):
+    source = TEST_FUNCTIONS_DIR / "cloudevents" / "main.py"
+    target = "function"
+
+    client = create_app(target, source, "cloudevent").test_client()
+
+    m = marshaller.NewDefaultHTTPMarshaller()
+
+    binary_headers, binary_data = m.ToRequest(
+        event_1_10, converters.TypeBinary, json.dumps)
+
+    resp = client.post(
+        "/", headers=binary_headers, data=binary_data)
+
+    assert resp.status_code == 200
+    assert resp.data == b"OK"
+
+
+def test_event_0_3(event_0_3):
+    source = TEST_FUNCTIONS_DIR / "cloudevents" / "main.py"
+    target = "function"
+
+    client = create_app(target, source, "cloudevent").test_client()
+
+    m = marshaller.NewDefaultHTTPMarshaller()
+    structured_headers, structured_data = m.ToRequest(
+        event_0_3, converters.TypeStructured, json.dumps
+    )
+
+    resp = client.post("/", headers=structured_headers,
+                        data=structured_data.getvalue())
+    assert resp.status_code == 200
+    assert resp.data == b"OK"