test_onion_mix.py 文件源码

def test_onion_transport():
    """
    integration test for onion transport
    """
    chutney_control_port = os.environ.get('CHUTNEY_CONTROL_PORT')
    if chutney_control_port is None:
        print "CHUTNEY_CONTROL_PORT not set, aborting test"
        return

    params = SphinxParams(max_hops=5, payload_size=1024)
    sphinx_packet_size = params.get_sphinx_forward_size()
    transport_factory = create_transport_factory(sphinx_packet_size, chutney_control_port)
    transport = yield transport_factory.build_transport()
    received_d = defer.Deferred()
    received_buffer = []

    def packet_received(packet):
        print "packet received of len %s" % len(packet)
        received_buffer.append(packet)
        received_d.callback(None)

    protocol = FakeMixProtocol(packet_received)
    yield protocol.make_connection(transport)
    onion_host, onion_port = transport.addr
    tor_endpoint = transport.tor.stream_via(onion_host, onion_port)
    send_message_protocol = Int32StringReceiver()
    remote_mix_protocol = yield endpoints.connectProtocol(tor_endpoint, send_message_protocol)
    message = b"A" * sphinx_packet_size
    remote_mix_protocol.sendString(message)
    remote_mix_protocol.transport.loseConnection()
    yield received_d
    assert received_buffer[0] == message