|
1 | 1 | # coding=utf-8 |
2 | 2 | """Shared utility functions""" |
3 | | - |
4 | 3 | import argparse |
5 | 4 | import collections |
6 | 5 | import functools |
|
17 | 16 | Enum, |
18 | 17 | ) |
19 | 18 | from typing import ( |
20 | | - IO, |
21 | 19 | TYPE_CHECKING, |
22 | 20 | Any, |
23 | 21 | Callable, |
|
45 | 43 | if TYPE_CHECKING: # pragma: no cover |
46 | 44 | import cmd2 # noqa: F401 |
47 | 45 |
|
| 46 | + PopenTextIO = subprocess.Popen[bytes] |
| 47 | + |
| 48 | +else: |
| 49 | + PopenTextIO = subprocess.Popen |
| 50 | + |
48 | 51 |
|
49 | 52 | _T = TypeVar('_T') |
50 | 53 |
|
@@ -445,7 +448,7 @@ class StdSim: |
445 | 448 |
|
446 | 449 | def __init__( |
447 | 450 | self, |
448 | | - inner_stream: Union[TextIO, 'StdSim', IO[str]], |
| 451 | + inner_stream: Union[TextIO, 'StdSim'], |
449 | 452 | *, |
450 | 453 | echo: bool = False, |
451 | 454 | encoding: str = 'utf-8', |
@@ -574,7 +577,7 @@ class ProcReader: |
574 | 577 | If neither are pipes, then the process will run normally and no output will be captured. |
575 | 578 | """ |
576 | 579 |
|
577 | | - def __init__(self, proc: subprocess.Popen, stdout: Union[StdSim, IO[str]], stderr: Union[StdSim, IO[str]]) -> None: |
| 580 | + def __init__(self, proc: PopenTextIO, stdout: Union[StdSim, TextIO], stderr: Union[StdSim, TextIO]) -> None: |
578 | 581 | """ |
579 | 582 | ProcReader initializer |
580 | 583 | :param proc: the Popen process being read from |
@@ -650,7 +653,7 @@ def _reader_thread_func(self, read_stdout: bool) -> None: |
650 | 653 | # Run until process completes |
651 | 654 | while self._proc.poll() is None: |
652 | 655 | # noinspection PyUnresolvedReferences |
653 | | - available = read_stream.peek() |
| 656 | + available = read_stream.peek() # type: ignore[attr-defined] |
654 | 657 | if available: |
655 | 658 | read_stream.read(len(available)) |
656 | 659 | self._write_bytes(write_stream, available) |
@@ -699,8 +702,8 @@ class RedirectionSavedState: |
699 | 702 |
|
700 | 703 | def __init__( |
701 | 704 | self, |
702 | | - self_stdout: Union[StdSim, IO[str]], |
703 | | - sys_stdout: Union[StdSim, IO[str]], |
| 705 | + self_stdout: Union[StdSim, TextIO], |
| 706 | + sys_stdout: Union[StdSim, TextIO], |
704 | 707 | pipe_proc_reader: Optional[ProcReader], |
705 | 708 | saved_redirecting: bool, |
706 | 709 | ) -> None: |
|
0 commit comments