Add EventHub metadata E2E testcases (#719)

* Fix syntax issue in azure-pipelinesyml

Fix an error in the elif statement
Fix an issue in nuspec_path variable reference

* Establish python_value and python_type contract with azure.functions.meta.Datum

* Add e2e test for eventhub generating by azure-eventhub library

* Remove unused datetime library

* Add batch trigger tests

* Replace eventhub using async interfaces

* Python 3.6 and 3.7 not support datetime.fromisostring

* EventHub timestamp is timezone aware

* Fix setup.py

* Add comments in EventHub trigger E2E tests

* Use docstring instead

Author: Hazhzeng

azure_functions_worker/bindings/datumdef.py

@@ -1,5 +1,8 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
+
+from typing import Any
+import json
 from .. import protos
 
 
@@ -8,6 +11,29 @@ def __init__(self, value, type):
         self.value = value
         self.type = type
 
+    @property
+    def python_value(self) -> Any:
+        if self.value is None or self.type is None:
+            return None
+        elif self.type in ('bytes', 'string', 'int', 'double'):
+            return self.value
+        elif self.type == 'json':
+            return json.loads(self.value)
+        elif self.type == 'collection_string':
+            return [v for v in self.value.string]
+        elif self.type == 'collection_bytes':
+            return [v for v in self.value.bytes]
+        elif self.type == 'collection_double':
+            return [v for v in self.value.double]
+        elif self.type == 'collection_sint64':
+            return [v for v in self.value.sint64]
+        else:
+            return self.value
+
+    @property
+    def python_type(self) -> type:
+        return type(self.python_value)
+
     def __eq__(self, other):
         if not isinstance(other, type(self)):
             return False

setup.py

@@ -286,6 +286,8 @@ def run(self):
     extras_require={
         'dev': [
             'azure-functions==1.3.0',
+            'azure-eventhub~=5.1.0',
+            'python-dateutil~=2.8.1',
             'flake8~=3.7.9',
             'mypy',
             'pytest',

tests/endtoend/eventhub_batch_functions/eventhub_multiple/__init__.py

@@ -3,6 +3,8 @@
 import json
 
 
+# This is an actual EventHub trigger which handles Eventhub events in batches.
+# It serializes multiple event data into a json and store it into a blob.
 def main(events):
     table_entries = []
     for event in events:

tests/endtoend/eventhub_batch_functions/eventhub_output_batch/__init__.py

@@ -3,6 +3,7 @@
 import azure.functions as func
 
 
+# An HttpTrigger to generating EventHub event from EventHub Output Binding
 def main(req: func.HttpRequest) -> str:
     events = req.get_body().decode('utf-8')
     return events

tests/endtoend/eventhub_batch_functions/get_eventhub_batch_triggered/__init__.py

@@ -3,5 +3,6 @@
 import azure.functions as func
 
 
+# Retrieve the event data from storage blob and return it as Http response
 def main(req: func.HttpRequest, testEntities):
     return func.HttpResponse(status_code=200, body=testEntities)

tests/endtoend/eventhub_batch_functions/get_eventhub_batch_triggered/function.json

@@ -1,27 +1,27 @@
-{
-  "scriptFile": "__init__.py",
-  "bindings": [
-    {
-      "type": "httpTrigger",
-      "direction": "in",
-      "authLevel": "anonymous",
-      "methods": [
-        "get"
-      ],
-      "name": "req"
-    },
-    {
-      "direction": "in",
-      "type": "table",
-      "name": "testEntities",
-      "partitionKey": "WillBePopulated",
-      "tableName": "EventHubBatchTest",
-      "connection": "AzureWebJobsStorage"
-    },
-    {
-      "type": "http",
-      "direction": "out",
-      "name": "$return"
-    }
-  ]
+{
+  "scriptFile": "__init__.py",
+  "bindings": [
+    {
+      "type": "httpTrigger",
+      "direction": "in",
+      "authLevel": "anonymous",
+      "methods": [
+        "get"
+      ],
+      "name": "req"
+    },
+    {
+      "direction": "in",
+      "type": "table",
+      "name": "testEntities",
+      "partitionKey": "WillBePopulated",
+      "tableName": "EventHubBatchTest",
+      "connection": "AzureWebJobsStorage"
+    },
+    {
+      "type": "http",
+      "direction": "out",
+      "name": "$return"
+    }
+  ]
 }

tests/endtoend/eventhub_batch_functions/get_metadata_batch_triggered/__init__.py

@@ -0,0 +1,10 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+import azure.functions as func
+
+
+# Retrieve the event data from storage blob and return it as Http response
+def main(req: func.HttpRequest, file: func.InputStream) -> str:
+    return func.HttpResponse(body=file.read().decode('utf-8'),
+                             status_code=200,
+                             mimetype='application/json')

tests/endtoend/eventhub_batch_functions/get_metadata_batch_triggered/function.json

@@ -0,0 +1,22 @@
+{
+  "scriptFile": "__init__.py",
+  "bindings": [
+    {
+      "type": "httpTrigger",
+      "direction": "in",
+      "name": "req"
+    },
+    {
+      "type": "blob",
+      "direction": "in",
+      "name": "file",
+      "connection": "AzureWebJobsStorage",
+      "path": "python-worker-tests/test-metadata-batch-triggered.txt"
+    },
+    {
+      "type": "http",
+      "direction": "out",
+      "name": "$return"
+    }
+  ]
+}

tests/endtoend/eventhub_batch_functions/metadata_multiple/__init__.py

@@ -0,0 +1,24 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import typing
+import json
+import azure.functions as func
+
+
+# This is an actual EventHub trigger which handles Eventhub events in batches.
+# It serializes multiple event data into a json and store it into a blob.
+def main(events: typing.List[func.EventHubEvent]) -> bytes:
+    event_list = []
+    for event in events:
+        event_dict: typing.Mapping[str, typing.Any] = {
+            'body': event.get_body().decode('utf-8'),
+            'enqueued_time': event.enqueued_time.isoformat(),
+            'partition_key': event.partition_key,
+            'sequence_number': event.sequence_number,
+            'offset': event.offset,
+            'metadata': event.metadata
+        }
+        event_list.append(event_dict)
+
+    return json.dumps(event_list)

tests/endtoend/eventhub_batch_functions/metadata_multiple/function.json

@@ -0,0 +1,21 @@
+{
+    "scriptFile": "__init__.py",
+    "bindings": [
+      {
+        "type": "eventHubTrigger",
+        "name": "events",
+        "direction": "in",
+        "cardinality": "many",
+        "dataType": "binary",
+        "eventHubName": "python-worker-ci-eventhub-batch-metadata",
+        "connection": "AzureWebJobsEventHubConnectionString"
+      },
+      {
+        "type": "blob",
+        "direction": "out",
+        "name": "$return",
+        "connection": "AzureWebJobsStorage",
+        "path": "python-worker-tests/test-metadata-batch-triggered.txt"
+      }
+    ]
+  }