test_twisted.py 文件源码

def test_error_deferred_body(self):
        d = defer.Deferred()

        def echo(x):
            d2 = defer.Deferred()

            def cb(result):
                raise IndexError

            reactor.callLater(0, lambda: d2.callback(None))

            d2.addCallback(cb)
            return d2

        p = self.getProcessor({'echo': echo}, expose_request=False)

        request = remoting.Request('echo', envelope=remoting.Envelope())
        request.body = ['a']

        def cb(result):
            self.assertTrue(isinstance(result, remoting.Response))
            self.assertTrue(result.status, remoting.STATUS_ERROR)
            self.assertTrue(isinstance(result.body, remoting.ErrorFault))
            self.assertEqual(result.body.code, 'IndexError')

        return p(request).addCallback(cb).addErrback(lambda x: d.errback())