Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_select_cancel_true(loop: asyncio.AbstractEventLoop):
event1 = asyncio.Event()
event2 = asyncio.Event()
event3 = asyncio.Event()
async def coro1():
await event1.wait()
event2.set()
async def coro2():
await asyncio.sleep(0)
event3.set()
await aiomisc.select(coro2(), coro1(), cancel=True)
assert not event1.is_set()
assert not event2.is_set()
async def test_select(loop: asyncio.AbstractEventLoop):
f_one = asyncio.Event()
f_two = asyncio.Event()
loop.call_soon(f_one.set)
loop.call_later(1, f_two.set)
two, one = await aiomisc.select(f_two.wait(), f_one.wait())
assert one
assert two is None
one, two = await aiomisc.select(f_one.wait(), f_two.wait())
assert one
async def test_select_exception(loop: asyncio.AbstractEventLoop):
event = asyncio.Event()
async def bad_coro():
if event.is_set():
await asyncio.sleep(10)
else:
await asyncio.sleep(0)
raise ZeroDivisionError
with pytest.raises(ZeroDivisionError):
await aiomisc.select(bad_coro(), bad_coro())
results = []
async def good_coro(wait):
nonlocal results
try:
if wait:
await asyncio.sleep(10)
else:
await asyncio.sleep(0)
results.append(True)
return True
finally:
results.append(None)
one, two = await aiomisc.select(good_coro(False), good_coro(True))
assert one
assert results[0]
assert results[1] is None