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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
|
import datetime
import re
from typing import Final, Optional
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 (
StringExpression,
StringLiteral,
)
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
TIMESTAMP_FORMAT: Final[str] = "%Y-%m-%dT%H:%M:%SZ"
# TODO: could be a bit more exact (e.g. 90 shouldn't be a valid minute)
TIMESTAMP_PATTERN: Final[str] = r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?Z$"
class Timestamp(WaitFunction):
string: Final[StringExpression]
def __init__(self, string: StringExpression):
self.string = string
# If a string literal, assert it encodes a valid timestamp.
if isinstance(string, StringLiteral):
timestamp = string.literal_value
if self._from_timestamp_string(timestamp) is None:
raise ValueError(
"The Timestamp value does not reference a valid ISO-8601 "
f"extended offset date-time format string: '{timestamp}'"
)
@staticmethod
def _is_valid_timestamp_pattern(timestamp: str) -> bool:
return re.match(TIMESTAMP_PATTERN, timestamp) is not None
@staticmethod
def _from_timestamp_string(timestamp: str) -> Optional[datetime.datetime]:
if not Timestamp._is_valid_timestamp_pattern(timestamp):
return None
try:
# anything lower than seconds is truncated
processed_timestamp = timestamp.rsplit(".", 2)[0]
# add back the "Z" suffix if we removed it
if not processed_timestamp.endswith("Z"):
processed_timestamp = f"{processed_timestamp}Z"
datetime_timestamp = datetime.datetime.strptime(
processed_timestamp, TIMESTAMP_FORMAT
)
return datetime_timestamp
except Exception:
return None
def _create_failure_event(
self, env: Environment, timestamp_str: str
) -> FailureEvent:
return FailureEvent(
env=env,
error_name=StatesErrorName(typ=StatesErrorNameType.StatesRuntime),
event_type=HistoryEventType.ExecutionFailed,
event_details=EventDetails(
executionFailedEventDetails=ExecutionFailedEventDetails(
error=StatesErrorNameType.StatesRuntime.to_name(),
cause="The Timestamp parameter does not reference a valid ISO-8601 "
f"extended offset date-time format string: {self.string.literal_value} == {timestamp_str}",
)
),
)
def _get_wait_seconds(self, env: Environment) -> int:
self.string.eval(env=env)
timestamp_str: str = env.stack.pop()
timestamp = self._from_timestamp_string(timestamp=timestamp_str)
if timestamp is None:
raise FailureEventException(self._create_failure_event(env, timestamp_str))
delta = timestamp - datetime.datetime.now()
delta_sec = int(delta.total_seconds())
return delta_sec
class TimestampPath(Timestamp):
def _create_failure_event(
self, env: Environment, timestamp_str: str
) -> FailureEvent:
return FailureEvent(
env=env,
error_name=StatesErrorName(typ=StatesErrorNameType.StatesRuntime),
event_type=HistoryEventType.ExecutionFailed,
event_details=EventDetails(
executionFailedEventDetails=ExecutionFailedEventDetails(
error=StatesErrorNameType.StatesRuntime.to_name(),
cause="The TimestampPath parameter does not reference a valid ISO-8601 "
f"extended offset date-time format string: {self.string.literal_value} == {timestamp_str}",
)
),
)
|
