test_all.py 文件源码

def test_client_make_real_stream():
    alice = env.clients[0]
    alice.transport.written = []
    alice.reactor = task.Clock()

    alice.make_real_stream()
    assert len(alice.transport.written) == 1
    alice.reactor.advance(100)
    assert len(alice.transport.written) == 2

    calls = alice.reactor.getDelayedCalls()
    for c in calls:
        assert c.func == alice.make_real_stream

    # Check whether a real message is sent if buffer non-empty
    receiver = env.pubs_clients[1]
    path = [alice.provider] + env.pubs_mixes + [receiver.provider]
    test_packet = 'ABCDefgHIJKlmnOPRstUWxyz'
    alice.output_buffer.put(test_packet)
    alice.make_real_stream()
    packet, addr = alice.transport.written[-1]
    assert petlib.pack.decode(packet) == 'ABCDefgHIJKlmnOPRstUWxyz'
    assert addr == (alice.provider.host, alice.provider.port)