test_twisted.py 文件源码

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

        def preprocessor(http_request, service_request):
            return reactor.callLater(0, lambda: True)

        preprocessor = gateway.expose_request(preprocessor)
        p = self.getProcessor({'echo': lambda x: x}, preprocessor=preprocessor)

        request = remoting.Request('echo', envelope=remoting.Envelope())

        def cb(result):
            self.assertTrue(result)
            d.callback(None)

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

        return d