p53209761
/
test_ok2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

some special handling of stdin capturing, unification, un-xfail the win32 test 

--HG--
branch : trunk
This commit is contained in:
holger krekel 2010-05-18 12:12:34 -07:00
parent 10296faff1
commit cf255cd643
2 changed files with 19 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
py/_io/capture.py
View File

@ -38,7 +38,7 @@ class FDCapture:
in text mode. in text mode.
""" """
self.targetfd = targetfd self.targetfd = targetfd
if tmpfile is None: if tmpfile is None and targetfd != 0:
f = tempfile.TemporaryFile('wb+') f = tempfile.TemporaryFile('wb+')
tmpfile = dupfile(f, encoding="UTF-8") tmpfile = dupfile(f, encoding="UTF-8")
f.close() f.close()
@ -55,16 +55,24 @@ class FDCapture:
except OSError: except OSError:
raise ValueError("saved filedescriptor not valid, " raise ValueError("saved filedescriptor not valid, "
"did you call start() twice?") "did you call start() twice?")
os.dup2(self.tmpfile.fileno(), self.targetfd) if self.targetfd == 0 and not self.tmpfile:
if hasattr(self, '_oldsys'): fd = os.open(devnullpath, os.O_RDONLY)
setattr(sys, patchsysdict[self.targetfd], self.tmpfile) os.dup2(fd, 0)
if hasattr(self, '_oldsys'):
setattr(sys, patchsysdict[self.targetfd], DontReadFromInput())
else:
fd = self.tmpfile.fileno()
os.dup2(self.tmpfile.fileno(), self.targetfd)
if hasattr(self, '_oldsys'):
setattr(sys, patchsysdict[self.targetfd], self.tmpfile)
def done(self): def done(self):
""" unpatch and clean up, returns the self.tmpfile (file object) """ unpatch and clean up, returns the self.tmpfile (file object)
""" """
os.dup2(self._savefd, self.targetfd) os.dup2(self._savefd, self.targetfd)
os.close(self._savefd) os.close(self._savefd)
self.tmpfile.seek(0) if self.targetfd != 0:
self.tmpfile.seek(0)
if hasattr(self, '_oldsys'): if hasattr(self, '_oldsys'):
setattr(sys, patchsysdict[self.targetfd], self._oldsys) setattr(sys, patchsysdict[self.targetfd], self._oldsys)
return self.tmpfile return self.tmpfile
@ -184,15 +192,9 @@ class StdCaptureFD(Capture):
mixed = self._options['mixed'] mixed = self._options['mixed']
patchsys = self._options['patchsys'] patchsys = self._options['patchsys']
if in_: if in_:
if hasattr(in_, 'read'):
tmpfile = in_
else:
fd = os.open(devnullpath, os.O_RDONLY)
tmpfile = os.fdopen(fd)
try: try:
self.in_ = FDCapture(0, tmpfile=tmpfile, now=False, self.in_ = FDCapture(0, tmpfile=None, now=False,
patchsys=patchsys) patchsys=patchsys)
self._options['in_'] = self.in_.tmpfile
except OSError: except OSError:
pass pass
if out: if out:
@ -222,14 +224,10 @@ class StdCaptureFD(Capture):
def startall(self): def startall(self):
if hasattr(self, 'in_'): if hasattr(self, 'in_'):
self.in_.start() self.in_.start()
sys.stdin = DontReadFromInput() if hasattr(self, 'out'):
self.out.start()
out = getattr(self, 'out', None) if hasattr(self, 'err'):
if out: self.err.start()
out.start()
err = getattr(self, 'err', None)
if err:
err.start()
def resume(self): def resume(self):
""" resume capturing with original temp files. """ """ resume capturing with original temp files. """

12
testing/io_/test_capture.py
View File

@ -372,15 +372,6 @@ class TestStdCaptureFDinvalidFD:
os.close(2) os.close(2)
cap = py.io.StdCaptureFD(out=False, err=True, in_=False) cap = py.io.StdCaptureFD(out=False, err=True, in_=False)
cap.done() cap.done()
""")
result = testdir.runpytest("--capture=fd")
assert result.ret == 0
assert result.parseoutcomes()['passed'] == 2
@py.test.mark.xfail("sys.platform == 'win32'", run=False)
def test_stdcapture_fd_invalid_fd_null(self, testdir):
testdir.makepyfile("""
import py, os
def test_stdin(): def test_stdin():
os.close(0) os.close(0)
cap = py.io.StdCaptureFD(out=False, err=False, in_=True) cap = py.io.StdCaptureFD(out=False, err=False, in_=True)
@ -388,8 +379,7 @@ class TestStdCaptureFDinvalidFD:
""") """)
result = testdir.runpytest("--capture=fd") result = testdir.runpytest("--capture=fd")
assert result.ret == 0 assert result.ret == 0
assert result.parseoutcomes()['passed'] == 1 assert result.parseoutcomes()['passed'] == 3
def test_capture_not_started_but_reset(): def test_capture_not_started_but_reset():
capsys = py.io.StdCapture(now=False) capsys = py.io.StdCapture(now=False)