intermediate changes

ref:cde9a383711a11544ce7e107a78147fb96cc4029

1 files changed, 853 insertions, 0 deletions

diff --git a/contrib/python/pytest/py3/_pytest/capture.py b/contrib/python/pytest/py3/_pytest/capture.py
new file mode 100644
index 0000000000..673bb07a9b
--- /dev/null
+++ b/contrib/python/pytest/py3/_pytest/capture.py
@@ -0,0 +1,853 @@
+"""
+per-test stdout/stderr capturing mechanism.
+
+"""
+import collections
+import contextlib
+import io
+import os
+import sys
+from io import UnsupportedOperation
+from tempfile import TemporaryFile
+from typing import BinaryIO
+from typing import Generator
+from typing import Iterable
+from typing import Optional
+
+import pytest
+from _pytest.compat import CaptureAndPassthroughIO
+from _pytest.compat import CaptureIO
+from _pytest.compat import TYPE_CHECKING
+from _pytest.config import Config
+from _pytest.fixtures import FixtureRequest
+
+if TYPE_CHECKING:
+    from typing_extensions import Literal
+
+    _CaptureMethod = Literal["fd", "sys", "no", "tee-sys"]
+
+patchsysdict = {0: "stdin", 1: "stdout", 2: "stderr"}
+
+
+def pytest_addoption(parser):
+    group = parser.getgroup("general")
+    group._addoption(
+        "--capture",
+        action="store",
+        default="fd" if hasattr(os, "dup") else "sys",
+        metavar="method",
+        choices=["fd", "sys", "no", "tee-sys"],
+        help="per-test capturing method: one of fd|sys|no|tee-sys.",
+    )
+    group._addoption(
+        "-s",
+        action="store_const",
+        const="no",
+        dest="capture",
+        help="shortcut for --capture=no.",
+    )
+
+
+@pytest.hookimpl(hookwrapper=True)
+def pytest_load_initial_conftests(early_config: Config):
+    ns = early_config.known_args_namespace
+    if ns.capture == "fd":
+        _py36_windowsconsoleio_workaround(sys.stdout)
+    _colorama_workaround()
+    _readline_workaround()
+    pluginmanager = early_config.pluginmanager
+    capman = CaptureManager(ns.capture)
+    pluginmanager.register(capman, "capturemanager")
+
+    # make sure that capturemanager is properly reset at final shutdown
+    early_config.add_cleanup(capman.stop_global_capturing)
+
+    # finally trigger conftest loading but while capturing (issue93)
+    capman.start_global_capturing()
+    outcome = yield
+    capman.suspend_global_capture()
+    if outcome.excinfo is not None:
+        out, err = capman.read_global_capture()
+        sys.stdout.write(out)
+        sys.stderr.write(err)
+
+
+def _get_multicapture(method: "_CaptureMethod") -> "MultiCapture":
+    if method == "fd":
+        return MultiCapture(out=True, err=True, Capture=FDCapture)
+    elif method == "sys":
+        return MultiCapture(out=True, err=True, Capture=SysCapture)
+    elif method == "no":
+        return MultiCapture(out=False, err=False, in_=False)
+    elif method == "tee-sys":
+        return MultiCapture(out=True, err=True, in_=False, Capture=TeeSysCapture)
+    raise ValueError("unknown capturing method: {!r}".format(method))
+
+
+class CaptureManager:
+    """
+    Capture plugin, manages that the appropriate capture method is enabled/disabled during collection and each
+    test phase (setup, call, teardown). After each of those points, the captured output is obtained and
+    attached to the collection/runtest report.
+
+    There are two levels of capture:
+    * global: which is enabled by default and can be suppressed by the ``-s`` option. This is always enabled/disabled
+      during collection and each test phase.
+    * fixture: when a test function or one of its fixture depend on the ``capsys`` or ``capfd`` fixtures. In this
+      case special handling is needed to ensure the fixtures take precedence over the global capture.
+    """
+
+    def __init__(self, method: "_CaptureMethod") -> None:
+        self._method = method
+        self._global_capturing = None
+        self._capture_fixture = None  # type: Optional[CaptureFixture]
+
+    def __repr__(self):
+        return "<CaptureManager _method={!r} _global_capturing={!r} _capture_fixture={!r}>".format(
+            self._method, self._global_capturing, self._capture_fixture
+        )
+
+    def is_capturing(self):
+        if self.is_globally_capturing():
+            return "global"
+        if self._capture_fixture:
+            return "fixture %s" % self._capture_fixture.request.fixturename
+        return False
+
+    # Global capturing control
+
+    def is_globally_capturing(self):
+        return self._method != "no"
+
+    def start_global_capturing(self):
+        assert self._global_capturing is None
+        self._global_capturing = _get_multicapture(self._method)
+        self._global_capturing.start_capturing()
+
+    def stop_global_capturing(self):
+        if self._global_capturing is not None:
+            self._global_capturing.pop_outerr_to_orig()
+            self._global_capturing.stop_capturing()
+            self._global_capturing = None
+
+    def resume_global_capture(self):
+        # During teardown of the python process, and on rare occasions, capture
+        # attributes can be `None` while trying to resume global capture.
+        if self._global_capturing is not None:
+            self._global_capturing.resume_capturing()
+
+    def suspend_global_capture(self, in_=False):
+        cap = getattr(self, "_global_capturing", None)
+        if cap is not None:
+            cap.suspend_capturing(in_=in_)
+
+    def suspend(self, in_=False):
+        # Need to undo local capsys-et-al if it exists before disabling global capture.
+        self.suspend_fixture()
+        self.suspend_global_capture(in_)
+
+    def resume(self):
+        self.resume_global_capture()
+        self.resume_fixture()
+
+    def read_global_capture(self):
+        return self._global_capturing.readouterr()
+
+    # Fixture Control (it's just forwarding, think about removing this later)
+
+    @contextlib.contextmanager
+    def _capturing_for_request(
+        self, request: FixtureRequest
+    ) -> Generator["CaptureFixture", None, None]:
+        """
+        Context manager that creates a ``CaptureFixture`` instance for the
+        given ``request``, ensuring there is only a single one being requested
+        at the same time.
+
+        This is used as a helper with ``capsys``, ``capfd`` etc.
+        """
+        if self._capture_fixture:
+            other_name = next(
+                k
+                for k, v in map_fixname_class.items()
+                if v is self._capture_fixture.captureclass
+            )
+            raise request.raiseerror(
+                "cannot use {} and {} at the same time".format(
+                    request.fixturename, other_name
+                )
+            )
+        capture_class = map_fixname_class[request.fixturename]
+        self._capture_fixture = CaptureFixture(capture_class, request)
+        self.activate_fixture()
+        yield self._capture_fixture
+        self._capture_fixture.close()
+        self._capture_fixture = None
+
+    def activate_fixture(self):
+        """If the current item is using ``capsys`` or ``capfd``, activate them so they take precedence over
+        the global capture.
+        """
+        if self._capture_fixture:
+            self._capture_fixture._start()
+
+    def deactivate_fixture(self):
+        """Deactivates the ``capsys`` or ``capfd`` fixture of this item, if any."""
+        if self._capture_fixture:
+            self._capture_fixture.close()
+
+    def suspend_fixture(self):
+        if self._capture_fixture:
+            self._capture_fixture._suspend()
+
+    def resume_fixture(self):
+        if self._capture_fixture:
+            self._capture_fixture._resume()
+
+    # Helper context managers
+
+    @contextlib.contextmanager
+    def global_and_fixture_disabled(self):
+        """Context manager to temporarily disable global and current fixture capturing."""
+        self.suspend()
+        try:
+            yield
+        finally:
+            self.resume()
+
+    @contextlib.contextmanager
+    def item_capture(self, when, item):
+        self.resume_global_capture()
+        self.activate_fixture()
+        try:
+            yield
+        finally:
+            self.deactivate_fixture()
+            self.suspend_global_capture(in_=False)
+
+        out, err = self.read_global_capture()
+        item.add_report_section(when, "stdout", out)
+        item.add_report_section(when, "stderr", err)
+
+    # Hooks
+
+    @pytest.hookimpl(hookwrapper=True)
+    def pytest_make_collect_report(self, collector):
+        if isinstance(collector, pytest.File):
+            self.resume_global_capture()
+            outcome = yield
+            self.suspend_global_capture()
+            out, err = self.read_global_capture()
+            rep = outcome.get_result()
+            if out:
+                rep.sections.append(("Captured stdout", out))
+            if err:
+                rep.sections.append(("Captured stderr", err))
+        else:
+            yield
+
+    @pytest.hookimpl(hookwrapper=True)
+    def pytest_runtest_setup(self, item):
+        with self.item_capture("setup", item):
+            yield
+
+    @pytest.hookimpl(hookwrapper=True)
+    def pytest_runtest_call(self, item):
+        with self.item_capture("call", item):
+            yield
+
+    @pytest.hookimpl(hookwrapper=True)
+    def pytest_runtest_teardown(self, item):
+        with self.item_capture("teardown", item):
+            yield
+
+    @pytest.hookimpl(tryfirst=True)
+    def pytest_keyboard_interrupt(self, excinfo):
+        self.stop_global_capturing()
+
+    @pytest.hookimpl(tryfirst=True)
+    def pytest_internalerror(self, excinfo):
+        self.stop_global_capturing()
+
+
+@pytest.fixture
+def capsys(request):
+    """Enable text capturing of writes to ``sys.stdout`` and ``sys.stderr``.
+
+    The captured output is made available via ``capsys.readouterr()`` method
+    calls, which return a ``(out, err)`` namedtuple.
+    ``out`` and ``err`` will be ``text`` objects.
+    """
+    capman = request.config.pluginmanager.getplugin("capturemanager")
+    with capman._capturing_for_request(request) as fixture:
+        yield fixture
+
+
+@pytest.fixture
+def capsysbinary(request):
+    """Enable bytes capturing of writes to ``sys.stdout`` and ``sys.stderr``.
+
+    The captured output is made available via ``capsysbinary.readouterr()``
+    method calls, which return a ``(out, err)`` namedtuple.
+    ``out`` and ``err`` will be ``bytes`` objects.
+    """
+    capman = request.config.pluginmanager.getplugin("capturemanager")
+    with capman._capturing_for_request(request) as fixture:
+        yield fixture
+
+
+@pytest.fixture
+def capfd(request):
+    """Enable text capturing of writes to file descriptors ``1`` and ``2``.
+
+    The captured output is made available via ``capfd.readouterr()`` method
+    calls, which return a ``(out, err)`` namedtuple.
+    ``out`` and ``err`` will be ``text`` objects.
+    """
+    if not hasattr(os, "dup"):
+        pytest.skip(
+            "capfd fixture needs os.dup function which is not available in this system"
+        )
+    capman = request.config.pluginmanager.getplugin("capturemanager")
+    with capman._capturing_for_request(request) as fixture:
+        yield fixture
+
+
+@pytest.fixture
+def capfdbinary(request):
+    """Enable bytes capturing of writes to file descriptors ``1`` and ``2``.
+
+    The captured output is made available via ``capfd.readouterr()`` method
+    calls, which return a ``(out, err)`` namedtuple.
+    ``out`` and ``err`` will be ``byte`` objects.
+    """
+    if not hasattr(os, "dup"):
+        pytest.skip(
+            "capfdbinary fixture needs os.dup function which is not available in this system"
+        )
+    capman = request.config.pluginmanager.getplugin("capturemanager")
+    with capman._capturing_for_request(request) as fixture:
+        yield fixture
+
+
+class CaptureFixture:
+    """
+    Object returned by :py:func:`capsys`, :py:func:`capsysbinary`, :py:func:`capfd` and :py:func:`capfdbinary`
+    fixtures.
+    """
+
+    def __init__(self, captureclass, request):
+        self.captureclass = captureclass
+        self.request = request
+        self._capture = None
+        self._captured_out = self.captureclass.EMPTY_BUFFER
+        self._captured_err = self.captureclass.EMPTY_BUFFER
+
+    def _start(self):
+        if self._capture is None:
+            self._capture = MultiCapture(
+                out=True, err=True, in_=False, Capture=self.captureclass
+            )
+            self._capture.start_capturing()
+
+    def close(self):
+        if self._capture is not None:
+            out, err = self._capture.pop_outerr_to_orig()
+            self._captured_out += out
+            self._captured_err += err
+            self._capture.stop_capturing()
+            self._capture = None
+
+    def readouterr(self):
+        """Read and return the captured output so far, resetting the internal buffer.
+
+        :return: captured content as a namedtuple with ``out`` and ``err`` string attributes
+        """
+        captured_out, captured_err = self._captured_out, self._captured_err
+        if self._capture is not None:
+            out, err = self._capture.readouterr()
+            captured_out += out
+            captured_err += err
+        self._captured_out = self.captureclass.EMPTY_BUFFER
+        self._captured_err = self.captureclass.EMPTY_BUFFER
+        return CaptureResult(captured_out, captured_err)
+
+    def _suspend(self):
+        """Suspends this fixture's own capturing temporarily."""
+        if self._capture is not None:
+            self._capture.suspend_capturing()
+
+    def _resume(self):
+        """Resumes this fixture's own capturing temporarily."""
+        if self._capture is not None:
+            self._capture.resume_capturing()
+
+    @contextlib.contextmanager
+    def disabled(self):
+        """Temporarily disables capture while inside the 'with' block."""
+        capmanager = self.request.config.pluginmanager.getplugin("capturemanager")
+        with capmanager.global_and_fixture_disabled():
+            yield
+
+
+def safe_text_dupfile(f, mode, default_encoding="UTF8"):
+    """ return an open text file object that's a duplicate of f on the
+        FD-level if possible.
+    """
+    encoding = getattr(f, "encoding", None)
+    try:
+        fd = f.fileno()
+    except Exception:
+        if "b" not in getattr(f, "mode", "") and hasattr(f, "encoding"):
+            # we seem to have a text stream, let's just use it
+            return f
+    else:
+        newfd = os.dup(fd)
+        if "b" not in mode:
+            mode += "b"
+        f = os.fdopen(newfd, mode, 0)  # no buffering
+    return EncodedFile(f, encoding or default_encoding)
+
+
+class EncodedFile:
+    errors = "strict"  # possibly needed by py3 code (issue555)
+
+    def __init__(self, buffer: BinaryIO, encoding: str) -> None:
+        self.buffer = buffer
+        self.encoding = encoding
+
+    def write(self, s: str) -> int:
+        if not isinstance(s, str):
+            raise TypeError(
+                "write() argument must be str, not {}".format(type(s).__name__)
+            )
+        return self.buffer.write(s.encode(self.encoding, "replace"))
+
+    def writelines(self, lines: Iterable[str]) -> None:
+        self.buffer.writelines(x.encode(self.encoding, "replace") for x in lines)
+
+    @property
+    def name(self) -> str:
+        """Ensure that file.name is a string."""
+        return repr(self.buffer)
+
+    @property
+    def mode(self) -> str:
+        return self.buffer.mode.replace("b", "")
+
+    def __getattr__(self, name):
+        return getattr(object.__getattribute__(self, "buffer"), name)
+
+
+CaptureResult = collections.namedtuple("CaptureResult", ["out", "err"])
+
+
+class MultiCapture:
+    out = err = in_ = None
+    _state = None
+    _in_suspended = False
+
+    def __init__(self, out=True, err=True, in_=True, Capture=None):
+        if in_:
+            self.in_ = Capture(0)
+        if out:
+            self.out = Capture(1)
+        if err:
+            self.err = Capture(2)
+
+    def __repr__(self):
+        return "<MultiCapture out={!r} err={!r} in_={!r} _state={!r} _in_suspended={!r}>".format(
+            self.out, self.err, self.in_, self._state, self._in_suspended,
+        )
+
+    def start_capturing(self):
+        self._state = "started"
+        if self.in_:
+            self.in_.start()
+        if self.out:
+            self.out.start()
+        if self.err:
+            self.err.start()
+
+    def pop_outerr_to_orig(self):
+        """ pop current snapshot out/err capture and flush to orig streams. """
+        out, err = self.readouterr()
+        if out:
+            self.out.writeorg(out)
+        if err:
+            self.err.writeorg(err)
+        return out, err
+
+    def suspend_capturing(self, in_=False):
+        self._state = "suspended"
+        if self.out:
+            self.out.suspend()
+        if self.err:
+            self.err.suspend()
+        if in_ and self.in_:
+            self.in_.suspend()
+            self._in_suspended = True
+
+    def resume_capturing(self):
+        self._state = "resumed"
+        if self.out:
+            self.out.resume()
+        if self.err:
+            self.err.resume()
+        if self._in_suspended:
+            self.in_.resume()
+            self._in_suspended = False
+
+    def stop_capturing(self):
+        """ stop capturing and reset capturing streams """
+        if self._state == "stopped":
+            raise ValueError("was already stopped")
+        self._state = "stopped"
+        if self.out:
+            self.out.done()
+        if self.err:
+            self.err.done()
+        if self.in_:
+            self.in_.done()
+
+    def readouterr(self) -> CaptureResult:
+        if self.out:
+            out = self.out.snap()
+        else:
+            out = ""
+        if self.err:
+            err = self.err.snap()
+        else:
+            err = ""
+        return CaptureResult(out, err)
+
+
+class NoCapture:
+    EMPTY_BUFFER = None
+    __init__ = start = done = suspend = resume = lambda *args: None
+
+
+class FDCaptureBinary:
+    """Capture IO to/from a given os-level filedescriptor.
+
+    snap() produces `bytes`
+    """
+
+    EMPTY_BUFFER = b""
+    _state = None
+
+    def __init__(self, targetfd, tmpfile=None):
+        self.targetfd = targetfd
+        try:
+            self.targetfd_save = os.dup(self.targetfd)
+        except OSError:
+            self.start = lambda: None
+            self.done = lambda: None
+        else:
+            self.start = self._start
+            self.done = self._done
+            if targetfd == 0:
+                assert not tmpfile, "cannot set tmpfile with stdin"
+                tmpfile = open(os.devnull, "r")
+                self.syscapture = SysCapture(targetfd)
+            else:
+                if tmpfile is None:
+                    f = TemporaryFile()
+                    with f:
+                        tmpfile = safe_text_dupfile(f, mode="wb+")
+                if targetfd in patchsysdict:
+                    self.syscapture = SysCapture(targetfd, tmpfile)
+                else:
+                    self.syscapture = NoCapture()
+            self.tmpfile = tmpfile
+            self.tmpfile_fd = tmpfile.fileno()
+
+    def __repr__(self):
+        return "<{} {} oldfd={} _state={!r} tmpfile={}>".format(
+            self.__class__.__name__,
+            self.targetfd,
+            getattr(self, "targetfd_save", "<UNSET>"),
+            self._state,
+            hasattr(self, "tmpfile") and repr(self.tmpfile) or "<UNSET>",
+        )
+
+    def _start(self):
+        """ Start capturing on targetfd using memorized tmpfile. """
+        try:
+            os.fstat(self.targetfd_save)
+        except (AttributeError, OSError):
+            raise ValueError("saved filedescriptor not valid anymore")
+        os.dup2(self.tmpfile_fd, self.targetfd)
+        self.syscapture.start()
+        self._state = "started"
+
+    def snap(self):
+        self.tmpfile.seek(0)
+        res = self.tmpfile.read()
+        self.tmpfile.seek(0)
+        self.tmpfile.truncate()
+        return res
+
+    def _done(self):
+        """ stop capturing, restore streams, return original capture file,
+        seeked to position zero. """
+        targetfd_save = self.__dict__.pop("targetfd_save")
+        os.dup2(targetfd_save, self.targetfd)
+        os.close(targetfd_save)
+        self.syscapture.done()
+        self.tmpfile.close()
+        self._state = "done"
+
+    def suspend(self):
+        self.syscapture.suspend()
+        os.dup2(self.targetfd_save, self.targetfd)
+        self._state = "suspended"
+
+    def resume(self):
+        self.syscapture.resume()
+        os.dup2(self.tmpfile_fd, self.targetfd)
+        self._state = "resumed"
+
+    def writeorg(self, data):
+        """ write to original file descriptor. """
+        os.write(self.targetfd_save, data)
+
+
+class FDCapture(FDCaptureBinary):
+    """Capture IO to/from a given os-level filedescriptor.
+
+    snap() produces text
+    """
+
+    # Ignore type because it doesn't match the type in the superclass (bytes).
+    EMPTY_BUFFER = str()  # type: ignore
+
+    def snap(self):
+        res = super().snap()
+        enc = getattr(self.tmpfile, "encoding", None)
+        if enc and isinstance(res, bytes):
+            res = str(res, enc, "replace")
+        return res
+
+    def writeorg(self, data):
+        """ write to original file descriptor. """
+        data = data.encode("utf-8")  # XXX use encoding of original stream
+        os.write(self.targetfd_save, data)
+
+
+class SysCaptureBinary:
+
+    EMPTY_BUFFER = b""
+    _state = None
+
+    def __init__(self, fd, tmpfile=None):
+        name = patchsysdict[fd]
+        self._old = getattr(sys, name)
+        self.name = name
+        if tmpfile is None:
+            if name == "stdin":
+                tmpfile = DontReadFromInput()
+            else:
+                tmpfile = CaptureIO()
+        self.tmpfile = tmpfile
+
+    def __repr__(self):
+        return "<{} {} _old={} _state={!r} tmpfile={!r}>".format(
+            self.__class__.__name__,
+            self.name,
+            hasattr(self, "_old") and repr(self._old) or "<UNSET>",
+            self._state,
+            self.tmpfile,
+        )
+
+    def start(self):
+        setattr(sys, self.name, self.tmpfile)
+        self._state = "started"
+
+    def snap(self):
+        res = self.tmpfile.buffer.getvalue()
+        self.tmpfile.seek(0)
+        self.tmpfile.truncate()
+        return res
+
+    def done(self):
+        setattr(sys, self.name, self._old)
+        del self._old
+        self.tmpfile.close()
+        self._state = "done"
+
+    def suspend(self):
+        setattr(sys, self.name, self._old)
+        self._state = "suspended"
+
+    def resume(self):
+        setattr(sys, self.name, self.tmpfile)
+        self._state = "resumed"
+
+    def writeorg(self, data):
+        self._old.flush()
+        self._old.buffer.write(data)
+        self._old.buffer.flush()
+
+
+class SysCapture(SysCaptureBinary):
+    EMPTY_BUFFER = str()  # type: ignore[assignment]  # noqa: F821
+
+    def snap(self):
+        res = self.tmpfile.getvalue()
+        self.tmpfile.seek(0)
+        self.tmpfile.truncate()
+        return res
+
+    def writeorg(self, data):
+        self._old.write(data)
+        self._old.flush()
+
+
+class TeeSysCapture(SysCapture):
+    def __init__(self, fd, tmpfile=None):
+        name = patchsysdict[fd]
+        self._old = getattr(sys, name)
+        self.name = name
+        if tmpfile is None:
+            if name == "stdin":
+                tmpfile = DontReadFromInput()
+            else:
+                tmpfile = CaptureAndPassthroughIO(self._old)
+        self.tmpfile = tmpfile
+
+
+map_fixname_class = {
+    "capfd": FDCapture,
+    "capfdbinary": FDCaptureBinary,
+    "capsys": SysCapture,
+    "capsysbinary": SysCaptureBinary,
+}
+
+
+class DontReadFromInput:
+    encoding = None
+
+    def read(self, *args):
+        raise IOError(
+            "pytest: reading from stdin while output is captured!  Consider using `-s`."
+        )
+
+    readline = read
+    readlines = read
+    __next__ = read
+
+    def __iter__(self):
+        return self
+
+    def fileno(self):
+        raise UnsupportedOperation("redirected stdin is pseudofile, has no fileno()")
+
+    def isatty(self):
+        return False
+
+    def close(self):
+        pass
+
+    @property
+    def buffer(self):
+        return self
+
+
+def _colorama_workaround():
+    """
+    Ensure colorama is imported so that it attaches to the correct stdio
+    handles on Windows.
+
+    colorama uses the terminal on import time. So if something does the
+    first import of colorama while I/O capture is active, colorama will
+    fail in various ways.
+    """
+    if sys.platform.startswith("win32"):
+        try:
+            import colorama  # noqa: F401
+        except ImportError:
+            pass
+
+
+def _readline_workaround():
+    """
+    Ensure readline is imported so that it attaches to the correct stdio
+    handles on Windows.
+
+    Pdb uses readline support where available--when not running from the Python
+    prompt, the readline module is not imported until running the pdb REPL.  If
+    running pytest with the --pdb option this means the readline module is not
+    imported until after I/O capture has been started.
+
+    This is a problem for pyreadline, which is often used to implement readline
+    support on Windows, as it does not attach to the correct handles for stdout
+    and/or stdin if they have been redirected by the FDCapture mechanism.  This
+    workaround ensures that readline is imported before I/O capture is setup so
+    that it can attach to the actual stdin/out for the console.
+
+    See https://github.com/pytest-dev/pytest/pull/1281
+    """
+    if sys.platform.startswith("win32"):
+        try:
+            import readline  # noqa: F401
+        except ImportError:
+            pass
+
+
+def _py36_windowsconsoleio_workaround(stream):
+    """
+    Python 3.6 implemented unicode console handling for Windows. This works
+    by reading/writing to the raw console handle using
+    ``{Read,Write}ConsoleW``.
+
+    The problem is that we are going to ``dup2`` over the stdio file
+    descriptors when doing ``FDCapture`` and this will ``CloseHandle`` the
+    handles used by Python to write to the console. Though there is still some
+    weirdness and the console handle seems to only be closed randomly and not
+    on the first call to ``CloseHandle``, or maybe it gets reopened with the
+    same handle value when we suspend capturing.
+
+    The workaround in this case will reopen stdio with a different fd which
+    also means a different handle by replicating the logic in
+    "Py_lifecycle.c:initstdio/create_stdio".
+
+    :param stream: in practice ``sys.stdout`` or ``sys.stderr``, but given
+        here as parameter for unittesting purposes.
+
+    See https://github.com/pytest-dev/py/issues/103
+    """
+    if (
+        not sys.platform.startswith("win32")
+        or sys.version_info[:2] < (3, 6)
+        or hasattr(sys, "pypy_version_info")
+    ):
+        return
+
+    # bail out if ``stream`` doesn't seem like a proper ``io`` stream (#2666)
+    if not hasattr(stream, "buffer"):
+        return
+
+    buffered = hasattr(stream.buffer, "raw")
+    raw_stdout = stream.buffer.raw if buffered else stream.buffer
+
+    if not isinstance(raw_stdout, io._WindowsConsoleIO):
+        return
+
+    def _reopen_stdio(f, mode):
+        if not buffered and mode[0] == "w":
+            buffering = 0
+        else:
+            buffering = -1
+
+        return io.TextIOWrapper(
+            open(os.dup(f.fileno()), mode, buffering),
+            f.encoding,
+            f.errors,
+            f.newlines,
+            f.line_buffering,
+        )
+
+    sys.stdin = _reopen_stdio(sys.stdin, "rb")
+    sys.stdout = _reopen_stdio(sys.stdout, "wb")
+    sys.stderr = _reopen_stdio(sys.stderr, "wb")