16820655284

Committed 07 Aug 2025 05:03PM UTC coverage: 86.841% (-0.05%) from 86.892%

Build # 16820655284

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: support CDK bootstrap and deployment (#12967)

Coverage Stats

32 of 38 new or added lines in 5 files covered. (84.21%)

2013 existing lines in 125 files now uncovered.

66606 of 76699 relevant lines covered (86.84%)

0.87 hits per line

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

90.91

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

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


def _import_submodules(
    package_name: str, module_regex: Pattern | None = 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

                if kwarg not in base_spec.annotations:
                    # Typically happens when the implementation uses '**kwargs: Any'
                    # This parameter is not part of the base spec, so we can't compare types
                    continue

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

localstack / localstack / 16820655284

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