test_twisted.py 文件源码

def test_deferred_service(self):
        def echo(data):
            x = defer.Deferred()
            reactor.callLater(0, x.callback, data)

            return x

        self.gw.addService(echo)
        d = self.doRequest('echo', 'hello')

        def cb(response):
            self.assertEqual(response.amfVersion, pyamf.AMF3)

            self.assertTrue('/1' in response)
            body_response = response['/1']

            self.assertEqual(body_response.status, remoting.STATUS_OK)
            self.assertEqual(body_response.body, 'hello')

        return d.addCallback(cb)