traverseda
/
cpython
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

bpo-31034: Reliable signal handler for test_asyncio (#2867) 

* bpo-31034: Reliable signal handler for test_asyncio

Don't rely on the current SIGHUP signal handler, make sure that it's
set to the "default" signal handler: SIG_DFL.

* Add comments
This commit is contained in:
Victor Stinner 2017-07-25 19:19:09 +02:00 committed by GitHub
parent 302bbbe9ba
commit 830080913c
2 changed files with 41 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
Lib/test/test_asyncio/test_events.py
View File

@ -1980,19 +1980,26 @@ class SubprocessTestsMixin:
@unittest.skipIf(sys.platform == 'win32', "Don't have SIGHUP") @unittest.skipIf(sys.platform == 'win32', "Don't have SIGHUP")
def test_subprocess_send_signal(self): def test_subprocess_send_signal(self):
prog = os.path.join(os.path.dirname(__file__), 'echo.py') # bpo-31034: Make sure that we get the default signal handler (killing
# the process). The parent process may have decided to ignore SIGHUP,
# and signal handlers are inherited.
old_handler = signal.signal(signal.SIGHUP, signal.SIG_DFL)
try:
prog = os.path.join(os.path.dirname(__file__), 'echo.py')
connect = self.loop.subprocess_exec( connect = self.loop.subprocess_exec(
functools.partial(MySubprocessProtocol, self.loop), functools.partial(MySubprocessProtocol, self.loop),
sys.executable, prog) sys.executable, prog)
transp, proto = self.loop.run_until_complete(connect) transp, proto = self.loop.run_until_complete(connect)
self.assertIsInstance(proto, MySubprocessProtocol) self.assertIsInstance(proto, MySubprocessProtocol)
self.loop.run_until_complete(proto.connected) self.loop.run_until_complete(proto.connected)
transp.send_signal(signal.SIGHUP) transp.send_signal(signal.SIGHUP)
self.loop.run_until_complete(proto.completed) self.loop.run_until_complete(proto.completed)
self.assertEqual(-signal.SIGHUP, proto.returncode) self.assertEqual(-signal.SIGHUP, proto.returncode)
transp.close() transp.close()
finally:
signal.signal(signal.SIGHUP, old_handler)
def test_subprocess_stderr(self): def test_subprocess_stderr(self):
prog = os.path.join(os.path.dirname(__file__), 'echo2.py') prog = os.path.join(os.path.dirname(__file__), 'echo2.py')

39
Lib/test/test_asyncio/test_subprocess.py
View File

@ -166,25 +166,32 @@ class SubprocessMixin:
@unittest.skipIf(sys.platform == 'win32', "Don't have SIGHUP") @unittest.skipIf(sys.platform == 'win32', "Don't have SIGHUP")
def test_send_signal(self): def test_send_signal(self):
code = 'import time; print("sleeping", flush=True); time.sleep(3600)' # bpo-31034: Make sure that we get the default signal handler (killing
args = [sys.executable, '-c', code] # the process). The parent process may have decided to ignore SIGHUP,
create = asyncio.create_subprocess_exec(*args, # and signal handlers are inherited.
stdout=subprocess.PIPE, old_handler = signal.signal(signal.SIGHUP, signal.SIG_DFL)
loop=self.loop) try:
proc = self.loop.run_until_complete(create) code = 'import time; print("sleeping", flush=True); time.sleep(3600)'
args = [sys.executable, '-c', code]
create = asyncio.create_subprocess_exec(*args,
stdout=subprocess.PIPE,
loop=self.loop)
proc = self.loop.run_until_complete(create)
@asyncio.coroutine @asyncio.coroutine
def send_signal(proc): def send_signal(proc):
# basic synchronization to wait until the program is sleeping # basic synchronization to wait until the program is sleeping
line = yield from proc.stdout.readline() line = yield from proc.stdout.readline()
self.assertEqual(line, b'sleeping\n') self.assertEqual(line, b'sleeping\n')
proc.send_signal(signal.SIGHUP) proc.send_signal(signal.SIGHUP)
returncode = (yield from proc.wait()) returncode = (yield from proc.wait())
return returncode return returncode
returncode = self.loop.run_until_complete(send_signal(proc)) returncode = self.loop.run_until_complete(send_signal(proc))
self.assertEqual(-signal.SIGHUP, returncode) self.assertEqual(-signal.SIGHUP, returncode)
finally:
signal.signal(signal.SIGHUP, old_handler)
def prepare_broken_pipe_test(self): def prepare_broken_pipe_test(self):
# buffer large enough to feed the whole pipe buffer # buffer large enough to feed the whole pipe buffer