diff options
author | Florent Xicluna <florent.xicluna@gmail.com> | 2010-03-07 15:27:39 (GMT) |
---|---|---|
committer | Florent Xicluna <florent.xicluna@gmail.com> | 2010-03-07 15:27:39 (GMT) |
commit | c083864fc8a03a045dfc0200e8456996395ad4dc (patch) | |
tree | 3c238970bdfeed3782e0386d914320633903010a /Lib | |
parent | 21e99f492419f97b4e5818081440742ecfde269f (diff) | |
download | cpython-c083864fc8a03a045dfc0200e8456996395ad4dc.zip cpython-c083864fc8a03a045dfc0200e8456996395ad4dc.tar.gz cpython-c083864fc8a03a045dfc0200e8456996395ad4dc.tar.bz2 |
Do not fail if returncode is 0 on send_signal/kill/terminate, for win32 platforms.
Do not hide the KeyboardInterrupt on POSIX platforms.
Diffstat (limited to 'Lib')
-rw-r--r-- | Lib/test/test_subprocess.py | 83 |
1 files changed, 44 insertions, 39 deletions
diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py index 5796fc0..6a23414 100644 --- a/Lib/test/test_subprocess.py +++ b/Lib/test/test_subprocess.py @@ -650,42 +650,39 @@ class POSIXProcessTestCase(unittest.TestCase): os.remove(fname) self.assertEqual(rc, 47) - def test_send_signal(self): + def _kill_process(self, method, *args): # Do not inherit file handles from the parent. # It should fix failures on some platforms. p = subprocess.Popen([sys.executable, "-c", "input()"], close_fds=True, - stdin=subprocess.PIPE, stderr=subprocess.PIPE) + stdin=subprocess.PIPE) - # Let the process initialize correctly (Issue #3137) + # Let the process initialize (Issue #3137) time.sleep(0.1) + # The process should not terminate prematurely self.assertIsNone(p.poll()) + # Retry if the process do not receive the signal. count, maxcount = 0, 3 - # Retry if the process do not receive the SIGINT signal. while count < maxcount and p.poll() is None: - p.send_signal(signal.SIGINT) + getattr(p, method)(*args) time.sleep(0.1) count += 1 - self.assertIsNotNone(p.poll(), "the subprocess did not receive " - "the signal SIGINT") + + self.assertIsNotNone(p.poll(), "the subprocess did not terminate") if count > 1: - print >>sys.stderr, ("p.send_signal(SIGINT) succeeded " - "after {} attempts".format(count)) + print >>sys.stderr, ("p.{}{} succeeded after " + "{} attempts".format(method, args, count)) + return p + + def test_send_signal(self): + p = self._kill_process('send_signal', signal.SIGINT) self.assertNotEqual(p.wait(), 0) def test_kill(self): - p = subprocess.Popen([sys.executable, "-c", "input()"], - stdin=subprocess.PIPE, close_fds=True) - - self.assertIsNone(p.poll()) - p.kill() + p = self._kill_process('kill') self.assertEqual(p.wait(), -signal.SIGKILL) def test_terminate(self): - p = subprocess.Popen([sys.executable, "-c", "input()"], - stdin=subprocess.PIPE, close_fds=True) - - self.assertIsNone(p.poll()) - p.terminate() + p = self._kill_process('terminate') self.assertEqual(p.wait(), -signal.SIGTERM) @@ -768,31 +765,39 @@ class Win32ProcessTestCase(unittest.TestCase): ' -c "import sys; sys.exit(47)"') self.assertEqual(rc, 47) - def test_send_signal(self): - # Redirect the stdin file handle. - # It should fix failure on some win32 platforms. - p = subprocess.Popen([sys.executable, "-c", "input()"], - stdin=subprocess.PIPE) + def _kill_process(self, method, *args): + # Do not inherit file handles from the parent. + p = subprocess.Popen([sys.executable, "-c", "input()"], close_fds=True) - self.assertIs(p.poll(), None) - p.send_signal(signal.SIGTERM) - self.assertNotEqual(p.wait(), 0) + # Let the process initialize + time.sleep(0.1) + # The process should not terminate prematurely + self.assertIsNone(p.poll()) + # Retry if the process do not receive the signal. + count, maxcount = 0, 3 + while count < maxcount and p.poll() is None: + getattr(p, method)(*args) + time.sleep(0.1) + count += 1 - def test_kill(self): - p = subprocess.Popen([sys.executable, "-c", "input()"], - stdin=subprocess.PIPE) + returncode = p.poll() + self.assertIsNotNone(returncode, "the subprocess did not terminate") + if count > 1: + print >>sys.stderr, ("p.{}{} succeeded after " + "{} attempts".format(method, args, count)) + if returncode == 0: + # On some win32 platforms, it returns 0. See #2777. + print >>sys.stderr, "p.{}{} returned 0".format(method, args) + self.assertEqual(p.wait(), returncode) - self.assertIs(p.poll(), None) - p.kill() - self.assertNotEqual(p.wait(), 0) + def test_send_signal(self): + self._kill_process('send_signal', signal.SIGTERM) - def test_terminate(self): - p = subprocess.Popen([sys.executable, "-c", "input()"], - stdin=subprocess.PIPE) + def test_kill(self): + self._kill_process('kill') - self.assertIs(p.poll(), None) - p.terminate() - self.assertNotEqual(p.wait(), 0) + def test_terminate(self): + self._kill_process('terminate') @unittest.skipUnless(getattr(subprocess, '_has_poll', False), |