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().

2 files changed, 78 insertions, 0 deletions

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 os
 import types
 import traceback
 import gc
+import signal
 
 # Exception classes used by this module.
 class CalledProcessError(Exception):
@@ -384,6 +385,7 @@ if mswindows:
         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 @@ class Popen(object):
             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 @@ class Popen(object):
             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 @@ class ProcessTestCase(unittest.TestCase):
             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 @@ class ProcessTestCase(unittest.TestCase):
                                  ' -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)