

def test_endpoint():
    (server, session, transport, conn) = _get_test_environment()
    transport.recv(proto.connect('/test?a=123&b=456'))
    eq_(transport.pop_outgoing(), '1::/test')
    conn_test = session.endpoints['/test']
    eq_(conn_test.endpoint, '/test')
    eq_(conn_test.is_open, True)
    eq_(conn_test.request.arguments, dict(a=['123'], b=['456']))
    eq_(conn_test.request.get_argument('a'), '123')
    transport.recv(proto.message('/test', 'abc'))
    eq_(conn_test.pop_incoming(), 'abc')
    eq_(transport.pop_outgoing(), '3::/test:abc')
    transport.recv(proto.disconnect('/test'))
    eq_(transport.pop_outgoing(), '0::/test')
    eq_(conn_test.is_open, False)
    eq_(conn.is_open, True)
    eq_(session.is_closed, False)
    eq_(session.endpoints, dict())
    transport.recv(proto.connect('/test2'))
    eq_(transport.pop_outgoing(), '1::/test2')
    conn_test = session.endpoints['/test2']
    eq_(conn.request.arguments, dict())
    transport.recv(proto.disconnect())
    eq_(transport.pop_outgoing(), '0::/test2')
    eq_(transport.pop_outgoing(), '0::')
    eq_(conn_test.is_open, False)
    eq_(conn.is_open, False)
    eq_(session.is_closed, True)

dataset/ETHPy150Open mrjoes/tornadio2/tests/session_test.py test_endpoint/VarMisuse@242/336 `conn_test`->`conn`