Test

Author: Evan Roman

tests/extension_tests/deferred_bindings_tests/deferred_bindings_eventhub_functions/function_app.py

@@ -0,0 +1,49 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+import json
+
+import azure.functions as func
+import azurefunctions.extensions.bindings.eventhub as eh
+
+app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
+
+# An HttpTrigger to generating EventHub event from EventHub Output Binding
+@app.function_name(name="eventhub_output")
+@app.route(route="eventhub_output")
+@app.event_hub_output(arg_name="event",
+                      event_hub_name="python-worker-ci-eventhub-one",
+                      connection="AzureWebJobsEventHubConnectionString")
+def eventhub_output(req: func.HttpRequest, event: func.Out[str]) -> str:
+    event.set(req.get_body().decode('utf-8'))
+    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.blob_output(arg_name="$return",
+                 path="python-worker-tests/test-eventhub-triggered.txt",
+                 connection="AzureWebJobsStorage")
+def eventhub_trigger(event: eh.EventData) -> bytes:
+    return bytes(event.body_as_str())
+
+# 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.blob_input(arg_name="file",
+                path="python-worker-tests/test-eventhub-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()

tests/extension_tests/deferred_bindings_tests/test_deferred_bindings_eventhub_functions.py

@@ -0,0 +1,43 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+import sys
+import time
+import unittest
+
+from tests.utils import testutils
+
+
+@unittest.skipIf(sys.version_info.minor <= 8, "The base extension"
+                                              "is only supported for 3.9+.")
+class TestDeferredBindingsEventHubFunctions(testutils.WebHostTestCase):
+
+    @classmethod
+    def get_script_dir(cls):
+        return testutils.EXTENSION_TESTS_FOLDER / 'deferred_bindings_tests' / \
+            'deferred_bindings_eventhub_functions'
+
+    @classmethod
+    def get_libraries_to_install(cls):
+        return ['azurefunctions-extensions-bindings-eventhub']
+    
+    def test_ed_eventhub_trigger(self):
+        data = "DummyData"
+
+        r = self.webhost.request('POST', 'eventhub_output',
+                                 data=data.encode('utf-8'))
+        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.
+        # time.sleep(5)
+
+        # # Call get_eventhub_triggered to retrieve event metadata from blob.
+        # r = self.webhost.request('GET', 'get_eventhub_triggered')
+
+        # # Waiting for the blob get updated with the latest data from the
+        # # eventhub output binding
+        # time.sleep(5)
+        # self.assertEqual(r.status_code, 200)
+        # response = r.json()