test_twisted.py 文件源码

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

        def auth(u, p):
            return reactor.callLater(0, lambda: True)

        p = self.getProcessor({'echo': lambda x: x}, authenticator=auth)

        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