test: add ServiceBus SDK tests

eng/templates/jobs/ci-emulator-tests.yml

@@ -82,8 +82,9 @@ jobs:
       - bash: |
             python -m pytest -q -n auto --dist loadfile --reruns 4 --ignore=tests/emulator_tests/test_servicebus_functions.py tests/emulator_tests
         env:
-          AzureWebJobsStorage: "UseDevelopmentStorage=true"
-          AzureWebJobsEventHubConnectionString: "Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;"
+          AzureWebJobsStorage: $(AzureWebJobsStorage)
+          AzureWebJobsEventHubConnectionString: $(AzureWebJobsEventHubConnectionString)
+          AZURE_STORAGE_CONNECTION_STRING: $(AZURE_STORAGE_CONNECTION_STRING)
         displayName: "Running $(PYTHON_VERSION) Python Linux Emulator Tests"
       - bash: |
           # Stop and remove EventHub Emulator container to free up the port
@@ -97,6 +98,7 @@ jobs:
       - bash: |
             python -m pytest -q -n auto --dist loadfile --reruns 4 tests/emulator_tests/test_servicebus_functions.py
         env:
-          AzureWebJobsStorage: "UseDevelopmentStorage=true"
-          AzureWebJobsServiceBusConnectionString: "Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;"
+          AzureWebJobsStorage: $(AzureWebJobsStorage)
+          AzureWebJobsServiceBusConnectionString: $(AzureWebJobsServiceBusConnectionString)
+          AzureWebJobsServiceBusSDKConnectionString: $(AzureWebJobsServiceBusSDKConnectionString)
         displayName: "Running $(PYTHON_VERSION) Python ServiceBus Linux Emulator Tests"

eng/templates/jobs/ci-unit-tests.yml

@@ -55,4 +55,5 @@ jobs:
         condition: and(eq(variables.isSdkRelease, false), eq(variables.isExtensionsRelease, false), eq(variables['USETESTPYTHONSDK'], false), eq(variables['USETESTPYTHONEXTENSIONS'], false))
         env:
           PYTHON_VERSION: $(PYTHON_VERSION)
+          AZURE_STORAGE_CONNECTION_STRING: $(AZURE_STORAGE_CONNECTION_STRING)
 

eng/templates/official/jobs/ci-e2e-tests.yml

@@ -150,7 +150,7 @@ jobs:
           Write-Host "skipTest: $(skipTest)"
         displayName: 'Display skipTest variable'
       - bash: |
-             python -m pytest -q -n auto --dist loadfile --reruns 4 --cov=./azure_functions_worker --cov-report xml --cov-branch --cov-append tests/endtoend tests/extension_tests/deferred_bindings_tests tests/extension_tests/http_v2_tests
+             python -m pytest -q -n auto --dist loadfile --reruns 4 --cov=./azure_functions_worker --cov-report xml --cov-branch --cov-append tests/endtoend
         env:
           AzureWebJobsStorage: $(STORAGE_CONNECTION)
           AzureWebJobsCosmosDBConnectionString: $(COSMOSDB_CONNECTION)

eng/templates/utils/variables.yml

@@ -1,4 +1,5 @@
 variables:
+  - group: python-emulator-resources
   - name: isSdkRelease
     value: $[startsWith(variables['Build.SourceBranch'], 'refs/heads/sdk/')] 
   - name: isExtensionsRelease

pyproject.toml

@@ -80,6 +80,7 @@ dev = [
     "pre-commit",
     "invoke",
     "cryptography",
+    "jsonpickle",
     "orjson"
 ]
 test-http-v2 = [
@@ -88,7 +89,8 @@ test-http-v2 = [
 ]
 test-deferred-bindings = [
     "azurefunctions-extensions-bindings-blob==1.0.0",
-    "azurefunctions-extensions-bindings-eventhub==1.0.0b1"
+    "azurefunctions-extensions-bindings-eventhub==1.0.0b1",
+    "azurefunctions-extensions-bindings-servicebus==1.0.0b1"
 ]
 
 [build-system]

tests/extension_tests/deferred_bindings_tests/deferred_bindings_blob_functions/function_app.py → tests/emulator_tests/blob_functions/blob_functions_sdk/function_app.py

@@ -11,7 +11,7 @@
 @app.function_name(name="put_bc_trigger")
 @app.blob_output(arg_name="file",
                  path="python-worker-tests/test-blobclient-trigger.txt",
-                 connection="AzureWebJobsStorage")
+                 connection="AZURE_STORAGE_CONNECTION_STRING")
 @app.route(route="put_bc_trigger")
 def put_bc_trigger(req: func.HttpRequest, file: func.Out[str]) -> str:
     file.set(req.get_body())
@@ -21,10 +21,10 @@ def put_bc_trigger(req: func.HttpRequest, file: func.Out[str]) -> str:
 @app.function_name(name="bc_blob_trigger")
 @app.blob_trigger(arg_name="client",
                   path="python-worker-tests/test-blobclient-trigger.txt",
-                  connection="AzureWebJobsStorage")
+                  connection="AZURE_STORAGE_CONNECTION_STRING")
 @app.blob_output(arg_name="$return",
                  path="python-worker-tests/test-blobclient-triggered.txt",
-                 connection="AzureWebJobsStorage")
+                 connection="AZURE_STORAGE_CONNECTION_STRING")
 def bc_blob_trigger(client: blob.BlobClient) -> str:
     blob_properties = client.get_blob_properties()
     file = client.download_blob(encoding='utf-8').readall()
@@ -38,7 +38,7 @@ def bc_blob_trigger(client: blob.BlobClient) -> str:
 @app.function_name(name="get_bc_blob_triggered")
 @app.blob_input(arg_name="client",
                 path="python-worker-tests/test-blobclient-triggered.txt",
-                connection="AzureWebJobsStorage")
+                connection="AZURE_STORAGE_CONNECTION_STRING")
 @app.route(route="get_bc_blob_triggered")
 def get_bc_blob_triggered(req: func.HttpRequest,
                           client: blob.BlobClient) -> str:
@@ -48,7 +48,7 @@ def get_bc_blob_triggered(req: func.HttpRequest,
 @app.function_name(name="put_cc_trigger")
 @app.blob_output(arg_name="file",
                  path="python-worker-tests/test-containerclient-trigger.txt",
-                 connection="AzureWebJobsStorage")
+                 connection="AZURE_STORAGE_CONNECTION_STRING")
 @app.route(route="put_cc_trigger")
 def put_cc_trigger(req: func.HttpRequest, file: func.Out[str]) -> str:
     file.set(req.get_body())
@@ -58,10 +58,10 @@ def put_cc_trigger(req: func.HttpRequest, file: func.Out[str]) -> str:
 @app.function_name(name="cc_blob_trigger")
 @app.blob_trigger(arg_name="client",
                   path="python-worker-tests/test-containerclient-trigger.txt",
-                  connection="AzureWebJobsStorage")
+                  connection="AZURE_STORAGE_CONNECTION_STRING")
 @app.blob_output(arg_name="$return",
                  path="python-worker-tests/test-containerclient-triggered.txt",
-                 connection="AzureWebJobsStorage")
+                 connection="AZURE_STORAGE_CONNECTION_STRING")
 def cc_blob_trigger(client: blob.ContainerClient) -> str:
     container_properties = client.get_container_properties()
     file = client.download_blob("test-containerclient-trigger.txt",
@@ -75,7 +75,7 @@ def cc_blob_trigger(client: blob.ContainerClient) -> str:
 @app.function_name(name="get_cc_blob_triggered")
 @app.blob_input(arg_name="client",
                 path="python-worker-tests/test-containerclient-triggered.txt",
-                connection="AzureWebJobsStorage")
+                connection="AZURE_STORAGE_CONNECTION_STRING")
 @app.route(route="get_cc_blob_triggered")
 def get_cc_blob_triggered(req: func.HttpRequest,
                           client: blob.ContainerClient) -> str:
@@ -86,7 +86,7 @@ def get_cc_blob_triggered(req: func.HttpRequest,
 @app.function_name(name="put_ssd_trigger")
 @app.blob_output(arg_name="file",
                  path="python-worker-tests/test-ssd-trigger.txt",
-                 connection="AzureWebJobsStorage")
+                 connection="AZURE_STORAGE_CONNECTION_STRING")
 @app.route(route="put_ssd_trigger")
 def put_ssd_trigger(req: func.HttpRequest, file: func.Out[str]) -> str:
     file.set(req.get_body())
@@ -96,10 +96,10 @@ def put_ssd_trigger(req: func.HttpRequest, file: func.Out[str]) -> str:
 @app.function_name(name="ssd_blob_trigger")
 @app.blob_trigger(arg_name="stream",
                   path="python-worker-tests/test-ssd-trigger.txt",
-                  connection="AzureWebJobsStorage")
+                  connection="AZURE_STORAGE_CONNECTION_STRING")
 @app.blob_output(arg_name="$return",
                  path="python-worker-tests/test-ssd-triggered.txt",
-                 connection="AzureWebJobsStorage")
+                 connection="AZURE_STORAGE_CONNECTION_STRING")
 def ssd_blob_trigger(stream: blob.StorageStreamDownloader) -> str:
     # testing chunking
     file = ""
@@ -113,7 +113,7 @@ def ssd_blob_trigger(stream: blob.StorageStreamDownloader) -> str:
 @app.function_name(name="get_ssd_blob_triggered")
 @app.blob_input(arg_name="stream",
                 path="python-worker-tests/test-ssd-triggered.txt",
-                connection="AzureWebJobsStorage")
+                connection="AZURE_STORAGE_CONNECTION_STRING")
 @app.route(route="get_ssd_blob_triggered")
 def get_ssd_blob_triggered(req: func.HttpRequest,
                            stream: blob.StorageStreamDownloader) -> str:
@@ -124,7 +124,7 @@ def get_ssd_blob_triggered(req: func.HttpRequest,
 @app.route(route="get_bc_bytes")
 @app.blob_input(arg_name="client",
                 path="python-worker-tests/test-blob-extension-bytes.txt",
-                connection="AzureWebJobsStorage")
+                connection="AZURE_STORAGE_CONNECTION_STRING")
 def get_bc_bytes(req: func.HttpRequest, client: blob.BlobClient) -> str:
     return client.download_blob(encoding='utf-8').readall()
 
@@ -133,7 +133,7 @@ def get_bc_bytes(req: func.HttpRequest, client: blob.BlobClient) -> str:
 @app.route(route="get_cc_bytes")
 @app.blob_input(arg_name="client",
                 path="python-worker-tests/test-blob-extension-bytes.txt",
-                connection="AzureWebJobsStorage")
+                connection="AZURE_STORAGE_CONNECTION_STRING")
 def get_cc_bytes(req: func.HttpRequest,
                  client: blob.ContainerClient) -> str:
     return client.download_blob("test-blob-extension-bytes.txt",
@@ -144,7 +144,7 @@ def get_cc_bytes(req: func.HttpRequest,
 @app.route(route="get_ssd_bytes")
 @app.blob_input(arg_name="stream",
                 path="python-worker-tests/test-blob-extension-bytes.txt",
-                connection="AzureWebJobsStorage")
+                connection="AZURE_STORAGE_CONNECTION_STRING")
 def get_ssd_bytes(req: func.HttpRequest,
                   stream: blob.StorageStreamDownloader) -> str:
     return stream.readall().decode('utf-8')
@@ -154,7 +154,7 @@ def get_ssd_bytes(req: func.HttpRequest,
 @app.route(route="get_bc_str")
 @app.blob_input(arg_name="client",
                 path="python-worker-tests/test-blob-extension-str.txt",
-                connection="AzureWebJobsStorage")
+                connection="AZURE_STORAGE_CONNECTION_STRING")
 def get_bc_str(req: func.HttpRequest, client: blob.BlobClient) -> str:
     return client.download_blob(encoding='utf-8').readall()
 
@@ -163,7 +163,7 @@ def get_bc_str(req: func.HttpRequest, client: blob.BlobClient) -> str:
 @app.route(route="get_cc_str")
 @app.blob_input(arg_name="client",
                 path="python-worker-tests",
-                connection="AzureWebJobsStorage")
+                connection="AZURE_STORAGE_CONNECTION_STRING")
 def get_cc_str(req: func.HttpRequest, client: blob.ContainerClient) -> str:
     return client.download_blob("test-blob-extension-str.txt",
                                 encoding='utf-8').readall()
@@ -173,7 +173,7 @@ def get_cc_str(req: func.HttpRequest, client: blob.ContainerClient) -> str:
 @app.route(route="get_ssd_str")
 @app.blob_input(arg_name="stream",
                 path="python-worker-tests/test-blob-extension-str.txt",
-                connection="AzureWebJobsStorage")
+                connection="AZURE_STORAGE_CONNECTION_STRING")
 def get_ssd_str(req: func.HttpRequest, stream: blob.StorageStreamDownloader) -> str:
     return stream.readall().decode('utf-8')
 
@@ -183,11 +183,11 @@ def get_ssd_str(req: func.HttpRequest, stream: blob.StorageStreamDownloader) ->
 @app.blob_input(arg_name="client",
                 path="python-worker-tests/test-blob-extension-str.txt",
                 data_type="STRING",
-                connection="AzureWebJobsStorage")
+                connection="AZURE_STORAGE_CONNECTION_STRING")
 @app.blob_input(arg_name="blob",
                 path="python-worker-tests/test-blob-extension-str.txt",
                 data_type="STRING",
-                connection="AzureWebJobsStorage")
+                connection="AZURE_STORAGE_CONNECTION_STRING")
 def bc_and_inputstream_input(req: func.HttpRequest, client: blob.BlobClient,
                              blob: func.InputStream) -> str:
     output_msg = ""
@@ -202,11 +202,11 @@ def bc_and_inputstream_input(req: func.HttpRequest, client: blob.BlobClient,
 @app.blob_input(arg_name="blob",
                 path="python-worker-tests/test-blob-extension-str.txt",
                 data_type="STRING",
-                connection="AzureWebJobsStorage")
+                connection="AZURE_STORAGE_CONNECTION_STRING")
 @app.blob_input(arg_name="client",
                 path="python-worker-tests/test-blob-extension-str.txt",
                 data_type="STRING",
-                connection="AzureWebJobsStorage")
+                connection="AZURE_STORAGE_CONNECTION_STRING")
 def inputstream_and_bc_input(req: func.HttpRequest, blob: func.InputStream,
                              client: blob.BlobClient) -> str:
     output_msg = ""
@@ -221,7 +221,7 @@ def inputstream_and_bc_input(req: func.HttpRequest, blob: func.InputStream,
 @app.blob_input(arg_name="file",
                 path="python-worker-tests/test-blob-extension-str.txt",
                 data_type="STRING",
-                connection="AzureWebJobsStorage")
+                connection="AZURE_STORAGE_CONNECTION_STRING")
 def type_undefined(req: func.HttpRequest, file) -> str:
     assert not isinstance(file, blob.BlobClient)
     assert not isinstance(file, blob.ContainerClient)
@@ -232,7 +232,7 @@ def type_undefined(req: func.HttpRequest, file) -> str:
 @app.function_name(name="put_blob_str")
 @app.blob_output(arg_name="file",
                  path="python-worker-tests/test-blob-extension-str.txt",
-                 connection="AzureWebJobsStorage")
+                 connection="AZURE_STORAGE_CONNECTION_STRING")
 @app.route(route="put_blob_str")
 def put_blob_str(req: func.HttpRequest, file: func.Out[str]) -> str:
     file.set(req.get_body())
@@ -242,7 +242,7 @@ def put_blob_str(req: func.HttpRequest, file: func.Out[str]) -> str:
 @app.function_name(name="put_blob_bytes")
 @app.blob_output(arg_name="file",
                  path="python-worker-tests/test-blob-extension-bytes.txt",
-                 connection="AzureWebJobsStorage")
+                 connection="AZURE_STORAGE_CONNECTION_STRING")
 @app.route(route="put_blob_bytes")
 def put_blob_bytes(req: func.HttpRequest, file: func.Out[bytes]) -> str:
     file.set(req.get_body())

tests/emulator_tests/eventhub_batch_functions/eventhub_batch_functions_stein/function_app.py

@@ -40,7 +40,7 @@ def eventhub_multiple(events) -> str:
                       connection="AzureWebJobsEventHubConnectionString",
                       event_hub_name="python-worker-ci-eventhub-batch")
 @app.route(route="eventhub_output_batch", binding_arg_name="out")
-def eventhub_output_batch(req: func.HttpRequest, out: func.Out[str]) -> str:
+def c(req: func.HttpRequest, out: func.Out[str]) -> str:
     events = req.get_body().decode('utf-8')
     return events
 

tests/emulator_tests/servicebus_batch_functions/servicebus_batch_functions_stein/function_app.py

@@ -0,0 +1,43 @@
+import json
+
+import azure.functions as func
+
+app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
+
+
+@app.route(route="servicebus_output_batch")
+@app.service_bus_queue_output(
+    arg_name="msg",
+    connection="AzureWebJobsServiceBusConnectionString",
+    queue_name="testqueue-batch")
+def servicebus_output_batch(req: func.HttpRequest, msg: func.Out[str]):
+    msg.set(req.get_body().decode('utf-8'))
+    return 'OK'
+
+
+@app.route(route="get_servicebus_batch_triggered")
+@app.blob_input(arg_name="file",
+                path="python-worker-tests/test-servicebus-batch-triggered.txt",
+                connection="AzureWebJobsStorage")
+def get_servicebus_batch_triggered(req: func.HttpRequest,
+                             file: func.InputStream) -> str:
+    return func.HttpResponse(
+        file.read().decode('utf-8'), mimetype='application/json')
+
+
+@app.service_bus_queue_trigger(
+    arg_name="events",
+    connection="AzureWebJobsServiceBusConnectionString",
+    queue_name="testqueue-batch",
+    cardinality="many")
+@app.blob_output(arg_name="$return",
+                 path="python-worker-tests/test-servicebus-batch-triggered.txt",
+                 connection="AzureWebJobsStorage")
+def servicebus_multiple(events) -> str:
+    table_entries = []
+    for event in events:
+        json_entry = event.get_body().decode('utf-8')
+        table_entry = json.loads(json_entry)
+        table_entries.append(table_entry)
+    table_json = json.dumps(table_entries)
+    return table_json

tests/emulator_tests/servicebus_functions/servicebus_functions_sdk/function_app.py

@@ -0,0 +1,51 @@
+import json
+import jsonpickle
+
+import azure.functions as func
+import azurefunctions.extensions.bindings.servicebus as sb
+
+app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
+
+
+@app.route(route="put_message_sdk")
+@app.service_bus_queue_output(
+    arg_name="msg",
+    connection="AzureWebJobsServiceBusSDKConnectionString",
+    queue_name="testqueue")
+def put_message_sdk(req: func.HttpRequest, msg: func.Out[str]):
+    msg.set(req.get_body().decode('utf-8'))
+    return 'OK'
+
+
+@app.route(route="get_servicebus_triggered_sdk")
+@app.blob_input(arg_name="file",
+                path="python-worker-tests/test-servicebus-sdk-triggered.txt",
+                connection="AzureWebJobsStorage")
+def get_servicebus_triggered_sdk(req: func.HttpRequest,
+                                 file: func.InputStream) -> str:
+    return func.HttpResponse(
+        file.read().decode('utf-8'), mimetype='application/json')
+
+
+@app.service_bus_queue_trigger(
+    arg_name="msg",
+    connection="AzureWebJobsServiceBusSDKConnectionString",
+    queue_name="testqueue")
+@app.blob_output(arg_name="$return",
+                 path="python-worker-tests/test-servicebus-sdk-triggered.txt",
+                 connection="AzureWebJobsStorage")
+def servicebus_trigger_sdk(msg: sb.ServiceBusReceivedMessage) -> str:
+    msg_json = jsonpickle.encode(msg)
+    body_json = jsonpickle.encode(msg.body)
+    enqueued_time_json = jsonpickle.encode(msg.enqueued_time_utc)
+    lock_token_json = jsonpickle.encode(msg.lock_token)
+    result = json.dumps({
+        'message': msg_json,
+        'body': body_json,
+        'enqueued_time_utc': enqueued_time_json,
+        'lock_token': lock_token_json,
+        'message_id': msg.message_id,
+        'sequence_number': msg.sequence_number
+    })
+
+    return result

tests/extension_tests/deferred_bindings_tests/test_deferred_bindings_blob_functions.py → tests/emulator_tests/test_deferred_bindings_blob_functions.py

@@ -13,8 +13,8 @@ class TestDeferredBindingsBlobFunctions(testutils.WebHostTestCase):
 
     @classmethod
     def get_script_dir(cls):
-        return testutils.EXTENSION_TESTS_FOLDER / 'deferred_bindings_tests' / \
-            'deferred_bindings_blob_functions'
+        return testutils.EMULATOR_TESTS_FOLDER / 'blob_functions' / \
+            'blob_functions_sdk'
 
     @classmethod
     def get_libraries_to_install(cls):