backing out Ronny's PR because it was merged too early (still has failing tests)
This commit is contained in:
holger krekel
parent
8976b3ee0e
commit
9aaf0fd340
|
|
@ -15,12 +15,6 @@ UNRELEASED
|
|||
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)
|
||||
|
||||
- make capfd/capsys.capture private, its unused and shouldnt be exposed
|
||||
|
||||
|
||||
2.5.1
|
||||
-----------------------------------
|
||||
|
||||
|
|
|
|||
|
|
@ -1,55 +1,17 @@
|
|||
"""
|
||||
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)
|
||||
""" per-test stdout/stderr capturing mechanisms, ``capsys`` and ``capfd`` function arguments. """
|
||||
|
||||
import pytest, py
|
||||
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)
|
||||
|
|
@ -60,16 +22,13 @@ def pytest_load_initial_conftests(early_config, parser, args, __multicall__):
|
|||
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:
|
||||
|
|
@ -88,27 +47,21 @@ def pytest_load_initial_conftests(early_config, parser, args, __multicall__):
|
|||
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 = {}
|
||||
|
|
@ -116,25 +69,21 @@ class CaptureManager:
|
|||
|
||||
def _maketempfile(self):
|
||||
f = py.std.tempfile.TemporaryFile()
|
||||
newf = dupfile(f, encoding="UTF-8")
|
||||
newf = py.io.dupfile(f, encoding="UTF-8")
|
||||
f.close()
|
||||
return newf
|
||||
|
||||
def _makestringio(self):
|
||||
return TextIO()
|
||||
return py.io.TextIO()
|
||||
|
||||
def _getcapture(self, method):
|
||||
if method == "fd":
|
||||
return StdCaptureFD(
|
||||
now=False,
|
||||
out=self._maketempfile(),
|
||||
err=self._maketempfile(),
|
||||
return py.io.StdCaptureFD(now=False,
|
||||
out=self._maketempfile(), err=self._maketempfile()
|
||||
)
|
||||
elif method == "sys":
|
||||
return StdCapture(
|
||||
now=False,
|
||||
out=self._makestringio(),
|
||||
err=self._makestringio(),
|
||||
return py.io.StdCapture(now=False,
|
||||
out=self._makestringio(), err=self._makestringio()
|
||||
)
|
||||
elif method == "no":
|
||||
return NoCapture()
|
||||
|
|
@ -154,7 +103,7 @@ class CaptureManager:
|
|||
return method
|
||||
|
||||
def reset_capturings(self):
|
||||
for cap in self._method2capture.values():
|
||||
for name, cap in self._method2capture.items():
|
||||
cap.reset()
|
||||
|
||||
def resumecapture_item(self, item):
|
||||
|
|
@ -165,8 +114,7 @@ class CaptureManager:
|
|||
|
||||
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
|
||||
|
|
@ -215,9 +163,8 @@ class CaptureManager:
|
|||
try:
|
||||
self.resumecapture(method)
|
||||
except ValueError:
|
||||
# recursive collect, XXX refactor capturing
|
||||
return # recursive collect, XXX refactor capturing
|
||||
# to allow for more lightweight recursive capturing
|
||||
return
|
||||
try:
|
||||
rep = __multicall__.execute()
|
||||
finally:
|
||||
|
|
@ -258,7 +205,6 @@ class CaptureManager:
|
|||
|
||||
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
|
||||
|
|
@ -266,8 +212,7 @@ def pytest_funcarg__capsys(request):
|
|||
"""
|
||||
if "capfd" in request._funcargs:
|
||||
raise request.raiseerror(error_capsysfderror)
|
||||
return CaptureFixture(StdCapture)
|
||||
|
||||
return CaptureFixture(py.io.StdCapture)
|
||||
|
||||
def pytest_funcarg__capfd(request):
|
||||
"""enables capturing of writes to file descriptors 1 and 2 and makes
|
||||
|
|
@ -278,366 +223,26 @@ def pytest_funcarg__capfd(request):
|
|||
request.raiseerror(error_capsysfderror)
|
||||
if not hasattr(os, 'dup'):
|
||||
pytest.skip("capfd funcarg needs os.dup")
|
||||
return CaptureFixture(StdCaptureFD)
|
||||
|
||||
return CaptureFixture(py.io.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
|
||||
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
|
||||
|
|
|
|||
|
|
@ -1,48 +1,8 @@
|
|||
# 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
|
||||
import pytest, py, os, sys
|
||||
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)
|
||||
|
|
@ -74,7 +34,7 @@ class TestCaptureManager:
|
|||
@needsosdup
|
||||
@pytest.mark.parametrize("method", ['no', 'fd', 'sys'])
|
||||
def test_capturing_basic_api(self, method):
|
||||
capouter = capture.StdCaptureFD()
|
||||
capouter = py.io.StdCaptureFD()
|
||||
old = sys.stdout, sys.stderr, sys.stdin
|
||||
try:
|
||||
capman = CaptureManager()
|
||||
|
|
@ -98,7 +58,7 @@ class TestCaptureManager:
|
|||
|
||||
@needsosdup
|
||||
def test_juggle_capturings(self, testdir):
|
||||
capouter = capture.StdCaptureFD()
|
||||
capouter = py.io.StdCaptureFD()
|
||||
try:
|
||||
#config = testdir.parseconfig(testdir.tmpdir)
|
||||
capman = CaptureManager()
|
||||
|
|
@ -120,7 +80,6 @@ class TestCaptureManager:
|
|||
finally:
|
||||
capouter.reset()
|
||||
|
||||
|
||||
@pytest.mark.xfail("hasattr(sys, 'pypy_version_info')")
|
||||
@pytest.mark.parametrize("method", ['fd', 'sys'])
|
||||
def test_capturing_unicode(testdir, method):
|
||||
|
|
@ -141,7 +100,6 @@ def test_capturing_unicode(testdir, method):
|
|||
"*1 passed*"
|
||||
])
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", ['fd', 'sys'])
|
||||
def test_capturing_bytes_in_utf8_encoding(testdir, method):
|
||||
testdir.makepyfile("""
|
||||
|
|
@ -153,7 +111,6 @@ def test_capturing_bytes_in_utf8_encoding(testdir, method):
|
|||
"*1 passed*"
|
||||
])
|
||||
|
||||
|
||||
def test_collect_capturing(testdir):
|
||||
p = testdir.makepyfile("""
|
||||
print ("collect %s failure" % 13)
|
||||
|
|
@ -165,7 +122,6 @@ def test_collect_capturing(testdir):
|
|||
"*collect 13 failure*",
|
||||
])
|
||||
|
||||
|
||||
class TestPerTestCapturing:
|
||||
def test_capture_and_fixtures(self, testdir):
|
||||
p = testdir.makepyfile("""
|
||||
|
|
@ -213,6 +169,7 @@ class TestPerTestCapturing:
|
|||
"in teardown*",
|
||||
])
|
||||
|
||||
|
||||
def test_no_carry_over(self, testdir):
|
||||
p = testdir.makepyfile("""
|
||||
def test_func1():
|
||||
|
|
@ -226,6 +183,7 @@ class TestPerTestCapturing:
|
|||
assert "in func1" not in s
|
||||
assert "in func2" in s
|
||||
|
||||
|
||||
def test_teardown_capturing(self, testdir):
|
||||
p = testdir.makepyfile("""
|
||||
def setup_function(function):
|
||||
|
|
@ -286,14 +244,13 @@ class TestPerTestCapturing:
|
|||
"2",
|
||||
])
|
||||
|
||||
|
||||
class TestLoggingInteraction:
|
||||
def test_logging_stream_ownership(self, testdir):
|
||||
p = testdir.makepyfile("""
|
||||
def test_logging():
|
||||
import logging
|
||||
import pytest
|
||||
stream = capture.TextIO()
|
||||
stream = py.io.TextIO()
|
||||
logging.basicConfig(stream=stream)
|
||||
stream.close() # to free memory/release resources
|
||||
""")
|
||||
|
|
@ -363,8 +320,7 @@ class TestLoggingInteraction:
|
|||
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([
|
||||
|
|
@ -505,7 +461,6 @@ def test_setup_failure_does_not_kill_capturing(testdir):
|
|||
"*1 error*"
|
||||
])
|
||||
|
||||
|
||||
def test_fdfuncarg_skips_on_no_osdup(testdir):
|
||||
testdir.makepyfile("""
|
||||
import os
|
||||
|
|
@ -519,7 +474,6 @@ def test_fdfuncarg_skips_on_no_osdup(testdir):
|
|||
"*1 skipped*"
|
||||
])
|
||||
|
||||
|
||||
def test_capture_conftest_runtest_setup(testdir):
|
||||
testdir.makeconftest("""
|
||||
def pytest_runtest_setup():
|
||||
|
|
@ -530,7 +484,6 @@ def test_capture_conftest_runtest_setup(testdir):
|
|||
assert result.ret == 0
|
||||
assert 'hello19' not in result.stdout.str()
|
||||
|
||||
|
||||
def test_capture_early_option_parsing(testdir):
|
||||
testdir.makeconftest("""
|
||||
def pytest_runtest_setup():
|
||||
|
|
@ -541,10 +494,9 @@ def test_capture_early_option_parsing(testdir):
|
|||
assert result.ret == 0
|
||||
assert 'hello19' in result.stdout.str()
|
||||
|
||||
|
||||
@pytest.mark.xfail(sys.version_info >= (3, 0), reason='encoding issues')
|
||||
def test_capture_binary_output(testdir):
|
||||
testdir.makepyfile(r"""
|
||||
testdir.makepyfile("""
|
||||
import pytest
|
||||
|
||||
def test_a():
|
||||
|
|
@ -564,490 +516,3 @@ def test_capture_binary_output(testdir):
|
|||
'*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
|
||||
|
||||
|
||||
@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()
|
||||
print ("hx\xc4\x85\xc4\x87")
|
||||
out, err = cap.readouterr()
|
||||
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
|
||||
|
||||
|
||||
@needsosdup
|
||||
def test_stdcapture_fd_tmpfile(tmpfile):
|
||||
capfd = capture.StdCaptureFD(out=tmpfile)
|
||||
os.write(1, "hello".encode("ascii"))
|
||||
os.write(2, "world".encode("ascii"))
|
||||
outf, errf = capfd.done()
|
||||
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()
|
||||
|
||||
|
||||
@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()
|
||||
|
||||
|
||||
@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")
|
||||
|
||||
|
||||
@needsosdup
|
||||
@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)
|
||||
cap.startall()
|
||||
capfile = cap.err.tmpfile
|
||||
cap.suspend()
|
||||
cap.resume()
|
||||
capfile2 = cap.err.tmpfile
|
||||
assert capfile2 == capfile
|
||||
|
||||
|
||||
@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()
|
||||
|
|
|
|||
Loading…
Reference in New Issue