16820655284

Committed 07 Aug 2025 05:03PM UTC coverage: 86.841% (-0.05%) from 86.892%

Build # 16820655284

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: support CDK bootstrap and deployment (#12967)

Coverage Stats

32 of 38 new or added lines in 5 files covered. (84.21%)

2013 existing lines in 125 files now uncovered.

66606 of 76699 relevant lines covered (86.84%)

0.87 hits per line

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

80.0

/localstack-core/localstack/services/stepfunctions/stepfunctions_utils.py

import base64
import logging

from localstack.aws.api.stepfunctions import ValidationException
from localstack.aws.connect import connect_to
from localstack.utils.aws.arns import parse_arn
from localstack.utils.common import retry
from localstack.utils.strings import to_bytes, to_str

LOG = logging.getLogger(__name__)


def await_sfn_execution_result(execution_arn: str, timeout_secs: int = 60) -> dict:
    """Wait until the given SFN execution ARN is no longer in RUNNING status, then return execution result."""

    arn_data = parse_arn(execution_arn)

    client = connect_to(
        aws_access_key_id=arn_data["account"], region_name=arn_data["region"]
    ).stepfunctions

    def _get_result():
        result = client.describe_execution(executionArn=execution_arn)
        assert result["status"] != "RUNNING"
        return result

    return retry(_get_result, sleep=2, retries=timeout_secs / 2)


def get_next_page_token_from_arn(resource_arn: str) -> str:
    return to_str(base64.b64encode(to_bytes(resource_arn)))


_DEFAULT_SFN_MAX_RESULTS: int = 100


def normalise_max_results(max_results: int = 100) -> int:
    if not max_results:
        return _DEFAULT_SFN_MAX_RESULTS
    return max_results


def assert_pagination_parameters_valid(
    max_results: int,
    next_token: str,
    next_token_length_limit: int = 1024,
    max_results_upper_limit: int = 1000,
) -> None:
    validation_errors = []

    match max_results:
        case int() if max_results > max_results_upper_limit:
            validation_errors.append(
                f"Value '{max_results}' at 'maxResults' failed to satisfy constraint: "
                f"Member must have value less than or equal to {max_results_upper_limit}"
            )

    match next_token:
        case str() if len(next_token) > next_token_length_limit:
            validation_errors.append(
                f"Value '{next_token}' at 'nextToken' failed to satisfy constraint: "
                f"Member must have length less than or equal to {next_token_length_limit}"
            )

    if validation_errors:
        errors_message = "; ".join(validation_errors)
        message = f"{len(validation_errors)} validation {'errors' if len(validation_errors) > 1 else 'error'} detected: {errors_message}"
        raise ValidationException(message)

1	import base64	1✔
2	import logging	1✔
3
4	from localstack.aws.api.stepfunctions import ValidationException	1✔
5	from localstack.aws.connect import connect_to	1✔
6	from localstack.utils.aws.arns import parse_arn	1✔
7	from localstack.utils.common import retry	1✔
8	from localstack.utils.strings import to_bytes, to_str	1✔
9
10	LOG = logging.getLogger(__name__)	1✔
11
12
13	def await_sfn_execution_result(execution_arn: str, timeout_secs: int = 60) -> dict:	1✔
14	"""Wait until the given SFN execution ARN is no longer in RUNNING status, then return execution result."""
15
UNCOV 16	arn_data = parse_arn(execution_arn)	×
17
UNCOV 18	client = connect_to(	×
19	aws_access_key_id=arn_data["account"], region_name=arn_data["region"]
20	).stepfunctions
21
UNCOV 22	def _get_result():	×
23	result = client.describe_execution(executionArn=execution_arn)	×
24	assert result["status"] != "RUNNING"	×
25	return result	×
26
UNCOV 27	return retry(_get_result, sleep=2, retries=timeout_secs / 2)	×
28
29
30	def get_next_page_token_from_arn(resource_arn: str) -> str:	1✔
31	return to_str(base64.b64encode(to_bytes(resource_arn)))	1✔
32
33
34	_DEFAULT_SFN_MAX_RESULTS: int = 100	1✔
35
36
37	def normalise_max_results(max_results: int = 100) -> int:	1✔
38	if not max_results:	1✔
39	return _DEFAULT_SFN_MAX_RESULTS	1✔
40	return max_results	1✔
41
42
43	def assert_pagination_parameters_valid(	1✔
44	max_results: int,
45	next_token: str,
46	next_token_length_limit: int = 1024,
47	max_results_upper_limit: int = 1000,
48	) -> None:
49	validation_errors = []	1✔
50
51	match max_results:	1✔
52	case int() if max_results > max_results_upper_limit:	1✔
53	validation_errors.append(	1✔
54	f"Value '{max_results}' at 'maxResults' failed to satisfy constraint: "
55	f"Member must have value less than or equal to {max_results_upper_limit}"
56	)
57
58	match next_token:	1✔
59	case str() if len(next_token) > next_token_length_limit:	1✔
60	validation_errors.append(	1✔
61	f"Value '{next_token}' at 'nextToken' failed to satisfy constraint: "
62	f"Member must have length less than or equal to {next_token_length_limit}"
63	)
64
65	if validation_errors:	1✔
66	errors_message = "; ".join(validation_errors)	1✔
67	message = f"{len(validation_errors)} validation {'errors' if len(validation_errors) > 1 else 'error'} detected: {errors_message}"	1✔
68	raise ValidationException(message)	1✔

localstack / localstack / 16820655284

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