278 lines
9.1 KiB
Python
278 lines
9.1 KiB
Python
"""
|
|
configurable per-test stdout/stderr capturing mechanisms.
|
|
|
|
This plugin captures stdout/stderr output for each test separately.
|
|
In case of test failures this captured output is shown grouped
|
|
togtther with the test.
|
|
|
|
The plugin also provides test function arguments that help to
|
|
assert stdout/stderr output from within your tests, see the
|
|
`funcarg example`_.
|
|
|
|
|
|
Capturing of input/output streams during tests
|
|
---------------------------------------------------
|
|
|
|
By default ``sys.stdout`` and ``sys.stderr`` are substituted with
|
|
temporary streams during the execution of tests and setup/teardown code.
|
|
During the whole testing process it will re-use the same temporary
|
|
streams allowing to play well with the logging module which easily
|
|
takes ownership on these streams.
|
|
|
|
Also, 'sys.stdin' is substituted with a file-like "null" object that
|
|
does not return any values. This is to immediately error out
|
|
on tests that wait on reading something from stdin.
|
|
|
|
You can influence output capturing mechanisms from the command line::
|
|
|
|
py.test -s # disable all capturing
|
|
py.test --capture=sys # set StringIO() to each of sys.stdout/stderr
|
|
py.test --capture=fd # capture stdout/stderr on Filedescriptors 1/2
|
|
|
|
If you set capturing values in a conftest file like this::
|
|
|
|
# conftest.py
|
|
conf_capture = 'fd'
|
|
|
|
then all tests in that directory will execute with "fd" style capturing.
|
|
|
|
sys-level capturing
|
|
------------------------------------------
|
|
|
|
Capturing on 'sys' level means that ``sys.stdout`` and ``sys.stderr``
|
|
will be replaced with StringIO() objects.
|
|
|
|
FD-level capturing and subprocesses
|
|
------------------------------------------
|
|
|
|
The ``fd`` based method means that writes going to system level files
|
|
based on the standard file descriptors will be captured, for example
|
|
writes such as ``os.write(1, 'hello')`` will be captured properly.
|
|
Capturing on fd-level will include output generated from
|
|
any subprocesses created during a test.
|
|
|
|
.. _`funcarg example`:
|
|
|
|
Example Usage of the capturing Function arguments
|
|
---------------------------------------------------
|
|
|
|
You can use the `capsys funcarg`_ and `capfd funcarg`_ to
|
|
capture writes to stdout and stderr streams. Using the
|
|
funcargs frees your test from having to care about setting/resetting
|
|
the old streams and also interacts well with py.test's own
|
|
per-test capturing. Here is an example test function:
|
|
|
|
.. sourcecode:: python
|
|
|
|
def test_myoutput(capsys):
|
|
print "hello"
|
|
print >>sys.stderr, "world"
|
|
out, err = capsys.readouterr()
|
|
assert out == "hello\\n"
|
|
assert err == "world\\n"
|
|
print "next"
|
|
out, err = capsys.readouterr()
|
|
assert out == "next\\n"
|
|
|
|
The ``readouterr()`` call snapshots the output so far -
|
|
and capturing will be continued. After the test
|
|
function finishes the original streams will
|
|
be restored. If you want to capture on
|
|
the filedescriptor level you can use the ``capfd`` function
|
|
argument which offers the same interface.
|
|
"""
|
|
|
|
import py
|
|
|
|
def pytest_addoption(parser):
|
|
group = parser.getgroup("general")
|
|
group._addoption('-s', action="store_const", const="no", dest="capture",
|
|
help="shortcut for --capture=no.")
|
|
group._addoption('--capture', action="store", default=None,
|
|
metavar="method", type="choice", choices=['fd', 'sys', 'no'],
|
|
help="set capturing method during tests: fd (default)|sys|no.")
|
|
|
|
def addouterr(rep, outerr):
|
|
repr = getattr(rep, 'longrepr', None)
|
|
if not hasattr(repr, 'addsection'):
|
|
return
|
|
for secname, content in zip(["out", "err"], outerr):
|
|
if content:
|
|
repr.addsection("Captured std%s" % secname, content.rstrip())
|
|
|
|
def pytest_configure(config):
|
|
config.pluginmanager.register(CaptureManager(), 'capturemanager')
|
|
|
|
class CaptureManager:
|
|
def __init__(self):
|
|
self._method2capture = {}
|
|
|
|
def _maketempfile(self):
|
|
f = py.std.tempfile.TemporaryFile()
|
|
newf = py.io.dupfile(f)
|
|
f.close()
|
|
return ustream(newf)
|
|
|
|
def _makestringio(self):
|
|
return py.std.StringIO.StringIO()
|
|
|
|
def _startcapture(self, method):
|
|
if method == "fd":
|
|
return py.io.StdCaptureFD(
|
|
out=self._maketempfile(), err=self._maketempfile()
|
|
)
|
|
elif method == "sys":
|
|
return py.io.StdCapture(
|
|
out=self._makestringio(), err=self._makestringio()
|
|
)
|
|
else:
|
|
raise ValueError("unknown capturing method: %r" % method)
|
|
|
|
def _getmethod(self, config, fspath):
|
|
if config.option.capture:
|
|
return config.option.capture
|
|
return config._conftest.rget("conf_capture", path=fspath)
|
|
|
|
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
|
|
return self.resumecapture(method)
|
|
|
|
def resumecapture(self, method):
|
|
if hasattr(self, '_capturing'):
|
|
raise ValueError("cannot resume, already capturing with %r" %
|
|
(self._capturing,))
|
|
if method != "no":
|
|
cap = self._method2capture.get(method)
|
|
if cap is None:
|
|
cap = self._startcapture(method)
|
|
self._method2capture[method] = cap
|
|
else:
|
|
cap.resume()
|
|
self._capturing = method
|
|
|
|
def suspendcapture(self):
|
|
self.deactivate_funcargs()
|
|
method = self._capturing
|
|
if method != "no":
|
|
cap = self._method2capture[method]
|
|
outerr = cap.suspend()
|
|
else:
|
|
outerr = "", ""
|
|
del self._capturing
|
|
return outerr
|
|
|
|
def activate_funcargs(self, pyfuncitem):
|
|
if not hasattr(pyfuncitem, 'funcargs'):
|
|
return
|
|
assert not hasattr(self, '_capturing_funcargs')
|
|
l = []
|
|
for name, obj in pyfuncitem.funcargs.items():
|
|
if name in ('capsys', 'capfd'):
|
|
obj._start()
|
|
l.append(obj)
|
|
if l:
|
|
self._capturing_funcargs = l
|
|
|
|
def deactivate_funcargs(self):
|
|
if hasattr(self, '_capturing_funcargs'):
|
|
for capfuncarg in self._capturing_funcargs:
|
|
capfuncarg._finalize()
|
|
del self._capturing_funcargs
|
|
|
|
def pytest_make_collect_report(self, __call__, collector):
|
|
method = self._getmethod(collector.config, collector.fspath)
|
|
self.resumecapture(method)
|
|
try:
|
|
rep = __call__.execute(firstresult=True)
|
|
finally:
|
|
outerr = self.suspendcapture()
|
|
addouterr(rep, outerr)
|
|
return rep
|
|
|
|
def pytest_runtest_setup(self, item):
|
|
self.resumecapture_item(item)
|
|
|
|
def pytest_runtest_call(self, item):
|
|
self.resumecapture_item(item)
|
|
self.activate_funcargs(item)
|
|
|
|
def pytest_runtest_teardown(self, item):
|
|
self.resumecapture_item(item)
|
|
|
|
def pytest_runtest_teardown(self, item):
|
|
self.resumecapture_item(item)
|
|
|
|
def pytest__teardown_final(self, __call__, session):
|
|
method = self._getmethod(session.config, None)
|
|
self.resumecapture(method)
|
|
try:
|
|
rep = __call__.execute(firstresult=True)
|
|
finally:
|
|
outerr = self.suspendcapture()
|
|
if rep:
|
|
addouterr(rep, outerr)
|
|
return rep
|
|
|
|
def pytest_keyboard_interrupt(self, excinfo):
|
|
if hasattr(self, '_capturing'):
|
|
self.suspendcapture()
|
|
|
|
def pytest_runtest_makereport(self, __call__, item, call):
|
|
self.deactivate_funcargs()
|
|
rep = __call__.execute(firstresult=True)
|
|
outerr = self.suspendcapture()
|
|
outerr = (item.outerr[0] + outerr[0], item.outerr[1] + outerr[1])
|
|
if not rep.passed:
|
|
addouterr(rep, outerr)
|
|
if not rep.passed or rep.when == "teardown":
|
|
outerr = ('', '')
|
|
item.outerr = outerr
|
|
return rep
|
|
|
|
def pytest_funcarg__capsys(request):
|
|
"""captures writes to sys.stdout/sys.stderr and makes
|
|
them available successively via a ``capsys.readouterr()`` method
|
|
which returns a ``(out, err)`` tuple of captured snapshot strings.
|
|
"""
|
|
return CaptureFuncarg(request, py.io.StdCapture)
|
|
|
|
def pytest_funcarg__capfd(request):
|
|
"""captures writes to file descriptors 1 and 2 and makes
|
|
snapshotted ``(out, err)`` string tuples available
|
|
via the ``capsys.readouterr()`` method.
|
|
"""
|
|
return CaptureFuncarg(request, py.io.StdCaptureFD)
|
|
|
|
|
|
class CaptureFuncarg:
|
|
def __init__(self, request, captureclass):
|
|
self._cclass = captureclass
|
|
#request.addfinalizer(self._finalize)
|
|
|
|
def _start(self):
|
|
self.capture = self._cclass()
|
|
|
|
def _finalize(self):
|
|
if hasattr(self, 'capture'):
|
|
self.capture.reset()
|
|
del self.capture
|
|
|
|
def readouterr(self):
|
|
return self.capture.readouterr()
|
|
|
|
def close(self):
|
|
self.capture.reset()
|
|
del self.capture
|
|
|
|
def ustream(f):
|
|
import codecs
|
|
encoding = getattr(f, 'encoding', None) or "UTF-8"
|
|
reader = codecs.getreader(encoding)
|
|
writer = codecs.getwriter(encoding)
|
|
srw = codecs.StreamReaderWriter(f, reader, writer)
|
|
srw.encoding = encoding
|
|
return srw
|
|
|