22048723723

Committed 13 Feb 2026 06:59PM UTC coverage: 87.006% (+0.1%) from 86.883%

Build # 22048723723

Build Type

push

github

Committed by

web-flow

Commit Message

CW Logs: Test suite for service internalization (#13692)

Run Details

22 of 22 new or added lines in 1 file covered. (100.0%)

928 existing lines in 33 files now uncovered.

69716 of 80128 relevant lines covered (87.01%)

0.87 hits per line

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

90.59

/localstack-core/localstack/testing/testselection/matching.py

import fnmatch
import pathlib
import re
from collections import defaultdict
from collections.abc import Callable, Iterable

from localstack.aws.scaffold import is_keyword

# TODO: extract API Dependencies and composites to constants or similar

SENTINEL_NO_TEST = "SENTINEL_NO_TEST"  # a line item which signals that we don't default to everything, we just don't want to actually want to run a test => useful to differentiate between empty / nothing
SENTINEL_ALL_TESTS = "SENTINEL_ALL_TESTS"  # a line item which signals that we don't default to everything, we just don't want to actually want to run a test => useful to differentiate between empty / nothing

DEFAULT_SEARCH_PATTERNS = (
    r"localstack/services/([^/]+)/.+",
    r"localstack/aws/api/([^/]+)/__init__\.py",
    r"tests/aws/services/([^/]+)/.+",
)


def _map_to_module_name(service_name: str) -> str:
    """sanitize a service name like we're doing when scaffolding, e.g. lambda => lambda_"""
    service_name = service_name.replace("-", "_")
    # handle service names which are reserved keywords in python (f.e. lambda)
    if is_keyword(service_name):
        service_name += "_"
    return service_name


def _map_to_service_name(module_name: str) -> str:
    """map a sanitized module name to a service name, e.g. lambda_ => lambda"""
    if module_name.endswith("_"):
        return module_name[:-1]
    return module_name.replace("_", "-")


def resolve_dependencies(module_name: str, api_dependencies: dict[str, Iterable[str]]) -> set[str]:
    """
    Resolves dependencies for a given service module name

    :param module_name: the name of the service to resolve (e.g. lambda_)
    :param api_dependencies: dict of API dependencies where each key is the service and its value a list of services it
                             depends on
    :return: set of resolved _service names_ that the service depends on (e.g. sts)
    """
    svc_name = _map_to_service_name(module_name)
    return set(_reverse_dependency_map(api_dependencies).get(svc_name, []))


# TODO: might want to cache that, but for now it shouldn't be too much overhead
def _reverse_dependency_map(dependency_map: dict[str, Iterable[str]]) -> dict[str, Iterable[str]]:
    """
    The current API_DEPENDENCIES actually maps the services to their own dependencies.
    In our case here we need the inverse of this, we need to of which other services this service is a dependency of.
    """
    result = {}
    for svc, deps in dependency_map.items():
        for dep in deps:
            result.setdefault(dep, set()).add(svc)
    return result


def get_test_dir_for_service(svc: str) -> str:
    return f"tests/aws/services/{svc}"


def get_directory(t: str) -> str:
    # we take the parent of the match file, and we split it in parts
    parent_parts = pathlib.PurePath(t).parent.parts
    # we remove any parts that can be present in front of the first `tests` folder, could be caused by namespacing
    root = parent_parts.index("tests")
    folder_path = "/".join(parent_parts[root:]) + "/"
    return folder_path


class Matcher:
    def __init__(self, matching_func: Callable[[str], bool]):
        self.matching_func = matching_func

    def full_suite(self):
        return lambda t: [SENTINEL_ALL_TESTS] if self.matching_func(t) else []

    def ignore(self):
        return lambda t: [SENTINEL_NO_TEST] if self.matching_func(t) else []

    def service_tests(self, services: list[str]):
        return lambda t: (
            [get_test_dir_for_service(svc) for svc in services] if self.matching_func(t) else []
        )

    def passthrough(self):
        return lambda t: [t] if self.matching_func(t) else []

    def directory(self, paths: list[str] = None):
        """Enables executing tests on a full directory if the file is matched.
        By default, it will return the directory of the modified file.
        If the argument `paths` is provided, it will instead return the provided list.
        """
        return lambda t: (paths or [get_directory(t)]) if self.matching_func(t) else []


class Matchers:
    @staticmethod
    def glob(glob: str) -> Matcher:
        return Matcher(lambda t: fnmatch.fnmatch(t, glob))

    @staticmethod
    def regex(glob: str) -> Matcher:
        return Matcher(lambda t: bool(re.match(t, glob)))

    @staticmethod
    def prefix(prefix: str) -> Matcher:
        return Matcher(lambda t: t.startswith(prefix))


def generic_service_test_matching_rule(
    changed_file_path: str,
    api_dependencies: dict[str, Iterable[str]] | None = None,
    search_patterns: Iterable[str] = DEFAULT_SEARCH_PATTERNS,
    test_dirs: Iterable[str] = ("tests/aws/services",),
) -> set[str]:
    """
    Generic matching of changes in service files to their tests

    :param api_dependencies: dict of API dependencies where each key is the service and its value a list of services it depends on
    :param changed_file_path: the file path of the detected change
    :param search_patterns: list of regex patterns to search for in the changed file path
    :param test_dirs: list of test directories to match for a changed service
    :return: list of partial test file path filters for the matching service and all services it depends on
    """
    # TODO: consider API_COMPOSITES

    if api_dependencies is None:
        from localstack.utils.bootstrap import API_DEPENDENCIES, API_DEPENDENCIES_OPTIONAL

        # merge the mandatory and optional service dependencies
        api_dependencies = defaultdict(set)
        for service, mandatory_dependencies in API_DEPENDENCIES.items():
            api_dependencies[service].update(mandatory_dependencies)

        for service, optional_dependencies in API_DEPENDENCIES_OPTIONAL.items():
            api_dependencies[service].update(optional_dependencies)

    match = None
    for pattern in search_patterns:
        match = re.findall(pattern, changed_file_path)
        if match:
            break

    if match:
        changed_service = match[0]
        changed_services = [changed_service]
        service_dependencies = resolve_dependencies(changed_service, api_dependencies)
        changed_services.extend(service_dependencies)
        changed_service_module_names = [_map_to_module_name(svc) for svc in changed_services]
        return {
            f"{test_dir}/{svc}/" for test_dir in test_dirs for svc in changed_service_module_names
        }

    return set()


MatchingRule = Callable[[str], Iterable[str]]


def check_rule_has_matches(rule: MatchingRule, files: Iterable[str]) -> bool:
    """maintenance utility to check if a rule has any matches at all in the given directory"""
    detected_tests = set()
    for file in files:
        detected_tests.update(rule(file))
    return len(detected_tests) > 0


MATCHING_RULES: list[MatchingRule] = [
    # Generic rules
    generic_service_test_matching_rule,  # always *at least* the service tests and dependencies
    Matchers.glob(
        "tests/**/test_*.py"
    ).passthrough(),  # changes in a test file should always at least test that file
    # CI
    Matchers.glob(".github/**").full_suite(),
    # dependencies / project setup
    Matchers.glob("requirements*.txt").full_suite(),
    Matchers.glob("setup.cfg").full_suite(),
    Matchers.glob("setup.py").full_suite(),
    Matchers.glob("pyproject.toml").full_suite(),
    Matchers.glob("Dockerfile").full_suite(),
    Matchers.glob("Makefile").full_suite(),
    Matchers.glob("bin/**").full_suite(),
    Matchers.glob("localstack/config.py").full_suite(),
    Matchers.glob("localstack/constants.py").full_suite(),
    Matchers.glob("localstack/plugins.py").full_suite(),
    Matchers.glob("localstack/utils/**").full_suite(),
    # testing
    Matchers.glob("localstack/testing/**").full_suite(),
    Matchers.glob("**/conftest.py").directory(),
    Matchers.glob("**/fixtures.py").full_suite(),
    # ignore
    Matchers.glob("**/*.md").ignore(),
    Matchers.glob("doc/**").ignore(),
    Matchers.glob("CODEOWNERS").ignore(),
    Matchers.glob(".gitignore").ignore(),
    Matchers.glob(".git-blame-ignore-revs").ignore(),
    # lambda
    Matchers.glob("tests/aws/services/lambda_/functions/**").service_tests(services=["lambda"]),
]

1	import fnmatch	1✔
2	import pathlib	1✔
3	import re	1✔
4	from collections import defaultdict	1✔
5	from collections.abc import Callable, Iterable	1✔
6
7	from localstack.aws.scaffold import is_keyword	1✔
8
9	# TODO: extract API Dependencies and composites to constants or similar
10
11	SENTINEL_NO_TEST = "SENTINEL_NO_TEST" # a line item which signals that we don't default to everything, we just don't want to actually want to run a test => useful to differentiate between empty / nothing	1✔
12	SENTINEL_ALL_TESTS = "SENTINEL_ALL_TESTS" # a line item which signals that we don't default to everything, we just don't want to actually want to run a test => useful to differentiate between empty / nothing	1✔
13
14	DEFAULT_SEARCH_PATTERNS = (	1✔
15	r"localstack/services/([^/]+)/.+",
16	r"localstack/aws/api/([^/]+)/__init__\.py",
17	r"tests/aws/services/([^/]+)/.+",
18	)
19
20
21	def _map_to_module_name(service_name: str) -> str:	1✔
22	"""sanitize a service name like we're doing when scaffolding, e.g. lambda => lambda_"""
23	service_name = service_name.replace("-", "_")	1✔
24	# handle service names which are reserved keywords in python (f.e. lambda)
25	if is_keyword(service_name):	1✔
26	service_name += "_"	1✔
27	return service_name	1✔
28
29
30	def _map_to_service_name(module_name: str) -> str:	1✔
31	"""map a sanitized module name to a service name, e.g. lambda_ => lambda"""
32	if module_name.endswith("_"):	1✔
33	return module_name[:-1]	1✔
34	return module_name.replace("_", "-")	1✔
35
36
37	def resolve_dependencies(module_name: str, api_dependencies: dict[str, Iterable[str]]) -> set[str]:	1✔
38	"""
39	Resolves dependencies for a given service module name
40
41	:param module_name: the name of the service to resolve (e.g. lambda_)
42	:param api_dependencies: dict of API dependencies where each key is the service and its value a list of services it
43	depends on
44	:return: set of resolved _service names_ that the service depends on (e.g. sts)
45	"""
46	svc_name = _map_to_service_name(module_name)	1✔
47	return set(_reverse_dependency_map(api_dependencies).get(svc_name, []))	1✔
48
49
50	# TODO: might want to cache that, but for now it shouldn't be too much overhead
51	def _reverse_dependency_map(dependency_map: dict[str, Iterable[str]]) -> dict[str, Iterable[str]]:	1✔
52	"""
53	The current API_DEPENDENCIES actually maps the services to their own dependencies.
54	In our case here we need the inverse of this, we need to of which other services this service is a dependency of.
55	"""
56	result = {}	1✔
57	for svc, deps in dependency_map.items():	1✔
58	for dep in deps:	1✔
59	result.setdefault(dep, set()).add(svc)	1✔
60	return result	1✔
61
62
63	def get_test_dir_for_service(svc: str) -> str:	1✔
64	return f"tests/aws/services/{svc}"	×
65
66
67	def get_directory(t: str) -> str:	1✔
68	# we take the parent of the match file, and we split it in parts
69	parent_parts = pathlib.PurePath(t).parent.parts	1✔
70	# we remove any parts that can be present in front of the first `tests` folder, could be caused by namespacing
71	root = parent_parts.index("tests")	1✔
72	folder_path = "/".join(parent_parts[root:]) + "/"	1✔
73	return folder_path	1✔
74
75
76	class Matcher:	1✔
77	def __init__(self, matching_func: Callable[[str], bool]):	1✔
78	self.matching_func = matching_func	1✔
79
80	def full_suite(self):	1✔
81	return lambda t: [SENTINEL_ALL_TESTS] if self.matching_func(t) else []	1✔
82
83	def ignore(self):	1✔
84	return lambda t: [SENTINEL_NO_TEST] if self.matching_func(t) else []	1✔
85
86	def service_tests(self, services: list[str]):	1✔
87	return lambda t: (	1✔
88	[get_test_dir_for_service(svc) for svc in services] if self.matching_func(t) else []
89	)
90
91	def passthrough(self):	1✔
92	return lambda t: [t] if self.matching_func(t) else []	1✔
93
94	def directory(self, paths: list[str] = None):	1✔
95	"""Enables executing tests on a full directory if the file is matched.
96	By default, it will return the directory of the modified file.
97	If the argument `paths` is provided, it will instead return the provided list.
98	"""
99	return lambda t: (paths or [get_directory(t)]) if self.matching_func(t) else []	1✔
100
101
102	class Matchers:	1✔
103	@staticmethod	1✔
104	def glob(glob: str) -> Matcher:	1✔
105	return Matcher(lambda t: fnmatch.fnmatch(t, glob))	1✔
106
107	@staticmethod	1✔
108	def regex(glob: str) -> Matcher:	1✔
UNCOV 109	return Matcher(lambda t: bool(re.match(t, glob)))	×
110
111	@staticmethod	1✔
112	def prefix(prefix: str) -> Matcher:	1✔
UNCOV 113	return Matcher(lambda t: t.startswith(prefix))	×
114
115
116	def generic_service_test_matching_rule(	1✔
117	changed_file_path: str,
118	api_dependencies: dict[str, Iterable[str]] \| None = None,
119	search_patterns: Iterable[str] = DEFAULT_SEARCH_PATTERNS,
120	test_dirs: Iterable[str] = ("tests/aws/services",),
121	) -> set[str]:
122	"""
123	Generic matching of changes in service files to their tests
124
125	:param api_dependencies: dict of API dependencies where each key is the service and its value a list of services it depends on
126	:param changed_file_path: the file path of the detected change
127	:param search_patterns: list of regex patterns to search for in the changed file path
128	:param test_dirs: list of test directories to match for a changed service
129	:return: list of partial test file path filters for the matching service and all services it depends on
130	"""
131	# TODO: consider API_COMPOSITES
132
133	if api_dependencies is None:	1✔
134	from localstack.utils.bootstrap import API_DEPENDENCIES, API_DEPENDENCIES_OPTIONAL	1✔
135
136	# merge the mandatory and optional service dependencies
137	api_dependencies = defaultdict(set)	1✔
138	for service, mandatory_dependencies in API_DEPENDENCIES.items():	1✔
139	api_dependencies[service].update(mandatory_dependencies)	1✔
140
141	for service, optional_dependencies in API_DEPENDENCIES_OPTIONAL.items():	1✔
142	api_dependencies[service].update(optional_dependencies)	1✔
143
144	match = None	1✔
145	for pattern in search_patterns:	1✔
146	match = re.findall(pattern, changed_file_path)	1✔
147	if match:	1✔
148	break	1✔
149
150	if match:	1✔
151	changed_service = match[0]	1✔
152	changed_services = [changed_service]	1✔
153	service_dependencies = resolve_dependencies(changed_service, api_dependencies)	1✔
154	changed_services.extend(service_dependencies)	1✔
155	changed_service_module_names = [_map_to_module_name(svc) for svc in changed_services]	1✔
156	return {	1✔
157	f"{test_dir}/{svc}/" for test_dir in test_dirs for svc in changed_service_module_names
158	}
159
UNCOV 160	return set()	×
161
162
163	MatchingRule = Callable[[str], Iterable[str]]	1✔
164
165
166	def check_rule_has_matches(rule: MatchingRule, files: Iterable[str]) -> bool:	1✔
167	"""maintenance utility to check if a rule has any matches at all in the given directory"""
UNCOV 168	detected_tests = set()	×
UNCOV 169	for file in files:	×
170	detected_tests.update(rule(file))	×
171	return len(detected_tests) > 0	×
172
173
174	MATCHING_RULES: list[MatchingRule] = [	1✔
175	# Generic rules
176	generic_service_test_matching_rule, # always at least the service tests and dependencies
177	Matchers.glob(
178	"tests/*/test_.py"
179	).passthrough(), # changes in a test file should always at least test that file
180	# CI
181	Matchers.glob(".github/**").full_suite(),
182	# dependencies / project setup
183	Matchers.glob("requirements*.txt").full_suite(),
184	Matchers.glob("setup.cfg").full_suite(),
185	Matchers.glob("setup.py").full_suite(),
186	Matchers.glob("pyproject.toml").full_suite(),
187	Matchers.glob("Dockerfile").full_suite(),
188	Matchers.glob("Makefile").full_suite(),
189	Matchers.glob("bin/**").full_suite(),
190	Matchers.glob("localstack/config.py").full_suite(),
191	Matchers.glob("localstack/constants.py").full_suite(),
192	Matchers.glob("localstack/plugins.py").full_suite(),
193	Matchers.glob("localstack/utils/**").full_suite(),
194	# testing
195	Matchers.glob("localstack/testing/**").full_suite(),
196	Matchers.glob("**/conftest.py").directory(),
197	Matchers.glob("**/fixtures.py").full_suite(),
198	# ignore
199	Matchers.glob("*/.md").ignore(),
200	Matchers.glob("doc/**").ignore(),
201	Matchers.glob("CODEOWNERS").ignore(),
202	Matchers.glob(".gitignore").ignore(),
203	Matchers.glob(".git-blame-ignore-revs").ignore(),
204	# lambda
205	Matchers.glob("tests/aws/services/lambda_/functions/**").service_tests(services=["lambda"]),
206	]

localstack / localstack / 22048723723

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