test_endpoints.py 文件源码

def test_execCancelled(self):
        """
        If execution of the command is cancelled via the L{Deferred} returned
        by L{SSHCommandClientEndpoint.connect}, the connection is closed
        immediately.
        """
        self.realm.channelLookup[b'session'] = UnsatisfiedExecSession
        endpoint = self.create()

        factory = Factory()
        factory.protocol = Protocol
        connected = endpoint.connect(factory)
        server, client, pump = self.finishConnection()

        connected.cancel()

        f = self.failureResultOf(connected)
        f.trap(CancelledError)

        self.assertClientTransportState(client, True)