a53c0273-9548-479e-ab3c-f3af40c0e980

Committed 13 May 2025 05:31PM UTC coverage: 86.624% (-0.03%) from 86.658%

Build # a53c0273-9548-479e-ab3c-f3af40c0e980

Build Type

push

circleci

Committed by

web-flow

Commit Message

ASF: Mark optional params as such (X | None) (#12614)

Coverage Stats

5 of 7 new or added lines in 2 files covered. (71.43%)

34 existing lines in 16 files now uncovered.

64347 of 74283 relevant lines covered (86.62%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.54

/localstack-core/localstack/testing/aws/asf_utils.py

import importlib
import importlib.util
import inspect
import pkgutil
import re
from types import FunctionType, ModuleType, NoneType, UnionType
from typing import Optional, Pattern, Union, get_args, get_origin


def _import_submodules(
    package_name: str, module_regex: Optional[Pattern] = None, recursive: bool = True
) -> dict[str, ModuleType]:
    """
    Imports all submodules of the given package with the defined (optional) module_suffix.

    :param package_name: To start the loading / importing at
    :param module_regex: Optional regex to filter the module names for
    :param recursive: True if the package should be loaded recursively
    :return:
    """
    package = importlib.import_module(package_name)
    results = {}
    for loader, name, is_pkg in pkgutil.walk_packages(package.__path__, package.__name__ + "."):
        if not module_regex or module_regex.match(name):
            results[name] = importlib.import_module(name)
        if recursive and is_pkg:
            results.update(_import_submodules(name, module_regex, recursive))
    return results


def _collect_provider_classes(
    provider_module: str, provider_module_regex: Pattern, provider_class_regex: Pattern
) -> list[type]:
    """
    Collects all provider implementation classes which should be tested.
    :param provider_module: module to start collecting in
    :param provider_module_regex: Regex to filter the module names for
    :param provider_class_regex: Regex to filter the provider class names for
    :return: list of classes to check the operation signatures of
    """
    provider_classes = []
    provider_modules = _import_submodules(provider_module, provider_module_regex)
    # check that all these files don't import any encrypted code
    for _, mod in provider_modules.items():
        # get all classes of the module which end with "Provider"
        classes = [
            cls_obj
            for cls_name, cls_obj in inspect.getmembers(mod)
            if inspect.isclass(cls_obj) and provider_class_regex.match(cls_name)
        ]
        provider_classes.extend(classes)
    return provider_classes


def collect_implemented_provider_operations(
    provider_module: str = "localstack.services",
    provider_module_regex: Pattern = re.compile(r".*\.provider[A-Za-z_0-9]*$"),
    provider_class_regex: Pattern = re.compile(r".*Provider$"),
    asf_api_module: str = "localstack.aws.api",
) -> list[tuple[type, type, str]]:
    """
    Collects all implemented operations on all provider classes together with their base classes (generated API classes).
    :param provider_module: module to start collecting in
    :param provider_module_regex: Regex to filter the module names for
    :param provider_class_regex: Regex to filter the provider class names for
    :param asf_api_module: module which contains the generated ASF APIs
    :return: list of tuple, where each tuple is (provider_class: type, base_class: type, provider_function: str)
    """
    results = []
    provider_classes = _collect_provider_classes(
        provider_module, provider_module_regex, provider_class_regex
    )
    for provider_class in provider_classes:
        for base_class in provider_class.__bases__:
            base_parent_module = ".".join(base_class.__module__.split(".")[:-1])
            if base_parent_module == asf_api_module:
                # find all functions on the provider class which are also defined in the super class and are not dunder functions
                provider_functions = [
                    method
                    for method in dir(provider_class)
                    if hasattr(base_class, method)
                    and isinstance(getattr(base_class, method), FunctionType)
                    and method.startswith("__") is False
                ]
                for provider_function in provider_functions:
                    results.append((provider_class, base_class, provider_function))
    return results


def check_provider_signature(sub_class: type, base_class: type, method_name: str) -> None:
    """
    Checks if the signature of a given provider method is equal to the signature of the function with the same name on the base class.

    :param sub_class: provider class to check the given method's signature of
    :param base_class: API class to check the given method's signature against
    :param method_name: name of the method on the sub_class and base_class to compare
    :raise: AssertionError if the two signatures are not equal
    """
    try:
        sub_function = getattr(sub_class, method_name)
    except AttributeError:
        raise AttributeError(
            f"Given method name ('{method_name}') is not a method of the sub class ('{sub_class.__name__}')."
        )

    if not isinstance(sub_function, FunctionType):
        raise AttributeError(
            f"Given method name ('{method_name}') is not a method of the sub class ('{sub_class.__name__}')."
        )

    if not getattr(sub_function, "expand_parameters", True):
        # if the operation on the subclass has the "expand_parameters" attribute (it has a handler decorator) set to False, we don't care
        return

    if wrapped := getattr(sub_function, "__wrapped__", False):
        # if the operation on the subclass has a decorator, unwrap it
        sub_function = wrapped

    try:
        base_function = getattr(base_class, method_name)
        # unwrap from the handler decorator
        base_function = base_function.__wrapped__

        sub_spec = inspect.getfullargspec(sub_function)
        base_spec = inspect.getfullargspec(base_function)

        error_msg = f"{sub_class.__name__}#{method_name} breaks with {base_class.__name__}#{method_name}. This can also be caused by 'from __future__ import annotations' in a provider file!"

        # Assert that the signature is correct
        assert sub_spec.args == base_spec.args, error_msg
        assert sub_spec.varargs == base_spec.varargs, error_msg
        assert sub_spec.varkw == base_spec.varkw, error_msg
        assert sub_spec.defaults == base_spec.defaults, (
            error_msg + f"\n{sub_spec.defaults} != {base_spec.defaults}"
        )
        assert sub_spec.kwonlyargs == base_spec.kwonlyargs, error_msg
        assert sub_spec.kwonlydefaults == base_spec.kwonlydefaults, error_msg

        # Assert that the typing of the implementation is equal to the base
        for kwarg in sub_spec.annotations:
            if kwarg == "return":
                assert sub_spec.annotations[kwarg] == base_spec.annotations[kwarg]
            else:
                # The API currently marks everything as required, and optional args are configured as:
                #    arg: ArgType = None
                # which is obviously incorrect.
                # Implementations sometimes do this correctly:
                #    arg: ArgType | None = None
                # These should be considered equal, so until the API is fixed, we remove any Optionals
                # This also gives us the flexibility to correct the API without fixing all implementations at the same time
                sub_type = _remove_optional(sub_spec.annotations[kwarg])
                base_type = _remove_optional(base_spec.annotations[kwarg])
                assert sub_type == base_type, (
                    f"Types for {kwarg} are different - {sub_type} instead of {base_type}"
                )

    except AttributeError:
        # the function is not defined in the superclass
        pass


def _remove_optional(_type: type) -> list[type]:
    if get_origin(_type) in [Union, UnionType]:
        union_types = list(get_args(_type))
        try:
            union_types.remove(NoneType)
        except ValueError:
            # Union of some other kind, like 'str | int'
            pass
        return union_types
    return [_type]

1	import importlib	1✔
2	import importlib.util	1✔
3	import inspect	1✔
4	import pkgutil	1✔
5	import re	1✔
6	from types import FunctionType, ModuleType, NoneType, UnionType	1✔
7	from typing import Optional, Pattern, Union, get_args, get_origin	1✔
8
9
10	def _import_submodules(	1✔
11	package_name: str, module_regex: Optional[Pattern] = None, recursive: bool = True
12	) -> dict[str, ModuleType]:
13	"""
14	Imports all submodules of the given package with the defined (optional) module_suffix.
15
16	:param package_name: To start the loading / importing at
17	:param module_regex: Optional regex to filter the module names for
18	:param recursive: True if the package should be loaded recursively
19	:return:
20	"""
21	package = importlib.import_module(package_name)	1✔
22	results = {}	1✔
23	for loader, name, is_pkg in pkgutil.walk_packages(package.__path__, package.__name__ + "."):	1✔
24	if not module_regex or module_regex.match(name):	1✔
25	results[name] = importlib.import_module(name)	1✔
26	if recursive and is_pkg:	1✔
27	results.update(_import_submodules(name, module_regex, recursive))	1✔
28	return results	1✔
29
30
31	def _collect_provider_classes(	1✔
32	provider_module: str, provider_module_regex: Pattern, provider_class_regex: Pattern
33	) -> list[type]:
34	"""
35	Collects all provider implementation classes which should be tested.
36	:param provider_module: module to start collecting in
37	:param provider_module_regex: Regex to filter the module names for
38	:param provider_class_regex: Regex to filter the provider class names for
39	:return: list of classes to check the operation signatures of
40	"""
41	provider_classes = []	1✔
42	provider_modules = _import_submodules(provider_module, provider_module_regex)	1✔
43	# check that all these files don't import any encrypted code
44	for _, mod in provider_modules.items():	1✔
45	# get all classes of the module which end with "Provider"
46	classes = [	1✔
47	cls_obj
48	for cls_name, cls_obj in inspect.getmembers(mod)
49	if inspect.isclass(cls_obj) and provider_class_regex.match(cls_name)
50	]
51	provider_classes.extend(classes)	1✔
52	return provider_classes	1✔
53
54
55	def collect_implemented_provider_operations(	1✔
56	provider_module: str = "localstack.services",
57	provider_module_regex: Pattern = re.compile(r".\.provider[A-Za-z_0-9]$"),
58	provider_class_regex: Pattern = re.compile(r".*Provider$"),
59	asf_api_module: str = "localstack.aws.api",
60	) -> list[tuple[type, type, str]]:
61	"""
62	Collects all implemented operations on all provider classes together with their base classes (generated API classes).
63	:param provider_module: module to start collecting in
64	:param provider_module_regex: Regex to filter the module names for
65	:param provider_class_regex: Regex to filter the provider class names for
66	:param asf_api_module: module which contains the generated ASF APIs
67	:return: list of tuple, where each tuple is (provider_class: type, base_class: type, provider_function: str)
68	"""
69	results = []	1✔
70	provider_classes = _collect_provider_classes(	1✔
71	provider_module, provider_module_regex, provider_class_regex
72	)
73	for provider_class in provider_classes:	1✔
74	for base_class in provider_class.__bases__:	1✔
75	base_parent_module = ".".join(base_class.__module__.split(".")[:-1])	1✔
76	if base_parent_module == asf_api_module:	1✔
77	# find all functions on the provider class which are also defined in the super class and are not dunder functions
78	provider_functions = [	1✔
79	method
80	for method in dir(provider_class)
81	if hasattr(base_class, method)
82	and isinstance(getattr(base_class, method), FunctionType)
83	and method.startswith("__") is False
84	]
85	for provider_function in provider_functions:	1✔
86	results.append((provider_class, base_class, provider_function))	1✔
87	return results	1✔
88
89
90	def check_provider_signature(sub_class: type, base_class: type, method_name: str) -> None:	1✔
91	"""
92	Checks if the signature of a given provider method is equal to the signature of the function with the same name on the base class.
93
94	:param sub_class: provider class to check the given method's signature of
95	:param base_class: API class to check the given method's signature against
96	:param method_name: name of the method on the sub_class and base_class to compare
97	:raise: AssertionError if the two signatures are not equal
98	"""
99	try:	1✔
100	sub_function = getattr(sub_class, method_name)	1✔
101	except AttributeError:	×
102	raise AttributeError(	×
103	f"Given method name ('{method_name}') is not a method of the sub class ('{sub_class.__name__}')."
104	)
105
106	if not isinstance(sub_function, FunctionType):	1✔
107	raise AttributeError(	×
108	f"Given method name ('{method_name}') is not a method of the sub class ('{sub_class.__name__}')."
109	)
110
111	if not getattr(sub_function, "expand_parameters", True):	1✔
112	# if the operation on the subclass has the "expand_parameters" attribute (it has a handler decorator) set to False, we don't care
113	return	1✔
114
115	if wrapped := getattr(sub_function, "__wrapped__", False):	1✔
116	# if the operation on the subclass has a decorator, unwrap it
117	sub_function = wrapped	1✔
118
119	try:	1✔
120	base_function = getattr(base_class, method_name)	1✔
121	# unwrap from the handler decorator
122	base_function = base_function.__wrapped__	1✔
123
124	sub_spec = inspect.getfullargspec(sub_function)	1✔
125	base_spec = inspect.getfullargspec(base_function)	1✔
126
127	error_msg = f"{sub_class.__name__}#{method_name} breaks with {base_class.__name__}#{method_name}. This can also be caused by 'from __future__ import annotations' in a provider file!"	1✔
128
129	# Assert that the signature is correct
130	assert sub_spec.args == base_spec.args, error_msg	1✔
131	assert sub_spec.varargs == base_spec.varargs, error_msg	1✔
132	assert sub_spec.varkw == base_spec.varkw, error_msg	1✔
133	assert sub_spec.defaults == base_spec.defaults, (	1✔
134	error_msg + f"\n{sub_spec.defaults} != {base_spec.defaults}"
135	)
136	assert sub_spec.kwonlyargs == base_spec.kwonlyargs, error_msg	1✔
137	assert sub_spec.kwonlydefaults == base_spec.kwonlydefaults, error_msg	1✔
138
139	# Assert that the typing of the implementation is equal to the base
140	for kwarg in sub_spec.annotations:	1✔
141	if kwarg == "return":	1✔
142	assert sub_spec.annotations[kwarg] == base_spec.annotations[kwarg]	1✔
143	else:
144	# The API currently marks everything as required, and optional args are configured as:
145	# arg: ArgType = None
146	# which is obviously incorrect.
147	# Implementations sometimes do this correctly:
148	# arg: ArgType \| None = None
149	# These should be considered equal, so until the API is fixed, we remove any Optionals
150	# This also gives us the flexibility to correct the API without fixing all implementations at the same time
151	sub_type = _remove_optional(sub_spec.annotations[kwarg])	1✔
152	base_type = _remove_optional(base_spec.annotations[kwarg])	1✔
153	assert sub_type == base_type, (	1✔
154	f"Types for {kwarg} are different - {sub_type} instead of {base_type}"
155	)
156
157	except AttributeError:	×
158	# the function is not defined in the superclass
159	pass	×
160
161
162	def _remove_optional(_type: type) -> list[type]:	1✔
163	if get_origin(_type) in [Union, UnionType]:	1✔
164	union_types = list(get_args(_type))	1✔
165	try:	1✔
166	union_types.remove(NoneType)	1✔
NEW 167	except ValueError:	×
168	# Union of some other kind, like 'str \| int'
NEW 169	pass	×
170	return union_types	1✔
171	return [_type]	1✔

localstack / localstack / a53c0273-9548-479e-ab3c-f3af40c0e980

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous