1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
|
from __future__ import annotations
from typing import Optional
from moto.stepfunctions.parser.api import Arn, InspectionData, StateMachineType
from moto.stepfunctions.parser.asl.eval.environment import Environment
from moto.stepfunctions.parser.asl.eval.evaluation_details import AWSExecutionDetails
from moto.stepfunctions.parser.asl.eval.event.event_manager import (
EventHistoryContext,
)
from moto.stepfunctions.parser.asl.eval.event.logging import (
CloudWatchLoggingSession,
)
from moto.stepfunctions.parser.asl.eval.program_state import (
ProgramRunning,
)
from moto.stepfunctions.parser.asl.eval.states import ContextObjectData
from moto.stepfunctions.parser.asl.eval.test_state.program_state import (
ProgramChoiceSelected,
)
from moto.stepfunctions.parser.asl.eval.variable_store import VariableStore
from moto.stepfunctions.parser.backend.activity import Activity
class TestStateEnvironment(Environment):
inspection_data: InspectionData
def __init__(
self,
aws_execution_details: AWSExecutionDetails,
execution_type: StateMachineType,
context: ContextObjectData,
event_history_context: EventHistoryContext,
activity_store: dict[Arn, Activity],
cloud_watch_logging_session: Optional[CloudWatchLoggingSession] = None,
):
super().__init__(
aws_execution_details=aws_execution_details,
execution_type=execution_type,
context=context,
event_history_context=event_history_context,
cloud_watch_logging_session=cloud_watch_logging_session,
activity_store=activity_store,
)
self.inspection_data = InspectionData()
def as_frame_of(
cls,
env: TestStateEnvironment,
event_history_frame_cache: Optional[EventHistoryContext] = None,
) -> Environment:
frame = super().as_frame_of(
env=env, event_history_frame_cache=event_history_frame_cache
)
frame.inspection_data = env.inspection_data
return frame
def as_inner_frame_of(
cls,
env: TestStateEnvironment,
variable_store: VariableStore,
event_history_frame_cache: Optional[EventHistoryContext] = None,
) -> Environment:
frame = super().as_inner_frame_of(
env=env,
event_history_frame_cache=event_history_frame_cache,
variable_store=variable_store,
)
frame.inspection_data = env.inspection_data
return frame
def set_choice_selected(self, next_state_name: str) -> None:
with self._state_mutex:
if isinstance(self._program_state, ProgramRunning):
self._program_state = ProgramChoiceSelected(
next_state_name=next_state_name
)
self.program_state_event.set()
self.program_state_event.clear()
else:
raise RuntimeError(
"Cannot set choice selected for non running ProgramState."
)
|
