@unittest.skipUnless(fcntl and hasattr(fcntl, 'F_GETPIPE_SZ'),
'fcntl.F_GETPIPE_SZ required for test.')
def test_pipesize_default(self):
- p = subprocess.Popen(
+ proc = subprocess.Popen(
[sys.executable, "-c",
'import sys; sys.stdin.read(); sys.stdout.write("out"); '
'sys.stderr.write("error!")'],
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, pipesize=-1)
- try:
- fp_r, fp_w = os.pipe()
+
+ with proc:
try:
- default_pipesize = fcntl.fcntl(fp_w, fcntl.F_GETPIPE_SZ)
- for fifo in [p.stdin, p.stdout, p.stderr]:
- self.assertEqual(
- fcntl.fcntl(fifo.fileno(), fcntl.F_GETPIPE_SZ),
- default_pipesize)
+ fp_r, fp_w = os.pipe()
+ try:
+ default_read_pipesize = fcntl.fcntl(fp_r, fcntl.F_GETPIPE_SZ)
+ default_write_pipesize = fcntl.fcntl(fp_w, fcntl.F_GETPIPE_SZ)
+ finally:
+ os.close(fp_r)
+ os.close(fp_w)
+
+ self.assertEqual(
+ fcntl.fcntl(proc.stdin.fileno(), fcntl.F_GETPIPE_SZ),
+ default_read_pipesize)
+ self.assertEqual(
+ fcntl.fcntl(proc.stdout.fileno(), fcntl.F_GETPIPE_SZ),
+ default_write_pipesize)
+ self.assertEqual(
+ fcntl.fcntl(proc.stderr.fileno(), fcntl.F_GETPIPE_SZ),
+ default_write_pipesize)
+ # On other platforms we cannot test the pipe size (yet). But above
+ # code using pipesize=-1 should not crash.
finally:
- os.close(fp_r)
- os.close(fp_w)
- # On other platforms we cannot test the pipe size (yet). But above
- # code using pipesize=-1 should not crash.
- p.stdin.close()
- p.stdout.close()
- p.stderr.close()
- finally:
- p.kill()
- p.wait()
+ proc.kill()
def test_env(self):
newenv = os.environ.copy()