deal gracefully with invalid file descriptors - don't capture the particular stream

--HG--
branch : trunk
This commit is contained in:
holger krekel 2010-05-18 16:52:56 +02:00
parent 4f5d7948f7
commit da097c9d67
2 changed files with 38 additions and 6 deletions

View File

@ -175,7 +175,8 @@ class Capture(object):
class StdCaptureFD(Capture):
""" This class allows to capture writes to FD1 and FD2
and may connect a NULL file to FD0 (and prevent
reads from sys.stdin)
reads from sys.stdin). If any of the 0,1,2 file descriptors
is invalid it will not be captured.
"""
def __init__(self, out=True, err=True, mixed=False,
in_=True, patchsys=True, now=True):
@ -192,13 +193,19 @@ class StdCaptureFD(Capture):
mixed = self._options['mixed']
self.in_ = in_
if in_:
try:
self._oldin = (sys.stdin, os.dup(0))
except OSError:
pass
if out:
tmpfile = None
if hasattr(out, 'write'):
tmpfile = out
self.out = py.io.FDCapture(1, tmpfile=tmpfile, now=False)
try:
self.out = FDCapture(1, tmpfile=tmpfile, now=False)
self._options['out'] = self.out.tmpfile
except OSError:
pass
if err:
if out and mixed:
tmpfile = self.out.tmpfile
@ -206,8 +213,11 @@ class StdCaptureFD(Capture):
tmpfile = err
else:
tmpfile = None
self.err = py.io.FDCapture(2, tmpfile=tmpfile, now=False)
try:
self.err = FDCapture(2, tmpfile=tmpfile, now=False)
self._options['err'] = self.err.tmpfile
except OSError:
pass
def startall(self):
if self.in_:

View File

@ -360,6 +360,28 @@ def test_stdcapture_fd_tmpfile(tmpfile):
outf, errf = capfd.done()
assert outf == tmpfile
class TestStdCaptureFDinvalidFD:
pytestmark = needsdup
def test_stdcapture_fd_invalid_fd(self, testdir):
testdir.makepyfile("""
import py, os
def test_stdout():
os.close(1)
cap = py.io.StdCaptureFD(out=True, err=False, in_=False)
cap.done()
def test_stderr():
os.close(2)
cap = py.io.StdCaptureFD(out=False, err=True, in_=False)
cap.done()
def test_stdin():
os.close(0)
cap = py.io.StdCaptureFD(out=False, err=False, in_=True)
cap.done()
""")
result = testdir.runpytest("--capture=fd")
assert result.ret == 0
assert result.parseoutcomes()['passed'] == 3
def test_capture_not_started_but_reset():
capsys = py.io.StdCapture(now=False)
capsys.done()