Method test_task_done

Lib/test/test_asyncio/test_queues.py:465–494

(self)

        self.assertRaises(ValueError, q.task_done)

    async def test_task_done(self):
        q = self.q_class()
        for i in range(100):
            q.put_nowait(i)

        accumulator = 0

        # Two workers get items from the queue and call task_done after each.
        # Join the queue and assert all items have been processed.
        running = True

        async def worker():
            nonlocal accumulator

            while running:
                item = await q.get()
                accumulator += item
                q.task_done()

        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(worker())
                     for index in range(2)]

            await q.join()
            self.assertEqual(sum(range(100)), accumulator)

            # close running generators
            running = False
            for i in range(len(tasks)):
                q.put_nowait(0)

    async def test_join_empty_queue(self):
        q = self.q_class()