test_amp.py 文件源码

def test_retry_with_timeout(self):
        """
        If a C{retryTimeout} is set, the L{RemoteObject} object will errback
        failed L{MethodCall}s after that amount of seconds, without retrying
        them when the connection established again.
        """
        self.client.retryOnReconnect = True
        self.client.retryTimeout = 0.1
        self.client.factor = 1  # Reconnect slower than timeout
        connector = reactor.connectUNIX(self.socket, self.client)
        remote = yield self.client.getRemoteObject()

        # Disconnect
        connector.disconnect()

        error = yield self.assertFailure(remote.method("foo"), MethodCallError)
        self.assertEqual("timeout", str(error))

        self.client.stopTrying()
        connector.disconnect()