Method test_put_timeout_preempted

tornado/test/queues_test.py:243–249

(self)

    @gen_test
    def test_put_timeout_preempted(self):
        q = queues.Queue(1)  # type: queues.Queue[int]
        q.put_nowait(0)
        put = q.put(1, timeout=timedelta(seconds=0.01))
        q.get()
        yield gen.sleep(0.02)
        yield put  # No TimeoutError.

    @gen_test
    def test_put_clears_timed_out_putters(self):