diff --git a/azure-functions-extension-base/azure/functions/extension/base/__init__.py b/azure-functions-extension-base/azure/functions/extension/base/__init__.py index a306af9..e9e0542 100644 --- a/azure-functions-extension-base/azure/functions/extension/base/__init__.py +++ b/azure-functions-extension-base/azure/functions/extension/base/__init__.py @@ -3,38 +3,38 @@ from .meta import ( Datum, - _ConverterMeta, - _BaseConverter, InConverter, OutConverter, + _BaseConverter, + _ConverterMeta, get_binding_registry, ) from .sdkType import SdkType from .web import ( - WebServer, - WebApp, + HttpV2FeatureChecker, ModuleTrackerMeta, RequestTrackerMeta, + ResponseLabels, ResponseTrackerMeta, - HttpV2FeatureChecker, - ResponseLabels + WebApp, + WebServer, ) __all__ = [ - 'Datum', - '_ConverterMeta', - '_BaseConverter', - 'InConverter', - 'OutConverter', - 'SdkType', - 'get_binding_registry', - 'ModuleTrackerMeta', - 'RequestTrackerMeta', - 'ResponseTrackerMeta', - 'HttpV2FeatureChecker', - 'ResponseLabels', - 'WebServer', - 'WebApp' + "Datum", + "_ConverterMeta", + "_BaseConverter", + "InConverter", + "OutConverter", + "SdkType", + "get_binding_registry", + "ModuleTrackerMeta", + "RequestTrackerMeta", + "ResponseTrackerMeta", + "HttpV2FeatureChecker", + "ResponseLabels", + "WebServer", + "WebApp", ] __version__ = "1.0.0a2" diff --git a/azure-functions-extension-base/azure/functions/extension/base/meta.py b/azure-functions-extension-base/azure/functions/extension/base/meta.py index d984627..e3646fd 100644 --- a/azure-functions-extension-base/azure/functions/extension/base/meta.py +++ b/azure-functions-extension-base/azure/functions/extension/base/meta.py @@ -4,11 +4,9 @@ import abc import inspect import json +from typing import Any, Dict, List, Mapping, Optional, Tuple, Union -from . import utils -from . import sdkType - -from typing import Any, Dict, List, Optional, Mapping, Union, Tuple +from . import sdkType, utils class Datum: @@ -20,17 +18,17 @@ def __init__(self, value: Any, type: Optional[str]): def python_value(self) -> Any: if self.value is None or self.type is None: return None - elif self.type in ('bytes', 'string', 'int', 'double'): + elif self.type in ("bytes", "string", "int", "double"): return self.value - elif self.type == 'json': + elif self.type == "json": return json.loads(self.value) - elif self.type == 'collection_string': + elif self.type == "collection_string": return [v for v in self.value.string] - elif self.type == 'collection_bytes': + elif self.type == "collection_bytes": return [v for v in self.value.bytes] - elif self.type == 'collection_double': + elif self.type == "collection_double": return [v for v in self.value.double] - elif self.type == 'collection_sint64': + elif self.type == "collection_sint64": return [v for v in self.value.sint64] else: return self.value @@ -51,17 +49,17 @@ def __hash__(self): def __repr__(self): val_repr = repr(self.value) if len(val_repr) > 10: - val_repr = val_repr[:10] + '...' - return ''.format(self.type, val_repr) + val_repr = val_repr[:10] + "..." + return "".format(self.type, val_repr) class _ConverterMeta(abc.ABCMeta): _bindings: Dict[str, type] = {} - def __new__(mcls, name, bases, dct, *, - binding: Optional[str], - trigger: Optional[str] = None): + def __new__( + mcls, name, bases, dct, *, binding: Optional[str], trigger: Optional[str] = None + ): cls = super().__new__(mcls, name, bases, dct) cls._trigger = trigger # type: ignore if binding is None: @@ -69,9 +67,10 @@ def __new__(mcls, name, bases, dct, *, if binding in mcls._bindings: raise RuntimeError( - f'cannot register a converter for {binding!r} binding: ' - f'another converter for this binding has already been ' - f'registered') + f"cannot register a converter for {binding!r} binding: " + f"another converter for this binding has already been " + f"registered" + ) mcls._bindings[binding] = cls if trigger is not None: @@ -101,51 +100,59 @@ class _BaseConverter(metaclass=_ConverterMeta, binding=None): @classmethod def _decode_typed_data( - cls, data: Datum, *, - python_type: Union[type, Tuple[type, ...]], - context: str = 'data') -> Any: + cls, + data: Datum, + *, + python_type: Union[type, Tuple[type, ...]], + context: str = "data", + ) -> Any: if data is None: return None data_type = data.type - if data_type == 'model_binding_data': + if data_type == "model_binding_data": result = data.value elif data_type is None: return None else: - raise ValueError( - f'unsupported type of {context}: {data_type}') + raise ValueError(f"unsupported type of {context}: {data_type}") if not isinstance(result, python_type): if isinstance(python_type, (tuple, list, dict)): raise ValueError( - f'unexpected value type in {context}: ' - f'{type(result).__name__}, expected one of: ' - f'{", ".join(t.__name__ for t in python_type)}') + f"unexpected value type in {context}: " + f"{type(result).__name__}, expected one of: " + f'{", ".join(t.__name__ for t in python_type)}' + ) else: try: # Try coercing into the requested type result = python_type(result) except (TypeError, ValueError) as e: raise ValueError( - f'cannot convert value of {context} into ' - f'{python_type.__name__}: {e}') from None + f"cannot convert value of {context} into " + f"{python_type.__name__}: {e}" + ) from None return result @classmethod def _decode_trigger_metadata_field( - cls, trigger_metadata: Mapping[str, Datum], - field: str, *, - python_type: Union[type, Tuple[type, ...]]) \ - -> Any: + cls, + trigger_metadata: Mapping[str, Datum], + field: str, + *, + python_type: Union[type, Tuple[type, ...]], + ) -> Any: data = trigger_metadata.get(field) if data is None: return None else: return cls._decode_typed_data( - data, python_type=python_type, - context=f'field {field!r} in trigger metadata') + data, + python_type=python_type, + context=f"field {field!r} in trigger metadata", + ) class InConverter(_BaseConverter, binding=None): @@ -175,8 +182,7 @@ def check_output_type_annotation(cls, pytype: type) -> bool: @classmethod @abc.abstractmethod - def encode(cls, obj: Any, *, - expected_type: Optional[type]) -> Optional[Datum]: + def encode(cls, obj: Any, *, expected_type: Optional[type]) -> Optional[Datum]: raise NotImplementedError diff --git a/azure-functions-extension-base/azure/functions/extension/base/utils.py b/azure-functions-extension-base/azure/functions/extension/base/utils.py index 73a2465..a95f7d2 100644 --- a/azure-functions-extension-base/azure/functions/extension/base/utils.py +++ b/azure-functions-extension-base/azure/functions/extension/base/utils.py @@ -4,15 +4,14 @@ import inspect import json import re - from abc import ABC from enum import Enum from typing import Any, Callable, Dict, List, Optional from . import meta -SNAKE_CASE_RE = re.compile(r'^([a-zA-Z]+\d*_|_+[a-zA-Z\d])\w*$') -WORD_RE = re.compile(r'^([a-zA-Z]+\d*)$') +SNAKE_CASE_RE = re.compile(r"^([a-zA-Z]+\d*_|_+[a-zA-Z\d])\w*$") +WORD_RE = re.compile(r"^([a-zA-Z]+\d*)$") class StringifyEnum(Enum): @@ -38,10 +37,8 @@ def __new__(mcs, name, bases, dct): value fields. It is needed for enabling binding param optionality. """ cls = super().__new__(mcs, name, bases, dct) - setattr(cls, '__init__', - cls.add_to_dict(getattr(cls, '__init__'))) - setattr(cls, 'get_dict_repr', - cls.skip_none(getattr(cls, 'get_dict_repr'))) + setattr(cls, "__init__", cls.add_to_dict(getattr(cls, "__init__"))) + setattr(cls, "get_dict_repr", cls.skip_none(getattr(cls, "get_dict_repr"))) return cls @staticmethod @@ -57,8 +54,9 @@ def add_to_dict(func: Callable[..., Any]): def wrapper(*args, **kwargs): if args is None or len(args) == 0: raise ValueError( - f'{func.__name__} has no args. Please ensure func is an ' - f'object method.') + f"{func.__name__} has no args. Please ensure func is an " + f"object method." + ) func(*args, **kwargs) @@ -70,7 +68,7 @@ def wrapper(*args, **kwargs): if not hasattr(self, key): setattr(self, key, kwargs[key]) - setattr(self, 'init_params', init_params) + setattr(self, "init_params", init_params) return wrapper @@ -82,8 +80,7 @@ def clean_nones(value): the result as a new dictionary or list. """ if isinstance(value, list): - return [BuildDictMeta.clean_nones(x) for x in value if - x is not None] + return [BuildDictMeta.clean_nones(x) for x in value if x is not None] elif isinstance(value, dict): return { key: BuildDictMeta.clean_nones(val) @@ -97,6 +94,7 @@ def clean_nones(value): # Enums class BindingDirection(StringifyEnum): """Direction of the binding used in function.json""" + IN = 0 """Input binding direction.""" OUT = 1 @@ -107,6 +105,7 @@ class BindingDirection(StringifyEnum): class DataType(StringifyEnum): """Data type of the binding used in function.json""" + """Parse binding argument as undefined.""" UNDEFINED = 0 """Parse binding argument as string.""" @@ -123,30 +122,34 @@ class Binding(ABC): every binding, the only restriction is ***ENSURE*** __init__ parameter names of any binding class are snake case form of corresponding attribute in function.json when new binding classes are created. - Ref: https://aka.ms/azure-function-binding-http """ + Ref: https://aka.ms/azure-function-binding-http""" - EXCLUDED_INIT_PARAMS = {'self', 'kwargs', 'type', 'data_type', 'direction'} + EXCLUDED_INIT_PARAMS = {"self", "kwargs", "type", "data_type", "direction"} @staticmethod def get_binding_name() -> str: pass - def __init__(self, name: str, - direction: BindingDirection, - data_type: Optional[DataType] = None, - type: Optional[str] = None): # NoQa + def __init__( + self, + name: str, + direction: BindingDirection, + data_type: Optional[DataType] = None, + type: Optional[str] = None, + ): # NoQa # For natively supported bindings, get_binding_name is always # implemented, and for generic bindings, type is a required argument # in decorator functions. - self.type = self.get_binding_name() \ - if self.get_binding_name() is not None else type + self.type = ( + self.get_binding_name() if self.get_binding_name() is not None else type + ) self.name = name self._direction = direction self._data_type = data_type self._dict = { "direction": self._direction, "dataType": self._data_type, - "type": self.type + "type": self.type, } @property @@ -166,7 +169,7 @@ def get_dict_repr(binding, input_types) -> Dict: :return: Dictionary representation of the binding. """ - params = list(dict.fromkeys(getattr(binding, 'init_params', []))) + params = list(dict.fromkeys(getattr(binding, "init_params", []))) for p in params: if p not in Binding.EXCLUDED_INIT_PARAMS: binding._dict[to_camel_case(p)] = getattr(binding, p, None) @@ -175,10 +178,13 @@ def get_dict_repr(binding, input_types) -> Dict: # 1. check if the binding is a supported type (blob, blobTrigger) # 2. check if the binding is an input binding # 3. check if the defined type is an SdkType - if (binding.type in meta._ConverterMeta._bindings - and binding.direction == 0 - and meta._ConverterMeta.check_supported_type( - input_types.get(binding.name).pytype)): + if ( + binding.type in meta._ConverterMeta._bindings + and binding.direction == 0 + and meta._ConverterMeta.check_supported_type( + input_types.get(binding.name).pytype + ) + ): binding._dict["properties"] = {"SupportsDeferredBinding": True} # if it isn't, we set the flag to false else: @@ -189,15 +195,15 @@ def get_dict_repr(binding, input_types) -> Dict: def to_camel_case(snake_case_str: str): if snake_case_str is None or len(snake_case_str) == 0: - raise ValueError( - f"Please ensure arg name {snake_case_str} is not empty!") + raise ValueError(f"Please ensure arg name {snake_case_str} is not empty!") if not is_snake_case(snake_case_str) and not is_word(snake_case_str): raise ValueError( f"Please ensure {snake_case_str} is a word or snake case " - f"string with underscore as separator.") - words = snake_case_str.split('_') - return words[0] + ''.join([ele.title() for ele in words[1:]]) + f"string with underscore as separator." + ) + words = snake_case_str.split("_") + return words[0] + "".join([ele.title() for ele in words[1:]]) def is_snake_case(input_string: str) -> bool: @@ -233,6 +239,7 @@ def is_word(input_string: str) -> bool: def get_raw_bindings(indexed_function, input_types) -> List[str]: - return [json.dumps(Binding.get_dict_repr(b, input_types), - cls=StringifyEnumJsonEncoder) - for b in indexed_function._bindings] + return [ + json.dumps(Binding.get_dict_repr(b, input_types), cls=StringifyEnumJsonEncoder) + for b in indexed_function._bindings + ] diff --git a/azure-functions-extension-base/azure/functions/extension/base/web.py b/azure-functions-extension-base/azure/functions/extension/base/web.py index 56d8ac9..389485a 100644 --- a/azure-functions-extension-base/azure/functions/extension/base/web.py +++ b/azure-functions-extension-base/azure/functions/extension/base/web.py @@ -1,6 +1,6 @@ +import inspect from abc import abstractmethod from enum import Enum -import inspect from typing import Callable base_extension_module = __name__ @@ -12,13 +12,15 @@ class ModuleTrackerMeta(type): def __new__(cls, name, bases, dct, **kwargs): new_class = super().__new__(cls, name, bases, dct) - new_module = dct.get('__module__') + new_module = dct.get("__module__") if new_module != base_extension_module: if cls._module is None: cls._module = new_module elif cls._module != new_module: - raise Exception(f'Only one web extension package shall be imported, ' - f'{cls._module} and {new_module} are imported') + raise Exception( + f"Only one web extension package shall be imported, " + f"{cls._module} and {new_module} are imported" + ) return new_class @classmethod @@ -36,14 +38,16 @@ class RequestTrackerMeta(type): def __new__(cls, name, bases, dct, **kwargs): new_class = super().__new__(cls, name, bases, dct) - request_type = dct.get('request_type') + request_type = dct.get("request_type") if request_type is None: - raise Exception(f'Request type not provided for class {name}') + raise Exception(f"Request type not provided for class {name}") if cls._request_type is not None and cls._request_type != request_type: - raise Exception(f'Only one request type shall be recorded for class {name} ' - f'but found {cls._request_type} and {request_type}') + raise Exception( + f"Only one request type shall be recorded for class {name} " + f"but found {cls._request_type} and {request_type}" + ) cls._request_type = request_type return new_class @@ -55,7 +59,9 @@ def get_request_type(cls): @classmethod def check_type(cls, pytype: type) -> bool: if pytype is not None and inspect.isclass(pytype): - return cls._request_type is not None and issubclass(pytype, cls._request_type) + return cls._request_type is not None and issubclass( + pytype, cls._request_type + ) return False @@ -65,16 +71,21 @@ class ResponseTrackerMeta(type): def __new__(cls, name, bases, dct, **kwargs): new_class = super().__new__(cls, name, bases, dct) - label = dct.get('label') - response_type = dct.get('response_type') + label = dct.get("label") + response_type = dct.get("response_type") if label is None: - raise Exception(f'Response label not provided for class {name}') + raise Exception(f"Response label not provided for class {name}") if response_type is None: - raise Exception(f'Response type not provided for class {name}') - if cls._response_types.get(label) is not None and cls._response_types.get(label) != response_type: - raise Exception(f'Only one response type shall be recorded for class {name} ' - f'but found {cls._response_types.get(label)} and {response_type}') + raise Exception(f"Response type not provided for class {name}") + if ( + cls._response_types.get(label) is not None + and cls._response_types.get(label) != response_type + ): + raise Exception( + f"Only one response type shall be recorded for class {name} " + f"but found {cls._response_types.get(label)} and {response_type}" + ) cls._response_types[label] = response_type @@ -91,10 +102,13 @@ def get_response_type(cls, label): @classmethod def check_type(cls, pytype: type) -> bool: if pytype is not None and inspect.isclass(pytype): - return cls._response_types is not None and any(issubclass(pytype, response_type) - for response_type in cls._response_types.values()) + return cls._response_types is not None and any( + issubclass(pytype, response_type) + for response_type in cls._response_types.values() + ) return False + class WebApp(metaclass=ModuleTrackerMeta): @abstractmethod def route(self, func: Callable): @@ -115,18 +129,20 @@ def __init__(self, hostname, port, web_app: WebApp): async def serve(self): pass + class HttpV2FeatureChecker: @staticmethod def http_v2_enabled(): return ModuleTrackerMeta.module_imported() + class ResponseLabels(Enum): - STANDARD = 'standard' - STREAMING = 'streaming' - FILE = 'file' - HTML = 'html' - JSON = 'json' - ORJSON = 'orjson' - PLAIN_TEXT = 'plain_text' - REDIRECT = 'redirect' - UJSON = 'ujson' + STANDARD = "standard" + STREAMING = "streaming" + FILE = "file" + HTML = "html" + JSON = "json" + ORJSON = "orjson" + PLAIN_TEXT = "plain_text" + REDIRECT = "redirect" + UJSON = "ujson" diff --git a/azure-functions-extension-base/tests/__init__.py b/azure-functions-extension-base/tests/__init__.py index d138d18..528a01b 100644 --- a/azure-functions-extension-base/tests/__init__.py +++ b/azure-functions-extension-base/tests/__init__.py @@ -11,11 +11,10 @@ def suite(): test_loader = unittest.TestLoader() - return test_loader.discover( - os.path.dirname(__file__), pattern='test_*.py') + return test_loader.discover(os.path.dirname(__file__), pattern="test_*.py") -if __name__ == '__main__': +if __name__ == "__main__": runner = unittest.runner.TextTestRunner() result = runner.run(suite()) sys.exit(not result.wasSuccessful()) diff --git a/azure-functions-extension-base/tests/test_meta.py b/azure-functions-extension-base/tests/test_meta.py index db5155f..1d7c036 100644 --- a/azure-functions-extension-base/tests/test_meta.py +++ b/azure-functions-extension-base/tests/test_meta.py @@ -1,7 +1,7 @@ # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. -from typing import Mapping, List import unittest +from typing import List, Mapping from unittest.mock import patch from azure.functions.extension.base import meta, sdkType @@ -23,7 +23,7 @@ def test_datum_single_level_python_value(self): self.assertEqual(datum.python_type, bytes) datum = meta.Datum(value="awesome string", type="string") - self.assertEqual(datum.python_value, 'awesome string') + self.assertEqual(datum.python_value, "awesome string") self.assertEqual(datum.python_type, str) datum = meta.Datum(value=42, type="int") @@ -41,88 +41,86 @@ def test_datum_collections_python_value(self): class DatumCollectionString: def __init__(self, *args: List[str]): self.string = args - datum = meta.Datum(value=DatumCollectionString("string 1", - "string 2"), - type="collection_string") + + datum = meta.Datum( + value=DatumCollectionString("string 1", "string 2"), + type="collection_string", + ) self.assertListEqual(datum.python_value, ["string 1", "string 2"]) self.assertEqual(datum.python_type, list) class DatumCollectionBytes: def __init__(self, *args: List[bytes]): self.bytes = args - datum = meta.Datum(value=DatumCollectionBytes(b"bytes 1", b"bytes 2"), - type="collection_bytes") + + datum = meta.Datum( + value=DatumCollectionBytes(b"bytes 1", b"bytes 2"), type="collection_bytes" + ) self.assertListEqual(datum.python_value, [b"bytes 1", b"bytes 2"]) self.assertEqual(datum.python_type, list) class DatumCollectionDouble: def __init__(self, *args: List[bytes]): self.double = args - datum = meta.Datum(value=DatumCollectionDouble(43.2103, 45.601), - type="collection_double") + + datum = meta.Datum( + value=DatumCollectionDouble(43.2103, 45.601), type="collection_double" + ) self.assertListEqual(datum.python_value, [43.2103, 45.601]) self.assertEqual(datum.python_type, list) class DatumCollectionSint64: def __init__(self, *args: List[int]): self.sint64 = args - datum = meta.Datum(value=DatumCollectionSint64(1234567, 8901234), - type="collection_sint64") + + datum = meta.Datum( + value=DatumCollectionSint64(1234567, 8901234), type="collection_sint64" + ) self.assertListEqual(datum.python_value, [1234567, 8901234]) self.assertEqual(datum.python_type, list) def test_datum_json_python_value(self): # None - datum = meta.Datum(value='null', - type="json") + datum = meta.Datum(value="null", type="json") self.assertEqual(datum.python_value, None) self.assertEqual(datum.python_type, type(None)) # Int - datum = meta.Datum(value='123', - type="json") + datum = meta.Datum(value="123", type="json") self.assertEqual(datum.python_value, 123) self.assertEqual(datum.python_type, int) # Float - datum = meta.Datum(value='456.789', - type="json") + datum = meta.Datum(value="456.789", type="json") self.assertEqual(datum.python_value, 456.789) self.assertEqual(datum.python_type, float) # String - datum = meta.Datum(value='"string in json"', - type="json") + datum = meta.Datum(value='"string in json"', type="json") self.assertEqual(datum.python_value, "string in json") self.assertEqual(datum.python_type, str) # List - datum = meta.Datum(value='["a", "b", "c"]', - type="json") + datum = meta.Datum(value='["a", "b", "c"]', type="json") self.assertListEqual(datum.python_value, ["a", "b", "c"]) self.assertEqual(datum.python_type, list) # Object - datum = meta.Datum(value='{"name": "awesome", "value": "cool"}', - type="json") - self.assertDictEqual(datum.python_value, { - "name": "awesome", - "value": "cool"}) + datum = meta.Datum(value='{"name": "awesome", "value": "cool"}', type="json") + self.assertDictEqual(datum.python_value, {"name": "awesome", "value": "cool"}) self.assertEqual(datum.python_type, dict) # Should ignore Newlines and Spaces - datum = meta.Datum(value='{ "name" : "awesome",\n "value": "cool"\n}', - type="json") - self.assertDictEqual(datum.python_value, { - "name": "awesome", - "value": "cool"}) + datum = meta.Datum( + value='{ "name" : "awesome",\n "value": "cool"\n}', type="json" + ) + self.assertDictEqual(datum.python_value, {"name": "awesome", "value": "cool"}) self.assertEqual(datum.python_type, dict) def test_equals(self): str_datum = meta.Datum(value="awesome string", type="string") str_datum_copy = meta.Datum(value="awesome string", type="string") - str_datum_wrong_copy = meta.Datum(value="not awesome string", - type="string") + str_datum_wrong_copy = meta.Datum(value="not awesome string", type="string") self.assertFalse(str_datum.__eq__(dict)) self.assertTrue(str_datum.__eq__(str_datum_copy)) self.assertFalse(str_datum.__eq__(str_datum_wrong_copy)) @@ -137,20 +135,18 @@ def test_repr(self): self.assertEqual(str_datum.__repr__(), "") long_str_datum = meta.Datum(value="awesome string", type="string") - self.assertEqual(long_str_datum.__repr__(), - "") + self.assertEqual(long_str_datum.__repr__(), "") def test_registry(self): registry = meta.get_binding_registry() self.assertIsInstance(registry, type(meta._ConverterMeta)) - self.assertIsNone(registry.get('test')) + self.assertIsNone(registry.get("test")) class MockIndexedFunction: _bindings = {} _trigger = None - self.assertEqual(registry.get_raw_bindings(MockIndexedFunction, []), - []) + self.assertEqual(registry.get_raw_bindings(MockIndexedFunction, []), []) self.assertFalse(registry.check_supported_type(None)) self.assertFalse(registry.check_supported_type("hello")) @@ -160,78 +156,86 @@ class MockIndexedFunction: def test_decode_typed_data(self): # Case 1: data is None - self.assertIsNone(meta._BaseConverter._decode_typed_data( - data=None, python_type=str)) + self.assertIsNone( + meta._BaseConverter._decode_typed_data(data=None, python_type=str) + ) # Case 2: data.type is model_binding_data - datum_mbd = meta.Datum(value='{}', type='model_binding_data') - self.assertEqual(meta._BaseConverter._decode_typed_data( - datum_mbd, python_type=str), '{}') + datum_mbd = meta.Datum(value="{}", type="model_binding_data") + self.assertEqual( + meta._BaseConverter._decode_typed_data(datum_mbd, python_type=str), "{}" + ) # Case 3: data.type is None - datum_none = meta.Datum(value='{}', type=None) - self.assertIsNone(meta._BaseConverter._decode_typed_data( - datum_none, python_type=str)) + datum_none = meta.Datum(value="{}", type=None) + self.assertIsNone( + meta._BaseConverter._decode_typed_data(datum_none, python_type=str) + ) # Case 4: data.type is unsupported - datum_unsupp = meta.Datum(value='{}', type=dict) + datum_unsupp = meta.Datum(value="{}", type=dict) with self.assertRaises(ValueError): - meta._BaseConverter._decode_typed_data( - datum_unsupp, python_type=str) + meta._BaseConverter._decode_typed_data(datum_unsupp, python_type=str) # Case 5: can't coerce - datum_coerce_fail = meta.Datum(value='{}', type='model_binding_data') + datum_coerce_fail = meta.Datum(value="{}", type="model_binding_data") with self.assertRaises(ValueError): meta._BaseConverter._decode_typed_data( - datum_coerce_fail, python_type=(tuple, list, dict)) + datum_coerce_fail, python_type=(tuple, list, dict) + ) # Case 6: attempt coerce & fail - datum_attempt_coerce = meta.Datum(value=1, type='model_binding_data') + datum_attempt_coerce = meta.Datum(value=1, type="model_binding_data") with self.assertRaises(ValueError): meta._BaseConverter._decode_typed_data( - datum_attempt_coerce, python_type=dict) + datum_attempt_coerce, python_type=dict + ) # Case 7: attempt to coerce and pass - datum_coerce_pass = meta.Datum(value=1, type='model_binding_data') - self.assertEquals(meta._BaseConverter._decode_typed_data( - datum_coerce_pass, python_type=str), '1') + datum_coerce_pass = meta.Datum(value=1, type="model_binding_data") + self.assertEquals( + meta._BaseConverter._decode_typed_data(datum_coerce_pass, python_type=str), + "1", + ) def test_decode_trigger_metadata_field(self): - datum_mbd = meta.Datum(value='{}', type='model_binding_data') - mock_trigger_metadata = {'key': datum_mbd} + datum_mbd = meta.Datum(value="{}", type="model_binding_data") + mock_trigger_metadata = {"key": datum_mbd} - self.assertIsNone(meta._BaseConverter._decode_trigger_metadata_field( - trigger_metadata=mock_trigger_metadata, - field="fakeKey", - python_type=str)) + self.assertIsNone( + meta._BaseConverter._decode_trigger_metadata_field( + trigger_metadata=mock_trigger_metadata, field="fakeKey", python_type=str + ) + ) self.assertEqual( meta._BaseConverter._decode_trigger_metadata_field( - trigger_metadata=mock_trigger_metadata, - field="key", - python_type=str), - '{}') - - @patch("azure.functions.extension.base.meta." - "InConverter.__abstractmethods__", set()) + trigger_metadata=mock_trigger_metadata, field="key", python_type=str + ), + "{}", + ) + + @patch( + "azure.functions.extension.base.meta." "InConverter.__abstractmethods__", set() + ) def test_in_converter(self): - class MockInConverter(meta.InConverter, binding='test1'): - _sample = '' + class MockInConverter(meta.InConverter, binding="test1"): + _sample = "" mock_converter = MockInConverter() - self.assertIsNone(mock_converter.check_input_type_annotation( - pytype=str)) + self.assertIsNone(mock_converter.check_input_type_annotation(pytype=str)) with self.assertRaises(NotImplementedError): mock_converter.decode(data=None, trigger_metadata={}) self.assertFalse(mock_converter.has_implicit_output()) - @patch("azure.functions.extension.base.meta." - "OutConverter.__abstractmethods__", set()) + @patch( + "azure.functions.extension.base.meta." "OutConverter.__abstractmethods__", set() + ) def test_out_converter(self): - class MockOutConverter(meta.OutConverter, binding='test2'): - _sample = '' + class MockOutConverter(meta.OutConverter, binding="test2"): + _sample = "" mock_converter = MockOutConverter() mock_converter.check_output_type_annotation(pytype=str) @@ -243,42 +247,46 @@ def test_get_registry(self): registry = meta.get_binding_registry() self.assertEqual(registry, meta._ConverterMeta) - @patch("azure.functions.extension.base.meta." - "OutConverter.__abstractmethods__", set()) + @patch( + "azure.functions.extension.base.meta." "OutConverter.__abstractmethods__", set() + ) def test_converter_meta(self): class BindingNoneConverter(meta.OutConverter, binding=None): - _sample = '' + _sample = "" registry = meta.get_binding_registry() self.assertEqual(len(registry._bindings), 0) - class BindingBlobConverter(meta.OutConverter, binding='blob'): - _sample = '' + class BindingBlobConverter(meta.OutConverter, binding="blob"): + _sample = "" registry = meta.get_binding_registry() self.assertEqual(len(registry._bindings), 1) - self.assertIsNotNone(registry._bindings.get('blob')) - self.assertEqual(registry._bindings.get('blob'), BindingBlobConverter) + self.assertIsNotNone(registry._bindings.get("blob")) + self.assertEqual(registry._bindings.get("blob"), BindingBlobConverter) with self.assertRaises(RuntimeError): - class BindingBlob2Converter(meta.OutConverter, binding='blob'): - _sample = '' + + class BindingBlob2Converter(meta.OutConverter, binding="blob"): + _sample = "" registry = meta.get_binding_registry() self.assertEqual(len(registry._bindings), 1) - self.assertIsNotNone(registry._bindings.get('blob')) - self.assertEqual(registry._bindings.get('blob'), BindingBlobConverter) + self.assertIsNotNone(registry._bindings.get("blob")) + self.assertEqual(registry._bindings.get("blob"), BindingBlobConverter) - class BindingServiceBusConverter(meta.OutConverter, - binding='serviceBus', - trigger='serviceBusTrigger'): - _sample = '' + class BindingServiceBusConverter( + meta.OutConverter, binding="serviceBus", trigger="serviceBusTrigger" + ): + _sample = "" registry = meta.get_binding_registry() self.assertEqual(len(registry._bindings), 3) - self.assertIsNotNone(registry._bindings.get('serviceBus')) - self.assertEqual(registry._bindings.get('serviceBus'), - BindingServiceBusConverter) - self.assertIsNotNone(registry._bindings.get('serviceBusTrigger')) - self.assertEqual(registry._bindings.get('serviceBusTrigger'), - BindingServiceBusConverter) + self.assertIsNotNone(registry._bindings.get("serviceBus")) + self.assertEqual( + registry._bindings.get("serviceBus"), BindingServiceBusConverter + ) + self.assertIsNotNone(registry._bindings.get("serviceBusTrigger")) + self.assertEqual( + registry._bindings.get("serviceBusTrigger"), BindingServiceBusConverter + ) diff --git a/azure-functions-extension-base/tests/test_sdk_type.py b/azure-functions-extension-base/tests/test_sdk_type.py index 21e25f8..68c863f 100644 --- a/azure-functions-extension-base/tests/test_sdk_type.py +++ b/azure-functions-extension-base/tests/test_sdk_type.py @@ -8,15 +8,15 @@ class TestSdkType(unittest.TestCase): def test_init(self): - data_populated = sdkType.SdkType(data={'key': 'value'}) - self.assertEqual(data_populated._data, {'key': 'value'}) + data_populated = sdkType.SdkType(data={"key": "value"}) + self.assertEqual(data_populated._data, {"key": "value"}) data_empty = sdkType.SdkType() self.assertEqual(data_empty._data, {}) def test_get_sdk_type(self): class MockSdkType(sdkType.SdkType): - _sample = '' + _sample = "" mock_type = MockSdkType() self.assertIsNone(mock_type.get_sdk_type()) diff --git a/azure-functions-extension-base/tests/test_utils.py b/azure-functions-extension-base/tests/test_utils.py index 659fbc7..6227225 100644 --- a/azure-functions-extension-base/tests/test_utils.py +++ b/azure-functions-extension-base/tests/test_utils.py @@ -2,10 +2,9 @@ # Licensed under the MIT License. import unittest - from abc import ABC -from azure.functions.extension.base import meta, utils, sdkType +from azure.functions.extension.base import meta, sdkType, utils class MockParamTypeInfo: @@ -20,16 +19,15 @@ def __init__(self, bindings: utils.Binding): class MockInitParams(utils.Binding): - def __init__(self, name, direction, data_type, - type, init_params): - self.type = 'blob' + def __init__(self, name, direction, data_type, type, init_params): + self.type = "blob" self.name = name self._direction = direction self._data_type = data_type self._dict = { "direction": self._direction, "dataType": self._data_type, - "type": self.type + "type": self.type, } self.init_params = init_params @@ -41,93 +39,121 @@ def test_get_dict_repr_sdk(self): meta._ConverterMeta._bindings = {"blob"} # Create test binding - mock_blob = utils.Binding(name="client", - direction=utils.BindingDirection.IN, - data_type=None, type='blob') + mock_blob = utils.Binding( + name="client", + direction=utils.BindingDirection.IN, + data_type=None, + type="blob", + ) # Create test input_types dict - mock_input_types = {"client": MockParamTypeInfo( - binding_name='blobTrigger', pytype=sdkType.SdkType)} + mock_input_types = { + "client": MockParamTypeInfo( + binding_name="blobTrigger", pytype=sdkType.SdkType + ) + } # Create test indexed_function mock_indexed_functions = MockFunction(bindings=[mock_blob]) - dict_repr = utils.get_raw_bindings(mock_indexed_functions, - mock_input_types) - self.assertEqual(dict_repr, - ['{"direction": "IN", ' - '"dataType": null, "type": "blob", ' - '"properties": ' - '{"SupportsDeferredBinding": true}}']) + dict_repr = utils.get_raw_bindings(mock_indexed_functions, mock_input_types) + self.assertEqual( + dict_repr, + [ + '{"direction": "IN", ' + '"dataType": null, "type": "blob", ' + '"properties": ' + '{"SupportsDeferredBinding": true}}' + ], + ) def test_get_dict_repr_non_sdk(self): # Create mock blob meta._ConverterMeta._bindings = {"blob"} # Create test binding - mock_blob = utils.Binding(name="blob", - direction=utils.BindingDirection.IN, - data_type=None, type='blob') + mock_blob = utils.Binding( + name="blob", + direction=utils.BindingDirection.IN, + data_type=None, + type="blob", + ) # Create test input_types dict - mock_input_types = {"blob": MockParamTypeInfo( - binding_name='blobTrigger', pytype=bytes)} + mock_input_types = { + "blob": MockParamTypeInfo(binding_name="blobTrigger", pytype=bytes) + } # Create test indexed_function mock_indexed_functions = MockFunction(bindings=[mock_blob]) - dict_repr = utils.get_raw_bindings(mock_indexed_functions, - mock_input_types) - self.assertEqual(dict_repr, - ['{"direction": "IN", ' - '"dataType": null, "type": "blob", ' - '"properties": ' - '{"SupportsDeferredBinding": false}}']) + dict_repr = utils.get_raw_bindings(mock_indexed_functions, mock_input_types) + self.assertEqual( + dict_repr, + [ + '{"direction": "IN", ' + '"dataType": null, "type": "blob", ' + '"properties": ' + '{"SupportsDeferredBinding": false}}' + ], + ) def test_get_dict_repr_init_params(self): # Create mock blob meta._ConverterMeta._bindings = {"blob"} # Create test binding - mock_blob = MockInitParams(name="client", - direction=utils.BindingDirection.IN, - data_type=None, type='blob', - init_params=['test', 'type', 'direction']) + mock_blob = MockInitParams( + name="client", + direction=utils.BindingDirection.IN, + data_type=None, + type="blob", + init_params=["test", "type", "direction"], + ) # Create test input_types dict - mock_input_types = {"client": MockParamTypeInfo( - binding_name='blobTrigger', pytype=sdkType.SdkType)} + mock_input_types = { + "client": MockParamTypeInfo( + binding_name="blobTrigger", pytype=sdkType.SdkType + ) + } # Create test indexed_function mock_indexed_functions = MockFunction(bindings=[mock_blob]) - dict_repr = utils.get_raw_bindings(mock_indexed_functions, - mock_input_types) - self.assertEqual(dict_repr, - ['{"direction": "IN", "dataType": null, ' - '"type": "blob", "test": null, "properties": ' - '{"SupportsDeferredBinding": true}}']) + dict_repr = utils.get_raw_bindings(mock_indexed_functions, mock_input_types) + self.assertEqual( + dict_repr, + [ + '{"direction": "IN", "dataType": null, ' + '"type": "blob", "test": null, "properties": ' + '{"SupportsDeferredBinding": true}}' + ], + ) def test_binding_data_type(self): - mock_blob = utils.Binding(name="blob", - direction=utils.BindingDirection.IN, - data_type=None, type='blob') + mock_blob = utils.Binding( + name="blob", + direction=utils.BindingDirection.IN, + data_type=None, + type="blob", + ) self.assertIsNone(mock_blob.data_type) - mock_data_type = utils.Binding(name="blob", - direction=utils.BindingDirection.IN, - data_type=utils.DataType.STRING, - type='blob') + mock_data_type = utils.Binding( + name="blob", + direction=utils.BindingDirection.IN, + data_type=utils.DataType.STRING, + type="blob", + ) self.assertEqual(mock_data_type.data_type, 1) def test_to_camel_case(self): test_str = "" - self.assertRaises(ValueError, - utils.to_camel_case, test_str) + self.assertRaises(ValueError, utils.to_camel_case, test_str) test_str = "1iAmNotAWord" - self.assertRaises(ValueError, - utils.to_camel_case, test_str) + self.assertRaises(ValueError, utils.to_camel_case, test_str) test_str = utils.to_camel_case("string_in_correct_format") self.assertEqual(test_str, "stringInCorrectFormat") @@ -164,23 +190,25 @@ def test_snake_case_to_camel_case_single(self): def test_snake_case_to_camel_case_empty_str(self): with self.assertRaises(ValueError) as err: utils.to_camel_case("") - self.assertEqual(err.exception.args[0], - 'Please ensure arg name is not ' - 'empty!') + self.assertEqual( + err.exception.args[0], "Please ensure arg name is not " "empty!" + ) def test_snake_case_to_camel_case_none(self): with self.assertRaises(ValueError) as err: utils.to_camel_case(None) - self.assertEqual(err.exception.args[0], - 'Please ensure arg name None is not ' - 'empty!') + self.assertEqual( + err.exception.args[0], "Please ensure arg name None is not " "empty!" + ) def test_snake_case_to_camel_case_not_one_word_nor_snake_case(self): with self.assertRaises(ValueError) as err: utils.to_camel_case("data-type") - self.assertEqual(err.exception.args[0], - 'Please ensure data-type is a word or snake case ' - 'string with underscore as separator.') + self.assertEqual( + err.exception.args[0], + "Please ensure data-type is a word or snake case " + "string with underscore as separator.", + ) def test_is_snake_case_letters_only(self): self.assertTrue(utils.is_snake_case("dataType_foo")) @@ -216,33 +244,39 @@ def test_clean_nones_none(self): self.assertEqual(utils.BuildDictMeta.clean_nones(None), None) def test_clean_nones_nested(self): - self.assertEqual(utils.BuildDictMeta.clean_nones( - { - "hello": None, - "hello2": ["dummy1", None, "dummy2", ["dummy3", None], - {"hello3": None}], - "hello4": { - "dummy5": "pass1", - "dummy6": None + self.assertEqual( + utils.BuildDictMeta.clean_nones( + { + "hello": None, + "hello2": [ + "dummy1", + None, + "dummy2", + ["dummy3", None], + {"hello3": None}, + ], + "hello4": {"dummy5": "pass1", "dummy6": None}, } - }), + ), { "hello2": ["dummy1", "dummy2", ["dummy3"], {}], - "hello4": {"dummy5": "pass1"} - } # NoQA + "hello4": {"dummy5": "pass1"}, + }, # NoQA ) def test_add_to_dict_no_args(self): with self.assertRaises(ValueError) as err: + @utils.BuildDictMeta.add_to_dict def dummy(): pass dummy() - self.assertEqual(err.exception.args[0], - 'dummy has no args. Please ensure func is an object ' - 'method.') + self.assertEqual( + err.exception.args[0], + "dummy has no args. Please ensure func is an object " "method.", + ) def test_add_to_dict_valid(self): class TestDict: @@ -251,11 +285,12 @@ def __init__(self, arg1, arg2, **kwargs): self.arg1 = arg1 self.arg2 = arg2 - test_obj = TestDict('val1', 'val2', dummy1="dummy1", dummy2="dummy2") + test_obj = TestDict("val1", "val2", dummy1="dummy1", dummy2="dummy2") - self.assertCountEqual(getattr(test_obj, 'init_params'), - {'self', 'arg1', 'arg2', 'kwargs', 'dummy1', - 'dummy2'}) + self.assertCountEqual( + getattr(test_obj, "init_params"), + {"self", "arg1", "arg2", "kwargs", "dummy1", "dummy2"}, + ) self.assertEqual(getattr(test_obj, "arg1", None), "val1") self.assertEqual(getattr(test_obj, "arg2", None), "val2") self.assertEqual(getattr(test_obj, "dummy1", None), "dummy1") @@ -269,10 +304,11 @@ def __init__(self, arg1, arg2, **kwargs): self.arg2 = arg2 self.arg3 = arg1 - test_obj = TestDict('val1', 'val2', arg3="dummy1") + test_obj = TestDict("val1", "val2", arg3="dummy1") - self.assertCountEqual(getattr(test_obj, 'init_params'), - {'self', 'arg1', 'arg2', 'kwargs', 'arg3'}) + self.assertCountEqual( + getattr(test_obj, "init_params"), {"self", "arg1", "arg2", "kwargs", "arg3"} + ) self.assertEqual(getattr(test_obj, "arg1", None), "val1") self.assertEqual(getattr(test_obj, "arg2", None), "val2") self.assertEqual(getattr(test_obj, "arg3", None), "val1") @@ -283,13 +319,11 @@ def __init__(self, arg1, arg2): pass def get_dict_repr(self): - return { - "hello": None, - "world": ["dummy", None] - } + return {"hello": None, "world": ["dummy", None]} - test_obj = TestBuildDict('val1', 'val2') + test_obj = TestBuildDict("val1", "val2") - self.assertCountEqual(getattr(test_obj, 'init_params'), - {'self', 'arg1', 'arg2'}) + self.assertCountEqual( + getattr(test_obj, "init_params"), {"self", "arg1", "arg2"} + ) self.assertEqual(test_obj.get_dict_repr(), {"world": ["dummy"]}) diff --git a/azure-functions-extension-base/tests/test_web.py b/azure-functions-extension-base/tests/test_web.py index 98bfaf5..eaf2079 100644 --- a/azure-functions-extension-base/tests/test_web.py +++ b/azure-functions-extension-base/tests/test_web.py @@ -1,9 +1,15 @@ import unittest from unittest.mock import patch -from azure.functions.extension.base import (ModuleTrackerMeta, RequestTrackerMeta, - ResponseTrackerMeta, WebApp, WebServer, - HttpV2FeatureChecker, ResponseLabels) +from azure.functions.extension.base import ( + HttpV2FeatureChecker, + ModuleTrackerMeta, + RequestTrackerMeta, + ResponseLabels, + ResponseTrackerMeta, + WebApp, + WebServer, +) class TestModuleTrackerMeta(unittest.TestCase): @@ -15,6 +21,7 @@ def setUp(self): def test_classes_imported_from_same_module(self): class TestClass1(metaclass=ModuleTrackerMeta): pass + class TestClass2(metaclass=ModuleTrackerMeta): pass @@ -25,6 +32,7 @@ class TestClass2(metaclass=ModuleTrackerMeta): def test_class_imported_from_a_module(self): class TestClass1(metaclass=ModuleTrackerMeta): pass + self.assertEqual(ModuleTrackerMeta.get_module(), __name__) self.assertTrue(ModuleTrackerMeta.module_imported()) self.assertTrue(HttpV2FeatureChecker.http_v2_enabled()) @@ -37,12 +45,16 @@ class TestClass1(metaclass=ModuleTrackerMeta): self.assertTrue(ModuleTrackerMeta.module_imported()) with self.assertRaises(Exception) as context: + class TestClass2(metaclass=ModuleTrackerMeta): __module__ = "module2" - self.assertEqual(str(context.exception), - 'Only one web extension package shall be imported, ' - 'module1 and module2 are imported') + self.assertEqual( + str(context.exception), + "Only one web extension package shall be imported, " + "module1 and module2 are imported", + ) + class TestRequestTrackerMeta(unittest.TestCase): class TestRequest1: @@ -61,9 +73,13 @@ def setUp(self): def test_request_type_not_provided(self): # Define a class without providing the request_type attribute with self.assertRaises(Exception) as context: + class TestClass(metaclass=RequestTrackerMeta): pass - self.assertEqual(str(context.exception), 'Request type not provided for class TestClass') + + self.assertEqual( + str(context.exception), "Request type not provided for class TestClass" + ) def test_single_request_type(self): # Define a class providing a request_type attribute @@ -106,17 +122,22 @@ class TestClass1(metaclass=RequestTrackerMeta): # Define another class providing a different request_type attribute with self.assertRaises(Exception) as context: + class TestClass2(metaclass=RequestTrackerMeta): request_type = self.TestRequest2 - self.assertEqual(str(context.exception), - f'Only one request type shall be recorded for class TestClass2' - f' but found {self.TestRequest1} and {self.TestRequest2}') + + self.assertEqual( + str(context.exception), + f"Only one request type shall be recorded for class TestClass2" + f" but found {self.TestRequest1} and {self.TestRequest2}", + ) # Ensure the request_type remains the same after the exception self.assertEqual(RequestTrackerMeta.get_request_type(), self.TestRequest1) # Ensure check_type still returns True for the original request_type self.assertTrue(RequestTrackerMeta.check_type(self.TestRequest1)) + class TestResponseTrackerMeta(unittest.TestCase): class MockResponse1: pass @@ -126,44 +147,53 @@ class MockResponse2: def test_classes_imported_from_same_module(self): class TestResponse1(metaclass=ResponseTrackerMeta): - label = 'test_label_1' + label = "test_label_1" response_type = self.MockResponse1 class TestResponse2(metaclass=ResponseTrackerMeta): - label = 'test_label_2' + label = "test_label_2" response_type = self.MockResponse2 - self.assertEqual(ResponseTrackerMeta.get_response_type('test_label_1'), self.MockResponse1) - self.assertEqual(ResponseTrackerMeta.get_response_type('test_label_2'), self.MockResponse2) - self.assertIsNone(ResponseTrackerMeta.get_response_type('non_existing_label')) + self.assertEqual( + ResponseTrackerMeta.get_response_type("test_label_1"), self.MockResponse1 + ) + self.assertEqual( + ResponseTrackerMeta.get_response_type("test_label_2"), self.MockResponse2 + ) + self.assertIsNone(ResponseTrackerMeta.get_response_type("non_existing_label")) self.assertTrue(ResponseTrackerMeta.check_type(self.MockResponse1)) self.assertTrue(ResponseTrackerMeta.check_type(self.MockResponse2)) def test_class_imported_from_a_module(self): class TestResponse1(metaclass=ResponseTrackerMeta): - label = 'test_label_1' + label = "test_label_1" response_type = self.MockResponse1 - self.assertEqual(ResponseTrackerMeta.get_response_type('test_label_1'), self.MockResponse1) - self.assertIsNone(ResponseTrackerMeta.get_response_type('non_existing_label')) + self.assertEqual( + ResponseTrackerMeta.get_response_type("test_label_1"), self.MockResponse1 + ) + self.assertIsNone(ResponseTrackerMeta.get_response_type("non_existing_label")) self.assertTrue(ResponseTrackerMeta.check_type(self.MockResponse1)) self.assertFalse(ResponseTrackerMeta.check_type(self.MockResponse2)) def test_classes_imported_from_different_modules(self): class TestResponse1(metaclass=ResponseTrackerMeta): __module__ = "module1" - label = 'test_label_1' + label = "test_label_1" response_type = self.MockResponse1 with self.assertRaises(Exception) as context: + class TestResponse2(metaclass=ResponseTrackerMeta): __module__ = "module2" - label = 'test_label_1' + label = "test_label_1" response_type = self.MockResponse2 - self.assertEqual(str(context.exception), - 'Only one response type shall be recorded for class TestResponse2 ' - f'but found {self.MockResponse1} and {self.MockResponse2}') + self.assertEqual( + str(context.exception), + "Only one response type shall be recorded for class TestResponse2 " + f"but found {self.MockResponse1} and {self.MockResponse2}", + ) def test_different_labels(self): class TestResponse1(metaclass=ResponseTrackerMeta): @@ -174,16 +204,23 @@ class TestResponse2(metaclass=ResponseTrackerMeta): label = ResponseLabels.STREAMING response_type = self.MockResponse2 - self.assertEqual(ResponseTrackerMeta.get_response_type(ResponseLabels.STANDARD), self.MockResponse1) - self.assertEqual(ResponseTrackerMeta.get_response_type(ResponseLabels.STREAMING), self.MockResponse2) + self.assertEqual( + ResponseTrackerMeta.get_response_type(ResponseLabels.STANDARD), + self.MockResponse1, + ) + self.assertEqual( + ResponseTrackerMeta.get_response_type(ResponseLabels.STREAMING), + self.MockResponse2, + ) self.assertTrue(ResponseTrackerMeta.check_type(self.MockResponse1)) self.assertTrue(ResponseTrackerMeta.check_type(self.MockResponse2)) + class TestWebApp(unittest.TestCase): def test_route_and_get_app(self): class MockWebApp(WebApp): def route(self, func): - return + return def get_app(self): return "MockApp" @@ -191,6 +228,7 @@ def get_app(self): app = MockWebApp() self.assertEqual(app.get_app(), "MockApp") + class TestWebServer(unittest.TestCase): def test_web_server_initialization(self): class MockWebApp(WebApp): @@ -206,11 +244,12 @@ def get_app(self): self.assertEqual(server.port, 8080) self.assertEqual(server.web_app, "MockApp") + class TestHttpV2Enabled(unittest.TestCase): - @patch('azure.functions.extension.base.ModuleTrackerMeta.module_imported') + @patch("azure.functions.extension.base.ModuleTrackerMeta.module_imported") def test_http_v2_enabled(self, mock_module_imported): mock_module_imported.return_value = True self.assertTrue(HttpV2FeatureChecker.http_v2_enabled()) mock_module_imported.return_value = False - self.assertFalse(HttpV2FeatureChecker.http_v2_enabled()) \ No newline at end of file + self.assertFalse(HttpV2FeatureChecker.http_v2_enabled())