forked from decentralized-identity/aries-rfcs
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_state_machine.py
164 lines (140 loc) · 4.83 KB
/
test_state_machine.py
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
import pytest
from state_machine import *
class NeverDone:
def is_done(self):
return False
class BeDone:
def is_done(self):
return True
@pytest.fixture
def sm():
return StateMachine(NeverDone())
def test_no_state_on_start(sm):
assert sm.state is None
def test_my_move_first(sm):
sm.handle(SEND_MOVE_EVENT)
assert sm.state == THEIR_MOVE_STATE
def test_their_move_first(sm):
sm.handle(RECEIVE_MOVE_EVENT)
assert sm.state == MY_MOVE_STATE
def test_illegal_move_by_me(sm):
sm.handle(SEND_MOVE_EVENT)
# Sending a move when it's the other person's turn
# is our programmer error, so it should assert.
with pytest.raises(AssertionError):
sm.handle(SEND_MOVE_EVENT)
def test_wrapup_from_them(sm):
sm.state = THEIR_MOVE_STATE
sm.logic = BeDone()
sm.handle(RECEIVE_MOVE_EVENT)
assert sm.state == WRAP_UP_STATE
sm.handle(SEND_OUTCOME_EVENT)
assert sm.state == DONE_STATE
def test_wrapup_from_me(sm):
sm.state = MY_MOVE_STATE
sm.logic = BeDone()
sm.handle(SEND_MOVE_EVENT)
assert sm.state == WRAP_UP_STATE
sm.handle(SEND_OUTCOME_EVENT)
assert sm.state == DONE_STATE
def test_send_move_in_wrapup(sm):
sm.state = WRAP_UP_STATE
# Sending a move when it's time to send an outcome
# is our programmer error, so this should assert.
with pytest.raises(AssertionError):
sm.handle(SEND_MOVE_EVENT)
def test_receive_move_in_wrapup(sm):
sm.state = WRAP_UP_STATE
# Receiving a move when it's time to send an outcome
# is a programmer error on the other side. It shouldn't
# assert in our code, but should generate an error.
# This is an error by the other party, so it should
# trigger on_error.
sm.on_error = ErrorHandler()
sm.handle(RECEIVE_MOVE_EVENT)
assert bool(sm.on_error.msg)
assert sm.state == WRAP_UP_STATE
class ErrorHandler:
def __init__(self):
self.msg = None
def __call__(self, msg):
self.msg = msg
def test_illegal_move_by_them(sm):
# This is an error by the other party, so it should
# trigger on_error.
sm.on_error = ErrorHandler()
sm.handle(RECEIVE_MOVE_EVENT)
assert sm.on_error.msg is None
assert sm.state == MY_MOVE_STATE
sm.handle(RECEIVE_MOVE_EVENT)
assert bool(sm.on_error.msg)
assert sm.state == MY_MOVE_STATE
def test_early_exit_by_me(sm):
sm.handle(RECEIVE_MOVE_EVENT)
sm.handle(SEND_OUTCOME_EVENT)
assert sm.state == DONE_STATE
def test_early_exit_by_them(sm):
sm.handle(SEND_MOVE_EVENT)
sm.handle(SEND_OUTCOME_EVENT)
assert sm.state == DONE_STATE
def test_unrecognized_event(sm):
with pytest.raises(AssertionError):
sm.handle(25)
def test_event_while_done(sm):
for i in range(len(EVENT_NAMES)):
sm.state = DONE_STATE
sm.on_error = ErrorHandler()
if i == SEND_MOVE_EVENT:
with pytest.raises(AssertionError):
sm.handle(i)
assert sm.on_error.msg is None
elif i in [SEND_OUTCOME_EVENT, RECEIVE_OUTCOME_EVENT]:
# should be ignored
sm.handle(i)
assert sm.on_error.msg is None
else: #i == RECEIVE_MOVE_EVENT:
sm.handle(i)
assert bool(sm.on_error.msg)
class HookHandler:
def __init__(self, response):
self.event = None
self.response = response
self.state = None
def __call__(self, state, event):
self.state = state
self.event = event
return self.response
def test_pre_hooks_allow(sm):
for i in range(DONE_STATE):
sm.state = i
sm.pre = HookHandler(True)
sm.handle(SEND_OUTCOME_EVENT)
# Because we're allowing a transition, the pre hook should
# be called with old state, but now we should have new state.
assert sm.pre.state != i
assert sm.state == sm.pre.state
assert sm.pre.event == SEND_OUTCOME_EVENT
def test_pre_hooks_deny(sm):
for i in range(DONE_STATE):
sm.state = i
# This hook handler should refuse to allow us to transition.
sm.pre = HookHandler(False)
sm.post = HookHandler(None)
sm.handle(SEND_OUTCOME_EVENT)
# Because we're denying a transition, the pre hook should
# be called with new state, and we should still have old state.
assert sm.state == i
assert sm.pre.state != i
assert sm.pre.event == SEND_OUTCOME_EVENT
# We should never have called the post event.
assert sm.post.event == None
def test_post_hooks(sm):
for i in range(DONE_STATE):
sm.state = i
sm.post = HookHandler(None)
sm.handle(SEND_OUTCOME_EVENT)
# Because we're allowing a transition, the post hook should
# be called with new state.
assert sm.state != i
assert sm.post.state == sm.state
assert sm.post.event == SEND_OUTCOME_EVENT