From 9aaf0fd340054004a33bf03eedfb8a8bcb60bd52 Mon Sep 17 00:00:00 2001 From: holger krekel Date: Wed, 22 Jan 2014 22:15:40 +0100 Subject: [PATCH] backing out Ronny's PR because it was merged too early (still has failing tests) --- CHANGELOG | 6 - _pytest/capture.py | 443 ++------------------------------ testing/test_capture.py | 555 +--------------------------------------- 3 files changed, 34 insertions(+), 970 deletions(-) diff --git a/CHANGELOG b/CHANGELOG index 05172a144..b0148b4f5 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -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 ----------------------------------- diff --git a/_pytest/capture.py b/_pytest/capture.py index bd5c010e4..be3194532 100644 --- a/_pytest/capture.py +++ b/_pytest/capture.py @@ -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() @@ -149,24 +98,23 @@ class CaptureManager: method = config._conftest.rget("option_capture", path=fspath) except KeyError: method = "fd" - if method == "fd" and not hasattr(os, 'dup'): # e.g. jython + if method == "fd" and not hasattr(os, 'dup'): # e.g. jython method = "sys" 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): method = self._getmethod(item.config, item.fspath) if not hasattr(item, 'outerr'): - item.outerr = ('', '') # we accumulate outerr on the item + item.outerr = ('', '') # we accumulate outerr on the item return self.resumecapture(method) 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 - # to allow for more lightweight recursive capturing - return + return # recursive collect, XXX refactor capturing + # to allow for more lightweight recursive capturing 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 diff --git a/testing/test_capture.py b/testing/test_capture.py index 9ffe7a790..cd1f2d865 100644 --- a/testing/test_capture.py +++ b/testing/test_capture.py @@ -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,11 +80,10 @@ 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): - if sys.version_info >= (3, 0): + if sys.version_info >= (3,0): obj = "'b\u00f6y'" else: obj = "u'\u00f6y'" @@ -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(): @@ -563,491 +515,4 @@ def test_capture_binary_output(testdir): result.stdout.fnmatch_lines([ '*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() +