22709357475

Committed 05 Mar 2026 08:35AM UTC coverage: 59.732% (-27.2%) from 86.974%

Build # 22709357475

Build Type

Pull #13880

github

Committed by

web-flow

Commit Message

Merge 28fcab93c into 710618057

Pull Request Pull Request #13880: Firehose: Replace TaggingService

Coverage Stats

12 of 12 new or added lines in 2 files covered. (100.0%)

20464 existing lines in 510 files now uncovered.

45290 of 75822 relevant lines covered (59.73%)

0.6 hits per line

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

96.0

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

import datetime
from threading import Thread
from typing import Final

from localstack.aws.api.stepfunctions import (
    Arn,
    ExecutionStartedEventDetails,
    HistoryEventExecutionDataDetails,
    HistoryEventType,
)
from localstack.services.stepfunctions.asl.component.eval_component import EvalComponent
from localstack.services.stepfunctions.asl.eval.environment import Environment
from localstack.services.stepfunctions.asl.eval.evaluation_details import EvaluationDetails
from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails
from localstack.services.stepfunctions.asl.eval.event.event_manager import (
    EventHistoryContext,
)
from localstack.services.stepfunctions.asl.eval.event.logging import (
    CloudWatchLoggingSession,
)
from localstack.services.stepfunctions.asl.eval.states import (
    ContextObjectData,
    ExecutionData,
    StateMachineData,
)
from localstack.services.stepfunctions.asl.parse.asl_parser import AmazonStateLanguageParser
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_comm import (
    ExecutionWorkerCommunication,
)
from localstack.services.stepfunctions.local_mocking.mock_config import LocalMockTestCase
from localstack.utils.common import TMP_THREADS


class ExecutionWorker:
    _evaluation_details: Final[EvaluationDetails]
    _execution_communication: Final[ExecutionWorkerCommunication]
    _cloud_watch_logging_session: Final[CloudWatchLoggingSession | None]
    _local_mock_test_case: Final[LocalMockTestCase | None]
    _activity_store: dict[Arn, Activity]

    env: Environment | None

    def __init__(
        self,
        evaluation_details: EvaluationDetails,
        exec_comm: ExecutionWorkerCommunication,
        cloud_watch_logging_session: CloudWatchLoggingSession | None,
        activity_store: dict[Arn, Activity],
        local_mock_test_case: LocalMockTestCase | None = None,
    ):
        self._evaluation_details = evaluation_details
        self._execution_communication = exec_comm
        self._cloud_watch_logging_session = cloud_watch_logging_session
        self._local_mock_test_case = local_mock_test_case
        self._activity_store = activity_store
        self.env = None

    def _get_evaluation_entrypoint(self) -> EvalComponent:
        return AmazonStateLanguageParser.parse(
            self._evaluation_details.state_machine_details.definition
        )[0]

    def _get_evaluation_environment(self) -> Environment:
        return Environment(
            aws_execution_details=self._evaluation_details.aws_execution_details,
            execution_type=self._evaluation_details.state_machine_details.typ,
            context=ContextObjectData(
                Execution=ExecutionData(
                    Id=self._evaluation_details.execution_details.arn,
                    Input=self._evaluation_details.execution_details.inpt,
                    Name=self._evaluation_details.execution_details.name,
                    RoleArn=self._evaluation_details.execution_details.role_arn,
                    StartTime=self._evaluation_details.execution_details.start_time,
                ),
                StateMachine=StateMachineData(
                    Id=self._evaluation_details.state_machine_details.arn,
                    Name=self._evaluation_details.state_machine_details.name,
                ),
            ),
            event_history_context=EventHistoryContext.of_program_start(),
            cloud_watch_logging_session=self._cloud_watch_logging_session,
            activity_store=self._activity_store,
            local_mock_test_case=self._local_mock_test_case,
        )

    def _execution_logic(self):
        program = self._get_evaluation_entrypoint()
        self.env = self._get_evaluation_environment()

        self.env.event_manager.add_event(
            context=self.env.event_history_context,
            event_type=HistoryEventType.ExecutionStarted,
            event_details=EventDetails(
                executionStartedEventDetails=ExecutionStartedEventDetails(
                    input=to_json_str(self._evaluation_details.execution_details.inpt),
                    inputDetails=HistoryEventExecutionDataDetails(
                        truncated=False
                    ),  # Always False for api calls.
                    roleArn=self._evaluation_details.aws_execution_details.role_arn,
                )
            ),
            update_source_event_id=False,
        )

        program.eval(self.env)

        self._execution_communication.terminated()

    def start(self):
        execution_logic_thread = Thread(target=self._execution_logic, daemon=True)
        TMP_THREADS.append(execution_logic_thread)
        execution_logic_thread.start()

    def stop(self, stop_date: datetime.datetime, error: str | None, cause: str | None):
        self.env.set_stop(stop_date=stop_date, cause=cause, error=error)


class SyncExecutionWorker(ExecutionWorker):
    def start(self):
        # bypass the native async execution of ASL programs.
        self._execution_logic()

1	import datetime	1✔
2	from threading import Thread	1✔
3	from typing import Final	1✔
4
5	from localstack.aws.api.stepfunctions import (	1✔
6	Arn,
7	ExecutionStartedEventDetails,
8	HistoryEventExecutionDataDetails,
9	HistoryEventType,
10	)
11	from localstack.services.stepfunctions.asl.component.eval_component import EvalComponent	1✔
12	from localstack.services.stepfunctions.asl.eval.environment import Environment	1✔
13	from localstack.services.stepfunctions.asl.eval.evaluation_details import EvaluationDetails	1✔
14	from localstack.services.stepfunctions.asl.eval.event.event_detail import EventDetails	1✔
15	from localstack.services.stepfunctions.asl.eval.event.event_manager import (	1✔
16	EventHistoryContext,
17	)
18	from localstack.services.stepfunctions.asl.eval.event.logging import (	1✔
19	CloudWatchLoggingSession,
20	)
21	from localstack.services.stepfunctions.asl.eval.states import (	1✔
22	ContextObjectData,
23	ExecutionData,
24	StateMachineData,
25	)
26	from localstack.services.stepfunctions.asl.parse.asl_parser import AmazonStateLanguageParser	1✔
27	from localstack.services.stepfunctions.asl.utils.encoding import to_json_str	1✔
28	from localstack.services.stepfunctions.backend.activity import Activity	1✔
29	from localstack.services.stepfunctions.backend.execution_worker_comm import (	1✔
30	ExecutionWorkerCommunication,
31	)
32	from localstack.services.stepfunctions.local_mocking.mock_config import LocalMockTestCase	1✔
33	from localstack.utils.common import TMP_THREADS	1✔
34
35
36	class ExecutionWorker:	1✔
37	_evaluation_details: Final[EvaluationDetails]	1✔
38	_execution_communication: Final[ExecutionWorkerCommunication]	1✔
39	_cloud_watch_logging_session: Final[CloudWatchLoggingSession \| None]	1✔
40	_local_mock_test_case: Final[LocalMockTestCase \| None]	1✔
41	_activity_store: dict[Arn, Activity]	1✔
42
43	env: Environment \| None	1✔
44
45	def __init__(	1✔
46	self,
47	evaluation_details: EvaluationDetails,
48	exec_comm: ExecutionWorkerCommunication,
49	cloud_watch_logging_session: CloudWatchLoggingSession \| None,
50	activity_store: dict[Arn, Activity],
51	local_mock_test_case: LocalMockTestCase \| None = None,
52	):
53	self._evaluation_details = evaluation_details	1✔
54	self._execution_communication = exec_comm	1✔
55	self._cloud_watch_logging_session = cloud_watch_logging_session	1✔
56	self._local_mock_test_case = local_mock_test_case	1✔
57	self._activity_store = activity_store	1✔
58	self.env = None	1✔
59
60	def _get_evaluation_entrypoint(self) -> EvalComponent:	1✔
61	return AmazonStateLanguageParser.parse(	1✔
62	self._evaluation_details.state_machine_details.definition
63	)[0]
64
65	def _get_evaluation_environment(self) -> Environment:	1✔
66	return Environment(	1✔
67	aws_execution_details=self._evaluation_details.aws_execution_details,
68	execution_type=self._evaluation_details.state_machine_details.typ,
69	context=ContextObjectData(
70	Execution=ExecutionData(
71	Id=self._evaluation_details.execution_details.arn,
72	Input=self._evaluation_details.execution_details.inpt,
73	Name=self._evaluation_details.execution_details.name,
74	RoleArn=self._evaluation_details.execution_details.role_arn,
75	StartTime=self._evaluation_details.execution_details.start_time,
76	),
77	StateMachine=StateMachineData(
78	Id=self._evaluation_details.state_machine_details.arn,
79	Name=self._evaluation_details.state_machine_details.name,
80	),
81	),
82	event_history_context=EventHistoryContext.of_program_start(),
83	cloud_watch_logging_session=self._cloud_watch_logging_session,
84	activity_store=self._activity_store,
85	local_mock_test_case=self._local_mock_test_case,
86	)
87
88	def _execution_logic(self):	1✔
89	program = self._get_evaluation_entrypoint()	1✔
90	self.env = self._get_evaluation_environment()	1✔
91
92	self.env.event_manager.add_event(	1✔
93	context=self.env.event_history_context,
94	event_type=HistoryEventType.ExecutionStarted,
95	event_details=EventDetails(
96	executionStartedEventDetails=ExecutionStartedEventDetails(
97	input=to_json_str(self._evaluation_details.execution_details.inpt),
98	inputDetails=HistoryEventExecutionDataDetails(
99	truncated=False
100	), # Always False for api calls.
101	roleArn=self._evaluation_details.aws_execution_details.role_arn,
102	)
103	),
104	update_source_event_id=False,
105	)
106
107	program.eval(self.env)	1✔
108
109	self._execution_communication.terminated()	1✔
110
111	def start(self):	1✔
112	execution_logic_thread = Thread(target=self._execution_logic, daemon=True)	1✔
113	TMP_THREADS.append(execution_logic_thread)	1✔
114	execution_logic_thread.start()	1✔
115
116	def stop(self, stop_date: datetime.datetime, error: str \| None, cause: str \| None):	1✔
UNCOV 117	self.env.set_stop(stop_date=stop_date, cause=cause, error=error)	×
118
119
120	class SyncExecutionWorker(ExecutionWorker):	1✔
121	def start(self):	1✔
122	# bypass the native async execution of ASL programs.
UNCOV 123	self._execution_logic()	×

localstack / localstack / 22709357475

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