test_eventlet_handler.py 文件源码

def test_async_exception(self):

        @raises(IOError)
        def check_exc(r):
            r.get()

        def broken():
            raise IOError("Failed")

        with start_stop_one() as handler:
            r = handler.async_result()
            w = handler.spawn(utils.wrap(r)(broken))
            w.join()

        self.assertFalse(r.successful())
        check_exc(r)