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
|
"""
These tests validate the state machines directly. Writing meaningful tests for
this case can be tricky, so the majority of these tests use Hypothesis to try
to talk about general behaviours rather than specific cases.
"""
from __future__ import annotations
import pytest
from hypothesis import given
from hypothesis.strategies import sampled_from
import h2.connection
import h2.exceptions
import h2.stream
class TestConnectionStateMachine:
"""
Tests of the connection state machine.
"""
@given(state=sampled_from(h2.connection.ConnectionState),
input_=sampled_from(h2.connection.ConnectionInputs))
def test_state_transitions(self, state, input_) -> None:
c = h2.connection.H2ConnectionStateMachine()
c.state = state
try:
c.process_input(input_)
except h2.exceptions.ProtocolError:
assert c.state == h2.connection.ConnectionState.CLOSED
else:
assert c.state in h2.connection.ConnectionState
def test_state_machine_only_allows_connection_states(self) -> None:
"""
The Connection state machine only allows ConnectionState inputs.
"""
c = h2.connection.H2ConnectionStateMachine()
with pytest.raises(ValueError):
c.process_input(1)
@pytest.mark.parametrize(
"state",
(
s for s in h2.connection.ConnectionState
if s != h2.connection.ConnectionState.CLOSED
),
)
@pytest.mark.parametrize(
"input_",
[
h2.connection.ConnectionInputs.RECV_PRIORITY,
h2.connection.ConnectionInputs.SEND_PRIORITY,
],
)
def test_priority_frames_allowed_in_all_states(self, state, input_) -> None:
"""
Priority frames can be sent/received in all connection states except
closed.
"""
c = h2.connection.H2ConnectionStateMachine()
c.state = state
c.process_input(input_)
class TestStreamStateMachine:
"""
Tests of the stream state machine.
"""
@given(state=sampled_from(h2.stream.StreamState),
input_=sampled_from(h2.stream.StreamInputs))
def test_state_transitions(self, state, input_) -> None:
s = h2.stream.H2StreamStateMachine(stream_id=1)
s.state = state
try:
s.process_input(input_)
except h2.exceptions.StreamClosedError:
# This can only happen for streams that started in the closed
# state OR where the input was RECV_DATA and the state was not
# OPEN or HALF_CLOSED_LOCAL OR where the state was
# HALF_CLOSED_REMOTE and a frame was received.
if state == h2.stream.StreamState.CLOSED:
assert s.state == h2.stream.StreamState.CLOSED
elif input_ == h2.stream.StreamInputs.RECV_DATA:
assert s.state == h2.stream.StreamState.CLOSED
assert state not in (
h2.stream.StreamState.OPEN,
h2.stream.StreamState.HALF_CLOSED_LOCAL,
)
elif state == h2.stream.StreamState.HALF_CLOSED_REMOTE:
assert input_ in (
h2.stream.StreamInputs.RECV_HEADERS,
h2.stream.StreamInputs.RECV_PUSH_PROMISE,
h2.stream.StreamInputs.RECV_DATA,
h2.stream.StreamInputs.RECV_CONTINUATION,
)
except h2.exceptions.ProtocolError:
assert s.state == h2.stream.StreamState.CLOSED
else:
assert s.state in h2.stream.StreamState
def test_state_machine_only_allows_stream_states(self) -> None:
"""
The Stream state machine only allows StreamState inputs.
"""
s = h2.stream.H2StreamStateMachine(stream_id=1)
with pytest.raises(ValueError):
s.process_input(1)
def test_stream_state_machine_forbids_pushes_on_server_streams(self) -> None:
"""
Streams where this peer is a server do not allow receiving pushed
frames.
"""
s = h2.stream.H2StreamStateMachine(stream_id=1)
s.process_input(h2.stream.StreamInputs.RECV_HEADERS)
with pytest.raises(h2.exceptions.ProtocolError):
s.process_input(h2.stream.StreamInputs.RECV_PUSH_PROMISE)
def test_stream_state_machine_forbids_sending_pushes_from_clients(self) -> None:
"""
Streams where this peer is a client do not allow sending pushed frames.
"""
s = h2.stream.H2StreamStateMachine(stream_id=1)
s.process_input(h2.stream.StreamInputs.SEND_HEADERS)
with pytest.raises(h2.exceptions.ProtocolError):
s.process_input(h2.stream.StreamInputs.SEND_PUSH_PROMISE)
@pytest.mark.parametrize(
"input_",
[
h2.stream.StreamInputs.SEND_HEADERS,
h2.stream.StreamInputs.SEND_PUSH_PROMISE,
h2.stream.StreamInputs.SEND_RST_STREAM,
h2.stream.StreamInputs.SEND_DATA,
h2.stream.StreamInputs.SEND_WINDOW_UPDATE,
h2.stream.StreamInputs.SEND_END_STREAM,
],
)
def test_cannot_send_on_closed_streams(self, input_) -> None:
"""
Sending anything but a PRIORITY frame is forbidden on closed streams.
"""
c = h2.stream.H2StreamStateMachine(stream_id=1)
c.state = h2.stream.StreamState.CLOSED
expected_error = (
h2.exceptions.ProtocolError
if input_ == h2.stream.StreamInputs.SEND_PUSH_PROMISE
else h2.exceptions.StreamClosedError
)
with pytest.raises(expected_error):
c.process_input(input_)
|
