def send_signal(self, sig):
"""Send a signal to the process
"""
if sig == signal.SIGTERM:
self.terminate()
elif sig == signal.CTRL_C_EVENT:
os.kill(self.pid, signal.CTRL_C_EVENT)
elif sig == signal.CTRL_BREAK_EVENT:
os.kill(self.pid, signal.CTRL_BREAK_EVENT)
else:
raise ValueError("Unsupported signal: {}".format(sig))
python类CTRL_BREAK_EVENT的实例源码
def send_signal(self, sig):
"""Send a signal to the process
"""
if sig == signal.SIGTERM:
self.terminate()
elif sig == signal.CTRL_C_EVENT:
os.kill(self.pid, signal.CTRL_C_EVENT)
elif sig == signal.CTRL_BREAK_EVENT:
os.kill(self.pid, signal.CTRL_BREAK_EVENT)
else:
raise ValueError("Unsupported signal: {}".format(sig))
def test_signal_waiting(self):
''' Fixture thine self.
'''
proc1 = ProcFixture(signal.SIGINT)
proc2 = ProcFixture(signal.SIGTERM)
proc3 = ProcFixture(signal.SIGABRT)
proc4 = ProcFixture(signal.CTRL_C_EVENT)
proc5 = ProcFixture(signal.CTRL_BREAK_EVENT)
self.assertEqual(_await_signal(proc1), signal.SIGINT)
self.assertEqual(_await_signal(proc2), signal.SIGTERM)
self.assertEqual(_await_signal(proc3), signal.SIGABRT)
self.assertEqual(_await_signal(proc4), signal.SIGINT)
self.assertEqual(_await_signal(proc5), signal.SIGINT)
def _await_signal(process):
''' Waits for the process to die, and then returns the exit code for
the process, converting CTRL_C_EVENT and CTRL_BREAK_EVENT into
SIGINT.
'''
# Note that this is implemented with a busy wait
process.wait()
code = process.returncode
if code == signal.CTRL_C_EVENT:
code = signal.SIGINT
elif code == signal.CTRL_BREAK_EVENT:
code = signal.SIGINT
return code
def kill(self, sig):
'''Sends a Unix signal to the subprocess.
Use constants from the :mod:`signal` module to specify which signal.
'''
if sys.platform == 'win32':
if sig in [signal.SIGINT, signal.CTRL_C_EVENT]:
sig = signal.CTRL_C_EVENT
elif sig in [signal.SIGBREAK, signal.CTRL_BREAK_EVENT]:
sig = signal.CTRL_BREAK_EVENT
else:
sig = signal.SIGTERM
os.kill(self.proc.pid, sig)
def _ctrl_handler(sig):
"""Handle a sig event and return 0 to terminate the process"""
if sig == signal.CTRL_C_EVENT:
pass
elif sig == signal.CTRL_BREAK_EVENT:
pass
else:
print("UNKNOWN EVENT")
return 0
def test_CTRL_BREAK_EVENT(self):
self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT")
def kill(self, sig):
'''Sends a Unix signal to the subprocess.
Use constants from the :mod:`signal` module to specify which signal.
'''
if sys.platform == 'win32':
if sig in [signal.SIGINT, signal.CTRL_C_EVENT]:
sig = signal.CTRL_C_EVENT
elif sig in [signal.SIGBREAK, signal.CTRL_BREAK_EVENT]:
sig = signal.CTRL_BREAK_EVENT
else:
sig = signal.SIGTERM
os.kill(self.proc.pid, sig)
def send_signal(self, sig):
"""Send a signal to the process."""
# Don't signal a process that we know has already died.
if self.returncode is not None:
return
if sig == signal.SIGTERM:
self.terminate()
elif sig == signal.CTRL_C_EVENT:
os.kill(self.pid, signal.CTRL_C_EVENT)
elif sig == signal.CTRL_BREAK_EVENT:
os.kill(self.pid, signal.CTRL_BREAK_EVENT)
else:
raise ValueError("Unsupported signal: {}".format(sig))
def _ctrl_handler(sig):
"""Handle a sig event and return 0 to terminate the process"""
if sig == signal.CTRL_C_EVENT:
pass
elif sig == signal.CTRL_BREAK_EVENT:
pass
else:
print("UNKNOWN EVENT")
return 0
def test_CTRL_BREAK_EVENT(self):
self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT")
def send_signal(self, sig):
"""Send a signal to the process
"""
if sig == signal.SIGTERM:
self.terminate()
elif sig == signal.CTRL_C_EVENT:
os.kill(self.pid, signal.CTRL_C_EVENT)
elif sig == signal.CTRL_BREAK_EVENT:
os.kill(self.pid, signal.CTRL_BREAK_EVENT)
else:
raise ValueError("Unsupported signal: {}".format(sig))
def _ctrl_handler(sig):
"""Handle a sig event and return 0 to terminate the process"""
if sig == signal.CTRL_C_EVENT:
pass
elif sig == signal.CTRL_BREAK_EVENT:
pass
else:
print("UNKNOWN EVENT")
return 0
def test_CTRL_BREAK_EVENT(self):
self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT")
def interrupt_process(process):
print("Due to limitations of the operating system, we can only kill the process.", file=sys.stderr)
process.send_signal(signal.CTRL_BREAK_EVENT)
def send_signal(self, sig):
"""Send a signal to the process
"""
if sig == signal.SIGTERM:
self.terminate()
elif sig == signal.CTRL_C_EVENT:
os.kill(self.pid, signal.CTRL_C_EVENT)
elif sig == signal.CTRL_BREAK_EVENT:
os.kill(self.pid, signal.CTRL_BREAK_EVENT)
else:
raise ValueError("Unsupported signal: {}".format(sig))
def send_signal(self, sig):
"""Send a signal to the process
"""
if sig == signal.SIGTERM:
self.terminate()
elif sig == signal.CTRL_C_EVENT:
os.kill(self.pid, signal.CTRL_C_EVENT)
elif sig == signal.CTRL_BREAK_EVENT:
os.kill(self.pid, signal.CTRL_BREAK_EVENT)
else:
raise ValueError("Unsupported signal: {}".format(sig))
def kill(self):
if sys.platform.startswith("win"):
# proc.terminate on Windows does not initiate a graceful shutdown
# through the processes signal handlers it just kills it hard. So
# this sends a SIGBREAK. You cannot sends a SIGINT (CTRL_C_EVENT)
# to a process group in Windows, otherwise Ctrl+C would be
# sent.
self.proc.send_signal(signal.CTRL_BREAK_EVENT)
else:
self.proc.terminate()
def test_ctrl_signals(self):
p = psutil.Process(get_test_subprocess().pid)
p.send_signal(signal.CTRL_C_EVENT)
p.send_signal(signal.CTRL_BREAK_EVENT)
p.kill()
p.wait()
self.assertRaises(psutil.NoSuchProcess,
p.send_signal, signal.CTRL_C_EVENT)
self.assertRaises(psutil.NoSuchProcess,
p.send_signal, signal.CTRL_BREAK_EVENT)
def send_signal(self, sig):
"""Send a signal to the process
"""
if sig == signal.SIGTERM:
self.terminate()
elif sig == signal.CTRL_C_EVENT:
os.kill(self.pid, signal.CTRL_C_EVENT)
elif sig == signal.CTRL_BREAK_EVENT:
os.kill(self.pid, signal.CTRL_BREAK_EVENT)
else:
raise ValueError("Unsupported signal: {}".format(sig))
