Method test_put_timeout

tornado/test/queues_test.py:227–240

(self)

    @gen_test
    def test_put_timeout(self):
        q = queues.Queue(1)  # type: queues.Queue[int]
        q.put_nowait(0)  # Now it's full.
        put_timeout = q.put(1, timeout=timedelta(seconds=0.01))
        put = q.put(2)
        with self.assertRaises(TimeoutError):
            yield put_timeout

        self.assertEqual(0, q.get_nowait())
        # 1 was never put in the queue.
        self.assertEqual(2, (yield q.get()))

        # Final get() unblocked this putter.
        yield put

    @gen_test
    def test_put_timeout_preempted(self):