5ab00ad5-db37-414f-8bc8-79add3c39e5f

Committed 29 May 2025 01:45PM UTC coverage: 86.677% (+0.03%) from 86.646%

Build # 5ab00ad5-db37-414f-8bc8-79add3c39e5f

Build Type

push

circleci

Committed by

web-flow

Commit Message

fix(esm/kinesis): Always store NextShardIterator from GetRecords (#12677)

Run Details

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

85 existing lines in 5 files now uncovered.

64650 of 74587 relevant lines covered (86.68%)

0.87 hits per line

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

52.27

/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
from typing import Dict, Optional

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

localstack / localstack / 5ab00ad5-db37-414f-8bc8-79add3c39e5f

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