diff --git a/src/trio/_subprocess.py b/src/trio/_subprocess.py index 710600dcf4..b384d18ffa 100644 --- a/src/trio/_subprocess.py +++ b/src/trio/_subprocess.py @@ -50,8 +50,6 @@ def pidfd_open(fd: int, flags: int) -> int: ... - from ._subprocess_platform import ClosableReceiveStream, ClosableSendStream - else: can_try_pidfd_open = True try: @@ -171,9 +169,9 @@ def __init__( if self.stdin is not None and self.stdout is not None: self.stdio = StapledStream(self.stdin, self.stdout) - self._wait_lock: Lock = Lock() + self._wait_lock = Lock() - self._pidfd: FileIO | None = None + self._pidfd = None if can_try_pidfd_open: try: fd: int = pidfd_open(self._proc.pid, 0) @@ -188,6 +186,8 @@ def __init__( # make sure it'll get closed. self._pidfd = FileIO(fd) + # TODO: remove the following two type hints, required only due to + # https://github.com/microsoft/pyright/issues/11917 self.args: StrOrBytesPath | Sequence[StrOrBytesPath] = self._proc.args self.pid: int = self._proc.pid @@ -381,9 +381,9 @@ async def _open_process( "shell=True on UNIX systems", ) - trio_stdin: ClosableSendStream | None = None - trio_stdout: ClosableReceiveStream | None = None - trio_stderr: ClosableReceiveStream | None = None + trio_stdin = None + trio_stdout = None + trio_stderr = None # Close the parent's handle for each child side of a pipe; we want the child to # have the only copy, so that when it exits we can read EOF on our side. The # trio ends of pipes will be transferred to the Process object, which will be diff --git a/src/trio/_tests/type_tests/subprocesses.py b/src/trio/_tests/type_tests/subprocesses.py index de3b5e8906..689ffd7d95 100644 --- a/src/trio/_tests/type_tests/subprocesses.py +++ b/src/trio/_tests/type_tests/subprocesses.py @@ -1,6 +1,12 @@ +import os import sys +from collections.abc import Sequence +from typing import TypeAlias import trio +from typing_extensions import assert_type + +StrOrBytesPath: TypeAlias = str | bytes | os.PathLike[str] | os.PathLike[bytes] async def test() -> None: @@ -21,3 +27,15 @@ async def test() -> None: # 3.11+: await trio.run_process("python", process_group=5) # type: ignore await trio.lowlevel.open_process("python", process_group=5) # type: ignore + + +def check_processes(proc: trio.Process) -> None: + assert_type(proc.stdin, trio.abc.SendStream | None) + assert_type(proc.stdout, trio.abc.ReceiveStream | None) + assert_type(proc.stderr, trio.abc.ReceiveStream | None) + assert_type( + proc.stdio, + trio.StapledStream[trio.abc.SendStream, trio.abc.ReceiveStream] | None, + ) + assert_type(proc.args, StrOrBytesPath | Sequence[StrOrBytesPath]) + assert_type(proc.pid, int)