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

94.44

/localstack-core/localstack/services/stepfunctions/backend/test_state/execution.py

from __future__ import annotations

import logging
import threading

from localstack.aws.api.stepfunctions import (
    Arn,
    ExecutionStatus,
    InspectionLevel,
    StateMachineType,
    TestExecutionStatus,
    TestStateOutput,
    Timestamp,
)
from localstack.services.stepfunctions.asl.eval.evaluation_details import EvaluationDetails
from localstack.services.stepfunctions.asl.eval.program_state import (
    ProgramEnded,
    ProgramError,
    ProgramState,
)
from localstack.services.stepfunctions.asl.eval.test_state.program_state import (
    ProgramChoiceSelected,
)
from localstack.services.stepfunctions.asl.utils.encoding import to_json_str
from localstack.services.stepfunctions.backend.activity import Activity
from localstack.services.stepfunctions.backend.execution import (
    BaseExecutionWorkerCommunication,
    Execution,
)
from localstack.services.stepfunctions.backend.state_machine import StateMachineInstance
from localstack.services.stepfunctions.backend.test_state.execution_worker import (
    TestStateExecutionWorker,
)

LOG = logging.getLogger(__name__)


class TestStateExecution(Execution):
    exec_worker: TestStateExecutionWorker | None
    next_state: str | None

    class TestCaseExecutionWorkerCommunication(BaseExecutionWorkerCommunication):
        _execution: TestStateExecution

        def terminated(self) -> None:
            exit_program_state: ProgramState = self.execution.exec_worker.env.program_state()
            if isinstance(exit_program_state, ProgramChoiceSelected):
                self.execution.exec_status = ExecutionStatus.SUCCEEDED
                self.execution.output = self.execution.exec_worker.env.states.get_input()
                self.execution.next_state = exit_program_state.next_state_name
            else:
                self._reflect_execution_status()

    def __init__(
        self,
        name: str,
        role_arn: Arn,
        exec_arn: Arn,
        account_id: str,
        region_name: str,
        state_machine: StateMachineInstance,
        start_date: Timestamp,
        activity_store: dict[Arn, Activity],
        input_data: dict | None = None,
    ):
        super().__init__(
            name=name,
            sm_type=StateMachineType.STANDARD,
            role_arn=role_arn,
            exec_arn=exec_arn,
            account_id=account_id,
            region_name=region_name,
            state_machine=state_machine,
            start_date=start_date,
            activity_store=activity_store,
            input_data=input_data,
            cloud_watch_logging_session=None,
            trace_header=None,
        )
        self._execution_terminated_event = threading.Event()
        self.next_state = None

    def _get_start_execution_worker_comm(self) -> BaseExecutionWorkerCommunication:
        return self.TestCaseExecutionWorkerCommunication(self)

    def _get_start_execution_worker(self) -> TestStateExecutionWorker:
        return TestStateExecutionWorker(
            evaluation_details=EvaluationDetails(
                aws_execution_details=self._get_start_aws_execution_details(),
                execution_details=self.get_start_execution_details(),
                state_machine_details=self.get_start_state_machine_details(),
            ),
            exec_comm=self._get_start_execution_worker_comm(),
            cloud_watch_logging_session=self._cloud_watch_logging_session,
            activity_store=self._activity_store,
        )

    def publish_execution_status_change_event(self):
        # Do not publish execution status change events during test state execution.
        pass

    def to_test_state_output(self, inspection_level: InspectionLevel) -> TestStateOutput:
        exit_program_state: ProgramState = self.exec_worker.env.program_state()
        if isinstance(exit_program_state, ProgramEnded):
            output_str = to_json_str(self.output)
            test_state_output = TestStateOutput(
                status=TestExecutionStatus.SUCCEEDED, output=output_str
            )
        elif isinstance(exit_program_state, ProgramError):
            test_state_output = TestStateOutput(
                status=TestExecutionStatus.FAILED,
                error=exit_program_state.error["error"],
                cause=exit_program_state.error["cause"],
            )
        elif isinstance(exit_program_state, ProgramChoiceSelected):
            output_str = to_json_str(self.output)
            test_state_output = TestStateOutput(
                status=TestExecutionStatus.SUCCEEDED, nextState=self.next_state, output=output_str
            )
        else:
            # TODO: handle other statuses
            LOG.warning(
                "Unsupported StateMachine exit type for TestState '%s'",
                type(exit_program_state),
            )
            output_str = to_json_str(self.output)
            test_state_output = TestStateOutput(
                status=TestExecutionStatus.FAILED, output=output_str
            )

        match inspection_level:
            case InspectionLevel.TRACE:
                test_state_output["inspectionData"] = self.exec_worker.env.inspection_data
            case InspectionLevel.DEBUG:
                test_state_output["inspectionData"] = self.exec_worker.env.inspection_data

        return test_state_output

1	from __future__ import annotations	1✔
2
3	import logging	1✔
4	import threading	1✔
5
6	from localstack.aws.api.stepfunctions import (	1✔
7	Arn,
8	ExecutionStatus,
9	InspectionLevel,
10	StateMachineType,
11	TestExecutionStatus,
12	TestStateOutput,
13	Timestamp,
14	)
15	from localstack.services.stepfunctions.asl.eval.evaluation_details import EvaluationDetails	1✔
16	from localstack.services.stepfunctions.asl.eval.program_state import (	1✔
17	ProgramEnded,
18	ProgramError,
19	ProgramState,
20	)
21	from localstack.services.stepfunctions.asl.eval.test_state.program_state import (	1✔
22	ProgramChoiceSelected,
23	)
24	from localstack.services.stepfunctions.asl.utils.encoding import to_json_str	1✔
25	from localstack.services.stepfunctions.backend.activity import Activity	1✔
26	from localstack.services.stepfunctions.backend.execution import (	1✔
27	BaseExecutionWorkerCommunication,
28	Execution,
29	)
30	from localstack.services.stepfunctions.backend.state_machine import StateMachineInstance	1✔
31	from localstack.services.stepfunctions.backend.test_state.execution_worker import (	1✔
32	TestStateExecutionWorker,
33	)
34
35	LOG = logging.getLogger(__name__)	1✔
36
37
38	class TestStateExecution(Execution):	1✔
39	exec_worker: TestStateExecutionWorker \| None	1✔
40	next_state: str \| None	1✔
41
42	class TestCaseExecutionWorkerCommunication(BaseExecutionWorkerCommunication):	1✔
43	_execution: TestStateExecution	1✔
44
45	def terminated(self) -> None:	1✔
46	exit_program_state: ProgramState = self.execution.exec_worker.env.program_state()	1✔
47	if isinstance(exit_program_state, ProgramChoiceSelected):	1✔
48	self.execution.exec_status = ExecutionStatus.SUCCEEDED	1✔
49	self.execution.output = self.execution.exec_worker.env.states.get_input()	1✔
50	self.execution.next_state = exit_program_state.next_state_name	1✔
51	else:
52	self._reflect_execution_status()	1✔
53
54	def __init__(	1✔
55	self,
56	name: str,
57	role_arn: Arn,
58	exec_arn: Arn,
59	account_id: str,
60	region_name: str,
61	state_machine: StateMachineInstance,
62	start_date: Timestamp,
63	activity_store: dict[Arn, Activity],
64	input_data: dict \| None = None,
65	):
66	super().__init__(	1✔
67	name=name,
68	sm_type=StateMachineType.STANDARD,
69	role_arn=role_arn,
70	exec_arn=exec_arn,
71	account_id=account_id,
72	region_name=region_name,
73	state_machine=state_machine,
74	start_date=start_date,
75	activity_store=activity_store,
76	input_data=input_data,
77	cloud_watch_logging_session=None,
78	trace_header=None,
79	)
80	self._execution_terminated_event = threading.Event()	1✔
81	self.next_state = None	1✔
82
83	def _get_start_execution_worker_comm(self) -> BaseExecutionWorkerCommunication:	1✔
84	return self.TestCaseExecutionWorkerCommunication(self)	1✔
85
86	def _get_start_execution_worker(self) -> TestStateExecutionWorker:	1✔
87	return TestStateExecutionWorker(	1✔
88	evaluation_details=EvaluationDetails(
89	aws_execution_details=self._get_start_aws_execution_details(),
90	execution_details=self.get_start_execution_details(),
91	state_machine_details=self.get_start_state_machine_details(),
92	),
93	exec_comm=self._get_start_execution_worker_comm(),
94	cloud_watch_logging_session=self._cloud_watch_logging_session,
95	activity_store=self._activity_store,
96	)
97
98	def publish_execution_status_change_event(self):	1✔
99	# Do not publish execution status change events during test state execution.
100	pass	1✔
101
102	def to_test_state_output(self, inspection_level: InspectionLevel) -> TestStateOutput:	1✔
103	exit_program_state: ProgramState = self.exec_worker.env.program_state()	1✔
104	if isinstance(exit_program_state, ProgramEnded):	1✔
105	output_str = to_json_str(self.output)	1✔
106	test_state_output = TestStateOutput(	1✔
107	status=TestExecutionStatus.SUCCEEDED, output=output_str
108	)
109	elif isinstance(exit_program_state, ProgramError):	1✔
110	test_state_output = TestStateOutput(	1✔
111	status=TestExecutionStatus.FAILED,
112	error=exit_program_state.error["error"],
113	cause=exit_program_state.error["cause"],
114	)
115	elif isinstance(exit_program_state, ProgramChoiceSelected):	1✔
116	output_str = to_json_str(self.output)	1✔
117	test_state_output = TestStateOutput(	1✔
118	status=TestExecutionStatus.SUCCEEDED, nextState=self.next_state, output=output_str
119	)
120	else:
121	# TODO: handle other statuses
UNCOV 122	LOG.warning(	×
123	"Unsupported StateMachine exit type for TestState '%s'",
124	type(exit_program_state),
125	)
UNCOV 126	output_str = to_json_str(self.output)	×
127	test_state_output = TestStateOutput(	×
128	status=TestExecutionStatus.FAILED, output=output_str
129	)
130
131	match inspection_level:	1✔
132	case InspectionLevel.TRACE:	1✔
133	test_state_output["inspectionData"] = self.exec_worker.env.inspection_data	1✔
134	case InspectionLevel.DEBUG:	1✔
135	test_state_output["inspectionData"] = self.exec_worker.env.inspection_data	1✔
136
137	return test_state_output	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