test_gipc.py 文件源码

def test_simpletimeout_expires(self):
        with pipe() as (r, w):
            t = gevent.Timeout.start_new(SHORTTIME)
            try:
                r.get(timeout=t)
                assert False
            except gevent.Timeout as raised_timeout:
                if t is not raised_timeout:
                    raise