18505123992

Committed 14 Oct 2025 05:30PM UTC coverage: 86.888% (-0.01%) from 86.899%

Build # 18505123992

Build Type

push

github

Committed by

web-flow

Commit Message

S3: fix `aws-global` validation in CreateBucket (#13250)

Run Details

10 of 10 new or added lines in 4 files covered. (100.0%)

831 existing lines in 40 files now uncovered.

68028 of 78294 relevant lines covered (86.89%)

0.87 hits per line

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

51.72

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

"""
When a test (in tests/aws) is executed against AWS, we want to track the date of the last successful run.

Keeping a record of how long ago a test was validated last,
we can periodically re-validate ALL AWS-targeting tests (and therefore not only just snapshot-using tests).
"""

import datetime
import json
import os
from pathlib import Path

import pytest
from pluggy import Result
from pytest import StashKey, TestReport

from localstack.testing.aws.util import is_aws_cloud

durations_key = StashKey[dict[str, float]]()
"""
Stores phase durations on the test node between execution phases.
See https://docs.pytest.org/en/latest/reference/reference.html#pytest.Stash
"""
test_failed_key = StashKey[bool]()
"""
Stores information from call execution phase about whether the test failed.
"""


def find_validation_data_for_item(item: pytest.Item) -> dict | None:
    base_path = os.path.join(item.fspath.dirname, item.fspath.purebasename)
    snapshot_path = f"{base_path}.validation.json"

    if not os.path.exists(snapshot_path):
        return None

    with open(snapshot_path) as fd:
        file_content = json.load(fd)
        return file_content.get(item.nodeid)


@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item: pytest.Item, call: pytest.CallInfo):
    """
    This hook is called after each test execution phase (setup, call, teardown).
    """
    result: Result = yield
    report: TestReport = result.get_result()

    if call.when == "setup":
        _makereport_setup(item, call)
    elif call.when == "call":
        _makereport_call(item, call)
    elif call.when == "teardown":
        _makereport_teardown(item, call)

    return report


def _stash_phase_duration(call, item):
    durations_by_phase = item.stash.setdefault(durations_key, {})
    durations_by_phase[call.when] = round(call.duration, 2)


def _makereport_setup(item: pytest.Item, call: pytest.CallInfo):
    _stash_phase_duration(call, item)


def _makereport_call(item: pytest.Item, call: pytest.CallInfo):
    _stash_phase_duration(call, item)
    item.stash[test_failed_key] = call.excinfo is not None


def _makereport_teardown(item: pytest.Item, call: pytest.CallInfo):
    _stash_phase_duration(call, item)

    # only update the file when running against AWS and the test finishes successfully
    if not is_aws_cloud() or item.stash.get(test_failed_key, True):
        return

    base_path = os.path.join(item.fspath.dirname, item.fspath.purebasename)
    file_path = Path(f"{base_path}.validation.json")
    file_path.touch()
    with file_path.open(mode="r+") as fd:
        # read existing state from file
        try:
            content = json.load(fd)
        except json.JSONDecodeError:  # expected on the first try (empty file)
            content = {}

        test_execution_data = content.setdefault(item.nodeid, {})

        timestamp = datetime.datetime.now(tz=datetime.timezone.utc)
        test_execution_data["last_validated_date"] = timestamp.isoformat(timespec="seconds")

        durations_by_phase = item.stash[durations_key]
        test_execution_data["durations_in_seconds"] = durations_by_phase

        total_duration = sum(durations_by_phase.values())
        durations_by_phase["total"] = round(total_duration, 2)

        # For json.dump sorted test entries enable consistent diffs.
        # But test execution data is more readable in insert order for each step (setup, call, teardown).
        # Hence, not using global sort_keys=True for json.dump but rather additionally sorting top-level dict only.
        content = dict(sorted(content.items()))

        # save updates
        fd.truncate(0)  # clear existing content
        fd.seek(0)
        json.dump(content, fd, indent=2)
        fd.write("\n")  # add trailing newline for linter and Git compliance


@pytest.hookimpl
def pytest_addoption(parser: pytest.Parser, pluginmanager: pytest.PytestPluginManager):
    parser.addoption("--validation-date-limit-days", action="store")
    parser.addoption("--validation-date-limit-timestamp", action="store")


@pytest.hookimpl(trylast=True)
def pytest_collection_modifyitems(
    session: pytest.Session, config: pytest.Config, items: list[pytest.Item]
):
    """
    Collect only items that have a validation timestamp earlier than the user-provided reference timestamp

    Example usage:
    - pytest ... --validation-date-limit-days=10
    - pytest ... --validation-date-limit-timestamp="2023-12-01T00:00:00"

    """
    # handle two potential config options (relative vs. absolute limits)
    if config.option.validation_date_limit_days is not None:
        reference_date = datetime.datetime.now(tz=datetime.timezone.utc) - datetime.timedelta(
            days=int(config.option.validation_date_limit_days)
        )
    elif config.option.validation_date_limit_timestamp is not None:
        reference_date = datetime.datetime.fromisoformat(
            config.option.validation_date_limit_timestamp
        )
    else:
        return

    selected = []  # items to collect
    deselected = []  # items to drop

    for item in items:
        validation_data = find_validation_data_for_item(item)
        if not validation_data:
            deselected.append(item)
            continue

        last_validated_date = datetime.datetime.fromisoformat(
            validation_data["last_validated_date"]
        )

        if last_validated_date < reference_date:
            selected.append(item)
        else:
            deselected.append(item)

    items[:] = selected
    config.hook.pytest_deselected(items=deselected)

1	"""
2	When a test (in tests/aws) is executed against AWS, we want to track the date of the last successful run.
3
4	Keeping a record of how long ago a test was validated last,
5	we can periodically re-validate ALL AWS-targeting tests (and therefore not only just snapshot-using tests).
6	"""
7
8	import datetime	1✔
9	import json	1✔
10	import os	1✔
11	from pathlib import Path	1✔
12
13	import pytest	1✔
14	from pluggy import Result	1✔
15	from pytest import StashKey, TestReport	1✔
16
17	from localstack.testing.aws.util import is_aws_cloud	1✔
18
19	durations_key = StashKey[dict[str, float]]()	1✔
20	"""	1✔
21	Stores phase durations on the test node between execution phases.
22	See https://docs.pytest.org/en/latest/reference/reference.html#pytest.Stash
23	"""
24	test_failed_key = StashKey[bool]()	1✔
25	"""	1✔
26	Stores information from call execution phase about whether the test failed.
27	"""
28
29
30	def find_validation_data_for_item(item: pytest.Item) -> dict \| None:	1✔
UNCOV 31	base_path = os.path.join(item.fspath.dirname, item.fspath.purebasename)	×
32	snapshot_path = f"{base_path}.validation.json"	×
33
UNCOV 34	if not os.path.exists(snapshot_path):	×
35	return None	×
36
UNCOV 37	with open(snapshot_path) as fd:	×
38	file_content = json.load(fd)	×
39	return file_content.get(item.nodeid)	×
40
41
42	@pytest.hookimpl(hookwrapper=True)	1✔
43	def pytest_runtest_makereport(item: pytest.Item, call: pytest.CallInfo):	1✔
44	"""
45	This hook is called after each test execution phase (setup, call, teardown).
46	"""
47	result: Result = yield	1✔
48	report: TestReport = result.get_result()	1✔
49
50	if call.when == "setup":	1✔
51	_makereport_setup(item, call)	1✔
52	elif call.when == "call":	1✔
53	_makereport_call(item, call)	1✔
54	elif call.when == "teardown":	1✔
55	_makereport_teardown(item, call)	1✔
56
57	return report	1✔
58
59
60	def _stash_phase_duration(call, item):	1✔
61	durations_by_phase = item.stash.setdefault(durations_key, {})	1✔
62	durations_by_phase[call.when] = round(call.duration, 2)	1✔
63
64
65	def _makereport_setup(item: pytest.Item, call: pytest.CallInfo):	1✔
66	_stash_phase_duration(call, item)	1✔
67
68
69	def _makereport_call(item: pytest.Item, call: pytest.CallInfo):	1✔
70	_stash_phase_duration(call, item)	1✔
71	item.stash[test_failed_key] = call.excinfo is not None	1✔
72
73
74	def _makereport_teardown(item: pytest.Item, call: pytest.CallInfo):	1✔
75	_stash_phase_duration(call, item)	1✔
76
77	# only update the file when running against AWS and the test finishes successfully
78	if not is_aws_cloud() or item.stash.get(test_failed_key, True):	1✔
79	return	1✔
80
UNCOV 81	base_path = os.path.join(item.fspath.dirname, item.fspath.purebasename)	×
82	file_path = Path(f"{base_path}.validation.json")	×
83	file_path.touch()	×
84	with file_path.open(mode="r+") as fd:	×
85	# read existing state from file
UNCOV 86	try:	×
87	content = json.load(fd)	×
88	except json.JSONDecodeError: # expected on the first try (empty file)	×
89	content = {}	×
90
UNCOV 91	test_execution_data = content.setdefault(item.nodeid, {})	×
92
UNCOV 93	timestamp = datetime.datetime.now(tz=datetime.timezone.utc)	×
94	test_execution_data["last_validated_date"] = timestamp.isoformat(timespec="seconds")	×
95
UNCOV 96	durations_by_phase = item.stash[durations_key]	×
97	test_execution_data["durations_in_seconds"] = durations_by_phase	×
98
UNCOV 99	total_duration = sum(durations_by_phase.values())	×
100	durations_by_phase["total"] = round(total_duration, 2)	×
101
102	# For json.dump sorted test entries enable consistent diffs.
103	# But test execution data is more readable in insert order for each step (setup, call, teardown).
104	# Hence, not using global sort_keys=True for json.dump but rather additionally sorting top-level dict only.
UNCOV 105	content = dict(sorted(content.items()))	×
106
107	# save updates
UNCOV 108	fd.truncate(0) # clear existing content	×
109	fd.seek(0)	×
110	json.dump(content, fd, indent=2)	×
111	fd.write("\n") # add trailing newline for linter and Git compliance	×
112
113
114	@pytest.hookimpl	1✔
115	def pytest_addoption(parser: pytest.Parser, pluginmanager: pytest.PytestPluginManager):	1✔
116	parser.addoption("--validation-date-limit-days", action="store")	1✔
117	parser.addoption("--validation-date-limit-timestamp", action="store")	1✔
118
119
120	@pytest.hookimpl(trylast=True)	1✔
121	def pytest_collection_modifyitems(	1✔
122	session: pytest.Session, config: pytest.Config, items: list[pytest.Item]
123	):
124	"""
125	Collect only items that have a validation timestamp earlier than the user-provided reference timestamp
126
127	Example usage:
128	- pytest ... --validation-date-limit-days=10
129	- pytest ... --validation-date-limit-timestamp="2023-12-01T00:00:00"
130
131	"""
132	# handle two potential config options (relative vs. absolute limits)
133	if config.option.validation_date_limit_days is not None:	1✔
UNCOV 134	reference_date = datetime.datetime.now(tz=datetime.timezone.utc) - datetime.timedelta(	×
135	days=int(config.option.validation_date_limit_days)
136	)
137	elif config.option.validation_date_limit_timestamp is not None:	1✔
UNCOV 138	reference_date = datetime.datetime.fromisoformat(	×
139	config.option.validation_date_limit_timestamp
140	)
141	else:
142	return	1✔
143
UNCOV 144	selected = [] # items to collect	×
145	deselected = [] # items to drop	×
146
UNCOV 147	for item in items:	×
148	validation_data = find_validation_data_for_item(item)	×
149	if not validation_data:	×
150	deselected.append(item)	×
151	continue	×
152
UNCOV 153	last_validated_date = datetime.datetime.fromisoformat(	×
154	validation_data["last_validated_date"]
155	)
156
UNCOV 157	if last_validated_date < reference_date:	×
158	selected.append(item)	×
159	else:
UNCOV 160	deselected.append(item)	×
161
UNCOV 162	items[:] = selected	×
163	config.hook.pytest_deselected(items=deselected)	×

localstack / localstack / 18505123992

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