We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
click.echo() raises UnsupportedOperation on Windows when using pytest capsys
### Actual behavior

When using `click.echo()` in a test function that uses the pytest `capsys` fixture, it raises `UnsupportedOperation` on Windows. It runs fine on Linux and macOS.

Here is a complete test module `tests/unit/test_error_click.py` that reproduces the issue:
```
import click
import pytest

def myfunc():
    """Function to be tested"""
    click.echo('bla')

def test_myfunc(capsys):
    myfunc()
    stdout, stderr = capsys.readouterr()
    assert stdout == 'bla\n'
```

Here is the failure on Windows (using Python 2.7):
```
$ pytest tests
. . .
_________________________________ test_myfunc _________________________________
capsys = <_pytest.capture.CaptureFixture object at 0x04865830>
    def test_myfunc(capsys):
>       myfunc()
tests\unit\test_error_click.py:15: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
tests\unit\test_error_click.py:11: in myfunc
    click.echo('bla')
.tox\win64_py27_32\lib\site-packages\click\utils.py:230: in echo
    file = _default_text_stdout()
.tox\win64_py27_32\lib\site-packages\click\_compat.py:760: in func
    rv = wrapper_func()
.tox\win64_py27_32\lib\site-packages\click\_compat.py:256: in get_text_stdout
    rv = _get_windows_console_stream(sys.stdout, encoding, errors)
.tox\win64_py27_32\lib\site-packages\click\_winconsole.py:356: in _get_windows_console_stream
    and _is_console(f)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
f = <_pytest.compat.CaptureIO object at 0x049156F0>
    def _is_console(f):
        if not hasattr(f, "fileno"):
            return False
    
        try:
>           fileno = f.fileno()
E           UnsupportedOperation: fileno
.tox\win64_py27_32\lib\site-packages\click\_winconsole.py:343: UnsupportedOperation
```

This is from this Appveyor run on Python 2.7: https://ci.appveyor.com/project/KSchopmeyer/pywbemtools/builds/33566529/job/j22nqi2rq10f7uk2#L1463

It also happens on Python 3.8: https://ci.appveyor.com/project/KSchopmeyer/pywbemtools/builds/33566529/job/0mo646j7ck1yidhs#L1473

### Expected behavior

This should succeed on Windows as it does on Linux and macOS.

### Possible solutions

I think `UnsupportedOperation` should be tolerated in the call to `f.fileno()` in `_winconsole._is_console()`. This seems to be done in other people's code calling fileno(), too:  https://www.programcreek.com/python/example/17474/io.UnsupportedOperation

### Environment

* Python version: 2.7, 3.8
* Click version: 7.1.2

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/click/_winconsole.py`
Content:
```
1 # This module is based on the excellent work by Adam Bartoš who
2 # provided a lot of what went into the implementation here in
3 # the discussion to issue1602 in the Python bug tracker.
4 #
5 # There are some general differences in regards to how this works
6 # compared to the original patches as we do not need to patch
7 # the entire interpreter but just work in our little world of
8 # echo and prompt.
9 import ctypes
10 import io
11 import time
12 from ctypes import byref
13 from ctypes import c_char
14 from ctypes import c_char_p
15 from ctypes import c_int
16 from ctypes import c_ssize_t
17 from ctypes import c_ulong
18 from ctypes import c_void_p
19 from ctypes import POINTER
20 from ctypes import py_object
21 from ctypes import windll
22 from ctypes import WINFUNCTYPE
23 from ctypes.wintypes import DWORD
24 from ctypes.wintypes import HANDLE
25 from ctypes.wintypes import LPCWSTR
26 from ctypes.wintypes import LPWSTR
27 
28 import msvcrt
29 
30 from ._compat import _NonClosingTextIOWrapper
31 
32 try:
33     from ctypes import pythonapi
34 except ImportError:
35     pythonapi = None
36 else:
37     PyObject_GetBuffer = pythonapi.PyObject_GetBuffer
38     PyBuffer_Release = pythonapi.PyBuffer_Release
39 
40 
41 c_ssize_p = POINTER(c_ssize_t)
42 
43 kernel32 = windll.kernel32
44 GetStdHandle = kernel32.GetStdHandle
45 ReadConsoleW = kernel32.ReadConsoleW
46 WriteConsoleW = kernel32.WriteConsoleW
47 GetConsoleMode = kernel32.GetConsoleMode
48 GetLastError = kernel32.GetLastError
49 GetCommandLineW = WINFUNCTYPE(LPWSTR)(("GetCommandLineW", windll.kernel32))
50 CommandLineToArgvW = WINFUNCTYPE(POINTER(LPWSTR), LPCWSTR, POINTER(c_int))(
51     ("CommandLineToArgvW", windll.shell32)
52 )
53 LocalFree = WINFUNCTYPE(ctypes.c_void_p, ctypes.c_void_p)(
54     ("LocalFree", windll.kernel32)
55 )
56 
57 
58 STDIN_HANDLE = GetStdHandle(-10)
59 STDOUT_HANDLE = GetStdHandle(-11)
60 STDERR_HANDLE = GetStdHandle(-12)
61 
62 
63 PyBUF_SIMPLE = 0
64 PyBUF_WRITABLE = 1
65 
66 ERROR_SUCCESS = 0
67 ERROR_NOT_ENOUGH_MEMORY = 8
68 ERROR_OPERATION_ABORTED = 995
69 
70 STDIN_FILENO = 0
71 STDOUT_FILENO = 1
72 STDERR_FILENO = 2
73 
74 EOF = b"\x1a"
75 MAX_BYTES_WRITTEN = 32767
76 
77 
78 class Py_buffer(ctypes.Structure):
79     _fields_ = [
80         ("buf", c_void_p),
81         ("obj", py_object),
82         ("len", c_ssize_t),
83         ("itemsize", c_ssize_t),
84         ("readonly", c_int),
85         ("ndim", c_int),
86         ("format", c_char_p),
87         ("shape", c_ssize_p),
88         ("strides", c_ssize_p),
89         ("suboffsets", c_ssize_p),
90         ("internal", c_void_p),
91     ]
92 
93 
94 # On PyPy we cannot get buffers so our ability to operate here is
95 # severely limited.
96 if pythonapi is None:
97     get_buffer = None
98 else:
99 
100     def get_buffer(obj, writable=False):
101         buf = Py_buffer()
102         flags = PyBUF_WRITABLE if writable else PyBUF_SIMPLE
103         PyObject_GetBuffer(py_object(obj), byref(buf), flags)
104         try:
105             buffer_type = c_char * buf.len
106             return buffer_type.from_address(buf.buf)
107         finally:
108             PyBuffer_Release(byref(buf))
109 
110 
111 class _WindowsConsoleRawIOBase(io.RawIOBase):
112     def __init__(self, handle):
113         self.handle = handle
114 
115     def isatty(self):
116         io.RawIOBase.isatty(self)
117         return True
118 
119 
120 class _WindowsConsoleReader(_WindowsConsoleRawIOBase):
121     def readable(self):
122         return True
123 
124     def readinto(self, b):
125         bytes_to_be_read = len(b)
126         if not bytes_to_be_read:
127             return 0
128         elif bytes_to_be_read % 2:
129             raise ValueError(
130                 "cannot read odd number of bytes from UTF-16-LE encoded console"
131             )
132 
133         buffer = get_buffer(b, writable=True)
134         code_units_to_be_read = bytes_to_be_read // 2
135         code_units_read = c_ulong()
136 
137         rv = ReadConsoleW(
138             HANDLE(self.handle),
139             buffer,
140             code_units_to_be_read,
141             byref(code_units_read),
142             None,
143         )
144         if GetLastError() == ERROR_OPERATION_ABORTED:
145             # wait for KeyboardInterrupt
146             time.sleep(0.1)
147         if not rv:
148             raise OSError(f"Windows error: {GetLastError()}")
149 
150         if buffer[0] == EOF:
151             return 0
152         return 2 * code_units_read.value
153 
154 
155 class _WindowsConsoleWriter(_WindowsConsoleRawIOBase):
156     def writable(self):
157         return True
158 
159     @staticmethod
160     def _get_error_message(errno):
161         if errno == ERROR_SUCCESS:
162             return "ERROR_SUCCESS"
163         elif errno == ERROR_NOT_ENOUGH_MEMORY:
164             return "ERROR_NOT_ENOUGH_MEMORY"
165         return f"Windows error {errno}"
166 
167     def write(self, b):
168         bytes_to_be_written = len(b)
169         buf = get_buffer(b)
170         code_units_to_be_written = min(bytes_to_be_written, MAX_BYTES_WRITTEN) // 2
171         code_units_written = c_ulong()
172 
173         WriteConsoleW(
174             HANDLE(self.handle),
175             buf,
176             code_units_to_be_written,
177             byref(code_units_written),
178             None,
179         )
180         bytes_written = 2 * code_units_written.value
181 
182         if bytes_written == 0 and bytes_to_be_written > 0:
183             raise OSError(self._get_error_message(GetLastError()))
184         return bytes_written
185 
186 
187 class ConsoleStream:
188     def __init__(self, text_stream, byte_stream):
189         self._text_stream = text_stream
190         self.buffer = byte_stream
191 
192     @property
193     def name(self):
194         return self.buffer.name
195 
196     def write(self, x):
197         if isinstance(x, str):
198             return self._text_stream.write(x)
199         try:
200             self.flush()
201         except Exception:
202             pass
203         return self.buffer.write(x)
204 
205     def writelines(self, lines):
206         for line in lines:
207             self.write(line)
208 
209     def __getattr__(self, name):
210         return getattr(self._text_stream, name)
211 
212     def isatty(self):
213         return self.buffer.isatty()
214 
215     def __repr__(self):
216         return f"<ConsoleStream name={self.name!r} encoding={self.encoding!r}>"
217 
218 
219 class WindowsChunkedWriter:
220     """
221     Wraps a stream (such as stdout), acting as a transparent proxy for all
222     attribute access apart from method 'write()' which we wrap to write in
223     limited chunks due to a Windows limitation on binary console streams.
224     """
225 
226     def __init__(self, wrapped):
227         # double-underscore everything to prevent clashes with names of
228         # attributes on the wrapped stream object.
229         self.__wrapped = wrapped
230 
231     def __getattr__(self, name):
232         return getattr(self.__wrapped, name)
233 
234     def write(self, text):
235         total_to_write = len(text)
236         written = 0
237 
238         while written < total_to_write:
239             to_write = min(total_to_write - written, MAX_BYTES_WRITTEN)
240             self.__wrapped.write(text[written : written + to_write])
241             written += to_write
242 
243 
244 def _get_text_stdin(buffer_stream):
245     text_stream = _NonClosingTextIOWrapper(
246         io.BufferedReader(_WindowsConsoleReader(STDIN_HANDLE)),
247         "utf-16-le",
248         "strict",
249         line_buffering=True,
250     )
251     return ConsoleStream(text_stream, buffer_stream)
252 
253 
254 def _get_text_stdout(buffer_stream):
255     text_stream = _NonClosingTextIOWrapper(
256         io.BufferedWriter(_WindowsConsoleWriter(STDOUT_HANDLE)),
257         "utf-16-le",
258         "strict",
259         line_buffering=True,
260     )
261     return ConsoleStream(text_stream, buffer_stream)
262 
263 
264 def _get_text_stderr(buffer_stream):
265     text_stream = _NonClosingTextIOWrapper(
266         io.BufferedWriter(_WindowsConsoleWriter(STDERR_HANDLE)),
267         "utf-16-le",
268         "strict",
269         line_buffering=True,
270     )
271     return ConsoleStream(text_stream, buffer_stream)
272 
273 
274 _stream_factories = {
275     0: _get_text_stdin,
276     1: _get_text_stdout,
277     2: _get_text_stderr,
278 }
279 
280 
281 def _is_console(f):
282     if not hasattr(f, "fileno"):
283         return False
284 
285     try:
286         fileno = f.fileno()
287     except OSError:
288         return False
289 
290     handle = msvcrt.get_osfhandle(fileno)
291     return bool(GetConsoleMode(handle, byref(DWORD())))
292 
293 
294 def _get_windows_console_stream(f, encoding, errors):
295     if (
296         get_buffer is not None
297         and encoding in {"utf-16-le", None}
298         and errors in {"strict", None}
299         and _is_console(f)
300     ):
301         func = _stream_factories.get(f.fileno())
302         if func is not None:
303             f = getattr(f, "buffer", None)
304 
305             if f is None:
306                 return None
307 
308             return func(f)
309
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/click/_winconsole.py b/src/click/_winconsole.py
--- a/src/click/_winconsole.py
+++ b/src/click/_winconsole.py
@@ -284,7 +284,7 @@
 
     try:
         fileno = f.fileno()
-    except OSError:
+    except (OSError, io.UnsupportedOperation):
         return False
 
     handle = msvcrt.get_osfhandle(fileno)

{"golden_diff": "diff --git a/src/click/_winconsole.py b/src/click/_winconsole.py\n--- a/src/click/_winconsole.py\n+++ b/src/click/_winconsole.py\n@@ -284,7 +284,7 @@\n \n     try:\n         fileno = f.fileno()\n-    except OSError:\n+    except (OSError, io.UnsupportedOperation):\n         return False\n \n     handle = msvcrt.get_osfhandle(fileno)\n", "issue": "click.echo() raises UnsupportedOperation on Windows when using pytest capsys\n### Actual behavior\r\n\r\nWhen using `click.echo()` in a test function that uses the pytest `capsys` fixture, it raises `UnsupportedOperation` on Windows. It runs fine on Linux and macOS.\r\n\r\nHere is a complete test module `tests/unit/test_error_click.py` that reproduces the issue:\r\n```\r\nimport click\r\nimport pytest\r\n\r\ndef myfunc():\r\n    \"\"\"Function to be tested\"\"\"\r\n    click.echo('bla')\r\n\r\ndef test_myfunc(capsys):\r\n    myfunc()\r\n    stdout, stderr = capsys.readouterr()\r\n    assert stdout == 'bla\\n'\r\n```\r\n\r\nHere is the failure on Windows (using Python 2.7):\r\n```\r\n$ pytest tests\r\n. . .\r\n_________________________________ test_myfunc _________________________________\r\ncapsys = <_pytest.capture.CaptureFixture object at 0x04865830>\r\n    def test_myfunc(capsys):\r\n>       myfunc()\r\ntests\\unit\\test_error_click.py:15: \r\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _\r\ntests\\unit\\test_error_click.py:11: in myfunc\r\n    click.echo('bla')\r\n.tox\\win64_py27_32\\lib\\site-packages\\click\\utils.py:230: in echo\r\n    file = _default_text_stdout()\r\n.tox\\win64_py27_32\\lib\\site-packages\\click\\_compat.py:760: in func\r\n    rv = wrapper_func()\r\n.tox\\win64_py27_32\\lib\\site-packages\\click\\_compat.py:256: in get_text_stdout\r\n    rv = _get_windows_console_stream(sys.stdout, encoding, errors)\r\n.tox\\win64_py27_32\\lib\\site-packages\\click\\_winconsole.py:356: in _get_windows_console_stream\r\n    and _is_console(f)\r\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _\r\nf = <_pytest.compat.CaptureIO object at 0x049156F0>\r\n    def _is_console(f):\r\n        if not hasattr(f, \"fileno\"):\r\n            return False\r\n    \r\n        try:\r\n>           fileno = f.fileno()\r\nE           UnsupportedOperation: fileno\r\n.tox\\win64_py27_32\\lib\\site-packages\\click\\_winconsole.py:343: UnsupportedOperation\r\n```\r\n\r\nThis is from this Appveyor run on Python 2.7: https://ci.appveyor.com/project/KSchopmeyer/pywbemtools/builds/33566529/job/j22nqi2rq10f7uk2#L1463\r\n\r\nIt also happens on Python 3.8: https://ci.appveyor.com/project/KSchopmeyer/pywbemtools/builds/33566529/job/0mo646j7ck1yidhs#L1473\r\n\r\n### Expected behavior\r\n\r\nThis should succeed on Windows as it does on Linux and macOS.\r\n\r\n### Possible solutions\r\n\r\nI think `UnsupportedOperation` should be tolerated in the call to `f.fileno()` in `_winconsole._is_console()`. This seems to be done in other people's code calling fileno(), too:  https://www.programcreek.com/python/example/17474/io.UnsupportedOperation\r\n\r\n### Environment\r\n\r\n* Python version: 2.7, 3.8\r\n* Click version: 7.1.2\n", "before_files": [{"content": "# This module is based on the excellent work by Adam Barto\u0161 who\n# provided a lot of what went into the implementation here in\n# the discussion to issue1602 in the Python bug tracker.\n#\n# There are some general differences in regards to how this works\n# compared to the original patches as we do not need to patch\n# the entire interpreter but just work in our little world of\n# echo and prompt.\nimport ctypes\nimport io\nimport time\nfrom ctypes import byref\nfrom ctypes import c_char\nfrom ctypes import c_char_p\nfrom ctypes import c_int\nfrom ctypes import c_ssize_t\nfrom ctypes import c_ulong\nfrom ctypes import c_void_p\nfrom ctypes import POINTER\nfrom ctypes import py_object\nfrom ctypes import windll\nfrom ctypes import WINFUNCTYPE\nfrom ctypes.wintypes import DWORD\nfrom ctypes.wintypes import HANDLE\nfrom ctypes.wintypes import LPCWSTR\nfrom ctypes.wintypes import LPWSTR\n\nimport msvcrt\n\nfrom ._compat import _NonClosingTextIOWrapper\n\ntry:\n    from ctypes import pythonapi\nexcept ImportError:\n    pythonapi = None\nelse:\n    PyObject_GetBuffer = pythonapi.PyObject_GetBuffer\n    PyBuffer_Release = pythonapi.PyBuffer_Release\n\n\nc_ssize_p = POINTER(c_ssize_t)\n\nkernel32 = windll.kernel32\nGetStdHandle = kernel32.GetStdHandle\nReadConsoleW = kernel32.ReadConsoleW\nWriteConsoleW = kernel32.WriteConsoleW\nGetConsoleMode = kernel32.GetConsoleMode\nGetLastError = kernel32.GetLastError\nGetCommandLineW = WINFUNCTYPE(LPWSTR)((\"GetCommandLineW\", windll.kernel32))\nCommandLineToArgvW = WINFUNCTYPE(POINTER(LPWSTR), LPCWSTR, POINTER(c_int))(\n    (\"CommandLineToArgvW\", windll.shell32)\n)\nLocalFree = WINFUNCTYPE(ctypes.c_void_p, ctypes.c_void_p)(\n    (\"LocalFree\", windll.kernel32)\n)\n\n\nSTDIN_HANDLE = GetStdHandle(-10)\nSTDOUT_HANDLE = GetStdHandle(-11)\nSTDERR_HANDLE = GetStdHandle(-12)\n\n\nPyBUF_SIMPLE = 0\nPyBUF_WRITABLE = 1\n\nERROR_SUCCESS = 0\nERROR_NOT_ENOUGH_MEMORY = 8\nERROR_OPERATION_ABORTED = 995\n\nSTDIN_FILENO = 0\nSTDOUT_FILENO = 1\nSTDERR_FILENO = 2\n\nEOF = b\"\\x1a\"\nMAX_BYTES_WRITTEN = 32767\n\n\nclass Py_buffer(ctypes.Structure):\n    _fields_ = [\n        (\"buf\", c_void_p),\n        (\"obj\", py_object),\n        (\"len\", c_ssize_t),\n        (\"itemsize\", c_ssize_t),\n        (\"readonly\", c_int),\n        (\"ndim\", c_int),\n        (\"format\", c_char_p),\n        (\"shape\", c_ssize_p),\n        (\"strides\", c_ssize_p),\n        (\"suboffsets\", c_ssize_p),\n        (\"internal\", c_void_p),\n    ]\n\n\n# On PyPy we cannot get buffers so our ability to operate here is\n# severely limited.\nif pythonapi is None:\n    get_buffer = None\nelse:\n\n    def get_buffer(obj, writable=False):\n        buf = Py_buffer()\n        flags = PyBUF_WRITABLE if writable else PyBUF_SIMPLE\n        PyObject_GetBuffer(py_object(obj), byref(buf), flags)\n        try:\n            buffer_type = c_char * buf.len\n            return buffer_type.from_address(buf.buf)\n        finally:\n            PyBuffer_Release(byref(buf))\n\n\nclass _WindowsConsoleRawIOBase(io.RawIOBase):\n    def __init__(self, handle):\n        self.handle = handle\n\n    def isatty(self):\n        io.RawIOBase.isatty(self)\n        return True\n\n\nclass _WindowsConsoleReader(_WindowsConsoleRawIOBase):\n    def readable(self):\n        return True\n\n    def readinto(self, b):\n        bytes_to_be_read = len(b)\n        if not bytes_to_be_read:\n            return 0\n        elif bytes_to_be_read % 2:\n            raise ValueError(\n                \"cannot read odd number of bytes from UTF-16-LE encoded console\"\n            )\n\n        buffer = get_buffer(b, writable=True)\n        code_units_to_be_read = bytes_to_be_read // 2\n        code_units_read = c_ulong()\n\n        rv = ReadConsoleW(\n            HANDLE(self.handle),\n            buffer,\n            code_units_to_be_read,\n            byref(code_units_read),\n            None,\n        )\n        if GetLastError() == ERROR_OPERATION_ABORTED:\n            # wait for KeyboardInterrupt\n            time.sleep(0.1)\n        if not rv:\n            raise OSError(f\"Windows error: {GetLastError()}\")\n\n        if buffer[0] == EOF:\n            return 0\n        return 2 * code_units_read.value\n\n\nclass _WindowsConsoleWriter(_WindowsConsoleRawIOBase):\n    def writable(self):\n        return True\n\n    @staticmethod\n    def _get_error_message(errno):\n        if errno == ERROR_SUCCESS:\n            return \"ERROR_SUCCESS\"\n        elif errno == ERROR_NOT_ENOUGH_MEMORY:\n            return \"ERROR_NOT_ENOUGH_MEMORY\"\n        return f\"Windows error {errno}\"\n\n    def write(self, b):\n        bytes_to_be_written = len(b)\n        buf = get_buffer(b)\n        code_units_to_be_written = min(bytes_to_be_written, MAX_BYTES_WRITTEN) // 2\n        code_units_written = c_ulong()\n\n        WriteConsoleW(\n            HANDLE(self.handle),\n            buf,\n            code_units_to_be_written,\n            byref(code_units_written),\n            None,\n        )\n        bytes_written = 2 * code_units_written.value\n\n        if bytes_written == 0 and bytes_to_be_written > 0:\n            raise OSError(self._get_error_message(GetLastError()))\n        return bytes_written\n\n\nclass ConsoleStream:\n    def __init__(self, text_stream, byte_stream):\n        self._text_stream = text_stream\n        self.buffer = byte_stream\n\n    @property\n    def name(self):\n        return self.buffer.name\n\n    def write(self, x):\n        if isinstance(x, str):\n            return self._text_stream.write(x)\n        try:\n            self.flush()\n        except Exception:\n            pass\n        return self.buffer.write(x)\n\n    def writelines(self, lines):\n        for line in lines:\n            self.write(line)\n\n    def __getattr__(self, name):\n        return getattr(self._text_stream, name)\n\n    def isatty(self):\n        return self.buffer.isatty()\n\n    def __repr__(self):\n        return f\"<ConsoleStream name={self.name!r} encoding={self.encoding!r}>\"\n\n\nclass WindowsChunkedWriter:\n    \"\"\"\n    Wraps a stream (such as stdout), acting as a transparent proxy for all\n    attribute access apart from method 'write()' which we wrap to write in\n    limited chunks due to a Windows limitation on binary console streams.\n    \"\"\"\n\n    def __init__(self, wrapped):\n        # double-underscore everything to prevent clashes with names of\n        # attributes on the wrapped stream object.\n        self.__wrapped = wrapped\n\n    def __getattr__(self, name):\n        return getattr(self.__wrapped, name)\n\n    def write(self, text):\n        total_to_write = len(text)\n        written = 0\n\n        while written < total_to_write:\n            to_write = min(total_to_write - written, MAX_BYTES_WRITTEN)\n            self.__wrapped.write(text[written : written + to_write])\n            written += to_write\n\n\ndef _get_text_stdin(buffer_stream):\n    text_stream = _NonClosingTextIOWrapper(\n        io.BufferedReader(_WindowsConsoleReader(STDIN_HANDLE)),\n        \"utf-16-le\",\n        \"strict\",\n        line_buffering=True,\n    )\n    return ConsoleStream(text_stream, buffer_stream)\n\n\ndef _get_text_stdout(buffer_stream):\n    text_stream = _NonClosingTextIOWrapper(\n        io.BufferedWriter(_WindowsConsoleWriter(STDOUT_HANDLE)),\n        \"utf-16-le\",\n        \"strict\",\n        line_buffering=True,\n    )\n    return ConsoleStream(text_stream, buffer_stream)\n\n\ndef _get_text_stderr(buffer_stream):\n    text_stream = _NonClosingTextIOWrapper(\n        io.BufferedWriter(_WindowsConsoleWriter(STDERR_HANDLE)),\n        \"utf-16-le\",\n        \"strict\",\n        line_buffering=True,\n    )\n    return ConsoleStream(text_stream, buffer_stream)\n\n\n_stream_factories = {\n    0: _get_text_stdin,\n    1: _get_text_stdout,\n    2: _get_text_stderr,\n}\n\n\ndef _is_console(f):\n    if not hasattr(f, \"fileno\"):\n        return False\n\n    try:\n        fileno = f.fileno()\n    except OSError:\n        return False\n\n    handle = msvcrt.get_osfhandle(fileno)\n    return bool(GetConsoleMode(handle, byref(DWORD())))\n\n\ndef _get_windows_console_stream(f, encoding, errors):\n    if (\n        get_buffer is not None\n        and encoding in {\"utf-16-le\", None}\n        and errors in {\"strict\", None}\n        and _is_console(f)\n    ):\n        func = _stream_factories.get(f.fileno())\n        if func is not None:\n            f = getattr(f, \"buffer\", None)\n\n            if f is None:\n                return None\n\n            return func(f)\n", "path": "src/click/_winconsole.py"}], "after_files": [{"content": "# This module is based on the excellent work by Adam Barto\u0161 who\n# provided a lot of what went into the implementation here in\n# the discussion to issue1602 in the Python bug tracker.\n#\n# There are some general differences in regards to how this works\n# compared to the original patches as we do not need to patch\n# the entire interpreter but just work in our little world of\n# echo and prompt.\nimport ctypes\nimport io\nimport time\nfrom ctypes import byref\nfrom ctypes import c_char\nfrom ctypes import c_char_p\nfrom ctypes import c_int\nfrom ctypes import c_ssize_t\nfrom ctypes import c_ulong\nfrom ctypes import c_void_p\nfrom ctypes import POINTER\nfrom ctypes import py_object\nfrom ctypes import windll\nfrom ctypes import WINFUNCTYPE\nfrom ctypes.wintypes import DWORD\nfrom ctypes.wintypes import HANDLE\nfrom ctypes.wintypes import LPCWSTR\nfrom ctypes.wintypes import LPWSTR\n\nimport msvcrt\n\nfrom ._compat import _NonClosingTextIOWrapper\n\ntry:\n    from ctypes import pythonapi\nexcept ImportError:\n    pythonapi = None\nelse:\n    PyObject_GetBuffer = pythonapi.PyObject_GetBuffer\n    PyBuffer_Release = pythonapi.PyBuffer_Release\n\n\nc_ssize_p = POINTER(c_ssize_t)\n\nkernel32 = windll.kernel32\nGetStdHandle = kernel32.GetStdHandle\nReadConsoleW = kernel32.ReadConsoleW\nWriteConsoleW = kernel32.WriteConsoleW\nGetConsoleMode = kernel32.GetConsoleMode\nGetLastError = kernel32.GetLastError\nGetCommandLineW = WINFUNCTYPE(LPWSTR)((\"GetCommandLineW\", windll.kernel32))\nCommandLineToArgvW = WINFUNCTYPE(POINTER(LPWSTR), LPCWSTR, POINTER(c_int))(\n    (\"CommandLineToArgvW\", windll.shell32)\n)\nLocalFree = WINFUNCTYPE(ctypes.c_void_p, ctypes.c_void_p)(\n    (\"LocalFree\", windll.kernel32)\n)\n\n\nSTDIN_HANDLE = GetStdHandle(-10)\nSTDOUT_HANDLE = GetStdHandle(-11)\nSTDERR_HANDLE = GetStdHandle(-12)\n\n\nPyBUF_SIMPLE = 0\nPyBUF_WRITABLE = 1\n\nERROR_SUCCESS = 0\nERROR_NOT_ENOUGH_MEMORY = 8\nERROR_OPERATION_ABORTED = 995\n\nSTDIN_FILENO = 0\nSTDOUT_FILENO = 1\nSTDERR_FILENO = 2\n\nEOF = b\"\\x1a\"\nMAX_BYTES_WRITTEN = 32767\n\n\nclass Py_buffer(ctypes.Structure):\n    _fields_ = [\n        (\"buf\", c_void_p),\n        (\"obj\", py_object),\n        (\"len\", c_ssize_t),\n        (\"itemsize\", c_ssize_t),\n        (\"readonly\", c_int),\n        (\"ndim\", c_int),\n        (\"format\", c_char_p),\n        (\"shape\", c_ssize_p),\n        (\"strides\", c_ssize_p),\n        (\"suboffsets\", c_ssize_p),\n        (\"internal\", c_void_p),\n    ]\n\n\n# On PyPy we cannot get buffers so our ability to operate here is\n# severely limited.\nif pythonapi is None:\n    get_buffer = None\nelse:\n\n    def get_buffer(obj, writable=False):\n        buf = Py_buffer()\n        flags = PyBUF_WRITABLE if writable else PyBUF_SIMPLE\n        PyObject_GetBuffer(py_object(obj), byref(buf), flags)\n        try:\n            buffer_type = c_char * buf.len\n            return buffer_type.from_address(buf.buf)\n        finally:\n            PyBuffer_Release(byref(buf))\n\n\nclass _WindowsConsoleRawIOBase(io.RawIOBase):\n    def __init__(self, handle):\n        self.handle = handle\n\n    def isatty(self):\n        io.RawIOBase.isatty(self)\n        return True\n\n\nclass _WindowsConsoleReader(_WindowsConsoleRawIOBase):\n    def readable(self):\n        return True\n\n    def readinto(self, b):\n        bytes_to_be_read = len(b)\n        if not bytes_to_be_read:\n            return 0\n        elif bytes_to_be_read % 2:\n            raise ValueError(\n                \"cannot read odd number of bytes from UTF-16-LE encoded console\"\n            )\n\n        buffer = get_buffer(b, writable=True)\n        code_units_to_be_read = bytes_to_be_read // 2\n        code_units_read = c_ulong()\n\n        rv = ReadConsoleW(\n            HANDLE(self.handle),\n            buffer,\n            code_units_to_be_read,\n            byref(code_units_read),\n            None,\n        )\n        if GetLastError() == ERROR_OPERATION_ABORTED:\n            # wait for KeyboardInterrupt\n            time.sleep(0.1)\n        if not rv:\n            raise OSError(f\"Windows error: {GetLastError()}\")\n\n        if buffer[0] == EOF:\n            return 0\n        return 2 * code_units_read.value\n\n\nclass _WindowsConsoleWriter(_WindowsConsoleRawIOBase):\n    def writable(self):\n        return True\n\n    @staticmethod\n    def _get_error_message(errno):\n        if errno == ERROR_SUCCESS:\n            return \"ERROR_SUCCESS\"\n        elif errno == ERROR_NOT_ENOUGH_MEMORY:\n            return \"ERROR_NOT_ENOUGH_MEMORY\"\n        return f\"Windows error {errno}\"\n\n    def write(self, b):\n        bytes_to_be_written = len(b)\n        buf = get_buffer(b)\n        code_units_to_be_written = min(bytes_to_be_written, MAX_BYTES_WRITTEN) // 2\n        code_units_written = c_ulong()\n\n        WriteConsoleW(\n            HANDLE(self.handle),\n            buf,\n            code_units_to_be_written,\n            byref(code_units_written),\n            None,\n        )\n        bytes_written = 2 * code_units_written.value\n\n        if bytes_written == 0 and bytes_to_be_written > 0:\n            raise OSError(self._get_error_message(GetLastError()))\n        return bytes_written\n\n\nclass ConsoleStream:\n    def __init__(self, text_stream, byte_stream):\n        self._text_stream = text_stream\n        self.buffer = byte_stream\n\n    @property\n    def name(self):\n        return self.buffer.name\n\n    def write(self, x):\n        if isinstance(x, str):\n            return self._text_stream.write(x)\n        try:\n            self.flush()\n        except Exception:\n            pass\n        return self.buffer.write(x)\n\n    def writelines(self, lines):\n        for line in lines:\n            self.write(line)\n\n    def __getattr__(self, name):\n        return getattr(self._text_stream, name)\n\n    def isatty(self):\n        return self.buffer.isatty()\n\n    def __repr__(self):\n        return f\"<ConsoleStream name={self.name!r} encoding={self.encoding!r}>\"\n\n\nclass WindowsChunkedWriter:\n    \"\"\"\n    Wraps a stream (such as stdout), acting as a transparent proxy for all\n    attribute access apart from method 'write()' which we wrap to write in\n    limited chunks due to a Windows limitation on binary console streams.\n    \"\"\"\n\n    def __init__(self, wrapped):\n        # double-underscore everything to prevent clashes with names of\n        # attributes on the wrapped stream object.\n        self.__wrapped = wrapped\n\n    def __getattr__(self, name):\n        return getattr(self.__wrapped, name)\n\n    def write(self, text):\n        total_to_write = len(text)\n        written = 0\n\n        while written < total_to_write:\n            to_write = min(total_to_write - written, MAX_BYTES_WRITTEN)\n            self.__wrapped.write(text[written : written + to_write])\n            written += to_write\n\n\ndef _get_text_stdin(buffer_stream):\n    text_stream = _NonClosingTextIOWrapper(\n        io.BufferedReader(_WindowsConsoleReader(STDIN_HANDLE)),\n        \"utf-16-le\",\n        \"strict\",\n        line_buffering=True,\n    )\n    return ConsoleStream(text_stream, buffer_stream)\n\n\ndef _get_text_stdout(buffer_stream):\n    text_stream = _NonClosingTextIOWrapper(\n        io.BufferedWriter(_WindowsConsoleWriter(STDOUT_HANDLE)),\n        \"utf-16-le\",\n        \"strict\",\n        line_buffering=True,\n    )\n    return ConsoleStream(text_stream, buffer_stream)\n\n\ndef _get_text_stderr(buffer_stream):\n    text_stream = _NonClosingTextIOWrapper(\n        io.BufferedWriter(_WindowsConsoleWriter(STDERR_HANDLE)),\n        \"utf-16-le\",\n        \"strict\",\n        line_buffering=True,\n    )\n    return ConsoleStream(text_stream, buffer_stream)\n\n\n_stream_factories = {\n    0: _get_text_stdin,\n    1: _get_text_stdout,\n    2: _get_text_stderr,\n}\n\n\ndef _is_console(f):\n    if not hasattr(f, \"fileno\"):\n        return False\n\n    try:\n        fileno = f.fileno()\n    except (OSError, io.UnsupportedOperation):\n        return False\n\n    handle = msvcrt.get_osfhandle(fileno)\n    return bool(GetConsoleMode(handle, byref(DWORD())))\n\n\ndef _get_windows_console_stream(f, encoding, errors):\n    if (\n        get_buffer is not None\n        and encoding in {\"utf-16-le\", None}\n        and errors in {\"strict\", None}\n        and _is_console(f)\n    ):\n        func = _stream_factories.get(f.fileno())\n        if func is not None:\n            f = getattr(f, \"buffer\", None)\n\n            if f is None:\n                return None\n\n            return func(f)\n", "path": "src/click/_winconsole.py"}]}