Do not fail if returncode is 0 on send_signal/kill/terminate, for win32 platforms.

Do not hide the KeyboardInterrupt on POSIX platforms.
This commit is contained in:
Florent Xicluna 2010-03-07 15:27:39 +00:00
parent 21e99f4924
commit c083864fc8
1 changed files with 44 additions and 39 deletions

View File

@ -650,42 +650,39 @@ class POSIXProcessTestCase(unittest.TestCase):
os.remove(fname) os.remove(fname)
self.assertEqual(rc, 47) self.assertEqual(rc, 47)
def test_send_signal(self): def _kill_process(self, method, *args):
# Do not inherit file handles from the parent. # Do not inherit file handles from the parent.
# It should fix failures on some platforms. # It should fix failures on some platforms.
p = subprocess.Popen([sys.executable, "-c", "input()"], close_fds=True, 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) time.sleep(0.1)
# The process should not terminate prematurely
self.assertIsNone(p.poll()) self.assertIsNone(p.poll())
# Retry if the process do not receive the signal.
count, maxcount = 0, 3 count, maxcount = 0, 3
# Retry if the process do not receive the SIGINT signal.
while count < maxcount and p.poll() is None: while count < maxcount and p.poll() is None:
p.send_signal(signal.SIGINT) getattr(p, method)(*args)
time.sleep(0.1) time.sleep(0.1)
count += 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: if count > 1:
print >>sys.stderr, ("p.send_signal(SIGINT) succeeded " print >>sys.stderr, ("p.{}{} succeeded after "
"after {} attempts".format(count)) "{} 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) self.assertNotEqual(p.wait(), 0)
def test_kill(self): def test_kill(self):
p = subprocess.Popen([sys.executable, "-c", "input()"], p = self._kill_process('kill')
stdin=subprocess.PIPE, close_fds=True)
self.assertIsNone(p.poll())
p.kill()
self.assertEqual(p.wait(), -signal.SIGKILL) self.assertEqual(p.wait(), -signal.SIGKILL)
def test_terminate(self): def test_terminate(self):
p = subprocess.Popen([sys.executable, "-c", "input()"], p = self._kill_process('terminate')
stdin=subprocess.PIPE, close_fds=True)
self.assertIsNone(p.poll())
p.terminate()
self.assertEqual(p.wait(), -signal.SIGTERM) self.assertEqual(p.wait(), -signal.SIGTERM)
@ -768,31 +765,39 @@ class Win32ProcessTestCase(unittest.TestCase):
' -c "import sys; sys.exit(47)"') ' -c "import sys; sys.exit(47)"')
self.assertEqual(rc, 47) self.assertEqual(rc, 47)
def test_send_signal(self): def _kill_process(self, method, *args):
# Redirect the stdin file handle. # Do not inherit file handles from the parent.
# It should fix failure on some win32 platforms. p = subprocess.Popen([sys.executable, "-c", "input()"], close_fds=True)
p = subprocess.Popen([sys.executable, "-c", "input()"],
stdin=subprocess.PIPE)
self.assertIs(p.poll(), None) # Let the process initialize
p.send_signal(signal.SIGTERM) time.sleep(0.1)
self.assertNotEqual(p.wait(), 0) # 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
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)
def test_send_signal(self):
self._kill_process('send_signal', signal.SIGTERM)
def test_kill(self): def test_kill(self):
p = subprocess.Popen([sys.executable, "-c", "input()"], self._kill_process('kill')
stdin=subprocess.PIPE)
self.assertIs(p.poll(), None)
p.kill()
self.assertNotEqual(p.wait(), 0)
def test_terminate(self): def test_terminate(self):
p = subprocess.Popen([sys.executable, "-c", "input()"], self._kill_process('terminate')
stdin=subprocess.PIPE)
self.assertIs(p.poll(), None)
p.terminate()
self.assertNotEqual(p.wait(), 0)
@unittest.skipUnless(getattr(subprocess, '_has_poll', False), @unittest.skipUnless(getattr(subprocess, '_has_poll', False),