test_ssl.py 文件源码

def _runTest(self, clientProto, serverProto, clientIsServer=False):
        self.clientProto = clientProto
        cf = self.clientFactory = protocol.ClientFactory()
        cf.protocol = lambda: clientProto
        if clientIsServer:
            cf.server = 0
        else:
            cf.client = 1

        self.serverProto = serverProto
        sf = self.serverFactory = protocol.ServerFactory()
        sf.protocol = lambda: serverProto
        if clientIsServer:
            sf.client = 0
        else:
            sf.server = 1

        if clientIsServer:
            inCharge = cf
        else:
            inCharge = sf
        inCharge.done = 0

        port = self.port = reactor.listenTCP(0, sf, interface="127.0.0.1")
        portNo = port.getHost().port

        reactor.connectTCP('127.0.0.1', portNo, cf)

        i = 0
        while i < 1000 and not inCharge.done:
            reactor.iterate(0.01)
            i += 1
        self.failUnless(
            inCharge.done,
            "Never finished reading all lines: %s" % (inCharge.lines,))