Add conformance tests

.github/workflows/conformance.yml

@@ -0,0 +1,48 @@
+name: Python Conformance CI
+on: [push, pull_request]
+jobs:
+  build:
+    runs-on: ubuntu-18.04
+    strategy:
+      matrix:
+        python-version: [3.8]
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v2
+
+    - name: Setup Python
+      uses: actions/setup-python@v2
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: Install the framework
+      run: python -m pip install -e .
+
+    - name: Setup Go
+      uses: actions/setup-go@v2
+      with:
+        go-version: '1.13'
+
+    - name: Run HTTP conformance tests
+      uses: GoogleCloudPlatform/functions-framework-conformance/action@v0.3.2
+      with:
+        functionType: 'http'
+        useBuildpacks: false
+        validateMapping: false
+        cmd: "'functions-framework --source tests/conformance/main.py --target write_http --signature-type http'"
+
+    - name: Run event conformance tests
+      uses: GoogleCloudPlatform/functions-framework-conformance/action@v0.3.2
+      with:
+        functionType: 'legacyevent'
+        useBuildpacks: false
+        validateMapping: false
+        cmd: "'functions-framework --source tests/conformance/main.py --target write_legacy_event --signature-type event'"
+
+    - name: Run cloudevent conformance tests
+      uses: GoogleCloudPlatform/functions-framework-conformance/action@v0.3.2
+      with:
+        functionType: 'cloudevent'
+        useBuildpacks: false
+        validateMapping: false
+        cmd: "'functions-framework --source tests/conformance/main.py --target write_cloud_event --signature-type cloudevent'"

tests/conformance/main.py

@@ -0,0 +1,35 @@
+import json
+
+from cloudevents.http import to_json
+
+filename = "function_output.json"
+
+
+def _write_output(content):
+    with open(filename, "w") as f:
+        f.write(content)
+
+
+def write_http(request):
+    _write_output(json.dumps(request.json))
+    return "OK", 200
+
+
+def write_legacy_event(data, context):
+    _write_output(
+        json.dumps(
+            {
+                "data": data,
+                "context": {
+                    "eventId": context.event_id,
+                    "timestamp": context.timestamp,
+                    "eventType": context.event_type,
+                    "resource": context.resource,
+                },
+            }
+        )
+    )
+
+
+def write_cloud_event(cloudevent):
+    _write_output(to_json(cloudevent).decode())