traverseda
/
cpython
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merged revisions 86185 via svnmerge from 

svn+ssh://pythondev@svn.python.org/python/branches/py3k

........
  r86185 | brian.curtin | 2010-11-04 22:58:52 -0500 (Thu, 04 Nov 2010) | 2 lines

  Add cleanups to stdout/stderr pipes to remove ResourceWarnings.
........
This commit is contained in:
Brian Curtin 2010-11-05 04:09:09 +00:00
parent ab016d21a6
commit d117b562ed
1 changed files with 16 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
Lib/test/test_subprocess.py
View File

@ -117,6 +117,8 @@ class ProcessTestCase(BaseTestCase):
# .stdin is None when not redirected
p = subprocess.Popen([sys.executable, "-c", 'print "banana"'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
self.addCleanup(p.stdout.close)
self.addCleanup(p.stderr.close)
p.wait()
self.assertEqual(p.stdin, None)
@ -127,6 +129,8 @@ class ProcessTestCase(BaseTestCase):
'test of stdout in a different '
'process ..."'],
stdin=subprocess.PIPE, stderr=subprocess.PIPE)
self.addCleanup(p.stdin.close)
self.addCleanup(p.stderr.close)
p.wait()
self.assertEqual(p.stdout, None)
@ -134,6 +138,8 @@ class ProcessTestCase(BaseTestCase):
# .stderr is None when not redirected
p = subprocess.Popen([sys.executable, "-c", 'print "banana"'],
stdin=subprocess.PIPE, stdout=subprocess.PIPE)
self.addCleanup(p.stdout.close)
self.addCleanup(p.stdin.close)
p.wait()
self.assertEqual(p.stderr, None)
@ -194,6 +200,7 @@ class ProcessTestCase(BaseTestCase):
p = subprocess.Popen([sys.executable, "-c",
'import sys; sys.stdout.write("orange")'],
stdout=subprocess.PIPE)
self.addCleanup(p.stdout.close)
self.assertEqual(p.stdout.read(), "orange")
def test_stdout_filedes(self):
@ -222,6 +229,7 @@ class ProcessTestCase(BaseTestCase):
p = subprocess.Popen([sys.executable, "-c",
'import sys; sys.stderr.write("strawberry")'],
stderr=subprocess.PIPE)
self.addCleanup(p.stderr.close)
self.assertStderrEqual(p.stderr.read(), "strawberry")
def test_stderr_filedes(self):
@ -254,6 +262,7 @@ class ProcessTestCase(BaseTestCase):
'sys.stderr.write("orange")'],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
self.addCleanup(p.stdout.close)
self.assertStderrEqual(p.stdout.read(), "appleorange")
def test_stdout_stderr_file(self):
@ -289,6 +298,7 @@ class ProcessTestCase(BaseTestCase):
'sys.stdout.write(os.getcwd())'],
stdout=subprocess.PIPE,
cwd=tmpdir)
self.addCleanup(p.stdout.close)
normcase = os.path.normcase
self.assertEqual(normcase(p.stdout.read()), normcase(tmpdir))
@ -300,6 +310,7 @@ class ProcessTestCase(BaseTestCase):
'sys.stdout.write(os.getenv("FRUIT"))'],
stdout=subprocess.PIPE,
env=newenv)
self.addCleanup(p.stdout.close)
self.assertEqual(p.stdout.read(), "orange")
def test_communicate_stdin(self):
@ -415,6 +426,7 @@ class ProcessTestCase(BaseTestCase):
'sys.stdout.write("\\nline6");'],
stdout=subprocess.PIPE,
universal_newlines=1)
self.addCleanup(p.stdout.close)
stdout = p.stdout.read()
if hasattr(file, 'newlines'):
# Interpreter with universal newline support
@ -639,6 +651,7 @@ class POSIXProcessTestCase(BaseTestCase):
"sys.stdout.write(os.getenv('FRUIT'))"],
stdout=subprocess.PIPE,
preexec_fn=lambda: os.putenv("FRUIT", "apple"))
self.addCleanup(p.stdout.close)
self.assertEqual(p.stdout.read(), "apple")
def test_args_string(self):
@ -672,6 +685,7 @@ class POSIXProcessTestCase(BaseTestCase):
p = subprocess.Popen(["echo $FRUIT"], shell=1,
stdout=subprocess.PIPE,
env=newenv)
self.addCleanup(p.stdout.close)
self.assertEqual(p.stdout.read().strip(), "apple")
def test_shell_string(self):
@ -681,6 +695,7 @@ class POSIXProcessTestCase(BaseTestCase):
p = subprocess.Popen("echo $FRUIT", shell=1,
stdout=subprocess.PIPE,
env=newenv)
self.addCleanup(p.stdout.close)
self.assertEqual(p.stdout.read().strip(), "apple")
def test_call_string(self):
@ -712,6 +727,7 @@ class POSIXProcessTestCase(BaseTestCase):
for sh in shells:
p = subprocess.Popen("echo $0", executable=sh, shell=True,
stdout=subprocess.PIPE)
self.addCleanup(p.stdout.close)
self.assertEqual(p.stdout.read().strip(), sh)
def _kill_process(self, method, *args):