backport to v2, add unit tests

Author: hallvictoria

runtimes/v2/azure_functions_runtime/bindings/meta.py

@@ -15,6 +15,7 @@
     CUSTOMER_PACKAGES_PATH,
     HTTP,
     HTTP_TRIGGER,
+    SERVICE_BUS_CLIENT_NAME
 )
 from ..utils.helpers import set_sdk_version
 
@@ -286,3 +287,39 @@ def get_deferred_raw_bindings(indexed_function, input_types):
     raw_bindings, bindings_logs = DEFERRED_BINDING_REGISTRY.get_raw_bindings(
         indexed_function, input_types)
     return raw_bindings, bindings_logs
+
+
+def get_settlement_client():
+    return DEFERRED_BINDING_REGISTRY.get(SERVICE_BUS_CLIENT_NAME).get_client()
+
+
+def validate_settlement_param(params: dict,
+                              bound_params: dict,
+                              annotations: dict) -> str:
+    """
+    Checks if the settlement client is enabled for a given function.
+
+    If there is more than one param that is not bound, return an empty string
+    indicating no settlement client support. This is a bad app.
+
+    If there is only one unbound param, check if it's a type that is supported
+    by the settlement client. If so, return the param_name, where param_name
+    is the name of the param that is supported. If not, return an empty string
+    indicating no settlement client support.
+
+    Note: If a param does not have a type annotation, it will be skipped and not
+    considered for settlement client support.
+    """
+    if len(set(params) - set(bound_params)) > 1:
+        return ''
+
+    # There is only one unbound param, check the type
+    settlement_param = next(iter(set(params) - set(bound_params)))
+    try:
+        param_type = annotations.get(settlement_param)
+        # Check if the type is a supported type for the settlement client
+        if DEFERRED_BINDING_REGISTRY.check_supported_grpc_client_type(param_type):
+            return settlement_param
+    except Exception:
+        param_type = None
+    return ''

runtimes/v2/azure_functions_runtime/functions.py

@@ -11,7 +11,8 @@
 from .bindings.meta import (has_implicit_output,
                             check_deferred_bindings_enabled,
                             check_output_type_annotation,
-                            check_input_type_annotation)
+                            check_input_type_annotation,
+                            validate_settlement_param)
 from .utils.constants import HTTP_TRIGGER
 from .utils.typing_inspect import is_generic_type, get_origin, get_args  # type: ignore
 
@@ -33,6 +34,7 @@ class FunctionInfo(typing.NamedTuple):
     has_return: bool
     is_http_func: bool
     deferred_bindings_enabled: bool
+    settlement_client_arg: str
 
     input_types: typing.Mapping[str, ParamTypeInfo]
     output_types: typing.Mapping[str, ParamTypeInfo]
@@ -138,7 +140,13 @@ def validate_function_params(params: dict, bound_params: dict,
                                  protos):
         logger.debug("Params: %s, BoundParams: %s, Annotations: %s, FuncName: %s",
                      params, bound_params, annotations, func_name)
+        settlement_client_arg = ''
         if set(params) - set(bound_params):
+            # Check for settlement client support for the missing parameters
+            settlement_client_arg = validate_settlement_param(
+                params, bound_params, annotations)
+            if settlement_client_arg != '':
+                params.pop(settlement_client_arg)
             raise FunctionLoadError(
                 func_name,
                 'Function parameter mismatch — the following trigger/input bindings '
@@ -278,7 +286,8 @@ def validate_function_params(params: dict, bound_params: dict,
                 output_types[param.name] = param_type_info
             else:
                 input_types[param.name] = param_type_info
-        return input_types, output_types, fx_deferred_bindings_enabled
+        return (input_types, output_types, fx_deferred_bindings_enabled,
+                settlement_client_arg)
 
     @staticmethod
     def get_function_return_type(annotations: dict, has_explicit_return: bool,
@@ -330,6 +339,7 @@ def add_func_to_registry_and_return_funcinfo(
             has_explicit_return: bool,
             has_implicit_return: bool,
             deferred_bindings_enabled: bool,
+            settlement_client_arg: str,
             input_types: typing.Dict[str, ParamTypeInfo],
             output_types: typing.Dict[str, ParamTypeInfo],
             return_type: str):
@@ -355,6 +365,7 @@ def add_func_to_registry_and_return_funcinfo(
             has_return=has_explicit_return or has_implicit_return,
             is_http_func=is_http_func,
             deferred_bindings_enabled=deferred_bindings_enabled,
+            settlement_client_arg=settlement_client_arg,
             input_types=input_types,
             output_types=output_types,
             return_type=return_type,
@@ -412,7 +423,8 @@ def add_indexed_function(self, function, protos):
                                                     func_name)
 
         (input_types, output_types,
-         deferred_bindings_enabled) = self.validate_function_params(
+         deferred_bindings_enabled,
+         settlement_client_arg) = self.validate_function_params(
             params,
             bound_params,
             annotations,
@@ -431,5 +443,6 @@ def add_indexed_function(self, function, protos):
                 func, func_name, function_id, func_dir,
                 requires_context, has_explicit_return,
                 has_implicit_return, deferred_bindings_enabled,
+                settlement_client_arg,
                 input_types, output_types,
                 return_type)

runtimes/v2/azure_functions_runtime/handle_event.py

@@ -22,6 +22,7 @@
 
 from .bindings.context import get_context
 from .bindings.meta import (from_incoming_proto,
+                            get_settlement_client,
                             is_trigger_binding,
                             load_binding_registry,
                             to_outgoing_param_binding,
@@ -224,6 +225,9 @@ async def invocation_request(request):
         if fi.requires_context:
             args['context'] = fi_context
 
+        if fi.settlement_client_arg != '':
+            args[fi.settlement_client_arg] = get_settlement_client()
+
         if fi.output_types:
             for name in fi.output_types:
                 args[name] = Out()

runtimes/v2/azure_functions_runtime/utils/constants.py

@@ -11,6 +11,7 @@
 MODULE_NOT_FOUND_TS_URL = "https://aka.ms/functions-modulenotfound"
 PYTHON_LANGUAGE_RUNTIME = "python"
 RETRY_POLICY = "retry_policy"
+SERVICE_BUS_CLIENT_NAME = "serviceBusClient"
 TRUE = "true"
 TRACEPARENT = "traceparent"
 TRACESTATE = "tracestate"

runtimes/v2/tests/unittests/test_deferred_bindings.py

@@ -16,6 +16,7 @@
                                                      ContainerClient,
                                                      StorageStreamDownloader)
 from azurefunctions.extensions.bindings.eventhub import EventData, EventDataConverter
+from azurefunctions.extensions.bindings.servicebus import ServiceBusMessageActions
 
 EVENTHUB_SAMPLE_CONTENT = b"\x00Sr\xc1\x8e\x08\xa3\x1bx-opt-sequence-number-epochT\xff\xa3\x15x-opt-sequence-numberU\x04\xa3\x0cx-opt-offset\x81\x00\x00\x00\x01\x00\x00\x010\xa3\x13x-opt-enqueued-time\x00\xa3\x1dcom.microsoft:datetime-offset\x81\x08\xddW\x05\xc3Q\xcf\x10\x00St\xc1I\x02\xa1\rDiagnostic-Id\xa1700-bdc3fde4889b4e907e0c9dcb46ff8d92-21f637af293ef13b-00\x00Su\xa0\x08message1"  # noqa: E501
 
@@ -98,3 +99,46 @@ async def test_check_deferred_bindings_enabled(self):
             ContainerClient, True), (True, True))
         self.assertEqual(meta.check_deferred_bindings_enabled(
             StorageStreamDownloader, True), (True, True))
+
+    async def test_valid_settlement_param():
+        params = {'param1', 'param2', 'param3'}
+        bound_params = {'param1', 'param2'}
+        annotations = {
+            'param1': func.InputStream,
+            'param2': func.Out[str],
+            'param3': ServiceBusMessageActions
+        }
+
+        settlement_client_arg = meta.validate_settlement_param(
+            params, bound_params, annotations)
+
+        assert settlement_client_arg == 'param3'
+
+    async def test_invalid_settlement_param():
+        params = {'param1', 'param2', 'param3'}
+        bound_params = {'param1', 'param2'}
+        annotations = {
+            'param1': func.InputStream,
+            'param2': func.Out[str],
+            'param3': str
+        }
+
+        settlement_client_arg = meta.validate_settlement_param(
+            params, bound_params, annotations)
+
+        assert settlement_client_arg == ''
+
+    async def test_invalid_settlement_param_multiple():
+        params = {'param1', 'param2', 'param3', 'param4'}
+        bound_params = {'param1', 'param2'}
+        annotations = {
+            'param1': func.InputStream,
+            'param2': func.Out[str],
+            'param3': ServiceBusMessageActions,
+            'param4': str
+        }
+
+        settlement_client_arg = meta.validate_settlement_param(
+            params, bound_params, annotations)
+
+        assert settlement_client_arg == ''

workers/azure_functions_worker/bindings/meta.py

@@ -10,6 +10,7 @@
     CUSTOMER_PACKAGES_PATH,
     HTTP,
     HTTP_TRIGGER,
+    SERVICE_BUS_CLIENT_NAME
 )
 from ..http_v2 import HttpV2Registry
 from ..logging import logger
@@ -317,7 +318,7 @@ def get_deferred_raw_bindings(indexed_function, input_types):
 
 
 def get_settlement_client():
-    return DEFERRED_BINDING_REGISTRY.get("serviceBusClient").get_client()
+    return DEFERRED_BINDING_REGISTRY.get(SERVICE_BUS_CLIENT_NAME).get_client()
 
 
 def validate_settlement_param(params: dict,

workers/azure_functions_worker/constants.py

@@ -115,3 +115,6 @@
     '3.11': '2027-04',
     '3.12': '2028-04'
 }
+
+# Service Bus Client Name
+SERVICE_BUS_CLIENT_NAME = "serviceBusClient"

workers/tests/extension_tests/deferred_bindings_tests/test_deferred_bindings.py

@@ -17,6 +17,7 @@
                                                          BlobClientConverter,
                                                          ContainerClient,
                                                          StorageStreamDownloader)
+    from azurefunctions.extensions.bindings.servicebus import ServiceBusMessageActions
 
 DEFERRED_BINDINGS_ENABLED_DIR = testutils.EXTENSION_TESTS_FOLDER / \
     'deferred_bindings_tests' / \
@@ -209,3 +210,46 @@ async def test_check_deferred_bindings_enabled(self):
                 ContainerClient, True), (True, True))
             self.assertEqual(meta.check_deferred_bindings_enabled(
                 StorageStreamDownloader, True), (True, True))
+
+    async def test_valid_settlement_param():
+        params = {'param1', 'param2', 'param3'}
+        bound_params = {'param1', 'param2'}
+        annotations = {
+            'param1': func.InputStream,
+            'param2': func.Out[str],
+            'param3': ServiceBusMessageActions
+        }
+
+        settlement_client_arg = meta.validate_settlement_param(
+            params, bound_params, annotations)
+
+        assert settlement_client_arg == 'param3'
+
+    async def test_invalid_settlement_param():
+        params = {'param1', 'param2', 'param3'}
+        bound_params = {'param1', 'param2'}
+        annotations = {
+            'param1': func.InputStream,
+            'param2': func.Out[str],
+            'param3': str
+        }
+
+        settlement_client_arg = meta.validate_settlement_param(
+            params, bound_params, annotations)
+
+        assert settlement_client_arg == ''
+
+    async def test_invalid_settlement_param_multiple():
+        params = {'param1', 'param2', 'param3', 'param4'}
+        bound_params = {'param1', 'param2'}
+        annotations = {
+            'param1': func.InputStream,
+            'param2': func.Out[str],
+            'param3': ServiceBusMessageActions,
+            'param4': str
+        }
+
+        settlement_client_arg = meta.validate_settlement_param(
+            params, bound_params, annotations)
+
+        assert settlement_client_arg == ''