Rooba
/
uvloop
mirror of https://github.com/MagicStack/uvloop.git
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

tests/signals: Silence resource warnings in subprocesses

This commit is contained in:
Yury Selivanov 2016-04-19 10:39:28 -04:00
parent 97a64505d1
commit e7202088a3
1 changed files with 25 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
tests/test_signals.py
View File

@ -97,7 +97,12 @@ async def worker():
loop = """ + self.NEW_LOOP + """ loop = """ + self.NEW_LOOP + """
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
loop.create_task(worker()) loop.create_task(worker())
try:
loop.run_forever() loop.run_forever()
finally:
srv.close()
loop.run_until_complete(srv.wait_closed())
loop.close()
""" """
proc = await asyncio.create_subprocess_exec( proc = await asyncio.create_subprocess_exec(
@ -141,7 +146,12 @@ loop.add_signal_handler(signal.SIGINT, handler_sig, '!s-int!')
loop.add_signal_handler(signal.SIGHUP, handler_hup, '!s-hup!') loop.add_signal_handler(signal.SIGHUP, handler_hup, '!s-hup!')
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
loop.create_task(worker()) loop.create_task(worker())
try:
loop.run_forever() loop.run_forever()
finally:
srv.close()
loop.run_until_complete(srv.wait_closed())
loop.close()
""" """
proc = await asyncio.create_subprocess_exec( proc = await asyncio.create_subprocess_exec(
@ -193,7 +203,13 @@ loop.add_signal_handler(signal.SIGUSR1, handler1)
loop.add_signal_handler(signal.SIGUSR2, handler2) loop.add_signal_handler(signal.SIGUSR2, handler2)
loop.add_signal_handler(signal.SIGHUP, handler_hup) loop.add_signal_handler(signal.SIGHUP, handler_hup)
loop.create_task(worker()) loop.create_task(worker())
try:
loop.run_forever() loop.run_forever()
finally:
srv.close()
loop.run_until_complete(srv.wait_closed())
loop.close()
""" """
proc = await asyncio.create_subprocess_exec( proc = await asyncio.create_subprocess_exec(
@ -261,7 +277,12 @@ loop = """ + self.NEW_LOOP + """
loop.add_signal_handler(signal.SIGUSR1, aio_handler) loop.add_signal_handler(signal.SIGUSR1, aio_handler)
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
loop.create_task(worker()) loop.create_task(worker())
try:
loop.run_forever() loop.run_forever()
finally:
srv.close()
loop.run_until_complete(srv.wait_closed())
loop.close()
print('step3', flush=True) print('step3', flush=True)
print(input(), flush=True) print(input(), flush=True)
""" """