test_timeout.py 文件源码

def test_timeout_enable_zero_coro_not_started(loop):
    coro_started = False

    @asyncio.coroutine
    def coro():
        nonlocal coro_started
        coro_started = True

    with pytest.raises(asyncio.TimeoutError):
        with timeout(0, loop=loop) as cm:
            yield from asyncio.sleep(0, loop=loop)
            yield from coro()

    assert cm.expired
    assert coro_started is False