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

simplify capturing funcarg handling

This commit is contained in:
holger krekel 2014-03-28 07:13:08 +01:00
parent a8f4f49a82
commit 859915dc5e
1 changed files with 23 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
_pytest/capture.py
View File

@ -155,29 +155,23 @@ class CaptureManager:
def suspendcapture(self, item=None): def suspendcapture(self, item=None):
self.deactivate_funcargs() self.deactivate_funcargs()
if hasattr(self, '_capturing'): method = self.__dict__.pop("_capturing", None)
method = self._capturing if method is not None:
del self._capturing
cap = self._method2capture.get(method) cap = self._method2capture.get(method)
if cap is not None: if cap is not None:
return cap.readouterr() return cap.readouterr()
return "", "" return "", ""
def activate_funcargs(self, pyfuncitem): def activate_funcargs(self, pyfuncitem):
funcargs = getattr(pyfuncitem, "funcargs", None) capfuncarg = pyfuncitem.__dict__.pop("_capfuncarg", None)
if funcargs is not None: if capfuncarg is not None:
for name, capfuncarg in funcargs.items(): capfuncarg._start()
if name in ('capsys', 'capfd'): self._capfuncarg = capfuncarg
assert not hasattr(self, '_capturing_funcarg')
self._capturing_funcarg = capfuncarg
capfuncarg._start()
def deactivate_funcargs(self): def deactivate_funcargs(self):
capturing_funcarg = getattr(self, '_capturing_funcarg', None) capfuncarg = self.__dict__.pop("_capfuncarg", None)
if capturing_funcarg: if capfuncarg is not None:
outerr = capturing_funcarg._finalize() capfuncarg.close()
del self._capturing_funcarg
return outerr
@pytest.mark.hookwrapper @pytest.mark.hookwrapper
def pytest_make_collect_report(self, __multicall__, collector): def pytest_make_collect_report(self, __multicall__, collector):
@ -210,7 +204,9 @@ class CaptureManager:
@pytest.mark.hookwrapper @pytest.mark.hookwrapper
def pytest_runtest_call(self, item): def pytest_runtest_call(self, item):
with self.item_capture_wrapper(item, "call"): with self.item_capture_wrapper(item, "call"):
self.activate_funcargs(item)
yield yield
#self.deactivate_funcargs() called from ctx's suspendcapture()
@pytest.mark.hookwrapper @pytest.mark.hookwrapper
def pytest_runtest_teardown(self, item): def pytest_runtest_teardown(self, item):
@ -228,17 +224,8 @@ class CaptureManager:
@contextlib.contextmanager @contextlib.contextmanager
def item_capture_wrapper(self, item, when): def item_capture_wrapper(self, item, when):
self.resumecapture_item(item) self.resumecapture_item(item)
if when == "call": yield
self.activate_funcargs(item)
yield
funcarg_outerr = self.deactivate_funcargs()
else:
yield
funcarg_outerr = None
out, err = self.suspendcapture(item) out, err = self.suspendcapture(item)
if funcarg_outerr is not None:
out += funcarg_outerr[0]
err += funcarg_outerr[1]
item.add_report_section(when, "out", out) item.add_report_section(when, "out", out)
item.add_report_section(when, "err", err) item.add_report_section(when, "err", err)
@ -252,7 +239,8 @@ def pytest_funcarg__capsys(request):
""" """
if "capfd" in request._funcargs: if "capfd" in request._funcargs:
raise request.raiseerror(error_capsysfderror) raise request.raiseerror(error_capsysfderror)
return CaptureFixture(SysCapture) request.node._capfuncarg = c = CaptureFixture(SysCapture)
return c
def pytest_funcarg__capfd(request): def pytest_funcarg__capfd(request):
"""enables capturing of writes to file descriptors 1 and 2 and makes """enables capturing of writes to file descriptors 1 and 2 and makes
@ -263,7 +251,8 @@ def pytest_funcarg__capfd(request):
request.raiseerror(error_capsysfderror) request.raiseerror(error_capsysfderror)
if not hasattr(os, 'dup'): if not hasattr(os, 'dup'):
pytest.skip("capfd funcarg needs os.dup") pytest.skip("capfd funcarg needs os.dup")
return CaptureFixture(FDCapture) request.node._capfuncarg = c = CaptureFixture(FDCapture)
return c
class CaptureFixture: class CaptureFixture:
@ -275,21 +264,17 @@ class CaptureFixture:
Capture=self.captureclass) Capture=self.captureclass)
self._capture.start_capturing() self._capture.start_capturing()
def _finalize(self): def close(self):
if hasattr(self, '_capture'): cap = self.__dict__.pop("_capture", None)
outerr = self._outerr = self._capture.stop_capturing() if cap is not None:
del self._capture cap.pop_outerr_to_orig()
return outerr cap.stop_capturing()
def readouterr(self): def readouterr(self):
try: try:
return self._capture.readouterr() return self._capture.readouterr()
except AttributeError: except AttributeError:
return self._outerr return "", ""
def close(self):
self._finalize()
def dupfile(f, mode=None, buffering=0, raising=False, encoding=None): def dupfile(f, mode=None, buffering=0, raising=False, encoding=None):
@ -448,8 +433,7 @@ class FDCapture:
self.tmpfile.close() self.tmpfile.close()
def writeorg(self, data): def writeorg(self, data):
""" write a string to the original file descriptor """ write to original file descriptor. """
"""
if py.builtin._istext(data): if py.builtin._istext(data):
data = data.encode("utf8") # XXX use encoding of original stream data = data.encode("utf8") # XXX use encoding of original stream
os.write(self._savefd, data) os.write(self._savefd, data)