Try with EHE

Author: Evan Roman

tests/extension_tests/deferred_bindings_tests/deferred_bindings_eventhub_functions/function_app.py

@@ -1,58 +1,52 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
 import json
+import typing
 
 import azure.functions as func
-import azurefunctions.extensions.bindings.eventhub as eh
+# import azurefunctions.extensions.bindings.eventhub as eh
 
 app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
 
 
-# @app.function_name(name="put_bc_trigger")
-# @app.blob_output(arg_name="file",
-#                  path="python-worker-tests/test-blobclient-trigger.txt",
-#                  connection="AzureWebJobsStorage")
-# @app.route(route="put_bc_trigger")
-# def put_bc_trigger(req: func.HttpRequest, file: func.Out[str]) -> str:
-#     file.set(req.get_body())
-#     return 'OK'
-
-# An HttpTrigger to generating EventHub event from EventHub Output Binding
-@app.function_name(name="eventhub_output")
+@app.function_name(name="put_eh_trigger")
 @app.event_hub_output(arg_name="event",
                       event_hub_name="python-worker-ci-eventhub-one",
                       connection="AzureWebJobsEventHubConnectionString")
-@app.route(route="eventhub_output")
-def eventhub_output(req: func.HttpRequest, event: func.Out[str]) -> str:
-    event.set('debug')
+@app.route(route="put_eh_trigger")
+def put_eh_trigger(req: func.HttpRequest, event: func.Out[str]) -> str:
+    event.set(req.get_body())
     return 'OK'
 
-# This is an actual EventHub trigger which will convert the event data
-# into a storage blob.
-@app.function_name(name="eventhub_trigger")
-@app.event_hub_message_trigger(arg_name="event",
-                               event_hub_name="python-worker-ci-eventhub-one",
-                               connection="AzureWebJobsEventHubConnectionString")
+@app.function_name(name="eh_ed_trigger")
+@app.event_hub_message_trigger(
+    arg_name="event",
+    event_hub_name="python-worker-ci-eventhub-one-metadata",
+    connection="AzureWebJobsEventHubConnectionString")
 @app.blob_output(arg_name="$return",
-                 path="python-worker-tests/test-eventhub-triggered.txt",
+                 path="python-worker-tests/test-metadata-triggered.txt",
                  connection="AzureWebJobsStorage")
-def eventhub_trigger(event: eh.EventData) -> bytes:
-    return bytes(event.body_as_str())
+async def eh_ed_trigger(event: func.EventHubEvent) -> bytes:
+    event_dict: typing.Mapping[str, typing.Any] = {
+        'body': event.get_body().decode('utf-8'),
+        # Uncomment this when the EnqueuedTimeUtc is fixed in azure-functions
+        # 'enqueued_time': event.enqueued_time.isoformat(),
+        'partition_key': event.partition_key,
+        'sequence_number': event.sequence_number,
+        'offset': event.offset,
+        'metadata': event.metadata
+    }
+
+    return json.dumps(event_dict)
 
 # Retrieve the event data from storage blob and return it as Http response
-@app.function_name(name="get_eventhub_triggered")
-@app.route(route="get_eventhub_triggered")
+@app.function_name(name="get_eh_ed_triggered")
+@app.route(route="get_eh_ed_triggered")
 @app.blob_input(arg_name="file",
-                path="python-worker-tests/test-eventhub-triggered.txt",
+                path="python-worker-tests/test-metadata-triggered.txt",
                 connection="AzureWebJobsStorage")
-def get_eventhub_triggered(req: func.HttpRequest,
-                           file: func.InputStream) -> str:
-    return file.read().decode('utf-8')
-
-# @app.event_hub_message_trigger(
-#     arg_name="event",
-#     event_hub_name="python-worker-ci-eventhub-one",
-#     connection="AzureWebJobsEventHubConnectionString"
-# )
-# def eventhub_trigger(event: eh.EventHubData) -> str:
-#     return event.body_as_str()
+async def get_eh_ed_triggered(req: func.HttpRequest,
+                                 file: func.InputStream) -> str:
+    return func.HttpResponse(body=file.read().decode('utf-8'),
+                             status_code=200,
+                             mimetype='application/json')

tests/extension_tests/deferred_bindings_tests/test_deferred_bindings_eventhub_functions.py

@@ -1,6 +1,6 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
-import os
+import json
 import sys
 import time
 import unittest
@@ -23,23 +23,29 @@ def get_libraries_to_install(cls):
 
     @testutils.retryable_test(3, 5)
     def test_ed_eventhub_trigger(self):
-        # data = "DummyData"
-
-        r = self.webhost.request('POST', 'eventhub_output',
-                                 data="test")
+        # Generate a unique event body for EventHub eventme
+        random_number = str(round(time.time()) % 1000)
+        req_body = {
+            'body': random_number
+        }
+
+        r = self.webhost.request('POST', 'put_eh_trigger',
+                                 data=json.dumps(req_body))
         self.assertEqual(r.status_code, 200)
         self.assertEqual(r.text, 'OK')
 
-        # # Once the event get generated, allow function host to poll from
-        # # EventHub and wait for eventhub_trigger to execute,
-        # # converting the event metadata into a blob.
+        # Once the event get generated, allow function host to poll from
+        # EventHub and wait for eventhub_trigger to execute,
+        # converting the event metadata into a blob.
         time.sleep(5)
 
-        # # Call get_eventhub_triggered to retrieve event metadata from blob.
-        r = self.webhost.request('GET', 'get_eventhub_triggered')
+        # Call get_eh_ed_triggered to retrieve event metadata from blob.
+        r = self.webhost.request('GET', 'get_eh_ed_triggered')
 
-        # # Waiting for the blob get updated with the latest data from the
+        # Waiting for the blob to get updated with the latest data from the
         # eventhub output binding
         time.sleep(5)
         self.assertEqual(r.status_code, 200)
-        # response = r.json()
+
+        event = r.json()
+        self.assertEqual(event['body'], random_number)