Added kill, terminate and send_signal to subprocess.Popen
The bits and pieces for the Windows side were already in place. The POSIX side is trivial (as usual) and uses os.kill().
diff --git a/Lib/subprocess.py b/Lib/subprocess.py
index 975a924..dc639c9 100644
--- a/Lib/subprocess.py
+++ b/Lib/subprocess.py
@@ -357,6 +357,7 @@
import types
import traceback
import gc
+import signal
# Exception classes used by this module.
class CalledProcessError(Exception):
@@ -384,6 +385,7 @@
from win32process import CreateProcess, STARTUPINFO, \
GetExitCodeProcess, STARTF_USESTDHANDLES, \
STARTF_USESHOWWINDOW, CREATE_NEW_CONSOLE
+ from win32process import TerminateProcess
from win32event import WaitForSingleObject, INFINITE, WAIT_OBJECT_0
else:
from _subprocess import *
@@ -906,6 +908,21 @@
self.wait()
return (stdout, stderr)
+ def send_signal(self, sig):
+ """Send a signal to the process
+ """
+ if sig == signal.SIGTERM:
+ self.terminate()
+ else:
+ raise ValueError("Only SIGTERM is supported on Windows")
+
+ def terminate(self):
+ """Terminates the process
+ """
+ TerminateProcess(self._handle, 1)
+
+ kill = terminate
+
else:
#
# POSIX methods
@@ -1184,6 +1201,21 @@
self.wait()
return (stdout, stderr)
+ def send_signal(self, sig):
+ """Send a signal to the process
+ """
+ os.kill(self.pid, sig)
+
+ def terminate(self):
+ """Terminate the process with SIGTERM
+ """
+ self.send_signal(signal.SIGTERM)
+
+ def kill(self):
+ """Kill the process with SIGKILL
+ """
+ self.send_signal(signal.SIGKILL)
+
def _demo_posix():
#
diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py
index 9886f3d..ccefb67 100644
--- a/Lib/test/test_subprocess.py
+++ b/Lib/test/test_subprocess.py
@@ -584,6 +584,29 @@
os.remove(fname)
self.assertEqual(rc, 47)
+ def test_send_signal(self):
+ p = subprocess.Popen([sys.executable,
+ "-c", "input()"])
+
+ self.assert_(p.poll() is None, p.poll())
+ p.send_signal(signal.SIGINT)
+ self.assertNotEqual(p.wait(), 0)
+
+ def test_kill(self):
+ p = subprocess.Popen([sys.executable,
+ "-c", "input()"])
+
+ self.assert_(p.poll() is None, p.poll())
+ p.kill()
+ self.assertEqual(p.wait(), -signal.SIGKILL)
+
+ def test_terminate(self):
+ p = subprocess.Popen([sys.executable,
+ "-c", "input()"])
+
+ self.assert_(p.poll() is None, p.poll())
+ p.terminate()
+ self.assertEqual(p.wait(), -signal.SIGTERM)
#
# Windows tests
@@ -655,6 +678,29 @@
' -c "import sys; sys.exit(47)"')
self.assertEqual(rc, 47)
+ def test_send_signal(self):
+ p = subprocess.Popen([sys.executable,
+ "-c", "input()"])
+
+ self.assert_(p.poll() is None, p.poll())
+ p.send_signal(signal.SIGTERM)
+ self.assertNotEqual(p.wait(), 0)
+
+ def test_kill(self):
+ p = subprocess.Popen([sys.executable,
+ "-c", "input()"])
+
+ self.assert_(p.poll() is None, p.poll())
+ p.kill()
+ self.assertNotEqual(p.wait(), 0)
+
+ def test_terminate(self):
+ p = subprocess.Popen([sys.executable,
+ "-c", "input()"])
+
+ self.assert_(p.poll() is None, p.poll())
+ p.terminate()
+ self.assertNotEqual(p.wait(), 0)
def test_main():
test_support.run_unittest(ProcessTestCase)