16981563750

Committed 14 Aug 2025 10:49PM UTC coverage: 86.896% (+0.04%) from 86.852%

Build # 16981563750

Build Type

push

github

Committed by

web-flow

Commit Message

add support for Fn::Tranform in CFnV2 (#12966)

Co-authored-by: Simon Walker <simon.walker@localstack.cloud>

Coverage Stats

181 of 195 new or added lines in 6 files covered. (92.82%)

348 existing lines in 22 files now uncovered.

66915 of 77006 relevant lines covered (86.9%)

0.87 hits per line

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

94.39

/localstack-core/localstack/testing/pytest/marking.py

"""
Custom pytest mark typings
"""

import os
from typing import TYPE_CHECKING, Callable, Optional

import pytest
from _pytest.config import PytestPluginManager
from _pytest.config.argparsing import Parser


class AwsCompatibilityMarkers:
    # test has been successfully run against AWS, ideally multiple times
    validated = pytest.mark.aws_validated

    # implies aws_validated. test needs additional setup, configuration or some other steps not included in the test setup itself
    manual_setup_required = pytest.mark.aws_manual_setup_required

    # fails against AWS but should be made runnable against AWS in the future, basically a TODO
    needs_fixing = pytest.mark.aws_needs_fixing

    # only runnable against localstack by design
    only_localstack = pytest.mark.aws_only_localstack

    # it's unknown if the test works (reliably) against AWS or not
    unknown = pytest.mark.aws_unknown


class ParityMarkers:
    aws_validated = pytest.mark.aws_validated
    only_localstack = pytest.mark.only_localstack


class SkipSnapshotVerifyMarker:
    def __call__(
        self,
        *,
        paths: "Optional[list[str]]" = None,
        condition: "Optional[Callable[[...], bool]]" = None,
    ): ...


class MultiRuntimeMarker:
    def __call__(self, *, scenario: str, runtimes: Optional[list[str]] = None): ...


class SnapshotMarkers:
    skip_snapshot_verify: SkipSnapshotVerifyMarker = pytest.mark.skip_snapshot_verify


class Markers:
    aws = AwsCompatibilityMarkers
    parity = ParityMarkers  # TODO: in here for compatibility sake. Remove when -ext has been refactored to use @markers.aws.*
    snapshot = SnapshotMarkers

    multiruntime: MultiRuntimeMarker = pytest.mark.multiruntime

    # test selection
    acceptance_test = pytest.mark.acceptance_test
    """This test is an acceptance test"""
    skip_offline = pytest.mark.skip_offline
    """Test is skipped if offline, as it requires some sort of internet connection to run"""
    only_on_amd64 = pytest.mark.only_on_amd64
    """Test requires ability of the system to execute amd64 binaries"""
    only_on_arm64 = pytest.mark.only_on_arm64
    """Test requires ability of the system to execute arm64 binaries"""
    resource_heavy = pytest.mark.resource_heavy
    """Test is very resource heavy, and might be skipped in CI"""
    requires_in_container = pytest.mark.requires_in_container
    """Test requires LocalStack to run inside a container"""
    requires_in_process = pytest.mark.requires_in_process
    """The test and the LS instance have to be run in the same process"""
    requires_docker = pytest.mark.requires_docker
    """The test requires docker or a compatible container engine - will not work on kubernetes"""
    lambda_runtime_update = pytest.mark.lambda_runtime_update
    """Tests to execute when updating snapshots for a new Lambda runtime"""


# pytest plugin
if TYPE_CHECKING:
    from _pytest.config import Config


@pytest.hookimpl
def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager):
    parser.addoption(
        "--offline",
        action="store_true",
        default=False,
        help="test run will not have an internet connection",
    )


def enforce_single_aws_marker(items: list[pytest.Item]):
    """Enforce that each test has exactly one aws compatibility marker"""
    marker_errors = []

    for item in items:
        # we should only concern ourselves with tests in tests/aws/
        if "tests/aws" not in item.fspath.dirname:
            continue

        aws_markers = []
        for mark in item.iter_markers():
            if mark.name.startswith("aws_"):
                aws_markers.append(mark.name)

        if len(aws_markers) > 1:
            marker_errors.append(f"{item.nodeid}: Too many aws markers specified: {aws_markers}")
        elif len(aws_markers) == 0:
            marker_errors.append(
                f"{item.nodeid}: Missing aws marker. Specify at least one marker, e.g. @markers.aws.validated"
            )

    if marker_errors:
        raise pytest.UsageError(*marker_errors)


def filter_by_markers(config: "Config", items: list[pytest.Item]):
    """Filter tests by markers."""
    from localstack import config as localstack_config
    from localstack.utils.bootstrap import in_ci
    from localstack.utils.platform import Arch, get_arch

    is_offline = config.getoption("--offline")
    is_in_docker = localstack_config.is_in_docker
    is_in_ci = in_ci()
    is_amd64 = get_arch() == Arch.amd64
    is_arm64 = get_arch() == Arch.arm64
    # Inlining `is_aws_cloud()` here because localstack.testing.aws.util imports boto3,
    # which is not installed for the CLI tests
    is_real_aws = os.environ.get("TEST_TARGET", "") == "AWS_CLOUD"

    if is_real_aws:
        # Do not skip any tests if they are executed against real AWS
        return

    skip_offline = pytest.mark.skip(
        reason="Test cannot be executed offline / in a restricted network environment. "
        "Add network connectivity and remove the --offline option when running "
        "the test."
    )
    requires_in_container = pytest.mark.skip(
        reason="Test requires execution inside a container (e.g., to install system packages)"
    )
    only_on_amd64 = pytest.mark.skip(
        reason="Test uses features that are currently only supported for AMD64. Skipping in CI."
    )
    only_on_arm64 = pytest.mark.skip(
        reason="Test uses features that are currently only supported for ARM64. Skipping in CI."
    )

    for item in items:
        if is_offline and "skip_offline" in item.keywords:
            item.add_marker(skip_offline)
        if not is_in_docker and "requires_in_container" in item.keywords:
            item.add_marker(requires_in_container)
        if is_in_ci and not is_amd64 and "only_on_amd64" in item.keywords:
            item.add_marker(only_on_amd64)
        if is_in_ci and not is_arm64 and "only_on_arm64" in item.keywords:
            item.add_marker(only_on_arm64)


@pytest.hookimpl
def pytest_collection_modifyitems(
    session: pytest.Session, config: "Config", items: list[pytest.Item]
) -> None:
    enforce_single_aws_marker(items)
    filter_by_markers(config, items)


@pytest.hookimpl
def pytest_configure(config):
    config.addinivalue_line(
        "markers",
        "skip_offline: mark the test to be skipped when the tests are run offline "
        "(this test explicitly / semantically needs an internet connection)",
    )
    config.addinivalue_line(
        "markers",
        "only_on_amd64: mark the test as running only in an amd64 (i.e., x86_64) environment",
    )
    config.addinivalue_line(
        "markers",
        "only_on_arm64: mark the test as running only in an arm64 environment",
    )
    config.addinivalue_line(
        "markers",
        "requires_in_container: mark the test as running only in a container (e.g., requires installation of system packages)",
    )
    config.addinivalue_line(
        "markers",
        "resource_heavy: mark the test as resource-heavy, e.g., downloading very large external dependencies, "
        "or requiring high amount of RAM/CPU (can be systematically sampled/optimized in the future)",
    )
    config.addinivalue_line(
        "markers",
        "aws_validated: mark the test as validated / verified against real AWS",
    )
    config.addinivalue_line(
        "markers",
        "aws_only_localstack: mark the test as inherently incompatible with AWS, e.g. when testing localstack-specific features",
    )
    config.addinivalue_line(
        "markers",
        "aws_needs_fixing: test fails against AWS but it shouldn't. Might need refactoring, additional permissions, etc.",
    )
    config.addinivalue_line(
        "markers",
        "aws_manual_setup_required: validated against real AWS but needs additional setup or account configuration (e.g. increased service quotas)",
    )
    config.addinivalue_line(
        "markers",
        "aws_unknown: it's unknown if the test works (reliably) against AWS or not",
    )
    config.addinivalue_line(
        "markers",
        "multiruntime: parametrize test against multiple Lambda runtimes",
    )
    config.addinivalue_line(
        "markers",
        "requires_docker: mark the test as requiring docker (or a compatible container engine) - will not work on kubernetes.",
    )
    config.addinivalue_line(
        "markers",
        "requires_in_process: mark the test as requiring the test to run inside the same process as LocalStack - will not work if tests are run against a running LS container.",
    )

1	"""
2	Custom pytest mark typings
3	"""
4
5	import os	1✔
6	from typing import TYPE_CHECKING, Callable, Optional	1✔
7
8	import pytest	1✔
9	from _pytest.config import PytestPluginManager	1✔
10	from _pytest.config.argparsing import Parser	1✔
11
12
13	class AwsCompatibilityMarkers:	1✔
14	# test has been successfully run against AWS, ideally multiple times
15	validated = pytest.mark.aws_validated	1✔
16
17	# implies aws_validated. test needs additional setup, configuration or some other steps not included in the test setup itself
18	manual_setup_required = pytest.mark.aws_manual_setup_required	1✔
19
20	# fails against AWS but should be made runnable against AWS in the future, basically a TODO
21	needs_fixing = pytest.mark.aws_needs_fixing	1✔
22
23	# only runnable against localstack by design
24	only_localstack = pytest.mark.aws_only_localstack	1✔
25
26	# it's unknown if the test works (reliably) against AWS or not
27	unknown = pytest.mark.aws_unknown	1✔
28
29
30	class ParityMarkers:	1✔
31	aws_validated = pytest.mark.aws_validated	1✔
32	only_localstack = pytest.mark.only_localstack	1✔
33
34
35	class SkipSnapshotVerifyMarker:	1✔
36	def __call__(	1✔
37	self,
38	*,
39	paths: "Optional[list[str]]" = None,
40	condition: "Optional[Callable[[...], bool]]" = None,
41	): ...
42
43
44	class MultiRuntimeMarker:	1✔
45	def __call__(self, *, scenario: str, runtimes: Optional[list[str]] = None): ...	1✔
46
47
48	class SnapshotMarkers:	1✔
49	skip_snapshot_verify: SkipSnapshotVerifyMarker = pytest.mark.skip_snapshot_verify	1✔
50
51
52	class Markers:	1✔
53	aws = AwsCompatibilityMarkers	1✔
54	parity = ParityMarkers # TODO: in here for compatibility sake. Remove when -ext has been refactored to use @markers.aws.*	1✔
55	snapshot = SnapshotMarkers	1✔
56
57	multiruntime: MultiRuntimeMarker = pytest.mark.multiruntime	1✔
58
59	# test selection
60	acceptance_test = pytest.mark.acceptance_test	1✔
61	"""This test is an acceptance test"""	1✔
62	skip_offline = pytest.mark.skip_offline	1✔
63	"""Test is skipped if offline, as it requires some sort of internet connection to run"""	1✔
64	only_on_amd64 = pytest.mark.only_on_amd64	1✔
65	"""Test requires ability of the system to execute amd64 binaries"""	1✔
66	only_on_arm64 = pytest.mark.only_on_arm64	1✔
67	"""Test requires ability of the system to execute arm64 binaries"""	1✔
68	resource_heavy = pytest.mark.resource_heavy	1✔
69	"""Test is very resource heavy, and might be skipped in CI"""	1✔
70	requires_in_container = pytest.mark.requires_in_container	1✔
71	"""Test requires LocalStack to run inside a container"""	1✔
72	requires_in_process = pytest.mark.requires_in_process	1✔
73	"""The test and the LS instance have to be run in the same process"""	1✔
74	requires_docker = pytest.mark.requires_docker	1✔
75	"""The test requires docker or a compatible container engine - will not work on kubernetes"""	1✔
76	lambda_runtime_update = pytest.mark.lambda_runtime_update	1✔
77	"""Tests to execute when updating snapshots for a new Lambda runtime"""	1✔
78
79
80	# pytest plugin
81	if TYPE_CHECKING:	1✔
UNCOV 82	from _pytest.config import Config	×
83
84
85	@pytest.hookimpl	1✔
86	def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager):	1✔
87	parser.addoption(	1✔
88	"--offline",
89	action="store_true",
90	default=False,
91	help="test run will not have an internet connection",
92	)
93
94
95	def enforce_single_aws_marker(items: list[pytest.Item]):	1✔
96	"""Enforce that each test has exactly one aws compatibility marker"""
97	marker_errors = []	1✔
98
99	for item in items:	1✔
100	# we should only concern ourselves with tests in tests/aws/
101	if "tests/aws" not in item.fspath.dirname:	1✔
102	continue	1✔
103
104	aws_markers = []	1✔
105	for mark in item.iter_markers():	1✔
106	if mark.name.startswith("aws_"):	1✔
107	aws_markers.append(mark.name)	1✔
108
109	if len(aws_markers) > 1:	1✔
UNCOV 110	marker_errors.append(f"{item.nodeid}: Too many aws markers specified: {aws_markers}")	×
111	elif len(aws_markers) == 0:	1✔
UNCOV 112	marker_errors.append(	×
113	f"{item.nodeid}: Missing aws marker. Specify at least one marker, e.g. @markers.aws.validated"
114	)
115
116	if marker_errors:	1✔
UNCOV 117	raise pytest.UsageError(*marker_errors)	×
118
119
120	def filter_by_markers(config: "Config", items: list[pytest.Item]):	1✔
121	"""Filter tests by markers."""
122	from localstack import config as localstack_config	1✔
123	from localstack.utils.bootstrap import in_ci	1✔
124	from localstack.utils.platform import Arch, get_arch	1✔
125
126	is_offline = config.getoption("--offline")	1✔
127	is_in_docker = localstack_config.is_in_docker	1✔
128	is_in_ci = in_ci()	1✔
129	is_amd64 = get_arch() == Arch.amd64	1✔
130	is_arm64 = get_arch() == Arch.arm64	1✔
131	# Inlining `is_aws_cloud()` here because localstack.testing.aws.util imports boto3,
132	# which is not installed for the CLI tests
133	is_real_aws = os.environ.get("TEST_TARGET", "") == "AWS_CLOUD"	1✔
134
135	if is_real_aws:	1✔
136	# Do not skip any tests if they are executed against real AWS
UNCOV 137	return	×
138
139	skip_offline = pytest.mark.skip(	1✔
140	reason="Test cannot be executed offline / in a restricted network environment. "
141	"Add network connectivity and remove the --offline option when running "
142	"the test."
143	)
144	requires_in_container = pytest.mark.skip(	1✔
145	reason="Test requires execution inside a container (e.g., to install system packages)"
146	)
147	only_on_amd64 = pytest.mark.skip(	1✔
148	reason="Test uses features that are currently only supported for AMD64. Skipping in CI."
149	)
150	only_on_arm64 = pytest.mark.skip(	1✔
151	reason="Test uses features that are currently only supported for ARM64. Skipping in CI."
152	)
153
154	for item in items:	1✔
155	if is_offline and "skip_offline" in item.keywords:	1✔
UNCOV 156	item.add_marker(skip_offline)	×
157	if not is_in_docker and "requires_in_container" in item.keywords:	1✔
158	item.add_marker(requires_in_container)	1✔
159	if is_in_ci and not is_amd64 and "only_on_amd64" in item.keywords:	1✔
160	item.add_marker(only_on_amd64)	1✔
161	if is_in_ci and not is_arm64 and "only_on_arm64" in item.keywords:	1✔
162	item.add_marker(only_on_arm64)	1✔
163
164
165	@pytest.hookimpl	1✔
166	def pytest_collection_modifyitems(	1✔
167	session: pytest.Session, config: "Config", items: list[pytest.Item]
168	) -> None:
169	enforce_single_aws_marker(items)	1✔
170	filter_by_markers(config, items)	1✔
171
172
173	@pytest.hookimpl	1✔
174	def pytest_configure(config):	1✔
175	config.addinivalue_line(	1✔
176	"markers",
177	"skip_offline: mark the test to be skipped when the tests are run offline "
178	"(this test explicitly / semantically needs an internet connection)",
179	)
180	config.addinivalue_line(	1✔
181	"markers",
182	"only_on_amd64: mark the test as running only in an amd64 (i.e., x86_64) environment",
183	)
184	config.addinivalue_line(	1✔
185	"markers",
186	"only_on_arm64: mark the test as running only in an arm64 environment",
187	)
188	config.addinivalue_line(	1✔
189	"markers",
190	"requires_in_container: mark the test as running only in a container (e.g., requires installation of system packages)",
191	)
192	config.addinivalue_line(	1✔
193	"markers",
194	"resource_heavy: mark the test as resource-heavy, e.g., downloading very large external dependencies, "
195	"or requiring high amount of RAM/CPU (can be systematically sampled/optimized in the future)",
196	)
197	config.addinivalue_line(	1✔
198	"markers",
199	"aws_validated: mark the test as validated / verified against real AWS",
200	)
201	config.addinivalue_line(	1✔
202	"markers",
203	"aws_only_localstack: mark the test as inherently incompatible with AWS, e.g. when testing localstack-specific features",
204	)
205	config.addinivalue_line(	1✔
206	"markers",
207	"aws_needs_fixing: test fails against AWS but it shouldn't. Might need refactoring, additional permissions, etc.",
208	)
209	config.addinivalue_line(	1✔
210	"markers",
211	"aws_manual_setup_required: validated against real AWS but needs additional setup or account configuration (e.g. increased service quotas)",
212	)
213	config.addinivalue_line(	1✔
214	"markers",
215	"aws_unknown: it's unknown if the test works (reliably) against AWS or not",
216	)
217	config.addinivalue_line(	1✔
218	"markers",
219	"multiruntime: parametrize test against multiple Lambda runtimes",
220	)
221	config.addinivalue_line(	1✔
222	"markers",
223	"requires_docker: mark the test as requiring docker (or a compatible container engine) - will not work on kubernetes.",
224	)
225	config.addinivalue_line(	1✔
226	"markers",
227	"requires_in_process: mark the test as requiring the test to run inside the same process as LocalStack - will not work if tests are run against a running LS container.",
228	)

localstack / localstack / 16981563750

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