Check issue

Author: Evan Roman

tests/extension_tests/deferred_bindings_tests/deferred_bindings_eventhub_functions/function_app.py

@@ -8,24 +8,24 @@
 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'
+# @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.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'
+@app.function_name(name="eventhub_output")
+@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(req.get_body().decode('utf-8'))
+    return 'OK'
 
 # # This is an actual EventHub trigger which will convert the event data
 # # into a storage blob.

tests/extension_tests/deferred_bindings_tests/test_deferred_bindings_eventhub_functions.py

@@ -1,5 +1,6 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
+import os
 import sys
 import time
 import unittest
@@ -23,10 +24,10 @@ def get_libraries_to_install(cls):
     def test_ed_eventhub_trigger(self):
         # data = "DummyData"
 
-        r = self.webhost.request('POST', 'put_bc_trigger',
+        r = self.webhost.request('POST', 'eventhub_output',
                                  data="test")
-        self.assertEqual(r.status_code, 200)
-        # self.assertEqual(r.text, 'OK')
+        # self.assertEqual(r.status_code, 200)
+        self.assertEqual(r.text, f"AzureWebJobsEventHubConnectionString: {os.getenv('AzureWebJobsEventHubConnectionString')}, AzureWebJobsStorage: {os.getenv('AzureWebJobsStorage')}")
 
         # # Once the event get generated, allow function host to poll from
         # # EventHub and wait for eventhub_trigger to execute,