1 new commit in pytest:

https://bitbucket.org/hpk42/pytest/commits/5ff86fc01cf5/
Changeset:   5ff86fc01cf5
User:        hpk42
Date:        2014-01-22 22:15:40
Summary:     backing out Ronny's PR because it was merged too early (still has 
failing tests)
Affected #:  3 files

diff -r 3ba17bae85389f17caa7a3bf0db73a09c41d0b12 -r 
5ff86fc01cf5c22ff19cd9409af677f3a411c6cc CHANGELOG
--- a/CHANGELOG
+++ b/CHANGELOG
@@ -15,12 +15,6 @@
   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 -r 3ba17bae85389f17caa7a3bf0db73a09c41d0b12 -r 
5ff86fc01cf5c22ff19cd9409af677f3a411c6cc _pytest/capture.py
--- 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 @@
         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 @@
         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 @@
 
     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 @@
                 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 @@
         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 @@
 
 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 @@
     """
     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 @@
         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 -r 3ba17bae85389f17caa7a3bf0db73a09c41d0b12 -r 
5ff86fc01cf5c22ff19cd9409af677f3a411c6cc testing/test_capture.py
--- 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 @@
     @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 @@
 
     @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 @@
         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 @@
         "*1 passed*"
     ])
 
-
 @pytest.mark.parametrize("method", ['fd', 'sys'])
 def test_capturing_bytes_in_utf8_encoding(testdir, method):
     testdir.makepyfile("""
@@ -153,7 +111,6 @@
         "*1 passed*"
     ])
 
-
 def test_collect_capturing(testdir):
     p = testdir.makepyfile("""
         print ("collect %s failure" % 13)
@@ -165,7 +122,6 @@
         "*collect 13 failure*",
     ])
 
-
 class TestPerTestCapturing:
     def test_capture_and_fixtures(self, testdir):
         p = testdir.makepyfile("""
@@ -213,6 +169,7 @@
             "in teardown*",
         ])
 
+
     def test_no_carry_over(self, testdir):
         p = testdir.makepyfile("""
             def test_func1():
@@ -226,6 +183,7 @@
         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 @@
             "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 @@
                 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 @@
         "*1 error*"
     ])
 
-
 def test_fdfuncarg_skips_on_no_osdup(testdir):
     testdir.makepyfile("""
         import os
@@ -519,7 +474,6 @@
         "*1 skipped*"
     ])
 
-
 def test_capture_conftest_runtest_setup(testdir):
     testdir.makeconftest("""
         def pytest_runtest_setup():
@@ -530,7 +484,6 @@
     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 @@
     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 @@
     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()
+

Repository URL: https://bitbucket.org/hpk42/pytest/

--

This is a commit notification from bitbucket.org. You are receiving
this because you have the service enabled, addressing the recipient of
this email.
_______________________________________________
pytest-commit mailing list
pytest-commit@python.org
https://mail.python.org/mailman/listinfo/pytest-commit

Reply via email to