Mirror
/
cpython
mirror of https://github.com/python/cpython
4
1
Fork 0
Code Issues Releases Wiki Activity

Attempt to speed up some subprocess tests (and hopefully keep them reliable).

This commit is contained in:
Ross Lagerwall 2012-02-12 09:01:30 +02:00
parent bc35bebb45
commit ab66d2a6cb
1 changed files with 21 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
Lib/test/test_subprocess.py
View File

@ -686,26 +686,19 @@ class ProcessTestCase(BaseTestCase):
self.assertEqual(subprocess.list2cmdline(['ab', '']), self.assertEqual(subprocess.list2cmdline(['ab', '']),
'ab ""') 'ab ""')
def test_poll(self): def test_poll(self):
p = subprocess.Popen([sys.executable, p = subprocess.Popen([sys.executable, "-c",
"-c", "import time; time.sleep(1)"]) "import os",
count = 0 "os.read(1)"], stdin=subprocess.PIPE)
while p.poll() is None: self.addCleanup(p.stdin.close)
time.sleep(0.1) self.assertIsNone(p.poll())
count += 1 os.write(p.stdin.fileno(), b'A')
# We expect that the poll loop probably went around about 10 times, p.wait()
# but, based on system scheduling we can't control, it's possible
# poll() never returned None. It "should be" very rare that it
# didn't go around at least twice.
self.assertGreaterEqual(count, 2)
# Subsequent invocations should just return the returncode # Subsequent invocations should just return the returncode
self.assertEqual(p.poll(), 0) self.assertEqual(p.poll(), 0)
def test_wait(self): def test_wait(self):
p = subprocess.Popen([sys.executable, p = subprocess.Popen([sys.executable, "-c", "pass"])
"-c", "import time; time.sleep(2)"])
self.assertEqual(p.wait(), 0) self.assertEqual(p.wait(), 0)
# Subsequent invocations should just return the returncode # Subsequent invocations should just return the returncode
self.assertEqual(p.wait(), 0) self.assertEqual(p.wait(), 0)
@ -797,25 +790,29 @@ class ProcessTestCase(BaseTestCase):
p = subprocess.Popen([sys.executable, "-c", 'pass'], p = subprocess.Popen([sys.executable, "-c", 'pass'],
stdin=subprocess.PIPE) stdin=subprocess.PIPE)
self.addCleanup(p.stdin.close) self.addCleanup(p.stdin.close)
time.sleep(2) p.wait()
p.communicate(b"x" * 2**20) p.communicate(b"x" * 2**20)
@unittest.skipUnless(hasattr(signal, 'SIGALRM'), @unittest.skipUnless(hasattr(signal, 'SIGUSR1'),
"Requires signal.SIGALRM") "Requires signal.SIGUSR1")
@unittest.skipUnless(hasattr(os, 'kill'),
"Requires os.kill")
@unittest.skipUnless(hasattr(os, 'getppid'),
"Requires os.getppid")
def test_communicate_eintr(self): def test_communicate_eintr(self):
# Issue #12493: communicate() should handle EINTR # Issue #12493: communicate() should handle EINTR
def handler(signum, frame): def handler(signum, frame):
pass pass
old_handler = signal.signal(signal.SIGALRM, handler) old_handler = signal.signal(signal.SIGUSR1, handler)
self.addCleanup(signal.signal, signal.SIGALRM, old_handler) self.addCleanup(signal.signal, signal.SIGUSR1, old_handler)
# the process is running for 2 seconds args = [sys.executable, "-c",
args = [sys.executable, "-c", 'import time; time.sleep(2)'] 'import os, signal;'
'os.kill(os.getppid(), signal.SIGUSR1)']
for stream in ('stdout', 'stderr'): for stream in ('stdout', 'stderr'):
kw = {stream: subprocess.PIPE} kw = {stream: subprocess.PIPE}
with subprocess.Popen(args, **kw) as process: with subprocess.Popen(args, **kw) as process:
signal.alarm(1) # communicate() will be interrupted by SIGUSR1
# communicate() will be interrupted by SIGALRM
process.communicate() process.communicate()