"""Tests for queues.py"""
-import unittest
import asyncio
+import unittest
from types import GenericAlias
-from test.test_asyncio import utils as test_utils
def tearDownModule():
asyncio.set_event_loop_policy(None)
-class _QueueTestBase(test_utils.TestCase):
-
- def setUp(self):
- super().setUp()
- self.loop = self.new_test_loop()
+class QueueBasicTests(unittest.IsolatedAsyncioTestCase):
-
-class QueueBasicTests(_QueueTestBase):
-
- def _test_repr_or_str(self, fn, expect_id):
+ async def _test_repr_or_str(self, fn, expect_id):
"""Test Queue's repr or str.
fn is repr or str. expect_id is True if we expect the Queue's id to
appear in fn(Queue()).
"""
- def gen():
- when = yield
- self.assertAlmostEqual(0.1, when)
- when = yield 0.1
- self.assertAlmostEqual(0.2, when)
- yield 0.1
-
- loop = self.new_test_loop(gen)
-
q = asyncio.Queue()
self.assertTrue(fn(q).startswith('<Queue'), fn(q))
id_is_present = hex(id(q)) in fn(q)
self.assertEqual(expect_id, id_is_present)
- async def add_getter():
- q = asyncio.Queue()
+ # getters
+ q = asyncio.Queue()
+ async with asyncio.TaskGroup() as tg:
# Start a task that waits to get.
- loop.create_task(q.get())
+ getter = tg.create_task(q.get())
# Let it start waiting.
await asyncio.sleep(0.1)
self.assertTrue('_getters[1]' in fn(q))
# resume q.get coroutine to finish generator
q.put_nowait(0)
- loop.run_until_complete(add_getter())
+ self.assertEqual(0, await getter)
- async def add_putter():
- q = asyncio.Queue(maxsize=1)
+ # putters
+ q = asyncio.Queue(maxsize=1)
+ async with asyncio.TaskGroup() as tg:
q.put_nowait(1)
# Start a task that waits to put.
- loop.create_task(q.put(2))
+ putter = tg.create_task(q.put(2))
# Let it start waiting.
await asyncio.sleep(0.1)
self.assertTrue('_putters[1]' in fn(q))
# resume q.put coroutine to finish generator
q.get_nowait()
- loop.run_until_complete(add_putter())
+ self.assertTrue(putter.done())
+
q = asyncio.Queue()
q.put_nowait(1)
self.assertTrue('_queue=[1]' in fn(q))
- def test_repr(self):
- self._test_repr_or_str(repr, True)
+ async def test_repr(self):
+ await self._test_repr_or_str(repr, True)
- def test_str(self):
- self._test_repr_or_str(str, False)
+ async def test_str(self):
+ await self._test_repr_or_str(str, False)
def test_generic_alias(self):
q = asyncio.Queue[int]
self.assertEqual(q.__args__, (int,))
self.assertIsInstance(q, GenericAlias)
- def test_empty(self):
+ async def test_empty(self):
q = asyncio.Queue()
self.assertTrue(q.empty())
- q.put_nowait(1)
+ await q.put(1)
self.assertFalse(q.empty())
- self.assertEqual(1, q.get_nowait())
+ self.assertEqual(1, await q.get())
self.assertTrue(q.empty())
- def test_full(self):
+ async def test_full(self):
q = asyncio.Queue()
self.assertFalse(q.full())
q = asyncio.Queue(maxsize=1)
- q.put_nowait(1)
+ await q.put(1)
self.assertTrue(q.full())
- def test_order(self):
+ async def test_order(self):
q = asyncio.Queue()
for i in [1, 3, 2]:
- q.put_nowait(i)
+ await q.put(i)
- items = [q.get_nowait() for _ in range(3)]
+ items = [await q.get() for _ in range(3)]
self.assertEqual([1, 3, 2], items)
- def test_maxsize(self):
-
- def gen():
- when = yield
- self.assertAlmostEqual(0.01, when)
- when = yield 0.01
- self.assertAlmostEqual(0.02, when)
- yield 0.01
-
- loop = self.new_test_loop(gen)
-
+ async def test_maxsize(self):
q = asyncio.Queue(maxsize=2)
self.assertEqual(2, q.maxsize)
have_been_put = []
have_been_put.append(i)
return True
- async def test():
- t = loop.create_task(putter())
- await asyncio.sleep(0.01)
+ t = asyncio.create_task(putter())
+ await asyncio.sleep(0.01)
- # The putter is blocked after putting two items.
- self.assertEqual([0, 1], have_been_put)
- self.assertEqual(0, q.get_nowait())
+ # The putter is blocked after putting two items.
+ self.assertEqual([0, 1], have_been_put)
+ self.assertEqual(0, await q.get())
- # Let the putter resume and put last item.
- await asyncio.sleep(0.01)
- self.assertEqual([0, 1, 2], have_been_put)
- self.assertEqual(1, q.get_nowait())
- self.assertEqual(2, q.get_nowait())
+ # Let the putter resume and put last item.
+ await asyncio.sleep(0.01)
+ self.assertEqual([0, 1, 2], have_been_put)
+ self.assertEqual(1, await q.get())
+ self.assertEqual(2, await q.get())
- self.assertTrue(t.done())
- self.assertTrue(t.result())
-
- loop.run_until_complete(test())
- self.assertAlmostEqual(0.02, loop.time())
+ self.assertTrue(t.done())
+ self.assertTrue(t.result())
-class QueueGetTests(_QueueTestBase):
+class QueueGetTests(unittest.IsolatedAsyncioTestCase):
- def test_blocking_get(self):
+ async def test_blocking_get(self):
q = asyncio.Queue()
q.put_nowait(1)
- async def queue_get():
- return await q.get()
+ self.assertEqual(1, await q.get())
- res = self.loop.run_until_complete(queue_get())
- self.assertEqual(1, res)
+ async def test_get_with_putters(self):
+ loop = asyncio.get_running_loop()
- def test_get_with_putters(self):
q = asyncio.Queue(1)
- q.put_nowait(1)
+ await q.put(1)
- waiter = self.loop.create_future()
+ waiter = loop.create_future()
q._putters.append(waiter)
- res = self.loop.run_until_complete(q.get())
- self.assertEqual(1, res)
+ self.assertEqual(1, await q.get())
self.assertTrue(waiter.done())
self.assertIsNone(waiter.result())
- def test_blocking_get_wait(self):
-
- def gen():
- when = yield
- self.assertAlmostEqual(0.01, when)
- yield 0.01
-
- loop = self.new_test_loop(gen)
-
+ async def test_blocking_get_wait(self):
+ loop = asyncio.get_running_loop()
q = asyncio.Queue()
started = asyncio.Event()
finished = False
finished = True
return res
- async def queue_put():
- loop.call_later(0.01, q.put_nowait, 1)
- queue_get_task = loop.create_task(queue_get())
- await started.wait()
- self.assertFalse(finished)
- res = await queue_get_task
- self.assertTrue(finished)
- return res
-
- res = loop.run_until_complete(queue_put())
+ loop.call_later(0.01, q.put_nowait, 1)
+ queue_get_task = asyncio.create_task(queue_get())
+ await started.wait()
+ self.assertFalse(finished)
+ res = await queue_get_task
+ self.assertTrue(finished)
self.assertEqual(1, res)
- self.assertAlmostEqual(0.01, loop.time())
def test_nonblocking_get(self):
q = asyncio.Queue()
q = asyncio.Queue()
self.assertRaises(asyncio.QueueEmpty, q.get_nowait)
- def test_get_cancelled(self):
-
- def gen():
- when = yield
- self.assertAlmostEqual(0.01, when)
- when = yield 0.01
- self.assertAlmostEqual(0.061, when)
- yield 0.05
-
- loop = self.new_test_loop(gen)
-
+ async def test_get_cancelled(self):
q = asyncio.Queue()
async def queue_get():
return await asyncio.wait_for(q.get(), 0.051)
- async def test():
- get_task = loop.create_task(queue_get())
- await asyncio.sleep(0.01) # let the task start
- q.put_nowait(1)
- return await get_task
-
- self.assertEqual(1, loop.run_until_complete(test()))
- self.assertAlmostEqual(0.06, loop.time())
+ get_task = asyncio.create_task(queue_get())
+ await asyncio.sleep(0.01) # let the task start
+ q.put_nowait(1)
+ self.assertEqual(1, await get_task)
- def test_get_cancelled_race(self):
+ async def test_get_cancelled_race(self):
q = asyncio.Queue()
- t1 = self.loop.create_task(q.get())
- t2 = self.loop.create_task(q.get())
+ t1 = asyncio.create_task(q.get())
+ t2 = asyncio.create_task(q.get())
- test_utils.run_briefly(self.loop)
+ await asyncio.sleep(0)
t1.cancel()
- test_utils.run_briefly(self.loop)
+ await asyncio.sleep(0)
self.assertTrue(t1.done())
- q.put_nowait('a')
- test_utils.run_briefly(self.loop)
- self.assertEqual(t2.result(), 'a')
+ await q.put('a')
+ await asyncio.sleep(0)
+ self.assertEqual('a', await t2)
- def test_get_with_waiting_putters(self):
+ async def test_get_with_waiting_putters(self):
q = asyncio.Queue(maxsize=1)
- self.loop.create_task(q.put('a'))
- self.loop.create_task(q.put('b'))
- test_utils.run_briefly(self.loop)
- self.assertEqual(self.loop.run_until_complete(q.get()), 'a')
- self.assertEqual(self.loop.run_until_complete(q.get()), 'b')
-
- def test_why_are_getters_waiting(self):
- # From issue #268.
- asyncio.set_event_loop(self.loop)
+ asyncio.create_task(q.put('a'))
+ asyncio.create_task(q.put('b'))
+ self.assertEqual(await q.get(), 'a')
+ self.assertEqual(await q.get(), 'b')
+ async def test_why_are_getters_waiting(self):
async def consumer(queue, num_expected):
for _ in range(num_expected):
await queue.get()
for i in range(num_items):
await queue.put(i)
- queue_size = 1
producer_num_items = 5
- async def create_queue():
- queue = asyncio.Queue(queue_size)
- queue._get_loop()
- return queue
-
- async def test():
- q = await create_queue()
- await asyncio.gather(producer(q, producer_num_items),
- consumer(q, producer_num_items))
-
- self.loop.run_until_complete(test())
-
- def test_cancelled_getters_not_being_held_in_self_getters(self):
- def a_generator():
- yield 0.1
- yield 0.2
+ q = asyncio.Queue(1)
+ async with asyncio.TaskGroup() as tg:
+ tg.create_task(producer(q, producer_num_items))
+ tg.create_task(consumer(q, producer_num_items))
- self.loop = self.new_test_loop(a_generator)
+ async def test_cancelled_getters_not_being_held_in_self_getters(self):
+ queue = asyncio.Queue(maxsize=5)
- async def consumer(queue):
- try:
- item = await asyncio.wait_for(queue.get(), 0.1)
- except asyncio.TimeoutError:
- pass
+ with self.assertRaises(TimeoutError):
+ await asyncio.wait_for(queue.get(), 0.1)
- queue = asyncio.Queue(maxsize=5)
- self.loop.run_until_complete(self.loop.create_task(consumer(queue)))
self.assertEqual(len(queue._getters), 0)
-class QueuePutTests(_QueueTestBase):
+class QueuePutTests(unittest.IsolatedAsyncioTestCase):
- def test_blocking_put(self):
+ async def test_blocking_put(self):
q = asyncio.Queue()
- async def queue_put():
- # No maxsize, won't block.
- await q.put(1)
-
- self.loop.run_until_complete(queue_put())
-
- def test_blocking_put_wait(self):
-
- def gen():
- when = yield
- self.assertAlmostEqual(0.01, when)
- yield 0.01
-
- loop = self.new_test_loop(gen)
+ # No maxsize, won't block.
+ await q.put(1)
+ self.assertEqual(1, await q.get())
+ async def test_blocking_put_wait(self):
q = asyncio.Queue(maxsize=1)
started = asyncio.Event()
finished = False
await q.put(2)
finished = True
- async def queue_get():
- loop.call_later(0.01, q.get_nowait)
- queue_put_task = loop.create_task(queue_put())
- await started.wait()
- self.assertFalse(finished)
- await queue_put_task
- self.assertTrue(finished)
-
- loop.run_until_complete(queue_get())
- self.assertAlmostEqual(0.01, loop.time())
+ loop = asyncio.get_running_loop()
+ loop.call_later(0.01, q.get_nowait)
+ queue_put_task = asyncio.create_task(queue_put())
+ await started.wait()
+ self.assertFalse(finished)
+ await queue_put_task
+ self.assertTrue(finished)
def test_nonblocking_put(self):
q = asyncio.Queue()
q.put_nowait(1)
self.assertEqual(1, q.get_nowait())
- def test_get_cancel_drop_one_pending_reader(self):
- def gen():
- yield 0.01
- yield 0.1
-
- loop = self.new_test_loop(gen)
-
+ async def test_get_cancel_drop_one_pending_reader(self):
q = asyncio.Queue()
- reader = loop.create_task(q.get())
+ reader = asyncio.create_task(q.get())
- loop.run_until_complete(asyncio.sleep(0.01))
+ await asyncio.sleep(0.01)
q.put_nowait(1)
q.put_nowait(2)
reader.cancel()
try:
- loop.run_until_complete(reader)
+ await reader
except asyncio.CancelledError:
# try again
- reader = loop.create_task(q.get())
- loop.run_until_complete(reader)
+ reader = asyncio.create_task(q.get())
+ await reader
result = reader.result()
# if we get 2, it means 1 got dropped!
self.assertEqual(1, result)
- def test_get_cancel_drop_many_pending_readers(self):
- def gen():
- yield 0.01
- yield 0.1
-
- loop = self.new_test_loop(gen)
- loop.set_debug(True)
-
+ async def test_get_cancel_drop_many_pending_readers(self):
q = asyncio.Queue()
- reader1 = loop.create_task(q.get())
- reader2 = loop.create_task(q.get())
- reader3 = loop.create_task(q.get())
+ async with asyncio.TaskGroup() as tg:
+ reader1 = tg.create_task(q.get())
+ reader2 = tg.create_task(q.get())
+ reader3 = tg.create_task(q.get())
- loop.run_until_complete(asyncio.sleep(0.01))
+ await asyncio.sleep(0.01)
- q.put_nowait(1)
- q.put_nowait(2)
- reader1.cancel()
+ q.put_nowait(1)
+ q.put_nowait(2)
+ reader1.cancel()
- try:
- loop.run_until_complete(reader1)
- except asyncio.CancelledError:
- pass
+ with self.assertRaises(asyncio.CancelledError):
+ await reader1
- loop.run_until_complete(reader3)
+ await reader3
# It is undefined in which order concurrent readers receive results.
self.assertEqual({reader2.result(), reader3.result()}, {1, 2})
- def test_put_cancel_drop(self):
-
- def gen():
- yield 0.01
- yield 0.1
-
- loop = self.new_test_loop(gen)
-
+ async def test_put_cancel_drop(self):
q = asyncio.Queue(1)
q.put_nowait(1)
# putting a second item in the queue has to block (qsize=1)
- writer = loop.create_task(q.put(2))
- loop.run_until_complete(asyncio.sleep(0.01))
+ writer = asyncio.create_task(q.put(2))
+ await asyncio.sleep(0.01)
value1 = q.get_nowait()
self.assertEqual(value1, 1)
writer.cancel()
try:
- loop.run_until_complete(writer)
+ await writer
except asyncio.CancelledError:
# try again
- writer = loop.create_task(q.put(2))
- loop.run_until_complete(writer)
+ writer = asyncio.create_task(q.put(2))
+ await writer
value2 = q.get_nowait()
self.assertEqual(value2, 2)
q.put_nowait(1)
self.assertRaises(asyncio.QueueFull, q.put_nowait, 2)
- def test_float_maxsize(self):
+ async def test_float_maxsize(self):
q = asyncio.Queue(maxsize=1.3, )
q.put_nowait(1)
q.put_nowait(2)
q = asyncio.Queue(maxsize=1.3, )
- async def queue_put():
- await q.put(1)
- await q.put(2)
- self.assertTrue(q.full())
- self.loop.run_until_complete(queue_put())
+ await q.put(1)
+ await q.put(2)
+ self.assertTrue(q.full())
- def test_put_cancelled(self):
+ async def test_put_cancelled(self):
q = asyncio.Queue()
async def queue_put():
await q.put(1)
return True
- async def test():
- return await q.get()
+ t = asyncio.create_task(queue_put())
- t = self.loop.create_task(queue_put())
- self.assertEqual(1, self.loop.run_until_complete(test()))
+ self.assertEqual(1, await q.get())
self.assertTrue(t.done())
self.assertTrue(t.result())
- def test_put_cancelled_race(self):
+ async def test_put_cancelled_race(self):
q = asyncio.Queue(maxsize=1)
- put_a = self.loop.create_task(q.put('a'))
- put_b = self.loop.create_task(q.put('b'))
- put_c = self.loop.create_task(q.put('X'))
+ put_a = asyncio.create_task(q.put('a'))
+ put_b = asyncio.create_task(q.put('b'))
+ put_c = asyncio.create_task(q.put('X'))
- test_utils.run_briefly(self.loop)
+ await asyncio.sleep(0)
self.assertTrue(put_a.done())
self.assertFalse(put_b.done())
put_c.cancel()
- test_utils.run_briefly(self.loop)
+ await asyncio.sleep(0)
self.assertTrue(put_c.done())
self.assertEqual(q.get_nowait(), 'a')
- test_utils.run_briefly(self.loop)
+ await asyncio.sleep(0)
self.assertEqual(q.get_nowait(), 'b')
- self.loop.run_until_complete(put_b)
+ await put_b
- def test_put_with_waiting_getters(self):
+ async def test_put_with_waiting_getters(self):
q = asyncio.Queue()
- t = self.loop.create_task(q.get())
- test_utils.run_briefly(self.loop)
- self.loop.run_until_complete(q.put('a'))
- self.assertEqual(self.loop.run_until_complete(t), 'a')
-
- def test_why_are_putters_waiting(self):
- # From issue #265.
- asyncio.set_event_loop(self.loop)
-
- async def create_queue():
- q = asyncio.Queue(2)
- q._get_loop()
- return q
+ t = asyncio.create_task(q.get())
+ await asyncio.sleep(0)
+ await q.put('a')
+ self.assertEqual(await t, 'a')
- queue = self.loop.run_until_complete(create_queue())
+ async def test_why_are_putters_waiting(self):
+ queue = asyncio.Queue(2)
async def putter(item):
await queue.put(item)
await asyncio.sleep(0)
num = queue.qsize()
for _ in range(num):
- item = queue.get_nowait()
-
- async def test():
- t0 = putter(0)
- t1 = putter(1)
- t2 = putter(2)
- t3 = putter(3)
- await asyncio.gather(getter(), t0, t1, t2, t3)
-
- self.loop.run_until_complete(test())
+ queue.get_nowait()
- def test_cancelled_puts_not_being_held_in_self_putters(self):
- def a_generator():
- yield 0.01
- yield 0.1
-
- loop = self.new_test_loop(a_generator)
+ async with asyncio.TaskGroup() as tg:
+ tg.create_task(getter())
+ tg.create_task(putter(0))
+ tg.create_task(putter(1))
+ tg.create_task(putter(2))
+ tg.create_task(putter(3))
+ async def test_cancelled_puts_not_being_held_in_self_putters(self):
# Full queue.
queue = asyncio.Queue(maxsize=1)
queue.put_nowait(1)
# Task waiting for space to put an item in the queue.
- put_task = loop.create_task(queue.put(1))
- loop.run_until_complete(asyncio.sleep(0.01))
+ put_task = asyncio.create_task(queue.put(1))
+ await asyncio.sleep(0.01)
# Check that the putter is correctly removed from queue._putters when
# the task is canceled.
self.assertEqual(len(queue._putters), 1)
put_task.cancel()
with self.assertRaises(asyncio.CancelledError):
- loop.run_until_complete(put_task)
+ await put_task
self.assertEqual(len(queue._putters), 0)
- def test_cancelled_put_silence_value_error_exception(self):
- def gen():
- yield 0.01
- yield 0.1
-
- loop = self.new_test_loop(gen)
-
+ async def test_cancelled_put_silence_value_error_exception(self):
# Full Queue.
queue = asyncio.Queue(1)
queue.put_nowait(1)
# Task waiting for space to put a item in the queue.
- put_task = loop.create_task(queue.put(1))
- loop.run_until_complete(asyncio.sleep(0.01))
+ put_task = asyncio.create_task(queue.put(1))
+ await asyncio.sleep(0.01)
# get_nowait() remove the future of put_task from queue._putters.
queue.get_nowait()
# inside queue.put should be silenced.
# If the ValueError is silenced we should catch a CancelledError.
with self.assertRaises(asyncio.CancelledError):
- loop.run_until_complete(put_task)
+ await put_task
-class LifoQueueTests(_QueueTestBase):
+class LifoQueueTests(unittest.IsolatedAsyncioTestCase):
- def test_order(self):
+ async def test_order(self):
q = asyncio.LifoQueue()
for i in [1, 3, 2]:
- q.put_nowait(i)
+ await q.put(i)
- items = [q.get_nowait() for _ in range(3)]
+ items = [await q.get() for _ in range(3)]
self.assertEqual([2, 3, 1], items)
-class PriorityQueueTests(_QueueTestBase):
+class PriorityQueueTests(unittest.IsolatedAsyncioTestCase):
- def test_order(self):
+ async def test_order(self):
q = asyncio.PriorityQueue()
for i in [1, 3, 2]:
- q.put_nowait(i)
+ await q.put(i)
- items = [q.get_nowait() for _ in range(3)]
+ items = [await q.get() for _ in range(3)]
self.assertEqual([1, 2, 3], items)
q = self.q_class()
self.assertRaises(ValueError, q.task_done)
- def test_task_done(self):
+ async def test_task_done(self):
q = self.q_class()
for i in range(100):
q.put_nowait(i)
accumulator += item
q.task_done()
- async def test():
- tasks = [self.loop.create_task(worker())
+ async with asyncio.TaskGroup() as tg:
+ tasks = [tg.create_task(worker())
for index in range(2)]
await q.join()
- return tasks
+ self.assertEqual(sum(range(100)), accumulator)
- tasks = self.loop.run_until_complete(test())
- self.assertEqual(sum(range(100)), accumulator)
+ # close running generators
+ running = False
+ for i in range(len(tasks)):
+ q.put_nowait(0)
- # close running generators
- running = False
- for i in range(len(tasks)):
- q.put_nowait(0)
- self.loop.run_until_complete(asyncio.wait(tasks))
-
- def test_join_empty_queue(self):
+ async def test_join_empty_queue(self):
q = self.q_class()
# Test that a queue join()s successfully, and before anything else
# (done twice for insurance).
- async def join():
- await q.join()
- await q.join()
-
- self.loop.run_until_complete(join())
+ await q.join()
+ await q.join()
- def test_format(self):
+ async def test_format(self):
q = self.q_class()
self.assertEqual(q._format(), 'maxsize=0')
self.assertEqual(q._format(), 'maxsize=0 tasks=2')
-class QueueJoinTests(_QueueJoinTestMixin, _QueueTestBase):
+class QueueJoinTests(_QueueJoinTestMixin, unittest.IsolatedAsyncioTestCase):
q_class = asyncio.Queue
-class LifoQueueJoinTests(_QueueJoinTestMixin, _QueueTestBase):
+class LifoQueueJoinTests(_QueueJoinTestMixin, unittest.IsolatedAsyncioTestCase):
q_class = asyncio.LifoQueue
-class PriorityQueueJoinTests(_QueueJoinTestMixin, _QueueTestBase):
+class PriorityQueueJoinTests(_QueueJoinTestMixin, unittest.IsolatedAsyncioTestCase):
q_class = asyncio.PriorityQueue
projects / thirdparty / Python / cpython.git / commitdiff
