22519085314

Committed 27 Feb 2026 11:47PM UTC coverage: 86.962% (+0.006%) from 86.956%

Build # 22519085314

Build Type

push

github

Committed by

web-flow

Commit Message

SNS: update store typing (#13866)

Run Details

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

388 existing lines in 19 files now uncovered.

69828 of 80297 relevant lines covered (86.96%)

0.87 hits per line

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

96.05

/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 (
    ProgramCaughtError,
    ProgramChoiceSelected,
    ProgramRetriable,
)
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,
)
from localstack.services.stepfunctions.backend.test_state.test_state_mock import TestStateMock

LOG = logging.getLogger(__name__)


class TestStateExecution(Execution):
    exec_worker: TestStateExecutionWorker | None
    next_state: str | None
    state_name: str | None
    mock: TestStateMock | None
    variables: dict | 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
            elif isinstance(exit_program_state, ProgramCaughtError):
                self.execution.exec_status = ExecutionStatus.SUCCEEDED
                self.execution.error = exit_program_state.error
                self.execution.cause = exit_program_state.cause
                self.execution.output = self.execution.exec_worker.env.states.get_input()
                self.execution.next_state = exit_program_state.next_state_name
            elif isinstance(exit_program_state, ProgramRetriable):
                self.execution.exec_status = ExecutionStatus.SUCCEEDED
                self.execution.error = exit_program_state.error
                self.execution.cause = exit_program_state.cause
            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],
        state_name: str | None = None,
        input_data: dict | None = None,
        mock: TestStateMock | None = None,
        variables: 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
        self.state_name = state_name
        self.mock = mock
        self.variables = variables

    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,
            state_name=self.state_name,
            mock=self.mock,
            variables=self.variables,
        )

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

localstack / localstack / 22519085314

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