20565403496

Committed 29 Dec 2025 05:11AM UTC coverage: 84.103% (-2.8%) from 86.921%

Build # 20565403496

Build Type

Pull #13567

github

Committed by

web-flow

Commit Message

Merge 4816837a5 into 2417384aa

Pull Request Pull Request #13567: Update ASF APIs

Run Details

67166 of 79862 relevant lines covered (84.1%)

0.84 hits per line

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

92.92

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

"""
Custom pytest mark typings
"""

import os
from collections.abc import Callable
from typing import TYPE_CHECKING

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: "list[str] | None" = None,
        condition: "Callable[[...], bool] | None" = None,
    ): ...


class MultiRuntimeMarker:
    def __call__(self, *, scenario: str, runtimes: list[str] | None = 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"""
    k8s_always_run = pytest.mark.k8s_always_run
    """This tests will always run against k8s environment"""
    skip_k8s = pytest.mark.skip_k8s
    """This test will be skipped in k8s environment"""


# 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.",
    )
    config.addinivalue_line(
        "markers",
        "k8s_always_run: mark the test to always run in k8s environment. This allows us to run tests that would otherwise be skipped, such as localstack_only tests.",
    )

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

localstack / localstack / 20565403496

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