mirror of
https://github.com/python/cpython.git
synced 2025-09-01 06:28:36 +00:00
Merged revisions 78736,78759,78761,78767,78788-78789 via svnmerge from
svn+ssh://pythondev@svn.python.org/python/trunk ........ r78736 | florent.xicluna | 2010-03-06 20:43:41 +0100 (sam, 06 mar 2010) | 2 lines Skip test_send_signal, test_kill, test_terminate on win32 platforms, for 2.7a4 release. ........ r78759 | florent.xicluna | 2010-03-07 13:21:36 +0100 (dim, 07 mar 2010) | 2 lines #2777: Enable test_send_signal, test_terminate and test_kill on win32 platforms. ........ r78761 | florent.xicluna | 2010-03-07 16:27:39 +0100 (dim, 07 mar 2010) | 4 lines Do not fail if returncode is 0 on send_signal/kill/terminate, for win32 platforms. Do not hide the KeyboardInterrupt on POSIX platforms. ........ r78767 | florent.xicluna | 2010-03-07 18:12:23 +0100 (dim, 07 mar 2010) | 2 lines #2777: Try hard to make Win7 buildbot happy... ........ r78788 | florent.xicluna | 2010-03-08 11:58:12 +0100 (lun, 08 mar 2010) | 2 lines Fix syntax: "rc != None" -> "rc is not None" ........ r78789 | florent.xicluna | 2010-03-08 11:59:33 +0100 (lun, 08 mar 2010) | 2 lines Replace the stderr logging with assertNotEqual(returncode, 0). ........
This commit is contained in:
parent
81c867c3fc
commit
4886d246a8
3 changed files with 46 additions and 39 deletions
|
@ -534,7 +534,7 @@ Return code handling translates as follows::
|
||||||
pipe = os.popen(cmd, 'w')
|
pipe = os.popen(cmd, 'w')
|
||||||
...
|
...
|
||||||
rc = pipe.close()
|
rc = pipe.close()
|
||||||
if rc != None and rc % 256:
|
if rc is not None and rc % 256:
|
||||||
print("There were some errors")
|
print("There were some errors")
|
||||||
==>
|
==>
|
||||||
process = Popen(cmd, 'w', stdin=PIPE)
|
process = Popen(cmd, 'w', stdin=PIPE)
|
||||||
|
|
|
@ -110,7 +110,7 @@ call(*popenargs, **kwargs):
|
||||||
|
|
||||||
The arguments are the same as for the Popen constructor. Example:
|
The arguments are the same as for the Popen constructor. Example:
|
||||||
|
|
||||||
>>> retcode = call(["ls", "-l"])
|
>>> retcode = subprocess.call(["ls", "-l"])
|
||||||
|
|
||||||
check_call(*popenargs, **kwargs):
|
check_call(*popenargs, **kwargs):
|
||||||
Run command with arguments. Wait for command to complete. If the
|
Run command with arguments. Wait for command to complete. If the
|
||||||
|
@ -120,7 +120,7 @@ check_call(*popenargs, **kwargs):
|
||||||
|
|
||||||
The arguments are the same as for the Popen constructor. Example:
|
The arguments are the same as for the Popen constructor. Example:
|
||||||
|
|
||||||
>>> check_call(["ls", "-l"])
|
>>> subprocess.check_call(["ls", "-l"])
|
||||||
0
|
0
|
||||||
|
|
||||||
getstatusoutput(cmd):
|
getstatusoutput(cmd):
|
||||||
|
|
|
@ -648,42 +648,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("p.send_signal(SIGINT) succeeded "
|
print("p.{}{} succeeded after "
|
||||||
"after {} attempts".format(count), file=sys.stderr)
|
"{} attempts".format(method, args, count), file=sys.stderr)
|
||||||
|
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)
|
||||||
|
|
||||||
|
|
||||||
|
@ -766,28 +763,38 @@ 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):
|
||||||
# Do not inherit file handles from the parent.
|
# Some win32 buildbot raises EOFError if stdin is inherited
|
||||||
# It should fix failure on some platforms.
|
p = subprocess.Popen([sys.executable, "-c", "input()"],
|
||||||
p = subprocess.Popen([sys.executable, "-c", "input()"], close_fds=True)
|
stdin=subprocess.PIPE)
|
||||||
|
|
||||||
self.assertIs(p.poll(), None)
|
# Let the process initialize (Issue #3137)
|
||||||
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("p.{}{} succeeded after "
|
||||||
|
"{} attempts".format(method, args, count), file=sys.stderr)
|
||||||
|
self.assertEqual(p.wait(), returncode)
|
||||||
|
self.assertNotEqual(returncode, 0)
|
||||||
|
|
||||||
|
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()"], close_fds=True)
|
self._kill_process('kill')
|
||||||
|
|
||||||
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()"], close_fds=True)
|
self._kill_process('terminate')
|
||||||
|
|
||||||
self.assertIs(p.poll(), None)
|
|
||||||
p.terminate()
|
|
||||||
self.assertNotEqual(p.wait(), 0)
|
|
||||||
|
|
||||||
|
|
||||||
# The module says:
|
# The module says:
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue