

def test_portforward(self):
    '\n        Test port forwarding through Echo protocol.\n        '
    realServerFactory = protocol.ServerFactory()
    realServerFactory.protocol = (lambda : self.serverProtocol)
    realServerPort = reactor.listenTCP(0, nBytes, interface='127.0.0.1')
    self.openPorts.append(realServerPort)
    self.proxyServerFactory = TestableProxyFactory('127.0.0.1', realServerPort.getHost().port)
    proxyServerPort = reactor.listenTCP(0, self.proxyServerFactory, interface='127.0.0.1')
    self.openPorts.append(proxyServerPort)
    nBytes = 1000
    received = []
    d = defer.Deferred()

    def testDataReceived(data):
        received.extend(data)
        if (len(received) >= nBytes):
            self.assertEqual(''.join(received), ('x' * nBytes))
            d.callback(None)
    self.clientProtocol.dataReceived = testDataReceived

    def testConnectionMade():
        self.clientProtocol.transport.write(('x' * nBytes))
    self.clientProtocol.connectionMade = testConnectionMade
    clientFactory = protocol.ClientFactory()
    clientFactory.protocol = (lambda : self.clientProtocol)
    reactor.connectTCP('127.0.0.1', proxyServerPort.getHost().port, clientFactory)
    return d

dataset/ETHPy150Open twisted/twisted/twisted/test/test_protocols.py PortforwardingTests.test_portforward/VarMisuse@40/70 `realServerFactory`->`nBytes`