floraachy/pytest2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

* reworked capturing to only capture once per runtest cycle

Browse Source 
* added readouterr() method to py.io capturing helpers

--HG--
branch : 1.0.x
This commit is contained in:
holger krekel
2009-07-31 14:21:02 +02:00
parent 2514b8faaf
commit be949f4037
29 changed files with 842 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
py/test/plugin/pytest_default.py
View File

@@ -81,9 +81,6 @@ def pytest_addoption(parser):
help="don't cut any tracebacks (default is to cut).")
group.addoption('--basetemp', dest="basetemp", default=None, metavar="dir",
help="base temporary directory for this test run.")
group._addoption('--iocapture', action="store", default="fd", metavar="method",
type="choice", choices=['fd', 'sys', 'no'],
help="set iocapturing method: fd|sys|no.")
group.addoption('--debug',
action="store_true", dest="debug", default=False,
help="generate and show debugging information.")

266
py/test/plugin/pytest_iocapture.py
View File

@@ -1,60 +1,97 @@
"""
convenient capturing of writes to stdout/stderror streams and file descriptors.
configurable per-test stdout/stderr capturing mechanisms.
Example Usage
----------------------
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.
You can use the `capsys funcarg`_ to capture writes
to stdout and stderr streams by using it in a test
likes this:
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.reset()
out, err = capsys.readouterr()
assert out == "hello\\n"
assert err == "world\\n"
print "next"
out, err = capsys.reset()
out, err = capsys.readouterr()
assert out == "next\\n"
The ``reset()`` call returns a tuple and will restart
capturing so that you can successively check for output.
After the test function finishes the original streams
will be restored.
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_true", dest="nocapture", default=False,
help="disable catching of stdout/stderr during test run.")
group._addoption('-s', action="store_const", const="no", dest="capture",
help="shortcut for --capture=no.")
group._addoption('--capture', action="store", default=None,
metavar="capture", type="choice", choices=['fd', 'sys', 'no'],
help="set IO capturing method during tests: sys|fd|no.")
def determine_capturing(config, path=None):
iocapture = config.getvalue("iocapture", path=path)
if iocapture == "fd":
return py.io.StdCaptureFD()
elif iocapture == "sys":
return py.io.StdCapture()
elif iocapture == "no":
return py.io.StdCapture(out=False, err=False, in_=False)
else:
# how to raise errors here?
raise config.Error("unknown io capturing: " + iocapture)
def pytest_make_collect_report(__call__, collector):
cap = determine_capturing(collector.config, collector.fspath)
try:
rep = __call__.execute(firstresult=True)
finally:
outerr = cap.reset()
addouterr(rep, outerr)
return rep
def addouterr(rep, outerr):
repr = getattr(rep, 'longrepr', None)
if not hasattr(repr, 'addsection'):
@@ -64,79 +101,140 @@ def addouterr(rep, outerr):
repr.addsection("Captured std%s" % secname, content.rstrip())
def pytest_configure(config):
if not config.option.nocapture:
config.pluginmanager.register(CapturePerTest())
config.pluginmanager.register(CaptureManager(), 'capturemanager')
class CapturePerTest:
class CaptureManager:
def __init__(self):
self.item2capture = {}
def _setcapture(self, item):
assert item not in self.item2capture
cap = determine_capturing(item.config, path=item.fspath)
self.item2capture[item] = cap
self._method2capture = {}
def _startcapture(self, method):
if method == "fd":
return py.io.StdCaptureFD()
elif method == "sys":
return py.io.StdCapture()
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._setcapture(item)
self.resumecapture_item(item)
def pytest_runtest_call(self, item):
self._setcapture(item)
self.resumecapture_item(item)
self.activate_funcargs(item)
def pytest_runtest_teardown(self, item):
self._setcapture(item)
self.resumecapture_item(item)
def pytest_keyboard_interrupt(self, excinfo):
for cap in self.item2capture.values():
cap.reset()
self.item2capture.clear()
if hasattr(self, '_capturing'):
self.suspendcapture()
def pytest_runtest_makereport(self, __call__, item, call):
capture = self.item2capture.pop(item)
outerr = capture.reset()
# XXX shift reporting elsewhere
self.deactivate_funcargs()
rep = __call__.execute(firstresult=True)
addouterr(rep, outerr)
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.reset()`` method
which returns a ``(out, err)`` tuple of captured strings.
them available successively via a ``capsys.readouterr()`` method
which returns a ``(out, err)`` tuple of captured snapshot strings.
"""
capture = CaptureFuncarg(py.io.StdCapture)
request.addfinalizer(capture.finalize)
return capture
return CaptureFuncarg(request, py.io.StdCapture)
def pytest_funcarg__capfd(request):
"""captures writes to file descriptors 1 and 2 and makes
them available successively via a ``capsys.reset()`` method
which returns a ``(out, err)`` tuple of captured strings.
snapshotted ``(out, err)`` string tuples available
via the ``capsys.readouterr()`` method.
"""
capture = CaptureFuncarg(py.io.StdCaptureFD)
request.addfinalizer(capture.finalize)
return capture
return CaptureFuncarg(request, py.io.StdCaptureFD)
def pytest_pyfunc_call(pyfuncitem):
if hasattr(pyfuncitem, 'funcargs'):
for funcarg, value in pyfuncitem.funcargs.items():
if funcarg == "capsys" or funcarg == "capfd":
value.reset()
class CaptureFuncarg:
_capture = None
def __init__(self, captureclass):
self._captureclass = captureclass
def __init__(self, request, captureclass):
self._cclass = captureclass
#request.addfinalizer(self._finalize)
def finalize(self):
if self._capture:
self._capture.reset()
def _start(self):
self.capture = self._cclass()
def reset(self):
res = None
if self._capture:
res = self._capture.reset()
self._capture = self._captureclass()
return res
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

7
py/test/plugin/pytest_pdb.py
View File

@@ -23,11 +23,18 @@ def pytest_configure(config):
class PdbInvoke:
def pytest_runtest_makereport(self, item, call):
if call.excinfo and not call.excinfo.errisinstance(Skipped):
# XXX hack hack hack to play well with capturing
capman = item.config.pluginmanager.impname2plugin['capturemanager']
capman.suspendcapture()
tw = py.io.TerminalWriter()
repr = call.excinfo.getrepr()
repr.toterminal(tw)
post_mortem(call.excinfo._excinfo[2])
# XXX hack end
capman.resumecapture_item(item)
class Pdb(py.std.pdb.Pdb):
def do_list(self, arg):
self.lastcmd = 'list'

13
py/test/plugin/pytest_pytester.py
View File

@@ -301,6 +301,9 @@ class TmpTestdir:
assert script.check()
return py.std.sys.executable, script
def runpython(self, script):
return self.run(py.std.sys.executable, script)
def runpytest(self, *args):
p = py.path.local.make_numbered_dir(prefix="runpytest-",
keep=None, rootdir=self.tmpdir)
@@ -520,7 +523,6 @@ def test_testdir_runs_with_plugin(testdir):
def pytest_funcarg__venv(request):
p = request.config.mktemp(request.function.__name__, numbered=True)
venv = VirtualEnv(str(p))
venv.create()
return venv
def pytest_funcarg__py_setup(request):
@@ -533,6 +535,7 @@ def pytest_funcarg__py_setup(request):
class SetupBuilder:
def __init__(self, setup_path):
self.setup_path = setup_path
assert setup_path.check()
def make_sdist(self, destdir=None):
temp = py.path.local.mkdtemp()
@@ -567,9 +570,9 @@ class VirtualEnv(object):
def _cmd(self, name):
return os.path.join(self.path, 'bin', name)
@property
def valid(self):
return os.path.exists(self._cmd('python'))
def ensure(self):
if not os.path.exists(self._cmd('python')):
self.create()
def create(self, sitepackages=False):
args = ['virtualenv', self.path]
@@ -582,7 +585,7 @@ class VirtualEnv(object):
return py.execnet.makegateway("popen//python=%s" %(python,))
def pcall(self, cmd, *args, **kw):
assert self.valid
self.ensure()
return subprocess.call([
self._cmd(cmd)
] + list(args),

3
py/test/plugin/pytest_terminal.py
View File

@@ -314,7 +314,8 @@ class TerminalReporter:
self.write_sep("_", msg)
if hasattr(rep, 'node'):
self.write_line(self.gateway2info.get(
rep.node.gateway, "node %r (platinfo not found? strange)")
rep.node.gateway,
"node %r (platinfo not found? strange)")
[:self._tw.fullwidth-1])
rep.toterminal(self._tw)

363
py/test/plugin/test_pytest_iocapture.py Normal file
View File

@@ -0,0 +1,363 @@
import py, os, sys
from py.__.test.plugin.pytest_iocapture import CaptureManager
class TestCaptureManager:
def test_configure_per_fspath(self, testdir):
config = testdir.parseconfig(testdir.tmpdir)
assert config.getvalue("capture") is None
capman = CaptureManager()
assert capman._getmethod(config, None) == "fd" # default
for name in ('no', 'fd', 'sys'):
sub = testdir.tmpdir.mkdir("dir" + name)
sub.ensure("__init__.py")
sub.join("conftest.py").write('conf_capture = %r' % name)
assert capman._getmethod(config, sub.join("test_hello.py")) == name
@py.test.mark.multi(method=['no', 'fd', 'sys'])
def test_capturing_basic_api(self, method):
capouter = py.io.StdCaptureFD()
old = sys.stdout, sys.stderr, sys.stdin
try:
capman = CaptureManager()
capman.resumecapture(method)
print "hello"
out, err = capman.suspendcapture()
if method == "no":
assert old == (sys.stdout, sys.stderr, sys.stdin)
else:
assert out == "hello\n"
capman.resumecapture(method)
out, err = capman.suspendcapture()
assert not out and not err
finally:
capouter.reset()
def test_juggle_capturings(self, testdir):
capouter = py.io.StdCaptureFD()
try:
config = testdir.parseconfig(testdir.tmpdir)
capman = CaptureManager()
capman.resumecapture("fd")
py.test.raises(ValueError, 'capman.resumecapture("fd")')
py.test.raises(ValueError, 'capman.resumecapture("sys")')
os.write(1, "hello\n")
out, err = capman.suspendcapture()
assert out == "hello\n"
capman.resumecapture("sys")
os.write(1, "hello\n")
print >>sys.stderr, "world"
out, err = capman.suspendcapture()
assert not out
assert err == "world\n"
finally:
capouter.reset()
def test_collect_capturing(testdir):
p = testdir.makepyfile("""
print "collect %s failure" % 13
import xyz42123
""")
result = testdir.runpytest(p)
result.stdout.fnmatch_lines([
"*Captured stdout*",
"*collect 13 failure*",
])
class TestPerTestCapturing:
def test_capture_and_fixtures(self, testdir):
p = testdir.makepyfile("""
def setup_module(mod):
print "setup module"
def setup_function(function):
print "setup", function.__name__
def test_func1():
print "in func1"
assert 0
def test_func2():
print "in func2"
assert 0
""")
result = testdir.runpytest(p)
result.stdout.fnmatch_lines([
"setup module*",
"setup test_func1*",
"in func1*",
"setup test_func2*",
"in func2*",
])
def test_no_carry_over(self, testdir):
p = testdir.makepyfile("""
def test_func1():
print "in func1"
def test_func2():
print "in func2"
assert 0
""")
result = testdir.runpytest(p)
s = result.stdout.str()
assert "in func1" not in s
assert "in func2" in s
def test_teardown_capturing(self, testdir):
p = testdir.makepyfile("""
def setup_function(function):
print "setup func1"
def teardown_function(function):
print "teardown func1"
assert 0
def test_func1():
print "in func1"
pass
""")
result = testdir.runpytest(p)
assert result.stdout.fnmatch_lines([
'*teardown_function*',
'*Captured stdout*',
"setup func1*",
"in func1*",
"teardown func1*",
#"*1 fixture failure*"
])
@py.test.mark.xfail
def test_teardown_final_capturing(self, testdir):
p = testdir.makepyfile("""
def teardown_module(mod):
print "teardown module"
assert 0
def test_func():
pass
""")
result = testdir.runpytest(p)
assert result.stdout.fnmatch_lines([
"teardown module*",
#"*1 fixture failure*"
])
def test_capturing_outerr(self, testdir):
p1 = testdir.makepyfile("""
import sys
def test_capturing():
print 42
print >>sys.stderr, 23
def test_capturing_error():
print 1
print >>sys.stderr, 2
raise ValueError
""")
result = testdir.runpytest(p1)
result.stdout.fnmatch_lines([
"*test_capturing_outerr.py .F",
"====* FAILURES *====",
"____*____",
"*test_capturing_outerr.py:8: ValueError",
"*--- Captured stdout ---*",
"1",
"*--- Captured stderr ---*",
"2",
])
class TestLoggingInteraction:
def test_logging_stream_ownership(self, testdir):
p = testdir.makepyfile("""
def test_logging():
import logging
import StringIO
stream = StringIO.StringIO()
logging.basicConfig(stream=stream)
stream.close() # to free memory/release resources
""")
result = testdir.runpytest(p)
result.stderr.str().find("atexit") == -1
def test_capturing_and_logging_fundamentals(self, testdir):
# here we check a fundamental feature
rootdir = str(py.path.local(py.__file__).dirpath().dirpath())
p = testdir.makepyfile("""
import sys
sys.path.insert(0, %r)
import py, logging
cap = py.io.StdCaptureFD(out=False, in_=False)
logging.warn("hello1")
outerr = cap.suspend()
print "suspeneded and captured", outerr
logging.warn("hello2")
cap.resume()
logging.warn("hello3")
outerr = cap.suspend()
print "suspend2 and captured", outerr
""" % rootdir)
result = testdir.runpython(p)
assert result.stdout.fnmatch_lines([
"suspeneded and captured*hello1*",
"suspend2 and captured*hello2*WARNING:root:hello3*",
])
assert "atexit" not in result.stderr.str()
def test_logging_and_immediate_setupteardown(self, testdir):
p = testdir.makepyfile("""
import logging
def setup_function(function):
logging.warn("hello1")
def test_logging():
logging.warn("hello2")
assert 0
def teardown_function(function):
logging.warn("hello3")
assert 0
""")
for optargs in (('--capture=sys',), ('--capture=fd',)):
print optargs
result = testdir.runpytest(p, *optargs)
s = result.stdout.str()
result.stdout.fnmatch_lines([
"*WARN*hello1",
"*WARN*hello2",
"*WARN*hello3",
])
# verify proper termination
assert "closed" not in s
@py.test.mark.xfail
def test_logging_and_crossscope_fixtures(self, testdir):
# XXX also needs final teardown reporting to work!
p = testdir.makepyfile("""
import logging
def setup_module(function):
logging.warn("hello1")
def test_logging():
logging.warn("hello2")
assert 0
def teardown_module(function):
logging.warn("hello3")
assert 0
""")
for optargs in (('--iocapture=sys',), ('--iocapture=fd',)):
print optargs
result = testdir.runpytest(p, *optargs)
s = result.stdout.str()
result.stdout.fnmatch_lines([
"*WARN*hello1",
"*WARN*hello2",
"*WARN*hello3",
])
# verify proper termination
assert "closed" not in s
class TestCaptureFuncarg:
def test_std_functional(self, testdir):
reprec = testdir.inline_runsource("""
def test_hello(capsys):
print 42
out, err = capsys.readouterr()
assert out.startswith("42")
""")
reprec.assertoutcome(passed=1)
def test_stdfd_functional(self, testdir):
reprec = testdir.inline_runsource("""
def test_hello(capfd):
import os
os.write(1, "42")
out, err = capfd.readouterr()
assert out.startswith("42")
capfd.close()
""")
reprec.assertoutcome(passed=1)
def test_partial_setup_failure(self, testdir):
p = testdir.makepyfile("""
def test_hello(capfd, missingarg):
pass
""")
result = testdir.runpytest(p)
assert result.stdout.fnmatch_lines([
"*test_partial_setup_failure*",
"*1 failed*",
])
def test_keyboardinterrupt_disables_capturing(self, testdir):
p = testdir.makepyfile("""
def test_hello(capfd):
import os
os.write(1, "42")
raise KeyboardInterrupt()
""")
result = testdir.runpytest(p)
result.stdout.fnmatch_lines([
"*KEYBOARD INTERRUPT*"
])
assert result.ret == 2
class TestFixtureReporting:
@py.test.mark.xfail
def test_setup_fixture_error(self, testdir):
p = testdir.makepyfile("""
def setup_function(function):
print "setup func"
assert 0
def test_nada():
pass
""")
result = testdir.runpytest()
result.stdout.fnmatch_lines([
"*FIXTURE ERROR at setup of test_nada*",
"*setup_function(function):*",
"*setup func*",
"*assert 0*",
"*0 passed*1 error*",
])
assert result.ret != 0
@py.test.mark.xfail
def test_teardown_fixture_error(self, testdir):
p = testdir.makepyfile("""
def test_nada():
pass
def teardown_function(function):
print "teardown func"
assert 0
""")
result = testdir.runpytest()
result.stdout.fnmatch_lines([
"*FIXTURE ERROR at teardown*",
"*teardown_function(function):*",
"*teardown func*",
"*assert 0*",
"*1 passed*1 error*",
])
@py.test.mark.xfail
def test_teardown_fixture_error_and_test_failure(self, testdir):
p = testdir.makepyfile("""
def test_fail():
assert 0, "failingfunc"
def teardown_function(function):
print "teardown func"
assert 0
""")
result = testdir.runpytest()
result.stdout.fnmatch_lines([
"*failingfunc*",
"*FIXTURE ERROR at teardown*",
"*teardown_function(function):*",
"*teardown func*",
"*assert 0*",
"*1 failed*1 error",
])