Skip to content

Commit

Permalink
Complete implementation of stdio stub
Browse files Browse the repository at this point in the history
This class replaced stdout and stderr to capture output, but caused
issues because it didn't implement the full API of sys.stdout and
sys.stderr.

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

Fixes #16678
  • Loading branch information
meshy committed Jan 20, 2024
1 parent 86190a0 commit a405475
Show file tree
Hide file tree
Showing 2 changed files with 64 additions and 5 deletions.
4 changes: 2 additions & 2 deletions mypy/dmypy_server.py
Expand Up @@ -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"}
Expand Down
65 changes: 62 additions & 3 deletions mypy/dmypy_util.py
Expand Up @@ -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

Expand Down Expand Up @@ -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) -> 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)

0 comments on commit a405475

Please sign in to comment.