1 import pytest 2 3 from .._events import ( 4 ConnectionClosed, 5 Data, 6 EndOfMessage, 7 Event, 8 InformationalResponse, 9 Request, 10 Response, 11 ) 12 from .._state import ( 13 _SWITCH_CONNECT, 14 _SWITCH_UPGRADE, 15 CLIENT, 16 CLOSED, 17 ConnectionState, 18 DONE, 19 IDLE, 20 MIGHT_SWITCH_PROTOCOL, 21 MUST_CLOSE, 22 SEND_BODY, 23 SEND_RESPONSE, 24 SERVER, 25 SWITCHED_PROTOCOL, 26 ) 27 from .._util import LocalProtocolError 28 29 30 def test_ConnectionState() -> None: 31 cs = ConnectionState() 32 33 # Basic event-triggered transitions 34 35 assert cs.states == {CLIENT: IDLE, SERVER: IDLE} 36 37 cs.process_event(CLIENT, Request) 38 # The SERVER-Request special case: 39 assert cs.states == {CLIENT: SEND_BODY, SERVER: SEND_RESPONSE} 40 41 # Illegal transitions raise an error and nothing happens 42 with pytest.raises(LocalProtocolError): 43 cs.process_event(CLIENT, Request) 44 assert cs.states == {CLIENT: SEND_BODY, SERVER: SEND_RESPONSE} 45 46 cs.process_event(SERVER, InformationalResponse) 47 assert cs.states == {CLIENT: SEND_BODY, SERVER: SEND_RESPONSE} 48 49 cs.process_event(SERVER, Response) 50 assert cs.states == {CLIENT: SEND_BODY, SERVER: SEND_BODY} 51 52 cs.process_event(CLIENT, EndOfMessage) 53 cs.process_event(SERVER, EndOfMessage) 54 assert cs.states == {CLIENT: DONE, SERVER: DONE} 55 56 # State-triggered transition 57 58 cs.process_event(SERVER, ConnectionClosed) 59 assert cs.states == {CLIENT: MUST_CLOSE, SERVER: CLOSED} 60 61 62 def test_ConnectionState_keep_alive() -> None: 63 # keep_alive = False 64 cs = ConnectionState() 65 cs.process_event(CLIENT, Request) 66 cs.process_keep_alive_disabled() 67 cs.process_event(CLIENT, EndOfMessage) 68 assert cs.states == {CLIENT: MUST_CLOSE, SERVER: SEND_RESPONSE} 69 70 cs.process_event(SERVER, Response) 71 cs.process_event(SERVER, EndOfMessage) 72 assert cs.states == {CLIENT: MUST_CLOSE, SERVER: MUST_CLOSE} 73 74 75 def test_ConnectionState_keep_alive_in_DONE() -> None: 76 # Check that if keep_alive is disabled when the CLIENT is already in DONE, 77 # then this is sufficient to immediately trigger the DONE -> MUST_CLOSE 78 # transition 79 cs = ConnectionState() 80 cs.process_event(CLIENT, Request) 81 cs.process_event(CLIENT, EndOfMessage) 82 assert cs.states[CLIENT] is DONE 83 cs.process_keep_alive_disabled() 84 assert cs.states[CLIENT] is MUST_CLOSE 85 86 87 def test_ConnectionState_switch_denied() -> None: 88 for switch_type in (_SWITCH_CONNECT, _SWITCH_UPGRADE): 89 for deny_early in (True, False): 90 cs = ConnectionState() 91 cs.process_client_switch_proposal(switch_type) 92 cs.process_event(CLIENT, Request) 93 cs.process_event(CLIENT, Data) 94 assert cs.states == {CLIENT: SEND_BODY, SERVER: SEND_RESPONSE} 95 96 assert switch_type in cs.pending_switch_proposals 97 98 if deny_early: 99 # before client reaches DONE 100 cs.process_event(SERVER, Response) 101 assert not cs.pending_switch_proposals 102 103 cs.process_event(CLIENT, EndOfMessage) 104 105 if deny_early: 106 assert cs.states == {CLIENT: DONE, SERVER: SEND_BODY} 107 else: 108 assert cs.states == { 109 CLIENT: MIGHT_SWITCH_PROTOCOL, 110 SERVER: SEND_RESPONSE, 111 } 112 113 cs.process_event(SERVER, InformationalResponse) 114 assert cs.states == { 115 CLIENT: MIGHT_SWITCH_PROTOCOL, 116 SERVER: SEND_RESPONSE, 117 } 118 119 cs.process_event(SERVER, Response) 120 assert cs.states == {CLIENT: DONE, SERVER: SEND_BODY} 121 assert not cs.pending_switch_proposals 122 123 124 _response_type_for_switch = { 125 _SWITCH_UPGRADE: InformationalResponse, 126 _SWITCH_CONNECT: Response, 127 None: Response, 128 } 129 130 131 def test_ConnectionState_protocol_switch_accepted() -> None: 132 for switch_event in [_SWITCH_UPGRADE, _SWITCH_CONNECT]: 133 cs = ConnectionState() 134 cs.process_client_switch_proposal(switch_event) 135 cs.process_event(CLIENT, Request) 136 cs.process_event(CLIENT, Data) 137 assert cs.states == {CLIENT: SEND_BODY, SERVER: SEND_RESPONSE} 138 139 cs.process_event(CLIENT, EndOfMessage) 140 assert cs.states == {CLIENT: MIGHT_SWITCH_PROTOCOL, SERVER: SEND_RESPONSE} 141 142 cs.process_event(SERVER, InformationalResponse) 143 assert cs.states == {CLIENT: MIGHT_SWITCH_PROTOCOL, SERVER: SEND_RESPONSE} 144 145 cs.process_event(SERVER, _response_type_for_switch[switch_event], switch_event) 146 assert cs.states == {CLIENT: SWITCHED_PROTOCOL, SERVER: SWITCHED_PROTOCOL} 147 148 149 def test_ConnectionState_double_protocol_switch() -> None: 150 # CONNECT + Upgrade is legal! Very silly, but legal. So we support 151 # it. Because sometimes doing the silly thing is easier than not. 152 for server_switch in [None, _SWITCH_UPGRADE, _SWITCH_CONNECT]: 153 cs = ConnectionState() 154 cs.process_client_switch_proposal(_SWITCH_UPGRADE) 155 cs.process_client_switch_proposal(_SWITCH_CONNECT) 156 cs.process_event(CLIENT, Request) 157 cs.process_event(CLIENT, EndOfMessage) 158 assert cs.states == {CLIENT: MIGHT_SWITCH_PROTOCOL, SERVER: SEND_RESPONSE} 159 cs.process_event( 160 SERVER, _response_type_for_switch[server_switch], server_switch 161 ) 162 if server_switch is None: 163 assert cs.states == {CLIENT: DONE, SERVER: SEND_BODY} 164 else: 165 assert cs.states == {CLIENT: SWITCHED_PROTOCOL, SERVER: SWITCHED_PROTOCOL} 166 167 168 def test_ConnectionState_inconsistent_protocol_switch() -> None: 169 for client_switches, server_switch in [ 170 ([], _SWITCH_CONNECT), 171 ([], _SWITCH_UPGRADE), 172 ([_SWITCH_UPGRADE], _SWITCH_CONNECT), 173 ([_SWITCH_CONNECT], _SWITCH_UPGRADE), 174 ]: 175 cs = ConnectionState() 176 for client_switch in client_switches: # type: ignore[attr-defined] 177 cs.process_client_switch_proposal(client_switch) 178 cs.process_event(CLIENT, Request) 179 with pytest.raises(LocalProtocolError): 180 cs.process_event(SERVER, Response, server_switch) 181 182 183 def test_ConnectionState_keepalive_protocol_switch_interaction() -> None: 184 # keep_alive=False + pending_switch_proposals 185 cs = ConnectionState() 186 cs.process_client_switch_proposal(_SWITCH_UPGRADE) 187 cs.process_event(CLIENT, Request) 188 cs.process_keep_alive_disabled() 189 cs.process_event(CLIENT, Data) 190 assert cs.states == {CLIENT: SEND_BODY, SERVER: SEND_RESPONSE} 191 192 # the protocol switch "wins" 193 cs.process_event(CLIENT, EndOfMessage) 194 assert cs.states == {CLIENT: MIGHT_SWITCH_PROTOCOL, SERVER: SEND_RESPONSE} 195 196 # but when the server denies the request, keep_alive comes back into play 197 cs.process_event(SERVER, Response) 198 assert cs.states == {CLIENT: MUST_CLOSE, SERVER: SEND_BODY} 199 200 201 def test_ConnectionState_reuse() -> None: 202 cs = ConnectionState() 203 204 with pytest.raises(LocalProtocolError): 205 cs.start_next_cycle() 206 207 cs.process_event(CLIENT, Request) 208 cs.process_event(CLIENT, EndOfMessage) 209 210 with pytest.raises(LocalProtocolError): 211 cs.start_next_cycle() 212 213 cs.process_event(SERVER, Response) 214 cs.process_event(SERVER, EndOfMessage) 215 216 cs.start_next_cycle() 217 assert cs.states == {CLIENT: IDLE, SERVER: IDLE} 218 219 # No keepalive 220 221 cs.process_event(CLIENT, Request) 222 cs.process_keep_alive_disabled() 223 cs.process_event(CLIENT, EndOfMessage) 224 cs.process_event(SERVER, Response) 225 cs.process_event(SERVER, EndOfMessage) 226 227 with pytest.raises(LocalProtocolError): 228 cs.start_next_cycle() 229 230 # One side closed 231 232 cs = ConnectionState() 233 cs.process_event(CLIENT, Request) 234 cs.process_event(CLIENT, EndOfMessage) 235 cs.process_event(CLIENT, ConnectionClosed) 236 cs.process_event(SERVER, Response) 237 cs.process_event(SERVER, EndOfMessage) 238 239 with pytest.raises(LocalProtocolError): 240 cs.start_next_cycle() 241 242 # Succesful protocol switch 243 244 cs = ConnectionState() 245 cs.process_client_switch_proposal(_SWITCH_UPGRADE) 246 cs.process_event(CLIENT, Request) 247 cs.process_event(CLIENT, EndOfMessage) 248 cs.process_event(SERVER, InformationalResponse, _SWITCH_UPGRADE) 249 250 with pytest.raises(LocalProtocolError): 251 cs.start_next_cycle() 252 253 # Failed protocol switch 254 255 cs = ConnectionState() 256 cs.process_client_switch_proposal(_SWITCH_UPGRADE) 257 cs.process_event(CLIENT, Request) 258 cs.process_event(CLIENT, EndOfMessage) 259 cs.process_event(SERVER, Response) 260 cs.process_event(SERVER, EndOfMessage) 261 262 cs.start_next_cycle() 263 assert cs.states == {CLIENT: IDLE, SERVER: IDLE} 264 265 266 def test_server_request_is_illegal() -> None: 267 # There used to be a bug in how we handled the Request special case that 268 # made this allowed... 269 cs = ConnectionState() 270 with pytest.raises(LocalProtocolError): 271 cs.process_event(SERVER, Request) 272