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
|
from typing import Any, Final
from moto.stepfunctions.parser.api import (
ExecutionFailedEventDetails,
HistoryEventType,
)
from moto.stepfunctions.parser.asl.component.common.error_name.failure_event import (
FailureEvent,
FailureEventException,
)
from moto.stepfunctions.parser.asl.component.common.error_name.states_error_name import (
StatesErrorName,
)
from moto.stepfunctions.parser.asl.component.common.error_name.states_error_name_type import (
StatesErrorNameType,
)
from moto.stepfunctions.parser.asl.component.common.string.string_expression import (
StringSampler,
)
from moto.stepfunctions.parser.asl.component.state.wait.wait_function.wait_function import (
WaitFunction,
)
from moto.stepfunctions.parser.asl.eval.environment import Environment
from moto.stepfunctions.parser.asl.eval.event.event_detail import EventDetails
from moto.stepfunctions.parser.asl.utils.json_path import NoSuchJsonPathError
class SecondsPath(WaitFunction):
# SecondsPath
# A time, in seconds, to state_wait before beginning the state specified in the Next
# field, specified using a path from the state's input data.
# You must specify an integer value for this field.
string_sampler: Final[StringSampler]
def __init__(self, string_sampler: StringSampler):
self.string_sampler = string_sampler
def _validate_seconds_value(self, env: Environment, seconds: Any):
if isinstance(seconds, int) and seconds >= 0:
return
error_type = StatesErrorNameType.StatesRuntime
assignment_description = f"{self.string_sampler.literal_value} == {seconds}"
if not isinstance(seconds, int):
cause = f"The SecondsPath parameter cannot be parsed as a long value: {assignment_description}"
else: # seconds < 0
cause = f"The SecondsPath parameter references a negative value: {assignment_description}"
raise FailureEventException(
failure_event=FailureEvent(
env=env,
error_name=StatesErrorName(typ=error_type),
event_type=HistoryEventType.ExecutionFailed,
event_details=EventDetails(
executionFailedEventDetails=ExecutionFailedEventDetails(
error=error_type.to_name(), cause=cause
)
),
)
)
def _get_wait_seconds(self, env: Environment) -> int:
try:
self.string_sampler.eval(env=env)
except NoSuchJsonPathError as no_such_json_path_error:
cause = f"The SecondsPath parameter does not reference an input value: {no_such_json_path_error.json_path}"
raise FailureEventException(
failure_event=FailureEvent(
env=env,
error_name=StatesErrorName(typ=StatesErrorNameType.StatesRuntime),
event_type=HistoryEventType.ExecutionFailed,
event_details=EventDetails(
executionFailedEventDetails=ExecutionFailedEventDetails(
error=StatesErrorNameType.StatesRuntime.to_name(),
cause=cause,
)
),
)
)
seconds = env.stack.pop()
self._validate_seconds_value(env=env, seconds=seconds)
return seconds
|
