Fix 'WriteToConn' object has no attribute 'flush' (#16801)

`WriteToConn` replaces stdout and stderr to capture output, but causes
issues because it doesn't implement the `TextIO` API (as expected of
`sys.stdout` and `sys.stderr`).

By stubbing the rest of the `TextIO` API we prevent issues with other
code which uses more of the API than we had previously accounted for.

Fixes #16678

Author: meshy

mypy/dmypy_server.py

@@ -221,8 +221,8 @@ def serve(self) -> None:
             while True:
                 with server:
                     data = receive(server)
-                    sys.stdout = WriteToConn(server, "stdout")  # type: ignore[assignment]
-                    sys.stderr = WriteToConn(server, "stderr")  # type: ignore[assignment]
+                    sys.stdout = WriteToConn(server, "stdout", sys.stdout.isatty())
+                    sys.stderr = WriteToConn(server, "stderr", sys.stderr.isatty())
                     resp: dict[str, Any] = {}
                     if "command" not in data:
                         resp = {"error": "No command found in request"}

mypy/dmypy_util.py

@@ -5,8 +5,10 @@
 
 from __future__ import annotations
 
+import io
 import json
-from typing import Any, Final, Iterable
+from types import TracebackType
+from typing import Any, Final, Iterable, Iterator, TextIO
 
 from mypy.ipc import IPCBase
 
@@ -40,19 +42,76 @@ def send(connection: IPCBase, data: Any) -> None:
     connection.write(json.dumps(data))
 
 
-class WriteToConn:
+class WriteToConn(TextIO):
     """Helper class to write to a connection instead of standard output."""
 
-    def __init__(self, server: IPCBase, output_key: str = "stdout") -> None:
+    def __init__(self, server: IPCBase, output_key: str, isatty: bool) -> None:
         self.server = server
         self.output_key = output_key
+        self._isatty = isatty
+
+    def __enter__(self) -> TextIO:
+        return self
+
+    def __exit__(
+        self,
+        t: type[BaseException] | None,
+        value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        pass
+
+    def __iter__(self) -> Iterator[str]:
+        raise io.UnsupportedOperation
+
+    def __next__(self) -> str:
+        raise io.UnsupportedOperation
+
+    def close(self) -> None:
+        pass
+
+    def fileno(self) -> int:
+        raise OSError
+
+    def flush(self) -> None:
+        pass
+
+    def isatty(self) -> bool:
+        return self._isatty
+
+    def read(self, n: int = 0) -> str:
+        raise io.UnsupportedOperation
+
+    def readable(self) -> bool:
+        return False
+
+    def readline(self, limit: int = 0) -> str:
+        raise io.UnsupportedOperation
+
+    def readlines(self, hint: int = 0) -> list[str]:
+        raise io.UnsupportedOperation
+
+    def seek(self, offset: int, whence: int = 0) -> int:
+        raise io.UnsupportedOperation
+
+    def seekable(self) -> bool:
+        return False
+
+    def tell(self) -> int:
+        raise io.UnsupportedOperation
+
+    def truncate(self, size: int | None = 0) -> int:
+        raise io.UnsupportedOperation
 
     def write(self, output: str) -> int:
         resp: dict[str, Any] = {}
         resp[self.output_key] = output
         send(self.server, resp)
         return len(output)
 
+    def writable(self) -> bool:
+        return True
+
     def writelines(self, lines: Iterable[str]) -> None:
         for s in lines:
             self.write(s)