EVOLUTION-MANAGER

Edit File: test_state.py

import pytest

from .._events import *
from .._state import *
from .._state import _SWITCH_CONNECT, _SWITCH_UPGRADE, ConnectionState
from .._util import LocalProtocolError

def test_ConnectionState():
    cs = ConnectionState()

# Basic event-triggered transitions

assert cs.states == {CLIENT: IDLE, SERVER: IDLE}

cs.process_event(CLIENT, Request)
    # The SERVER-Request special case:
    assert cs.states == {CLIENT: SEND_BODY, SERVER: SEND_RESPONSE}

# Illegal transitions raise an error and nothing happens
    with pytest.raises(LocalProtocolError):
        cs.process_event(CLIENT, Request)
    assert cs.states == {CLIENT: SEND_BODY, SERVER: SEND_RESPONSE}

cs.process_event(SERVER, InformationalResponse)
    assert cs.states == {CLIENT: SEND_BODY, SERVER: SEND_RESPONSE}

cs.process_event(SERVER, Response)
    assert cs.states == {CLIENT: SEND_BODY, SERVER: SEND_BODY}

cs.process_event(CLIENT, EndOfMessage)
    cs.process_event(SERVER, EndOfMessage)
    assert cs.states == {CLIENT: DONE, SERVER: DONE}

# State-triggered transition

cs.process_event(SERVER, ConnectionClosed)
    assert cs.states == {CLIENT: MUST_CLOSE, SERVER: CLOSED}

def test_ConnectionState_keep_alive():
    # keep_alive = False
    cs = ConnectionState()
    cs.process_event(CLIENT, Request)
    cs.process_keep_alive_disabled()
    cs.process_event(CLIENT, EndOfMessage)
    assert cs.states == {CLIENT: MUST_CLOSE, SERVER: SEND_RESPONSE}

cs.process_event(SERVER, Response)
    cs.process_event(SERVER, EndOfMessage)
    assert cs.states == {CLIENT: MUST_CLOSE, SERVER: MUST_CLOSE}

def test_ConnectionState_keep_alive_in_DONE():
    # Check that if keep_alive is disabled when the CLIENT is already in DONE,
    # then this is sufficient to immediately trigger the DONE -> MUST_CLOSE
    # transition
    cs = ConnectionState()
    cs.process_event(CLIENT, Request)
    cs.process_event(CLIENT, EndOfMessage)
    assert cs.states[CLIENT] is DONE
    cs.process_keep_alive_disabled()
    assert cs.states[CLIENT] is MUST_CLOSE

def test_ConnectionState_switch_denied():
    for switch_type in (_SWITCH_CONNECT, _SWITCH_UPGRADE):
        for deny_early in (True, False):
            cs = ConnectionState()
            cs.process_client_switch_proposal(switch_type)
            cs.process_event(CLIENT, Request)
            cs.process_event(CLIENT, Data)
            assert cs.states == {CLIENT: SEND_BODY, SERVER: SEND_RESPONSE}

assert switch_type in cs.pending_switch_proposals

if deny_early:
                # before client reaches DONE
                cs.process_event(SERVER, Response)
                assert not cs.pending_switch_proposals

cs.process_event(CLIENT, EndOfMessage)

if deny_early:
                assert cs.states == {CLIENT: DONE, SERVER: SEND_BODY}
            else:
                assert cs.states == {
                    CLIENT: MIGHT_SWITCH_PROTOCOL,
                    SERVER: SEND_RESPONSE,
                }

cs.process_event(SERVER, InformationalResponse)
                assert cs.states == {
                    CLIENT: MIGHT_SWITCH_PROTOCOL,
                    SERVER: SEND_RESPONSE,
                }

cs.process_event(SERVER, Response)
                assert cs.states == {CLIENT: DONE, SERVER: SEND_BODY}
                assert not cs.pending_switch_proposals

_response_type_for_switch = {
    _SWITCH_UPGRADE: InformationalResponse,
    _SWITCH_CONNECT: Response,
    None: Response,
}

def test_ConnectionState_protocol_switch_accepted():
    for switch_event in [_SWITCH_UPGRADE, _SWITCH_CONNECT]:
        cs = ConnectionState()
        cs.process_client_switch_proposal(switch_event)
        cs.process_event(CLIENT, Request)
        cs.process_event(CLIENT, Data)
        assert cs.states == {CLIENT: SEND_BODY, SERVER: SEND_RESPONSE}

cs.process_event(CLIENT, EndOfMessage)
        assert cs.states == {CLIENT: MIGHT_SWITCH_PROTOCOL, SERVER: SEND_RESPONSE}

cs.process_event(SERVER, InformationalResponse)
        assert cs.states == {CLIENT: MIGHT_SWITCH_PROTOCOL, SERVER: SEND_RESPONSE}

cs.process_event(SERVER, _response_type_for_switch[switch_event], switch_event)
        assert cs.states == {CLIENT: SWITCHED_PROTOCOL, SERVER: SWITCHED_PROTOCOL}

def test_ConnectionState_double_protocol_switch():
    # CONNECT + Upgrade is legal! Very silly, but legal. So we support
    # it. Because sometimes doing the silly thing is easier than not.
    for server_switch in [None, _SWITCH_UPGRADE, _SWITCH_CONNECT]:
        cs = ConnectionState()
        cs.process_client_switch_proposal(_SWITCH_UPGRADE)
        cs.process_client_switch_proposal(_SWITCH_CONNECT)
        cs.process_event(CLIENT, Request)
        cs.process_event(CLIENT, EndOfMessage)
        assert cs.states == {CLIENT: MIGHT_SWITCH_PROTOCOL, SERVER: SEND_RESPONSE}
        cs.process_event(
            SERVER, _response_type_for_switch[server_switch], server_switch
        )
        if server_switch is None:
            assert cs.states == {CLIENT: DONE, SERVER: SEND_BODY}
        else:
            assert cs.states == {CLIENT: SWITCHED_PROTOCOL, SERVER: SWITCHED_PROTOCOL}

def test_ConnectionState_inconsistent_protocol_switch():
    for client_switches, server_switch in [
        ([], _SWITCH_CONNECT),
        ([], _SWITCH_UPGRADE),
        ([_SWITCH_UPGRADE], _SWITCH_CONNECT),
        ([_SWITCH_CONNECT], _SWITCH_UPGRADE),
    ]:
        cs = ConnectionState()
        for client_switch in client_switches:
            cs.process_client_switch_proposal(client_switch)
        cs.process_event(CLIENT, Request)
        with pytest.raises(LocalProtocolError):
            cs.process_event(SERVER, Response, server_switch)

def test_ConnectionState_keepalive_protocol_switch_interaction():
    # keep_alive=False + pending_switch_proposals
    cs = ConnectionState()
    cs.process_client_switch_proposal(_SWITCH_UPGRADE)
    cs.process_event(CLIENT, Request)
    cs.process_keep_alive_disabled()
    cs.process_event(CLIENT, Data)
    assert cs.states == {CLIENT: SEND_BODY, SERVER: SEND_RESPONSE}

# the protocol switch "wins"
    cs.process_event(CLIENT, EndOfMessage)
    assert cs.states == {CLIENT: MIGHT_SWITCH_PROTOCOL, SERVER: SEND_RESPONSE}

# but when the server denies the request, keep_alive comes back into play
    cs.process_event(SERVER, Response)
    assert cs.states == {CLIENT: MUST_CLOSE, SERVER: SEND_BODY}

def test_ConnectionState_reuse():
    cs = ConnectionState()