[Pytest-commit] commit/pytest: RonnyPfannschmidt: Redo the Capture integration propperly
commits-noreply at bitbucket.org
commits-noreply at bitbucket.org
Fri Jan 24 21:46:08 CET 2014
1 new commit in pytest:
https://bitbucket.org/hpk42/pytest/commits/552a700dc71f/
Changeset: 552a700dc71f
User: RonnyPfannschmidt
Date: 2014-01-24 21:22:19
Summary: Redo the Capture integration propperly
Affected #: 3 files
diff -r f428ddd78673080a63c59887cb4f603d29d6ed4a -r 552a700dc71f69bc5b28970ec2ff500fbaea77cb CHANGELOG
--- a/CHANGELOG
+++ b/CHANGELOG
@@ -15,9 +15,15 @@
correctly also on python2 and with pytest-xdist runs. (the fix
requires py-1.4.20)
+- copy, cleanup and integrate py.io capture
+ from pylib 1.4.20.dev2 (rev 13d9af95547e)
+
- address issue416: clarify docs as to conftest.py loading semantics
+- make capfd/capsys.capture private, its unused and shouldnt be exposed
+
+
2.5.1
-----------------------------------
diff -r f428ddd78673080a63c59887cb4f603d29d6ed4a -r 552a700dc71f69bc5b28970ec2ff500fbaea77cb _pytest/capture.py
--- a/_pytest/capture.py
+++ b/_pytest/capture.py
@@ -1,17 +1,55 @@
-""" per-test stdout/stderr capturing mechanisms, ``capsys`` and ``capfd`` function arguments. """
-
-import pytest, py
+"""
+ per-test stdout/stderr capturing mechanisms,
+ ``capsys`` and ``capfd`` function arguments.
+"""
+# note: py.io capture was where copied from
+# pylib 1.4.20.dev2 (rev 13d9af95547e)
import sys
import os
+import tempfile
+
+import py
+import pytest
+
+try:
+ from io import StringIO
+except ImportError:
+ from StringIO import StringIO
+
+try:
+ from io import BytesIO
+except ImportError:
+ class BytesIO(StringIO):
+ def write(self, data):
+ if isinstance(data, unicode):
+ raise TypeError("not a byte value: %r" % (data,))
+ StringIO.write(self, data)
+
+if sys.version_info < (3, 0):
+ class TextIO(StringIO):
+ def write(self, data):
+ if not isinstance(data, unicode):
+ enc = getattr(self, '_encoding', 'UTF-8')
+ data = unicode(data, enc, 'replace')
+ StringIO.write(self, data)
+else:
+ TextIO = StringIO
+
+
+patchsysdict = {0: 'stdin', 1: 'stdout', 2: 'stderr'}
+
def pytest_addoption(parser):
group = parser.getgroup("general")
- group._addoption('--capture', action="store", default=None,
+ group._addoption(
+ '--capture', action="store", default=None,
metavar="method", choices=['fd', 'sys', 'no'],
help="per-test capturing method: one of fd (default)|sys|no.")
- group._addoption('-s', action="store_const", const="no", dest="capture",
+ group._addoption(
+ '-s', action="store_const", const="no", dest="capture",
help="shortcut for --capture=no.")
+
@pytest.mark.tryfirst
def pytest_load_initial_conftests(early_config, parser, args, __multicall__):
ns = parser.parse_known_args(args)
@@ -22,13 +60,16 @@
method = "sys"
capman = CaptureManager(method)
early_config.pluginmanager.register(capman, "capturemanager")
+
# make sure that capturemanager is properly reset at final shutdown
def teardown():
try:
capman.reset_capturings()
except ValueError:
pass
+
early_config.pluginmanager.add_shutdown(teardown)
+
# make sure logging does not raise exceptions at the end
def silence_logging_at_shutdown():
if "logging" in sys.modules:
@@ -47,21 +88,27 @@
sys.stderr.write(err)
raise
+
def addouterr(rep, outerr):
for secname, content in zip(["out", "err"], outerr):
if content:
rep.sections.append(("Captured std%s" % secname, content))
+
class NoCapture:
def startall(self):
pass
+
def resume(self):
pass
+
def reset(self):
pass
+
def suspend(self):
return "", ""
+
class CaptureManager:
def __init__(self, defaultmethod=None):
self._method2capture = {}
@@ -69,21 +116,25 @@
def _maketempfile(self):
f = py.std.tempfile.TemporaryFile()
- newf = py.io.dupfile(f, encoding="UTF-8")
+ newf = dupfile(f, encoding="UTF-8")
f.close()
return newf
def _makestringio(self):
- return py.io.TextIO()
+ return TextIO()
def _getcapture(self, method):
if method == "fd":
- return py.io.StdCaptureFD(now=False,
- out=self._maketempfile(), err=self._maketempfile()
+ return StdCaptureFD(
+ now=False,
+ out=self._maketempfile(),
+ err=self._maketempfile(),
)
elif method == "sys":
- return py.io.StdCapture(now=False,
- out=self._makestringio(), err=self._makestringio()
+ return StdCapture(
+ now=False,
+ out=self._makestringio(),
+ err=self._makestringio(),
)
elif method == "no":
return NoCapture()
@@ -98,23 +149,24 @@
method = config._conftest.rget("option_capture", path=fspath)
except KeyError:
method = "fd"
- if method == "fd" and not hasattr(os, 'dup'): # e.g. jython
+ if method == "fd" and not hasattr(os, 'dup'): # e.g. jython
method = "sys"
return method
def reset_capturings(self):
- for name, cap in self._method2capture.items():
+ for cap in self._method2capture.values():
cap.reset()
def resumecapture_item(self, item):
method = self._getmethod(item.config, item.fspath)
if not hasattr(item, 'outerr'):
- item.outerr = ('', '') # we accumulate outerr on the item
+ item.outerr = ('', '') # we accumulate outerr on the item
return self.resumecapture(method)
def resumecapture(self, method=None):
if hasattr(self, '_capturing'):
- raise ValueError("cannot resume, already capturing with %r" %
+ raise ValueError(
+ "cannot resume, already capturing with %r" %
(self._capturing,))
if method is None:
method = self._defaultmethod
@@ -163,8 +215,9 @@
try:
self.resumecapture(method)
except ValueError:
- return # recursive collect, XXX refactor capturing
- # to allow for more lightweight recursive capturing
+ # recursive collect, XXX refactor capturing
+ # to allow for more lightweight recursive capturing
+ return
try:
rep = __multicall__.execute()
finally:
@@ -205,6 +258,7 @@
error_capsysfderror = "cannot use capsys and capfd at the same time"
+
def pytest_funcarg__capsys(request):
"""enables capturing of writes to sys.stdout/sys.stderr and makes
captured output available via ``capsys.readouterr()`` method calls
@@ -212,7 +266,8 @@
"""
if "capfd" in request._funcargs:
raise request.raiseerror(error_capsysfderror)
- return CaptureFixture(py.io.StdCapture)
+ return CaptureFixture(StdCapture)
+
def pytest_funcarg__capfd(request):
"""enables capturing of writes to file descriptors 1 and 2 and makes
@@ -223,26 +278,366 @@
request.raiseerror(error_capsysfderror)
if not hasattr(os, 'dup'):
pytest.skip("capfd funcarg needs os.dup")
- return CaptureFixture(py.io.StdCaptureFD)
+ return CaptureFixture(StdCaptureFD)
+
class CaptureFixture:
def __init__(self, captureclass):
- self.capture = captureclass(now=False)
+ self._capture = captureclass(now=False)
def _start(self):
- self.capture.startall()
+ self._capture.startall()
def _finalize(self):
- if hasattr(self, 'capture'):
- outerr = self._outerr = self.capture.reset()
- del self.capture
+ if hasattr(self, '_capture'):
+ outerr = self._outerr = self._capture.reset()
+ del self._capture
return outerr
def readouterr(self):
try:
- return self.capture.readouterr()
+ return self._capture.readouterr()
except AttributeError:
return self._outerr
def close(self):
self._finalize()
+
+
+class FDCapture:
+ """ Capture IO to/from a given os-level filedescriptor. """
+
+ def __init__(self, targetfd, tmpfile=None, now=True, patchsys=False):
+ """ save targetfd descriptor, and open a new
+ temporary file there. If no tmpfile is
+ specified a tempfile.Tempfile() will be opened
+ in text mode.
+ """
+ self.targetfd = targetfd
+ if tmpfile is None and targetfd != 0:
+ f = tempfile.TemporaryFile('wb+')
+ tmpfile = dupfile(f, encoding="UTF-8")
+ f.close()
+ self.tmpfile = tmpfile
+ self._savefd = os.dup(self.targetfd)
+ if patchsys:
+ self._oldsys = getattr(sys, patchsysdict[targetfd])
+ if now:
+ self.start()
+
+ def start(self):
+ try:
+ os.fstat(self._savefd)
+ except OSError:
+ raise ValueError(
+ "saved filedescriptor not valid, "
+ "did you call start() twice?")
+ if self.targetfd == 0 and not self.tmpfile:
+ fd = os.open(os.devnull, os.O_RDONLY)
+ os.dup2(fd, 0)
+ os.close(fd)
+ if hasattr(self, '_oldsys'):
+ setattr(sys, patchsysdict[self.targetfd], DontReadFromInput())
+ else:
+ os.dup2(self.tmpfile.fileno(), self.targetfd)
+ if hasattr(self, '_oldsys'):
+ setattr(sys, patchsysdict[self.targetfd], self.tmpfile)
+
+ def done(self):
+ """ unpatch and clean up, returns the self.tmpfile (file object)
+ """
+ os.dup2(self._savefd, self.targetfd)
+ os.close(self._savefd)
+ if self.targetfd != 0:
+ self.tmpfile.seek(0)
+ if hasattr(self, '_oldsys'):
+ setattr(sys, patchsysdict[self.targetfd], self._oldsys)
+ return self.tmpfile
+
+ def writeorg(self, data):
+ """ write a string to the original file descriptor
+ """
+ tempfp = tempfile.TemporaryFile()
+ try:
+ os.dup2(self._savefd, tempfp.fileno())
+ tempfp.write(data)
+ finally:
+ tempfp.close()
+
+
+def dupfile(f, mode=None, buffering=0, raising=False, encoding=None):
+ """ return a new open file object that's a duplicate of f
+
+ mode is duplicated if not given, 'buffering' controls
+ buffer size (defaulting to no buffering) and 'raising'
+ defines whether an exception is raised when an incompatible
+ file object is passed in (if raising is False, the file
+ object itself will be returned)
+ """
+ try:
+ fd = f.fileno()
+ mode = mode or f.mode
+ except AttributeError:
+ if raising:
+ raise
+ return f
+ newfd = os.dup(fd)
+ if sys.version_info >= (3, 0):
+ if encoding is not None:
+ mode = mode.replace("b", "")
+ buffering = True
+ return os.fdopen(newfd, mode, buffering, encoding, closefd=True)
+ else:
+ f = os.fdopen(newfd, mode, buffering)
+ if encoding is not None:
+ return EncodedFile(f, encoding)
+ return f
+
+
+class EncodedFile(object):
+ def __init__(self, _stream, encoding):
+ self._stream = _stream
+ self.encoding = encoding
+
+ def write(self, obj):
+ if isinstance(obj, unicode):
+ obj = obj.encode(self.encoding)
+ self._stream.write(obj)
+
+ def writelines(self, linelist):
+ data = ''.join(linelist)
+ self.write(data)
+
+ def __getattr__(self, name):
+ return getattr(self._stream, name)
+
+
+class Capture(object):
+ def call(cls, func, *args, **kwargs):
+ """ return a (res, out, err) tuple where
+ out and err represent the output/error output
+ during function execution.
+ call the given function with args/kwargs
+ and capture output/error during its execution.
+ """
+ so = cls()
+ try:
+ res = func(*args, **kwargs)
+ finally:
+ out, err = so.reset()
+ return res, out, err
+ call = classmethod(call)
+
+ def reset(self):
+ """ reset sys.stdout/stderr and return captured output as strings. """
+ if hasattr(self, '_reset'):
+ raise ValueError("was already reset")
+ self._reset = True
+ outfile, errfile = self.done(save=False)
+ out, err = "", ""
+ if outfile and not outfile.closed:
+ out = outfile.read()
+ outfile.close()
+ if errfile and errfile != outfile and not errfile.closed:
+ err = errfile.read()
+ errfile.close()
+ return out, err
+
+ def suspend(self):
+ """ return current snapshot captures, memorize tempfiles. """
+ outerr = self.readouterr()
+ outfile, errfile = self.done()
+ return outerr
+
+
+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). 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):
+ self._options = {
+ "out": out,
+ "err": err,
+ "mixed": mixed,
+ "in_": in_,
+ "patchsys": patchsys,
+ "now": now,
+ }
+ self._save()
+ if now:
+ self.startall()
+
+ def _save(self):
+ in_ = self._options['in_']
+ out = self._options['out']
+ err = self._options['err']
+ mixed = self._options['mixed']
+ patchsys = self._options['patchsys']
+ if in_:
+ try:
+ self.in_ = FDCapture(
+ 0, tmpfile=None, now=False,
+ patchsys=patchsys)
+ except OSError:
+ pass
+ if out:
+ tmpfile = None
+ if hasattr(out, 'write'):
+ tmpfile = out
+ try:
+ self.out = FDCapture(
+ 1, tmpfile=tmpfile,
+ now=False, patchsys=patchsys)
+ self._options['out'] = self.out.tmpfile
+ except OSError:
+ pass
+ if err:
+ if out and mixed:
+ tmpfile = self.out.tmpfile
+ elif hasattr(err, 'write'):
+ tmpfile = err
+ else:
+ tmpfile = None
+ try:
+ self.err = FDCapture(
+ 2, tmpfile=tmpfile,
+ now=False, patchsys=patchsys)
+ self._options['err'] = self.err.tmpfile
+ except OSError:
+ pass
+
+ def startall(self):
+ if hasattr(self, 'in_'):
+ self.in_.start()
+ if hasattr(self, 'out'):
+ self.out.start()
+ if hasattr(self, 'err'):
+ self.err.start()
+
+ def resume(self):
+ """ resume capturing with original temp files. """
+ self.startall()
+
+ def done(self, save=True):
+ """ return (outfile, errfile) and stop capturing. """
+ outfile = errfile = None
+ if hasattr(self, 'out') and not self.out.tmpfile.closed:
+ outfile = self.out.done()
+ if hasattr(self, 'err') and not self.err.tmpfile.closed:
+ errfile = self.err.done()
+ if hasattr(self, 'in_'):
+ self.in_.done()
+ if save:
+ self._save()
+ return outfile, errfile
+
+ def readouterr(self):
+ """ return snapshot value of stdout/stderr capturings. """
+ out = self._readsnapshot('out')
+ err = self._readsnapshot('err')
+ return out, err
+
+ def _readsnapshot(self, name):
+ if hasattr(self, name):
+ f = getattr(self, name).tmpfile
+ else:
+ return ''
+
+ f.seek(0)
+ res = f.read()
+ enc = getattr(f, "encoding", None)
+ if enc:
+ res = py.builtin._totext(res, enc, "replace")
+ f.truncate(0)
+ f.seek(0)
+ return res
+
+
+class StdCapture(Capture):
+ """ This class allows to capture writes to sys.stdout|stderr "in-memory"
+ and will raise errors on tries to read from sys.stdin. It only
+ modifies sys.stdout|stderr|stdin attributes and does not
+ touch underlying File Descriptors (use StdCaptureFD for that).
+ """
+ def __init__(self, out=True, err=True, in_=True, mixed=False, now=True):
+ self._oldout = sys.stdout
+ self._olderr = sys.stderr
+ self._oldin = sys.stdin
+ if out and not hasattr(out, 'file'):
+ out = TextIO()
+ self.out = out
+ if err:
+ if mixed:
+ err = out
+ elif not hasattr(err, 'write'):
+ err = TextIO()
+ self.err = err
+ self.in_ = in_
+ if now:
+ self.startall()
+
+ def startall(self):
+ if self.out:
+ sys.stdout = self.out
+ if self.err:
+ sys.stderr = self.err
+ if self.in_:
+ sys.stdin = self.in_ = DontReadFromInput()
+
+ def done(self, save=True):
+ """ return (outfile, errfile) and stop capturing. """
+ outfile = errfile = None
+ if self.out and not self.out.closed:
+ sys.stdout = self._oldout
+ outfile = self.out
+ outfile.seek(0)
+ if self.err and not self.err.closed:
+ sys.stderr = self._olderr
+ errfile = self.err
+ errfile.seek(0)
+ if self.in_:
+ sys.stdin = self._oldin
+ return outfile, errfile
+
+ def resume(self):
+ """ resume capturing with original temp files. """
+ self.startall()
+
+ def readouterr(self):
+ """ return snapshot value of stdout/stderr capturings. """
+ out = err = ""
+ if self.out:
+ out = self.out.getvalue()
+ self.out.truncate(0)
+ self.out.seek(0)
+ if self.err:
+ err = self.err.getvalue()
+ self.err.truncate(0)
+ self.err.seek(0)
+ return out, err
+
+
+class DontReadFromInput:
+ """Temporary stub class. Ideally when stdin is accessed, the
+ capturing should be turned off, with possibly all data captured
+ so far sent to the screen. This should be configurable, though,
+ because in automated test runs it is better to crash than
+ hang indefinitely.
+ """
+ def read(self, *args):
+ raise IOError("reading from stdin while output is captured")
+ readline = read
+ readlines = read
+ __iter__ = read
+
+ def fileno(self):
+ raise ValueError("redirected Stdin is pseudofile, has no fileno()")
+
+ def isatty(self):
+ return False
+
+ def close(self):
+ pass
diff -r f428ddd78673080a63c59887cb4f603d29d6ed4a -r 552a700dc71f69bc5b28970ec2ff500fbaea77cb testing/test_capture.py
--- a/testing/test_capture.py
+++ b/testing/test_capture.py
@@ -1,8 +1,48 @@
-import pytest, py, os, sys
+# note: py.io capture tests where copied from
+# pylib 1.4.20.dev2 (rev 13d9af95547e)
+from __future__ import with_statement
+import os
+import sys
+import py
+import pytest
+
+from _pytest import capture
from _pytest.capture import CaptureManager
+from py.builtin import print_
needsosdup = pytest.mark.xfail("not hasattr(os, 'dup')")
+if sys.version_info >= (3, 0):
+ def tobytes(obj):
+ if isinstance(obj, str):
+ obj = obj.encode('UTF-8')
+ assert isinstance(obj, bytes)
+ return obj
+
+ def totext(obj):
+ if isinstance(obj, bytes):
+ obj = str(obj, 'UTF-8')
+ assert isinstance(obj, str)
+ return obj
+else:
+ def tobytes(obj):
+ if isinstance(obj, unicode):
+ obj = obj.encode('UTF-8')
+ assert isinstance(obj, str)
+ return obj
+
+ def totext(obj):
+ if isinstance(obj, str):
+ obj = unicode(obj, 'UTF-8')
+ assert isinstance(obj, unicode)
+ return obj
+
+
+def oswritebytes(fd, obj):
+ os.write(fd, tobytes(obj))
+
+
+
class TestCaptureManager:
def test_getmethod_default_no_fd(self, testdir, monkeypatch):
config = testdir.parseconfig(testdir.tmpdir)
@@ -34,7 +74,7 @@
@needsosdup
@pytest.mark.parametrize("method", ['no', 'fd', 'sys'])
def test_capturing_basic_api(self, method):
- capouter = py.io.StdCaptureFD()
+ capouter = capture.StdCaptureFD()
old = sys.stdout, sys.stderr, sys.stdin
try:
capman = CaptureManager()
@@ -58,7 +98,7 @@
@needsosdup
def test_juggle_capturings(self, testdir):
- capouter = py.io.StdCaptureFD()
+ capouter = capture.StdCaptureFD()
try:
#config = testdir.parseconfig(testdir.tmpdir)
capman = CaptureManager()
@@ -80,10 +120,11 @@
finally:
capouter.reset()
+
@pytest.mark.xfail("hasattr(sys, 'pypy_version_info')")
@pytest.mark.parametrize("method", ['fd', 'sys'])
def test_capturing_unicode(testdir, method):
- if sys.version_info >= (3,0):
+ if sys.version_info >= (3, 0):
obj = "'b\u00f6y'"
else:
obj = "u'\u00f6y'"
@@ -100,6 +141,7 @@
"*1 passed*"
])
+
@pytest.mark.parametrize("method", ['fd', 'sys'])
def test_capturing_bytes_in_utf8_encoding(testdir, method):
testdir.makepyfile("""
@@ -111,6 +153,7 @@
"*1 passed*"
])
+
def test_collect_capturing(testdir):
p = testdir.makepyfile("""
print ("collect %s failure" % 13)
@@ -122,6 +165,7 @@
"*collect 13 failure*",
])
+
class TestPerTestCapturing:
def test_capture_and_fixtures(self, testdir):
p = testdir.makepyfile("""
@@ -169,7 +213,6 @@
"in teardown*",
])
-
def test_no_carry_over(self, testdir):
p = testdir.makepyfile("""
def test_func1():
@@ -183,7 +226,6 @@
assert "in func1" not in s
assert "in func2" in s
-
def test_teardown_capturing(self, testdir):
p = testdir.makepyfile("""
def setup_function(function):
@@ -244,13 +286,14 @@
"2",
])
+
class TestLoggingInteraction:
def test_logging_stream_ownership(self, testdir):
p = testdir.makepyfile("""
def test_logging():
import logging
import pytest
- stream = py.io.TextIO()
+ stream = capture.TextIO()
logging.basicConfig(stream=stream)
stream.close() # to free memory/release resources
""")
@@ -320,7 +363,8 @@
logging.warn("hello432")
assert 0
""")
- result = testdir.runpytest(p, "--traceconfig",
+ result = testdir.runpytest(
+ p, "--traceconfig",
"-p", "no:capturelog")
assert result.ret != 0
result.stdout.fnmatch_lines([
@@ -461,6 +505,7 @@
"*1 error*"
])
+
def test_fdfuncarg_skips_on_no_osdup(testdir):
testdir.makepyfile("""
import os
@@ -474,6 +519,7 @@
"*1 skipped*"
])
+
def test_capture_conftest_runtest_setup(testdir):
testdir.makeconftest("""
def pytest_runtest_setup():
@@ -484,6 +530,7 @@
assert result.ret == 0
assert 'hello19' not in result.stdout.str()
+
def test_capture_early_option_parsing(testdir):
testdir.makeconftest("""
def pytest_runtest_setup():
@@ -494,9 +541,10 @@
assert result.ret == 0
assert 'hello19' in result.stdout.str()
- at pytest.mark.xfail(reason='encoding issues')
+
+ at pytest.mark.xfail(sys.version_info >= (3, 0), reason='encoding issues')
def test_capture_binary_output(testdir):
- testdir.makepyfile("""
+ testdir.makepyfile(r"""
import pytest
def test_a():
@@ -516,3 +564,499 @@
'*2 passed*',
])
+
+class TestTextIO:
+ def test_text(self):
+ f = capture.TextIO()
+ f.write("hello")
+ s = f.getvalue()
+ assert s == "hello"
+ f.close()
+
+ def test_unicode_and_str_mixture(self):
+ f = capture.TextIO()
+ if sys.version_info >= (3, 0):
+ f.write("\u00f6")
+ pytest.raises(TypeError, "f.write(bytes('hello', 'UTF-8'))")
+ else:
+ f.write(unicode("\u00f6", 'UTF-8'))
+ f.write("hello") # bytes
+ s = f.getvalue()
+ f.close()
+ assert isinstance(s, unicode)
+
+
+def test_bytes_io():
+ f = capture.BytesIO()
+ f.write(tobytes("hello"))
+ pytest.raises(TypeError, "f.write(totext('hello'))")
+ s = f.getvalue()
+ assert s == tobytes("hello")
+
+
+def test_dontreadfrominput():
+ from _pytest.capture import DontReadFromInput
+ f = DontReadFromInput()
+ assert not f.isatty()
+ pytest.raises(IOError, f.read)
+ pytest.raises(IOError, f.readlines)
+ pytest.raises(IOError, iter, f)
+ pytest.raises(ValueError, f.fileno)
+ f.close() # just for completeness
+
+
+def pytest_funcarg__tmpfile(request):
+ testdir = request.getfuncargvalue("testdir")
+ f = testdir.makepyfile("").open('wb+')
+ request.addfinalizer(f.close)
+ return f
+
+
+ at needsosdup
+def test_dupfile(tmpfile):
+ flist = []
+ for i in range(5):
+ nf = capture.dupfile(tmpfile, encoding="utf-8")
+ assert nf != tmpfile
+ assert nf.fileno() != tmpfile.fileno()
+ assert nf not in flist
+ print_(i, end="", file=nf)
+ flist.append(nf)
+ for i in range(5):
+ f = flist[i]
+ f.close()
+ tmpfile.seek(0)
+ s = tmpfile.read()
+ assert "01234" in repr(s)
+ tmpfile.close()
+
+
+def test_dupfile_no_mode():
+ """
+ dupfile should trap an AttributeError and return f if no mode is supplied.
+ """
+ class SomeFileWrapper(object):
+ "An object with a fileno method but no mode attribute"
+ def fileno(self):
+ return 1
+ tmpfile = SomeFileWrapper()
+ assert capture.dupfile(tmpfile) is tmpfile
+ with pytest.raises(AttributeError):
+ capture.dupfile(tmpfile, raising=True)
+
+
+def lsof_check(func):
+ pid = os.getpid()
+ try:
+ out = py.process.cmdexec("lsof -p %d" % pid)
+ except py.process.cmdexec.Error:
+ pytest.skip("could not run 'lsof'")
+ func()
+ out2 = py.process.cmdexec("lsof -p %d" % pid)
+ len1 = len([x for x in out.split("\n") if "REG" in x])
+ len2 = len([x for x in out2.split("\n") if "REG" in x])
+ assert len2 < len1 + 3, out2
+
+
+class TestFDCapture:
+ pytestmark = needsosdup
+
+ def test_not_now(self, tmpfile):
+ fd = tmpfile.fileno()
+ cap = capture.FDCapture(fd, now=False)
+ data = tobytes("hello")
+ os.write(fd, data)
+ f = cap.done()
+ s = f.read()
+ assert not s
+ cap = capture.FDCapture(fd, now=False)
+ cap.start()
+ os.write(fd, data)
+ f = cap.done()
+ s = f.read()
+ assert s == "hello"
+
+ def test_simple(self, tmpfile):
+ fd = tmpfile.fileno()
+ cap = capture.FDCapture(fd)
+ data = tobytes("hello")
+ os.write(fd, data)
+ f = cap.done()
+ s = f.read()
+ assert s == "hello"
+ f.close()
+
+ def test_simple_many(self, tmpfile):
+ for i in range(10):
+ self.test_simple(tmpfile)
+
+ def test_simple_many_check_open_files(self, tmpfile):
+ lsof_check(lambda: self.test_simple_many(tmpfile))
+
+ def test_simple_fail_second_start(self, tmpfile):
+ fd = tmpfile.fileno()
+ cap = capture.FDCapture(fd)
+ f = cap.done()
+ pytest.raises(ValueError, cap.start)
+ f.close()
+
+ def test_stderr(self):
+ cap = capture.FDCapture(2, patchsys=True)
+ print_("hello", file=sys.stderr)
+ f = cap.done()
+ s = f.read()
+ assert s == "hello\n"
+
+ def test_stdin(self, tmpfile):
+ tmpfile.write(tobytes("3"))
+ tmpfile.seek(0)
+ cap = capture.FDCapture(0, tmpfile=tmpfile)
+ # check with os.read() directly instead of raw_input(), because
+ # sys.stdin itself may be redirected (as pytest now does by default)
+ x = os.read(0, 100).strip()
+ cap.done()
+ assert x == tobytes("3")
+
+ def test_writeorg(self, tmpfile):
+ data1, data2 = tobytes("foo"), tobytes("bar")
+ try:
+ cap = capture.FDCapture(tmpfile.fileno())
+ tmpfile.write(data1)
+ cap.writeorg(data2)
+ finally:
+ tmpfile.close()
+ f = cap.done()
+ scap = f.read()
+ assert scap == totext(data1)
+ stmp = open(tmpfile.name, 'rb').read()
+ assert stmp == data2
+
+
+class TestStdCapture:
+ def getcapture(self, **kw):
+ return capture.StdCapture(**kw)
+
+ def test_capturing_done_simple(self):
+ cap = self.getcapture()
+ sys.stdout.write("hello")
+ sys.stderr.write("world")
+ outfile, errfile = cap.done()
+ s = outfile.read()
+ assert s == "hello"
+ s = errfile.read()
+ assert s == "world"
+
+ def test_capturing_reset_simple(self):
+ cap = self.getcapture()
+ print("hello world")
+ sys.stderr.write("hello error\n")
+ out, err = cap.reset()
+ assert out == "hello world\n"
+ assert err == "hello error\n"
+
+ def test_capturing_readouterr(self):
+ cap = self.getcapture()
+ try:
+ print ("hello world")
+ sys.stderr.write("hello error\n")
+ out, err = cap.readouterr()
+ assert out == "hello world\n"
+ assert err == "hello error\n"
+ sys.stderr.write("error2")
+ finally:
+ out, err = cap.reset()
+ assert err == "error2"
+
+ def test_capturing_readouterr_unicode(self):
+ cap = self.getcapture()
+ try:
+ print ("hx\xc4\x85\xc4\x87")
+ out, err = cap.readouterr()
+ finally:
+ cap.reset()
+ assert out == py.builtin._totext("hx\xc4\x85\xc4\x87\n", "utf8")
+
+ @pytest.mark.skipif('sys.version_info >= (3,)',
+ reason='text output different for bytes on python3')
+ def test_capturing_readouterr_decode_error_handling(self):
+ cap = self.getcapture()
+ # triggered a internal error in pytest
+ print('\xa6')
+ out, err = cap.readouterr()
+ assert out == py.builtin._totext('\ufffd\n', 'unicode-escape')
+
+ def test_capturing_mixed(self):
+ cap = self.getcapture(mixed=True)
+ sys.stdout.write("hello ")
+ sys.stderr.write("world")
+ sys.stdout.write(".")
+ out, err = cap.reset()
+ assert out.strip() == "hello world."
+ assert not err
+
+ def test_reset_twice_error(self):
+ cap = self.getcapture()
+ print ("hello")
+ out, err = cap.reset()
+ pytest.raises(ValueError, cap.reset)
+ assert out == "hello\n"
+ assert not err
+
+ def test_capturing_modify_sysouterr_in_between(self):
+ oldout = sys.stdout
+ olderr = sys.stderr
+ cap = self.getcapture()
+ sys.stdout.write("hello")
+ sys.stderr.write("world")
+ sys.stdout = capture.TextIO()
+ sys.stderr = capture.TextIO()
+ print ("not seen")
+ sys.stderr.write("not seen\n")
+ out, err = cap.reset()
+ assert out == "hello"
+ assert err == "world"
+ assert sys.stdout == oldout
+ assert sys.stderr == olderr
+
+ def test_capturing_error_recursive(self):
+ cap1 = self.getcapture()
+ print ("cap1")
+ cap2 = self.getcapture()
+ print ("cap2")
+ out2, err2 = cap2.reset()
+ out1, err1 = cap1.reset()
+ assert out1 == "cap1\n"
+ assert out2 == "cap2\n"
+
+ def test_just_out_capture(self):
+ cap = self.getcapture(out=True, err=False)
+ sys.stdout.write("hello")
+ sys.stderr.write("world")
+ out, err = cap.reset()
+ assert out == "hello"
+ assert not err
+
+ def test_just_err_capture(self):
+ cap = self.getcapture(out=False, err=True)
+ sys.stdout.write("hello")
+ sys.stderr.write("world")
+ out, err = cap.reset()
+ assert err == "world"
+ assert not out
+
+ def test_stdin_restored(self):
+ old = sys.stdin
+ cap = self.getcapture(in_=True)
+ newstdin = sys.stdin
+ out, err = cap.reset()
+ assert newstdin != sys.stdin
+ assert sys.stdin is old
+
+ def test_stdin_nulled_by_default(self):
+ print ("XXX this test may well hang instead of crashing")
+ print ("XXX which indicates an error in the underlying capturing")
+ print ("XXX mechanisms")
+ cap = self.getcapture()
+ pytest.raises(IOError, "sys.stdin.read()")
+ out, err = cap.reset()
+
+ def test_suspend_resume(self):
+ cap = self.getcapture(out=True, err=False, in_=False)
+ try:
+ print ("hello")
+ sys.stderr.write("error\n")
+ out, err = cap.suspend()
+ assert out == "hello\n"
+ assert not err
+ print ("in between")
+ sys.stderr.write("in between\n")
+ cap.resume()
+ print ("after")
+ sys.stderr.write("error_after\n")
+ finally:
+ out, err = cap.reset()
+ assert out == "after\n"
+ assert not err
+
+
+class TestStdCaptureNotNow(TestStdCapture):
+ def getcapture(self, **kw):
+ kw['now'] = False
+ cap = capture.StdCapture(**kw)
+ cap.startall()
+ return cap
+
+
+class TestStdCaptureFD(TestStdCapture):
+ pytestmark = needsosdup
+
+ def getcapture(self, **kw):
+ return capture.StdCaptureFD(**kw)
+
+ def test_intermingling(self):
+ cap = self.getcapture()
+ oswritebytes(1, "1")
+ sys.stdout.write(str(2))
+ sys.stdout.flush()
+ oswritebytes(1, "3")
+ oswritebytes(2, "a")
+ sys.stderr.write("b")
+ sys.stderr.flush()
+ oswritebytes(2, "c")
+ out, err = cap.reset()
+ assert out == "123"
+ assert err == "abc"
+
+ def test_callcapture(self):
+ def func(x, y):
+ print (x)
+ sys.stderr.write(str(y))
+ return 42
+
+ res, out, err = capture.StdCaptureFD.call(func, 3, y=4)
+ assert res == 42
+ assert out.startswith("3")
+ assert err.startswith("4")
+
+ def test_many(self, capfd):
+ def f():
+ for i in range(10):
+ cap = capture.StdCaptureFD()
+ cap.reset()
+ lsof_check(f)
+
+
+class TestStdCaptureFDNotNow(TestStdCaptureFD):
+ pytestmark = needsosdup
+
+ def getcapture(self, **kw):
+ kw['now'] = False
+ cap = capture.StdCaptureFD(**kw)
+ cap.startall()
+ return cap
+
+
+ at needsosdup
+def test_stdcapture_fd_tmpfile(tmpfile):
+ capfd = capture.StdCaptureFD(out=tmpfile)
+ try:
+ os.write(1, "hello".encode("ascii"))
+ os.write(2, "world".encode("ascii"))
+ outf, errf = capfd.done()
+ finally:
+ capfd.reset()
+ assert outf == tmpfile
+
+
+class TestStdCaptureFDinvalidFD:
+ pytestmark = needsosdup
+
+ def test_stdcapture_fd_invalid_fd(self, testdir):
+ testdir.makepyfile("""
+ import os
+ from _pytest.capture import StdCaptureFD
+ def test_stdout():
+ os.close(1)
+ cap = StdCaptureFD(out=True, err=False, in_=False)
+ cap.done()
+ def test_stderr():
+ os.close(2)
+ cap = StdCaptureFD(out=False, err=True, in_=False)
+ cap.done()
+ def test_stdin():
+ os.close(0)
+ cap = 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 = capture.StdCapture(now=False)
+ capsys.done()
+ capsys.done()
+ capsys.reset()
+
+
+ at needsosdup
+def test_capture_no_sys():
+ capsys = capture.StdCapture()
+ try:
+ cap = capture.StdCaptureFD(patchsys=False)
+ sys.stdout.write("hello")
+ sys.stderr.write("world")
+ oswritebytes(1, "1")
+ oswritebytes(2, "2")
+ out, err = cap.reset()
+ assert out == "1"
+ assert err == "2"
+ finally:
+ capsys.reset()
+
+
+ at needsosdup
+def test_callcapture_nofd():
+ def func(x, y):
+ oswritebytes(1, "hello")
+ oswritebytes(2, "hello")
+ print (x)
+ sys.stderr.write(str(y))
+ return 42
+
+ capfd = capture.StdCaptureFD(patchsys=False)
+ try:
+ res, out, err = capture.StdCapture.call(func, 3, y=4)
+ finally:
+ capfd.reset()
+ assert res == 42
+ assert out.startswith("3")
+ assert err.startswith("4")
+
+
+ at needsosdup
+ at pytest.mark.parametrize('use', [True, False])
+def test_fdcapture_tmpfile_remains_the_same(tmpfile, use):
+ if not use:
+ tmpfile = True
+ cap = capture.StdCaptureFD(out=False, err=tmpfile, now=False)
+ try:
+ cap.startall()
+ capfile = cap.err.tmpfile
+ cap.suspend()
+ cap.resume()
+ finally:
+ cap.reset()
+ capfile2 = cap.err.tmpfile
+ assert capfile2 == capfile
+
+
+ at pytest.mark.parametrize('method', ['StdCapture', 'StdCaptureFD'])
+def test_capturing_and_logging_fundamentals(testdir, method):
+ if method == "StdCaptureFD" and not hasattr(os, 'dup'):
+ pytest.skip("need os.dup")
+ # here we check a fundamental feature
+ p = testdir.makepyfile("""
+ import sys, os
+ import py, logging
+ from _pytest import capture
+ cap = capture.%s(out=False, in_=False)
+
+ logging.warn("hello1")
+ outerr = cap.suspend()
+ print ("suspend, captured %%s" %%(outerr,))
+ logging.warn("hello2")
+
+ cap.resume()
+ logging.warn("hello3")
+
+ outerr = cap.suspend()
+ print ("suspend2, captured %%s" %% (outerr,))
+ """ % (method,))
+ result = testdir.runpython(p)
+ result.stdout.fnmatch_lines([
+ "suspend, captured*hello1*",
+ "suspend2, captured*hello2*WARNING:root:hello3*",
+ ])
+ assert "atexit" not in result.stderr.str()
Repository URL: https://bitbucket.org/hpk42/pytest/
--
This is a commit notification from bitbucket.org. You are receiving
this because you have the service enabled, addressing the recipient of
this email.
More information about the pytest-commit
mailing list