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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
from moto.moto_api import state_manager
from moto.moto_api._internal.managed_state_model import ManagedState
class ExampleModel(ManagedState):
def __init__(self):
super().__init__(
model_name="example::model", transitions=[("frist_status", "second_status")]
)
state_manager.register_default_transition(
model_name="example::model", transition={"progression": "manual", "times": 999}
)
def test_initial_state():
assert ExampleModel().status == "frist_status"
def test_advancing_without_specifying_configuration_does_nothing():
model = ExampleModel()
for _ in range(5):
assert model.status == "frist_status"
model.advance()
def test_advance_immediately():
model = ExampleModel()
model._transitions = [
("frist_status", "second"),
("second", "third"),
("third", "fourth"),
("fourth", "fifth"),
]
state_manager.set_transition(
model_name="example::model", transition={"progression": "immediate"}
)
assert model.status == "fifth"
model.advance()
assert model.status == "fifth"
def test_advance_x_times():
model = ExampleModel()
state_manager.set_transition(
model_name="example::model", transition={"progression": "manual", "times": 3}
)
for _ in range(2):
model.advance()
assert model.status == "frist_status"
# 3rd time is a charm
model.advance()
assert model.status == "second_status"
# Status is still the same if we keep asking for it
assert model.status == "second_status"
# Advancing more does not make a difference - there's nothing to advance to
model.advance()
assert model.status == "second_status"
def test_advance_multiple_stages():
model = ExampleModel()
model._transitions = [
("frist_status", "second"),
("second", "third"),
("third", "fourth"),
("fourth", "fifth"),
]
state_manager.set_transition(
model_name="example::model", transition={"progression": "manual", "times": 1}
)
assert model.status == "frist_status"
assert model.status == "frist_status"
model.advance()
assert model.status == "second"
assert model.status == "second"
model.advance()
assert model.status == "third"
assert model.status == "third"
model.advance()
assert model.status == "fourth"
assert model.status == "fourth"
model.advance()
assert model.status == "fifth"
assert model.status == "fifth"
def test_override_status():
model = ExampleModel()
model.status = "creating"
model._transitions = [("creating", "ready"), ("updating", "ready")]
state_manager.set_transition(
model_name="example::model", transition={"progression": "manual", "times": 1}
)
assert model.status == "creating"
model.advance()
assert model.status == "ready"
model.advance()
# We're still ready
assert model.status == "ready"
# Override status manually
model.status = "updating"
assert model.status == "updating"
model.advance()
assert model.status == "ready"
assert model.status == "ready"
model.advance()
assert model.status == "ready"
class SlowModel(ManagedState):
def __init__(self):
super().__init__(
model_name="example::slowmodel", transitions=[("first", "second")]
)
def test_realworld_delay():
model = SlowModel()
state_manager.set_transition(
model_name="example::slowmodel",
transition={"progression": "time", "seconds": 2},
)
assert model.status == "first"
# The status will stick to 'first' for a long time
# Advancing the model doesn't do anything, really
for _ in range(10):
model.advance()
assert model.status == "first"
import time
time.sleep(2)
# Status has only progressed after 2 seconds have passed
assert model.status == "second"
|
