16981563750

Committed 14 Aug 2025 10:49PM UTC coverage: 86.896% (+0.04%) from 86.852%

Build # 16981563750

Build Type

push

github

Committed by

web-flow

Commit Message

add support for Fn::Tranform in CFnV2 (#12966)

Co-authored-by: Simon Walker <simon.walker@localstack.cloud>

Coverage Stats

181 of 195 new or added lines in 6 files covered. (92.82%)

348 existing lines in 22 files now uncovered.

66915 of 77006 relevant lines covered (86.9%)

0.87 hits per line

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

97.51

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

from __future__ import annotations

import datetime
import json
import logging
from typing import Final

from localstack.aws.api.events import PutEventsRequestEntry
from localstack.aws.api.stepfunctions import (
    Arn,
    CloudWatchEventsExecutionDataDetails,
    DescribeExecutionOutput,
    DescribeStateMachineForExecutionOutput,
    ExecutionListItem,
    ExecutionStatus,
    GetExecutionHistoryOutput,
    HistoryEventList,
    InvalidName,
    SensitiveCause,
    SensitiveError,
    StartExecutionOutput,
    StartSyncExecutionOutput,
    StateMachineType,
    SyncExecutionStatus,
    Timestamp,
    TraceHeader,
    VariableReferences,
)
from localstack.aws.connect import connect_to
from localstack.services.stepfunctions.asl.eval.evaluation_details import (
    AWSExecutionDetails,
    EvaluationDetails,
    ExecutionDetails,
    StateMachineDetails,
)
from localstack.services.stepfunctions.asl.eval.event.logging import (
    CloudWatchLoggingSession,
)
from localstack.services.stepfunctions.asl.eval.program_state import (
    ProgramEnded,
    ProgramError,
    ProgramState,
    ProgramStopped,
    ProgramTimedOut,
)
from localstack.services.stepfunctions.asl.static_analyser.variable_references_static_analyser import (
    VariableReferencesStaticAnalyser,
)
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_worker import (
    ExecutionWorker,
    SyncExecutionWorker,
)
from localstack.services.stepfunctions.backend.execution_worker_comm import (
    ExecutionWorkerCommunication,
)
from localstack.services.stepfunctions.backend.state_machine import (
    StateMachineInstance,
    StateMachineVersion,
)
from localstack.services.stepfunctions.mocking.mock_config import MockTestCase

LOG = logging.getLogger(__name__)


class BaseExecutionWorkerCommunication(ExecutionWorkerCommunication):
    execution: Final[Execution]

    def __init__(self, execution: Execution):
        self.execution = execution

    def _reflect_execution_status(self):
        exit_program_state: ProgramState = self.execution.exec_worker.env.program_state()
        self.execution.stop_date = datetime.datetime.now(tz=datetime.UTC)
        if isinstance(exit_program_state, ProgramEnded):
            self.execution.exec_status = ExecutionStatus.SUCCEEDED
            self.execution.output = self.execution.exec_worker.env.states.get_input()
        elif isinstance(exit_program_state, ProgramStopped):
            self.execution.exec_status = ExecutionStatus.ABORTED
        elif isinstance(exit_program_state, ProgramError):
            self.execution.exec_status = ExecutionStatus.FAILED
            self.execution.error = exit_program_state.error.get("error")
            self.execution.cause = exit_program_state.error.get("cause")
        elif isinstance(exit_program_state, ProgramTimedOut):
            self.execution.exec_status = ExecutionStatus.TIMED_OUT
        else:
            raise RuntimeWarning(
                f"Execution ended with unsupported ProgramState type '{type(exit_program_state)}'."
            )

    def terminated(self) -> None:
        self._reflect_execution_status()
        self.execution.publish_execution_status_change_event()


class Execution:
    name: Final[str]
    sm_type: Final[StateMachineType]
    role_arn: Final[Arn]
    exec_arn: Final[Arn]

    account_id: str
    region_name: str

    state_machine: Final[StateMachineInstance]
    state_machine_arn: Final[Arn]
    state_machine_version_arn: Final[Arn | None]
    state_machine_alias_arn: Final[Arn | None]

    mock_test_case: Final[MockTestCase | None]

    start_date: Final[Timestamp]
    input_data: Final[json | None]
    input_details: Final[CloudWatchEventsExecutionDataDetails | None]
    trace_header: Final[TraceHeader | None]
    _cloud_watch_logging_session: Final[CloudWatchLoggingSession | None]

    exec_status: ExecutionStatus | None
    stop_date: Timestamp | None

    output: json | None
    output_details: CloudWatchEventsExecutionDataDetails | None

    error: SensitiveError | None
    cause: SensitiveCause | None

    exec_worker: ExecutionWorker | None

    _activity_store: dict[Arn, Activity]

    def __init__(
        self,
        name: str,
        sm_type: StateMachineType,
        role_arn: Arn,
        exec_arn: Arn,
        account_id: str,
        region_name: str,
        state_machine: StateMachineInstance,
        start_date: Timestamp,
        cloud_watch_logging_session: CloudWatchLoggingSession | None,
        activity_store: dict[Arn, Activity],
        input_data: json | None = None,
        trace_header: TraceHeader | None = None,
        state_machine_alias_arn: Arn | None = None,
        mock_test_case: MockTestCase | None = None,
    ):
        self.name = name
        self.sm_type = sm_type
        self.role_arn = role_arn
        self.exec_arn = exec_arn
        self.account_id = account_id
        self.region_name = region_name
        self.state_machine = state_machine
        if isinstance(state_machine, StateMachineVersion):
            self.state_machine_arn = state_machine.source_arn
            self.state_machine_version_arn = state_machine.arn
        else:
            self.state_machine_arn = state_machine.arn
            self.state_machine_version_arn = None
        self.state_machine_alias_arn = state_machine_alias_arn
        self.start_date = start_date
        self._cloud_watch_logging_session = cloud_watch_logging_session
        self.input_data = input_data
        self.input_details = CloudWatchEventsExecutionDataDetails(included=True)
        self.trace_header = trace_header
        self.exec_status = None
        self.stop_date = None
        self.output = None
        self.output_details = CloudWatchEventsExecutionDataDetails(included=True)
        self.exec_worker = None
        self.error = None
        self.cause = None
        self._activity_store = activity_store
        self.mock_test_case = mock_test_case

    def _get_events_client(self):
        return connect_to(aws_access_key_id=self.account_id, region_name=self.region_name).events

    def to_start_output(self) -> StartExecutionOutput:
        return StartExecutionOutput(executionArn=self.exec_arn, startDate=self.start_date)

    def to_describe_output(self) -> DescribeExecutionOutput:
        describe_output = DescribeExecutionOutput(
            executionArn=self.exec_arn,
            stateMachineArn=self.state_machine_arn,
            name=self.name,
            status=self.exec_status,
            startDate=self.start_date,
            stopDate=self.stop_date,
            input=to_json_str(self.input_data, separators=(",", ":")),
            inputDetails=self.input_details,
            traceHeader=self.trace_header,
        )
        if describe_output["status"] == ExecutionStatus.SUCCEEDED:
            describe_output["output"] = to_json_str(self.output, separators=(",", ":"))
            describe_output["outputDetails"] = self.output_details
        if self.error is not None:
            describe_output["error"] = self.error
        if self.cause is not None:
            describe_output["cause"] = self.cause
        if self.state_machine_version_arn is not None:
            describe_output["stateMachineVersionArn"] = self.state_machine_version_arn
        if self.state_machine_alias_arn is not None:
            describe_output["stateMachineAliasArn"] = self.state_machine_alias_arn
        return describe_output

    def to_describe_state_machine_for_execution_output(
        self,
    ) -> DescribeStateMachineForExecutionOutput:
        state_machine: StateMachineInstance = self.state_machine
        state_machine_arn = (
            state_machine.source_arn
            if isinstance(state_machine, StateMachineVersion)
            else state_machine.arn
        )
        out = DescribeStateMachineForExecutionOutput(
            stateMachineArn=state_machine_arn,
            name=state_machine.name,
            definition=state_machine.definition,
            roleArn=self.role_arn,
            # The date and time the state machine associated with an execution was updated.
            updateDate=state_machine.create_date,
            loggingConfiguration=state_machine.logging_config,
        )
        revision_id = self.state_machine.revision_id
        if self.state_machine.revision_id:
            out["revisionId"] = revision_id
        variable_references: VariableReferences = VariableReferencesStaticAnalyser.process_and_get(
            definition=self.state_machine.definition
        )
        if variable_references:
            out["variableReferences"] = variable_references
        return out

    def to_execution_list_item(self) -> ExecutionListItem:
        if isinstance(self.state_machine, StateMachineVersion):
            state_machine_arn = self.state_machine.source_arn
            state_machine_version_arn = self.state_machine.arn
        else:
            state_machine_arn = self.state_machine.arn
            state_machine_version_arn = None

        item = ExecutionListItem(
            executionArn=self.exec_arn,
            stateMachineArn=state_machine_arn,
            name=self.name,
            status=self.exec_status,
            startDate=self.start_date,
            stopDate=self.stop_date,
        )
        if state_machine_version_arn is not None:
            item["stateMachineVersionArn"] = state_machine_version_arn
        if self.state_machine_alias_arn is not None:
            item["stateMachineAliasArn"] = self.state_machine_alias_arn
        return item

    def to_history_output(self) -> GetExecutionHistoryOutput:
        env = self.exec_worker.env
        event_history: HistoryEventList = []
        if env is not None:
            # The execution has not started yet.
            event_history: HistoryEventList = env.event_manager.get_event_history()
        return GetExecutionHistoryOutput(events=event_history)

    @staticmethod
    def _to_serialized_date(timestamp: datetime.datetime) -> str:
        """See test in tests.aws.services.stepfunctions.v2.base.test_base.TestSnfBase.test_execution_dateformat"""
        return f"{timestamp.astimezone(datetime.UTC).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3]}Z"

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

    def _get_start_aws_execution_details(self) -> AWSExecutionDetails:
        return AWSExecutionDetails(
            account=self.account_id, region=self.region_name, role_arn=self.role_arn
        )

    def get_start_execution_details(self) -> ExecutionDetails:
        return ExecutionDetails(
            arn=self.exec_arn,
            name=self.name,
            role_arn=self.role_arn,
            inpt=self.input_data,
            start_time=self._to_serialized_date(self.start_date),
        )

    def get_start_state_machine_details(self) -> StateMachineDetails:
        return StateMachineDetails(
            arn=self.state_machine.arn,
            name=self.state_machine.name,
            typ=self.state_machine.sm_type,
            definition=self.state_machine.definition,
        )

    def _get_start_execution_worker(self) -> ExecutionWorker:
        return ExecutionWorker(
            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,
            mock_test_case=self.mock_test_case,
        )

    def start(self) -> None:
        # TODO: checks exec_worker does not exists already?
        if self.exec_worker:
            raise InvalidName()  # TODO.
        self.exec_worker = self._get_start_execution_worker()
        self.exec_status = ExecutionStatus.RUNNING
        self.publish_execution_status_change_event()
        self.exec_worker.start()

    def stop(self, stop_date: datetime.datetime, error: str | None, cause: str | None):
        exec_worker: ExecutionWorker | None = self.exec_worker
        if exec_worker:
            exec_worker.stop(stop_date=stop_date, cause=cause, error=error)

    def publish_execution_status_change_event(self):
        input_value = (
            {} if not self.input_data else to_json_str(self.input_data, separators=(",", ":"))
        )
        output_value = (
            None if self.output is None else to_json_str(self.output, separators=(",", ":"))
        )
        output_details = None if output_value is None else self.output_details
        entry = PutEventsRequestEntry(
            Source="aws.states",
            Resources=[self.exec_arn],
            DetailType="Step Functions Execution Status Change",
            Detail=to_json_str(
                # Note: this operation carries significant changes from a describe_execution request.
                DescribeExecutionOutput(
                    executionArn=self.exec_arn,
                    stateMachineArn=self.state_machine.arn,
                    stateMachineAliasArn=None,
                    stateMachineVersionArn=None,
                    name=self.name,
                    status=self.exec_status,
                    startDate=self.start_date,
                    stopDate=self.stop_date,
                    input=input_value,
                    inputDetails=self.input_details,
                    output=output_value,
                    outputDetails=output_details,
                    error=self.error,
                    cause=self.cause,
                )
            ),
        )
        try:
            self._get_events_client().put_events(Entries=[entry])
        except Exception:
            LOG.error(
                "Unable to send notification of Entry='%s' for Step Function execution with Arn='%s' to EventBridge.",
                entry,
                self.exec_arn,
                exc_info=LOG.isEnabledFor(logging.DEBUG),
            )


class SyncExecutionWorkerCommunication(BaseExecutionWorkerCommunication):
    execution: Final[SyncExecution]

    def _reflect_execution_status(self) -> None:
        super()._reflect_execution_status()
        exit_status: ExecutionStatus = self.execution.exec_status
        if exit_status == ExecutionStatus.SUCCEEDED:
            self.execution.sync_execution_status = SyncExecutionStatus.SUCCEEDED
        elif exit_status == ExecutionStatus.TIMED_OUT:
            self.execution.sync_execution_status = SyncExecutionStatus.TIMED_OUT
        else:
            self.execution.sync_execution_status = SyncExecutionStatus.FAILED


class SyncExecution(Execution):
    sync_execution_status: SyncExecutionStatus | None = None

    def _get_start_execution_worker(self) -> SyncExecutionWorker:
        return SyncExecutionWorker(
            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,
            mock_test_case=self.mock_test_case,
        )

    def _get_start_execution_worker_comm(self) -> BaseExecutionWorkerCommunication:
        return SyncExecutionWorkerCommunication(self)

    def to_start_sync_execution_output(self) -> StartSyncExecutionOutput:
        start_output = StartSyncExecutionOutput(
            executionArn=self.exec_arn,
            stateMachineArn=self.state_machine.arn,
            name=self.name,
            status=self.sync_execution_status,
            startDate=self.start_date,
            stopDate=self.stop_date,
            input=to_json_str(self.input_data, separators=(",", ":")),
            inputDetails=self.input_details,
            traceHeader=self.trace_header,
        )
        if self.sync_execution_status == SyncExecutionStatus.SUCCEEDED:
            start_output["output"] = to_json_str(self.output, separators=(",", ":"))
        if self.output_details:
            start_output["outputDetails"] = self.output_details
        if self.error is not None:
            start_output["error"] = self.error
        if self.cause is not None:
            start_output["cause"] = self.cause
        return start_output

1	from __future__ import annotations	1✔
2
3	import datetime	1✔
4	import json	1✔
5	import logging	1✔
6	from typing import Final	1✔
7
8	from localstack.aws.api.events import PutEventsRequestEntry	1✔
9	from localstack.aws.api.stepfunctions import (	1✔
10	Arn,
11	CloudWatchEventsExecutionDataDetails,
12	DescribeExecutionOutput,
13	DescribeStateMachineForExecutionOutput,
14	ExecutionListItem,
15	ExecutionStatus,
16	GetExecutionHistoryOutput,
17	HistoryEventList,
18	InvalidName,
19	SensitiveCause,
20	SensitiveError,
21	StartExecutionOutput,
22	StartSyncExecutionOutput,
23	StateMachineType,
24	SyncExecutionStatus,
25	Timestamp,
26	TraceHeader,
27	VariableReferences,
28	)
29	from localstack.aws.connect import connect_to	1✔
30	from localstack.services.stepfunctions.asl.eval.evaluation_details import (	1✔
31	AWSExecutionDetails,
32	EvaluationDetails,
33	ExecutionDetails,
34	StateMachineDetails,
35	)
36	from localstack.services.stepfunctions.asl.eval.event.logging import (	1✔
37	CloudWatchLoggingSession,
38	)
39	from localstack.services.stepfunctions.asl.eval.program_state import (	1✔
40	ProgramEnded,
41	ProgramError,
42	ProgramState,
43	ProgramStopped,
44	ProgramTimedOut,
45	)
46	from localstack.services.stepfunctions.asl.static_analyser.variable_references_static_analyser import (	1✔
47	VariableReferencesStaticAnalyser,
48	)
49	from localstack.services.stepfunctions.asl.utils.encoding import to_json_str	1✔
50	from localstack.services.stepfunctions.backend.activity import Activity	1✔
51	from localstack.services.stepfunctions.backend.execution_worker import (	1✔
52	ExecutionWorker,
53	SyncExecutionWorker,
54	)
55	from localstack.services.stepfunctions.backend.execution_worker_comm import (	1✔
56	ExecutionWorkerCommunication,
57	)
58	from localstack.services.stepfunctions.backend.state_machine import (	1✔
59	StateMachineInstance,
60	StateMachineVersion,
61	)
62	from localstack.services.stepfunctions.mocking.mock_config import MockTestCase	1✔
63
64	LOG = logging.getLogger(__name__)	1✔
65
66
67	class BaseExecutionWorkerCommunication(ExecutionWorkerCommunication):	1✔
68	execution: Final[Execution]	1✔
69
70	def __init__(self, execution: Execution):	1✔
71	self.execution = execution	1✔
72
73	def _reflect_execution_status(self):	1✔
74	exit_program_state: ProgramState = self.execution.exec_worker.env.program_state()	1✔
75	self.execution.stop_date = datetime.datetime.now(tz=datetime.UTC)	1✔
76	if isinstance(exit_program_state, ProgramEnded):	1✔
77	self.execution.exec_status = ExecutionStatus.SUCCEEDED	1✔
78	self.execution.output = self.execution.exec_worker.env.states.get_input()	1✔
79	elif isinstance(exit_program_state, ProgramStopped):	1✔
80	self.execution.exec_status = ExecutionStatus.ABORTED	1✔
81	elif isinstance(exit_program_state, ProgramError):	1✔
82	self.execution.exec_status = ExecutionStatus.FAILED	1✔
83	self.execution.error = exit_program_state.error.get("error")	1✔
84	self.execution.cause = exit_program_state.error.get("cause")	1✔
85	elif isinstance(exit_program_state, ProgramTimedOut):	1✔
86	self.execution.exec_status = ExecutionStatus.TIMED_OUT	1✔
87	else:
88	raise RuntimeWarning(	×
89	f"Execution ended with unsupported ProgramState type '{type(exit_program_state)}'."
90	)
91
92	def terminated(self) -> None:	1✔
93	self._reflect_execution_status()	1✔
94	self.execution.publish_execution_status_change_event()	1✔
95
96
97	class Execution:	1✔
98	name: Final[str]	1✔
99	sm_type: Final[StateMachineType]	1✔
100	role_arn: Final[Arn]	1✔
101	exec_arn: Final[Arn]	1✔
102
103	account_id: str	1✔
104	region_name: str	1✔
105
106	state_machine: Final[StateMachineInstance]	1✔
107	state_machine_arn: Final[Arn]	1✔
108	state_machine_version_arn: Final[Arn \| None]	1✔
109	state_machine_alias_arn: Final[Arn \| None]	1✔
110
111	mock_test_case: Final[MockTestCase \| None]	1✔
112
113	start_date: Final[Timestamp]	1✔
114	input_data: Final[json \| None]	1✔
115	input_details: Final[CloudWatchEventsExecutionDataDetails \| None]	1✔
116	trace_header: Final[TraceHeader \| None]	1✔
117	_cloud_watch_logging_session: Final[CloudWatchLoggingSession \| None]	1✔
118
119	exec_status: ExecutionStatus \| None	1✔
120	stop_date: Timestamp \| None	1✔
121
122	output: json \| None	1✔
123	output_details: CloudWatchEventsExecutionDataDetails \| None	1✔
124
125	error: SensitiveError \| None	1✔
126	cause: SensitiveCause \| None	1✔
127
128	exec_worker: ExecutionWorker \| None	1✔
129
130	_activity_store: dict[Arn, Activity]	1✔
131
132	def __init__(	1✔
133	self,
134	name: str,
135	sm_type: StateMachineType,
136	role_arn: Arn,
137	exec_arn: Arn,
138	account_id: str,
139	region_name: str,
140	state_machine: StateMachineInstance,
141	start_date: Timestamp,
142	cloud_watch_logging_session: CloudWatchLoggingSession \| None,
143	activity_store: dict[Arn, Activity],
144	input_data: json \| None = None,
145	trace_header: TraceHeader \| None = None,
146	state_machine_alias_arn: Arn \| None = None,
147	mock_test_case: MockTestCase \| None = None,
148	):
149	self.name = name	1✔
150	self.sm_type = sm_type	1✔
151	self.role_arn = role_arn	1✔
152	self.exec_arn = exec_arn	1✔
153	self.account_id = account_id	1✔
154	self.region_name = region_name	1✔
155	self.state_machine = state_machine	1✔
156	if isinstance(state_machine, StateMachineVersion):	1✔
157	self.state_machine_arn = state_machine.source_arn	1✔
158	self.state_machine_version_arn = state_machine.arn	1✔
159	else:
160	self.state_machine_arn = state_machine.arn	1✔
161	self.state_machine_version_arn = None	1✔
162	self.state_machine_alias_arn = state_machine_alias_arn	1✔
163	self.start_date = start_date	1✔
164	self._cloud_watch_logging_session = cloud_watch_logging_session	1✔
165	self.input_data = input_data	1✔
166	self.input_details = CloudWatchEventsExecutionDataDetails(included=True)	1✔
167	self.trace_header = trace_header	1✔
168	self.exec_status = None	1✔
169	self.stop_date = None	1✔
170	self.output = None	1✔
171	self.output_details = CloudWatchEventsExecutionDataDetails(included=True)	1✔
172	self.exec_worker = None	1✔
173	self.error = None	1✔
174	self.cause = None	1✔
175	self._activity_store = activity_store	1✔
176	self.mock_test_case = mock_test_case	1✔
177
178	def _get_events_client(self):	1✔
179	return connect_to(aws_access_key_id=self.account_id, region_name=self.region_name).events	1✔
180
181	def to_start_output(self) -> StartExecutionOutput:	1✔
182	return StartExecutionOutput(executionArn=self.exec_arn, startDate=self.start_date)	1✔
183
184	def to_describe_output(self) -> DescribeExecutionOutput:	1✔
185	describe_output = DescribeExecutionOutput(	1✔
186	executionArn=self.exec_arn,
187	stateMachineArn=self.state_machine_arn,
188	name=self.name,
189	status=self.exec_status,
190	startDate=self.start_date,
191	stopDate=self.stop_date,
192	input=to_json_str(self.input_data, separators=(",", ":")),
193	inputDetails=self.input_details,
194	traceHeader=self.trace_header,
195	)
196	if describe_output["status"] == ExecutionStatus.SUCCEEDED:	1✔
197	describe_output["output"] = to_json_str(self.output, separators=(",", ":"))	1✔
198	describe_output["outputDetails"] = self.output_details	1✔
199	if self.error is not None:	1✔
200	describe_output["error"] = self.error	1✔
201	if self.cause is not None:	1✔
202	describe_output["cause"] = self.cause	1✔
203	if self.state_machine_version_arn is not None:	1✔
204	describe_output["stateMachineVersionArn"] = self.state_machine_version_arn	1✔
205	if self.state_machine_alias_arn is not None:	1✔
206	describe_output["stateMachineAliasArn"] = self.state_machine_alias_arn	1✔
207	return describe_output	1✔
208
209	def to_describe_state_machine_for_execution_output(	1✔
210	self,
211	) -> DescribeStateMachineForExecutionOutput:
212	state_machine: StateMachineInstance = self.state_machine	1✔
213	state_machine_arn = (	1✔
214	state_machine.source_arn
215	if isinstance(state_machine, StateMachineVersion)
216	else state_machine.arn
217	)
218	out = DescribeStateMachineForExecutionOutput(	1✔
219	stateMachineArn=state_machine_arn,
220	name=state_machine.name,
221	definition=state_machine.definition,
222	roleArn=self.role_arn,
223	# The date and time the state machine associated with an execution was updated.
224	updateDate=state_machine.create_date,
225	loggingConfiguration=state_machine.logging_config,
226	)
227	revision_id = self.state_machine.revision_id	1✔
228	if self.state_machine.revision_id:	1✔
229	out["revisionId"] = revision_id	1✔
230	variable_references: VariableReferences = VariableReferencesStaticAnalyser.process_and_get(	1✔
231	definition=self.state_machine.definition
232	)
233	if variable_references:	1✔
234	out["variableReferences"] = variable_references	1✔
235	return out	1✔
236
237	def to_execution_list_item(self) -> ExecutionListItem:	1✔
238	if isinstance(self.state_machine, StateMachineVersion):	1✔
239	state_machine_arn = self.state_machine.source_arn	1✔
240	state_machine_version_arn = self.state_machine.arn	1✔
241	else:
242	state_machine_arn = self.state_machine.arn	1✔
243	state_machine_version_arn = None	1✔
244
245	item = ExecutionListItem(	1✔
246	executionArn=self.exec_arn,
247	stateMachineArn=state_machine_arn,
248	name=self.name,
249	status=self.exec_status,
250	startDate=self.start_date,
251	stopDate=self.stop_date,
252	)
253	if state_machine_version_arn is not None:	1✔
254	item["stateMachineVersionArn"] = state_machine_version_arn	1✔
255	if self.state_machine_alias_arn is not None:	1✔
256	item["stateMachineAliasArn"] = self.state_machine_alias_arn	1✔
257	return item	1✔
258
259	def to_history_output(self) -> GetExecutionHistoryOutput:	1✔
260	env = self.exec_worker.env	1✔
261	event_history: HistoryEventList = []	1✔
262	if env is not None:	1✔
263	# The execution has not started yet.
264	event_history: HistoryEventList = env.event_manager.get_event_history()	1✔
265	return GetExecutionHistoryOutput(events=event_history)	1✔
266
267	@staticmethod	1✔
268	def _to_serialized_date(timestamp: datetime.datetime) -> str:	1✔
269	"""See test in tests.aws.services.stepfunctions.v2.base.test_base.TestSnfBase.test_execution_dateformat"""
270	return f"{timestamp.astimezone(datetime.UTC).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3]}Z"	1✔
271
272	def _get_start_execution_worker_comm(self) -> BaseExecutionWorkerCommunication:	1✔
273	return BaseExecutionWorkerCommunication(self)	1✔
274
275	def _get_start_aws_execution_details(self) -> AWSExecutionDetails:	1✔
276	return AWSExecutionDetails(	1✔
277	account=self.account_id, region=self.region_name, role_arn=self.role_arn
278	)
279
280	def get_start_execution_details(self) -> ExecutionDetails:	1✔
281	return ExecutionDetails(	1✔
282	arn=self.exec_arn,
283	name=self.name,
284	role_arn=self.role_arn,
285	inpt=self.input_data,
286	start_time=self._to_serialized_date(self.start_date),
287	)
288
289	def get_start_state_machine_details(self) -> StateMachineDetails:	1✔
290	return StateMachineDetails(	1✔
291	arn=self.state_machine.arn,
292	name=self.state_machine.name,
293	typ=self.state_machine.sm_type,
294	definition=self.state_machine.definition,
295	)
296
297	def _get_start_execution_worker(self) -> ExecutionWorker:	1✔
298	return ExecutionWorker(	1✔
299	evaluation_details=EvaluationDetails(
300	aws_execution_details=self._get_start_aws_execution_details(),
301	execution_details=self.get_start_execution_details(),
302	state_machine_details=self.get_start_state_machine_details(),
303	),
304	exec_comm=self._get_start_execution_worker_comm(),
305	cloud_watch_logging_session=self._cloud_watch_logging_session,
306	activity_store=self._activity_store,
307	mock_test_case=self.mock_test_case,
308	)
309
310	def start(self) -> None:	1✔
311	# TODO: checks exec_worker does not exists already?
312	if self.exec_worker:	1✔
313	raise InvalidName() # TODO.	×
314	self.exec_worker = self._get_start_execution_worker()	1✔
315	self.exec_status = ExecutionStatus.RUNNING	1✔
316	self.publish_execution_status_change_event()	1✔
317	self.exec_worker.start()	1✔
318
319	def stop(self, stop_date: datetime.datetime, error: str \| None, cause: str \| None):	1✔
320	exec_worker: ExecutionWorker \| None = self.exec_worker	1✔
321	if exec_worker:	1✔
322	exec_worker.stop(stop_date=stop_date, cause=cause, error=error)	1✔
323
324	def publish_execution_status_change_event(self):	1✔
325	input_value = (	1✔
326	{} if not self.input_data else to_json_str(self.input_data, separators=(",", ":"))
327	)
328	output_value = (	1✔
329	None if self.output is None else to_json_str(self.output, separators=(",", ":"))
330	)
331	output_details = None if output_value is None else self.output_details	1✔
332	entry = PutEventsRequestEntry(	1✔
333	Source="aws.states",
334	Resources=[self.exec_arn],
335	DetailType="Step Functions Execution Status Change",
336	Detail=to_json_str(
337	# Note: this operation carries significant changes from a describe_execution request.
338	DescribeExecutionOutput(
339	executionArn=self.exec_arn,
340	stateMachineArn=self.state_machine.arn,
341	stateMachineAliasArn=None,
342	stateMachineVersionArn=None,
343	name=self.name,
344	status=self.exec_status,
345	startDate=self.start_date,
346	stopDate=self.stop_date,
347	input=input_value,
348	inputDetails=self.input_details,
349	output=output_value,
350	outputDetails=output_details,
351	error=self.error,
352	cause=self.cause,
353	)
354	),
355	)
356	try:	1✔
357	self._get_events_client().put_events(Entries=[entry])	1✔
358	except Exception:	×
359	LOG.error(	×
360	"Unable to send notification of Entry='%s' for Step Function execution with Arn='%s' to EventBridge.",
361	entry,
362	self.exec_arn,
363	exc_info=LOG.isEnabledFor(logging.DEBUG),
364	)
365
366
367	class SyncExecutionWorkerCommunication(BaseExecutionWorkerCommunication):	1✔
368	execution: Final[SyncExecution]	1✔
369
370	def _reflect_execution_status(self) -> None:	1✔
371	super()._reflect_execution_status()	1✔
372	exit_status: ExecutionStatus = self.execution.exec_status	1✔
373	if exit_status == ExecutionStatus.SUCCEEDED:	1✔
374	self.execution.sync_execution_status = SyncExecutionStatus.SUCCEEDED	1✔
375	elif exit_status == ExecutionStatus.TIMED_OUT:	1✔
UNCOV 376	self.execution.sync_execution_status = SyncExecutionStatus.TIMED_OUT	×
377	else:
378	self.execution.sync_execution_status = SyncExecutionStatus.FAILED	1✔
379
380
381	class SyncExecution(Execution):	1✔
382	sync_execution_status: SyncExecutionStatus \| None = None	1✔
383
384	def _get_start_execution_worker(self) -> SyncExecutionWorker:	1✔
385	return SyncExecutionWorker(	1✔
386	evaluation_details=EvaluationDetails(
387	aws_execution_details=self._get_start_aws_execution_details(),
388	execution_details=self.get_start_execution_details(),
389	state_machine_details=self.get_start_state_machine_details(),
390	),
391	exec_comm=self._get_start_execution_worker_comm(),
392	cloud_watch_logging_session=self._cloud_watch_logging_session,
393	activity_store=self._activity_store,
394	mock_test_case=self.mock_test_case,
395	)
396
397	def _get_start_execution_worker_comm(self) -> BaseExecutionWorkerCommunication:	1✔
398	return SyncExecutionWorkerCommunication(self)	1✔
399
400	def to_start_sync_execution_output(self) -> StartSyncExecutionOutput:	1✔
401	start_output = StartSyncExecutionOutput(	1✔
402	executionArn=self.exec_arn,
403	stateMachineArn=self.state_machine.arn,
404	name=self.name,
405	status=self.sync_execution_status,
406	startDate=self.start_date,
407	stopDate=self.stop_date,
408	input=to_json_str(self.input_data, separators=(",", ":")),
409	inputDetails=self.input_details,
410	traceHeader=self.trace_header,
411	)
412	if self.sync_execution_status == SyncExecutionStatus.SUCCEEDED:	1✔
413	start_output["output"] = to_json_str(self.output, separators=(",", ":"))	1✔
414	if self.output_details:	1✔
415	start_output["outputDetails"] = self.output_details	1✔
416	if self.error is not None:	1✔
417	start_output["error"] = self.error	1✔
418	if self.cause is not None:	1✔
419	start_output["cause"] = self.cause	1✔
420	return start_output	1✔

localstack / localstack / 16981563750

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