test_ssl.py 文件源码

def test_bothSidesLoseConnection(self):
        """
        Both sides of SSL connection close connection; the connections should
        close cleanly, and only after the underlying TCP connection has
        disconnected.
        """
        @implementer(interfaces.IHandshakeListener)
        class CloseAfterHandshake(protocol.Protocol):
            gotData = False

            def __init__(self):
                self.done = defer.Deferred()

            def handshakeCompleted(self):
                self.transport.loseConnection()

            def connectionLost(self, reason):
                self.done.errback(reason)
                del self.done

        org = "twisted.test.test_ssl"
        self.setupServerAndClient(
            (org, org + ", client"), {},
            (org, org + ", server"), {})

        serverProtocol = CloseAfterHandshake()
        serverProtocolFactory = protocol.ServerFactory()
        serverProtocolFactory.protocol = lambda: serverProtocol
        serverPort = reactor.listenSSL(0,
            serverProtocolFactory, self.serverCtxFactory)
        self.addCleanup(serverPort.stopListening)

        clientProtocol = CloseAfterHandshake()
        clientProtocolFactory = protocol.ClientFactory()
        clientProtocolFactory.protocol = lambda: clientProtocol
        reactor.connectSSL('127.0.0.1',
            serverPort.getHost().port, clientProtocolFactory, self.clientCtxFactory)

        def checkResult(failure):
            failure.trap(ConnectionDone)
        return defer.gatherResults(
            [clientProtocol.done.addErrback(checkResult),
             serverProtocol.done.addErrback(checkResult)])