On 28/10/2009, at 19:44 , holger krekel holger-at-merlinux.eu |py-dev + execnet-dev| wrote: Hi Holger,I started looking into this a bit but ran into a problem when Istarted to probe the py.io module.I tried running the py.io examples on the website <http://codespeak.net/py/dist/io.htmlbut when I run the py.io.StdCaptureFD example, python dies on thesecond line:stak...@okum:~$ pythonPython 2.6.2 (release26-maint, Apr 19 2009, 01:56:41)[GCC 4.3.3] on linux2Type "help", "copyright", "credits" or "license" for more information.History restored: /auto/home/stakita/.pyhistory, Max Length: 500import py, syscapture = py.io.StdCaptureFD() My guess is that an exception is being thrown, but the capturedsys.stderr descriptor is not passing any meaningful traceback. Thanks for the suggestion, and here is a first pass. Attached is a modified version of pytest_capture.py from the 1.0.2 release. In this file, I just push the options into the CaptureManager so that the command line args are available. This does mean that the config file options stuff is not hooked up, but at least it works from the command line at the moment. Additional plugin command line options are: --log-file <filename> When used with the two capture options ('fd' and 'sys') this will enable pushing the stdout and stderr writes to the specified file. --append-log This allows appending to the file as opposed to the default clobber behaviour. Does this fit with your model of how the plugin should be structured? Regards, Simon |
""" 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
option_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
print "*******************************************************************************"
print "WARNING: USING LOCAL COPY OF pytest_capture.py"
print "*******************************************************************************"
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.")
group._addoption('--iocapturelog', action="store", default=None,
dest="capture_file", help="File for logging test output.")
group._addoption('--log-file', action="store", default=None,
dest="capture_file", help="File for logging test output.")
group._addoption('--append-log', action="store_true", default=False,
dest="append_log", help="Select appending to log file. Default is clobber.")
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(config.option), 'capturemanager')
class CaptureLogFile(object):
def __init__(self, _stream, encoding, logfd):
self.fd = logfd
self._stream = _stream
self.encoding = encoding
def write(self, obj):
if isinstance(obj, unicode):
self.fd.write(obj.encode(self.encoding))
self.fd.flush()
self._stream.write(obj.encode(self.encoding))
else:
self.fd.write(obj.encode(self.encoding))
self.fd.flush()
self._stream.write(obj)
def writelines(self, linelist):
data = ''.join(linelist)
self.write(data)
def __getattr__(self, name):
return getattr(self._stream, name)
class CaptureManager:
def __init__(self, option):
self._method2capture = {}
self.option = option
def _maketempfile(self):
f = py.std.tempfile.TemporaryFile()
newf = py.io.dupfile(f)
encoding = getattr(newf, 'encoding', None) or "UTF-8"
return EncodedFile(newf, encoding)
def _makestringio(self):
return py.std.StringIO.StringIO()
def _startcapture(self, method):
if self.option.append_log:
opt_string = 'a+'
else:
opt_string = 'w+'
# Here we use the same log file descriptor for both stderr and stdout.
self.logfd = open(self.option.capture_file, opt_string)
if self.option.capture_file:
# We wrap the two tempfiles so that they work the same as before, but log file
# data is syphoned off the write calls to the temp files.
outlog = CaptureLogFile(self._maketempfile(), "UTF-8", self.logfd)
errlog = CaptureLogFile(self._maketempfile(), "UTF-8", self.logfd)
else:
outlog = self._maketempfile()
errlog = self._maketempfile()
if method == "fd":
return py.io.StdCaptureFD(
out=outlog, err=errlog
)
elif method == "sys":
return py.io.StdCapture(
out=outlog, err=errlog
)
else:
raise ValueError("unknown capturing method: %r" % method)
def _getmethod(self, config, fspath):
if config.option.capture:
return config.option.capture
try:
return config._conftest.rget("option_capture", path=fspath)
except KeyError:
return "fd"
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, __multicall__, collector):
method = self._getmethod(collector.config, collector.fspath)
self.resumecapture(method)
try:
rep = __multicall__.execute()
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, __multicall__, session):
method = self._getmethod(session.config, None)
self.resumecapture(method)
try:
rep = __multicall__.execute()
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, __multicall__, item, call):
self.deactivate_funcargs()
rep = __multicall__.execute()
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
class EncodedFile(object):
def __init__(self, _stream, encoding):
self._stream = _stream
self.encoding = encoding
def write(self, obj):
if isinstance(obj, unicode):
self._stream.write(obj.encode(self.encoding))
else:
self._stream.write(obj)
def writelines(self, linelist):
data = ''.join(linelist)
self.write(data)
def __getattr__(self, name):
return getattr(self._stream, name)
_______________________________________________ py-dev mailing list py-dev@codespeak.net http://codespeak.net/mailman/listinfo/py-dev
