Method test_async_for

tornado/test/queues_test.py:161–174

(self)

    @gen_test
    def test_async_for(self):
        q = queues.Queue()  # type: queues.Queue[int]
        for i in range(5):
            q.put(i)

        async def f():
            results = []
            async for i in q:
                results.append(i)
                if i == 4:
                    return results

        results = yield f()
        self.assertEqual(results, list(range(5)))


class QueuePutTest(AsyncTestCase):