Add pexpect stubs (#10300)

pyrightconfig.stricter.json

@@ -56,6 +56,7 @@
         "stubs/inifile",
         "stubs/passlib",
         "stubs/peewee",
+        "stubs/pexpect",
         "stubs/pika",
         "stubs/psutil",
         "stubs/psycopg2",

stubs/pexpect/METADATA.toml

@@ -0,0 +1 @@
+version = "4.8.*"

stubs/pexpect/pexpect/ANSI.pyi

@@ -0,0 +1,43 @@
+from _typeshed import Incomplete
+
+from . import screen
+
+def DoEmit(fsm) -> None: ...
+def DoStartNumber(fsm) -> None: ...
+def DoBuildNumber(fsm) -> None: ...
+def DoBackOne(fsm) -> None: ...
+def DoBack(fsm) -> None: ...
+def DoDownOne(fsm) -> None: ...
+def DoDown(fsm) -> None: ...
+def DoForwardOne(fsm) -> None: ...
+def DoForward(fsm) -> None: ...
+def DoUpReverse(fsm) -> None: ...
+def DoUpOne(fsm) -> None: ...
+def DoUp(fsm) -> None: ...
+def DoHome(fsm) -> None: ...
+def DoHomeOrigin(fsm) -> None: ...
+def DoEraseDown(fsm) -> None: ...
+def DoErase(fsm) -> None: ...
+def DoEraseEndOfLine(fsm) -> None: ...
+def DoEraseLine(fsm) -> None: ...
+def DoEnableScroll(fsm) -> None: ...
+def DoCursorSave(fsm) -> None: ...
+def DoCursorRestore(fsm) -> None: ...
+def DoScrollRegion(fsm) -> None: ...
+def DoMode(fsm) -> None: ...
+def DoLog(fsm) -> None: ...
+
+class term(screen.screen):
+    def __init__(self, r: int = 24, c: int = 80, *args, **kwargs) -> None: ...
+
+class ANSI(term):
+    state: Incomplete
+    def __init__(self, r: int = 24, c: int = 80, *args, **kwargs) -> None: ...
+    def process(self, c) -> None: ...
+    def process_list(self, l) -> None: ...
+    def write(self, s) -> None: ...
+    def flush(self) -> None: ...
+    def write_ch(self, ch) -> None: ...
+    def do_sgr(self, fsm) -> None: ...
+    def do_decsca(self, fsm) -> None: ...
+    def do_modecrap(self, fsm) -> None: ...

stubs/pexpect/pexpect/FSM.pyi

@@ -0,0 +1,39 @@
+from _typeshed import Incomplete
+
+class ExceptionFSM(Exception):
+    value: Incomplete
+    def __init__(self, value) -> None: ...
+
+class FSM:
+    state_transitions: Incomplete
+    state_transitions_any: Incomplete
+    default_transition: Incomplete
+    input_symbol: Incomplete
+    initial_state: Incomplete
+    current_state: Incomplete
+    next_state: Incomplete
+    action: Incomplete
+    memory: Incomplete
+    def __init__(self, initial_state, memory: Incomplete | None = None) -> None: ...
+    def reset(self) -> None: ...
+    def add_transition(
+        self, input_symbol, state, action: Incomplete | None = None, next_state: Incomplete | None = None
+    ) -> None: ...
+    def add_transition_list(
+        self, list_input_symbols, state, action: Incomplete | None = None, next_state: Incomplete | None = None
+    ) -> None: ...
+    def add_transition_any(self, state, action: Incomplete | None = None, next_state: Incomplete | None = None) -> None: ...
+    def set_default_transition(self, action, next_state) -> None: ...
+    def get_transition(self, input_symbol, state): ...
+    def process(self, input_symbol) -> None: ...
+    def process_list(self, input_symbols) -> None: ...
+
+PY3: Incomplete
+
+def BeginBuildNumber(fsm) -> None: ...
+def BuildNumber(fsm) -> None: ...
+def EndBuildNumber(fsm) -> None: ...
+def DoOperator(fsm) -> None: ...
+def DoEqual(fsm) -> None: ...
+def Error(fsm) -> None: ...
+def main() -> None: ...

stubs/pexpect/pexpect/__init__.pyi

@@ -0,0 +1,20 @@
+from .exceptions import EOF as EOF, TIMEOUT as TIMEOUT, ExceptionPexpect as ExceptionPexpect
+from .pty_spawn import spawn as spawn, spawnu as spawnu
+from .run import run as run, runu as runu
+from .utils import split_command_line as split_command_line, which as which
+
+__version__: str
+__revision__: str
+__all__ = [
+    "ExceptionPexpect",
+    "EOF",
+    "TIMEOUT",
+    "spawn",
+    "spawnu",
+    "run",
+    "runu",
+    "which",
+    "split_command_line",
+    "__version__",
+    "__revision__",
+]

stubs/pexpect/pexpect/_async.pyi

@@ -0,0 +1,17 @@
+import asyncio
+from _typeshed import Incomplete
+
+async def expect_async(expecter, timeout: Incomplete | None = None): ...
+async def repl_run_command_async(repl, cmdlines, timeout: int = -1): ...
+
+class PatternWaiter(asyncio.Protocol):
+    transport: Incomplete
+    expecter: Incomplete
+    fut: Incomplete
+    def set_expecter(self, expecter) -> None: ...
+    def found(self, result) -> None: ...
+    def error(self, exc) -> None: ...
+    def connection_made(self, transport) -> None: ...
+    def data_received(self, data) -> None: ...
+    def eof_received(self) -> None: ...
+    def connection_lost(self, exc) -> None: ...

stubs/pexpect/pexpect/exceptions.pyi

@@ -0,0 +1,6 @@
+class ExceptionPexpect(Exception):
+    def __init__(self, value: str) -> None: ...
+    def get_trace(self): ...
+
+class EOF(ExceptionPexpect): ...
+class TIMEOUT(ExceptionPexpect): ...

stubs/pexpect/pexpect/expect.pyi

@@ -0,0 +1,38 @@
+from _typeshed import Incomplete
+from io import BytesIO, StringIO
+
+from .exceptions import EOF, TIMEOUT
+from .spawnbase import SpawnBase
+
+class searcher_string:
+    eof_index: int
+    timeout_index: int
+    longest_string: int
+    def __init__(self, strings) -> None: ...
+    match: Incomplete
+    start: Incomplete
+    end: Incomplete
+    def search(self, buffer, freshlen, searchwindowsize: Incomplete | None = None): ...
+
+class searcher_re:
+    eof_index: int
+    timeout_index: int
+    def __init__(self, patterns) -> None: ...
+    start: Incomplete
+    match: Incomplete
+    end: Incomplete
+    def search(self, buffer, freshlen: int, searchwindowsize: int | None = None): ...
+
+class Expecter:
+    spawn: BytesIO | StringIO
+    searcher: searcher_re | searcher_string
+    searchwindowsize: int | None
+    lookback: searcher_string | searcher_re | int | None
+    def __init__(self, spawn: SpawnBase, searcher: searcher_re | searcher_string, searchwindowsize: int = -1) -> None: ...
+    def do_search(self, window: str, freshlen: int): ...
+    def existing_data(self): ...
+    def new_data(self, data: Incomplete): ...
+    def eof(self, err: Incomplete | None = None) -> int | EOF: ...
+    def timeout(self, err: object | None = None) -> int | TIMEOUT: ...
+    def errored(self) -> None: ...
+    def expect_loop(self, timeout: int = -1): ...

stubs/pexpect/pexpect/fdpexpect.pyi

@@ -0,0 +1,32 @@
+from _typeshed import Incomplete
+
+from .spawnbase import SpawnBase
+
+class fdspawn(SpawnBase):
+    args: Incomplete
+    command: Incomplete
+    child_fd: Incomplete
+    own_fd: bool
+    closed: bool
+    name: Incomplete
+    use_poll: Incomplete
+    def __init__(
+        self,
+        fd,
+        args: Incomplete | None = None,
+        timeout: int = 30,
+        maxread: int = 2000,
+        searchwindowsize: Incomplete | None = None,
+        logfile: Incomplete | None = None,
+        encoding: Incomplete | None = None,
+        codec_errors: str = "strict",
+        use_poll: bool = False,
+    ) -> None: ...
+    def close(self) -> None: ...
+    def isalive(self) -> bool: ...
+    def terminate(self, force: bool = False) -> None: ...
+    def send(self, s: str | bytes) -> bytes: ...
+    def sendline(self, s: str | bytes) -> bytes: ...
+    def write(self, s) -> None: ...
+    def writelines(self, sequence) -> None: ...
+    def read_nonblocking(self, size: int = 1, timeout: int | None = -1) -> bytes: ...

stubs/pexpect/pexpect/popen_spawn.pyi

@@ -0,0 +1,34 @@
+from _typeshed import Incomplete
+
+from .spawnbase import SpawnBase
+
+class PopenSpawn(SpawnBase):
+    crlf: Incomplete
+    proc: Incomplete
+    pid: Incomplete
+    closed: bool
+    def __init__(
+        self,
+        cmd,
+        timeout: int = 30,
+        maxread: int = 2000,
+        searchwindowsize: Incomplete | None = None,
+        logfile: Incomplete | None = None,
+        cwd: Incomplete | None = None,
+        env: Incomplete | None = None,
+        encoding: Incomplete | None = None,
+        codec_errors: str = "strict",
+        preexec_fn: Incomplete | None = None,
+    ) -> None: ...
+    flag_eof: bool
+    def read_nonblocking(self, size, timeout): ...
+    def write(self, s) -> None: ...
+    def writelines(self, sequence) -> None: ...
+    def send(self, s): ...
+    def sendline(self, s: str = ""): ...
+    exitstatus: Incomplete
+    signalstatus: Incomplete
+    terminated: bool
+    def wait(self): ...
+    def kill(self, sig) -> None: ...
+    def sendeof(self) -> None: ...

stubs/pexpect/pexpect/pty_spawn.pyi

@@ -0,0 +1,73 @@
+from _typeshed import Incomplete
+from collections.abc import Callable
+from io import TextIOWrapper
+from os import _Environ
+
+from .spawnbase import SpawnBase
+
+PY3: Incomplete
+
+class spawn(SpawnBase):
+    use_native_pty_fork: bool
+    STDIN_FILENO: Incomplete
+    STDOUT_FILENO: Incomplete
+    STDERR_FILENO: Incomplete
+    str_last_chars: int
+    env: Incomplete
+    name: str
+    def __init__(
+        self,
+        command: str,
+        args: list[str] = [],
+        timeout: int = 30,
+        maxread: int = 2000,
+        searchwindowsize: int | None = None,
+        logfile: TextIOWrapper | None = None,
+        cwd: str | bytes | None = None,
+        env: _Environ[Incomplete] | None = None,
+        ignore_sighup: bool = False,
+        echo: bool = True,
+        preexec_fn: Callable[[Incomplete], Incomplete] | None = None,
+        encoding: str | None = None,
+        codec_errors: str = "strict",
+        dimensions: tuple[int, int] | None = None,
+        use_poll: bool = False,
+    ) -> None: ...
+    child_fd: int
+    closed: bool
+    def close(self, force: bool = True) -> None: ...
+    def isatty(self): ...
+    def waitnoecho(self, timeout: int = -1): ...
+    def getecho(self): ...
+    def setecho(self, state: bool): ...
+    def read_nonblocking(self, size: int = 1, timeout: int | None = -1) -> bytes: ...
+    def write(self, s: str | bytes) -> None: ...
+    def writelines(self, sequence: list[str | bytes]) -> None: ...
+    def send(self, s: str | bytes): ...
+    def sendline(self, s: str | bytes = ""): ...
+    def sendcontrol(self, char: str): ...
+    def sendeof(self) -> None: ...
+    def sendintr(self) -> None: ...
+    @property
+    def flag_eof(self) -> bool: ...
+    @flag_eof.setter
+    def flag_eof(self, value: bool) -> None: ...
+    def eof(self) -> bool: ...
+    def terminate(self, force: bool = False): ...
+    status: int | None
+    exitstatus: bool | None
+    signalstatus: int | None
+    terminated: bool
+    def wait(self) -> int: ...
+    def isalive(self) -> bool: ...
+    def kill(self, sig: int) -> None: ...
+    def getwinsize(self) -> tuple[int, int]: ...
+    def setwinsize(self, rows, cols) -> None: ...
+    def interact(
+        self,
+        escape_character="\x1d",
+        input_filter: Callable[[Incomplete], Incomplete] | None = None,
+        output_filter: Callable[[Incomplete], Incomplete] | None = None,
+    ) -> None: ...
+
+def spawnu(*args: str, **kwargs: str): ...

stubs/pexpect/pexpect/pxssh.pyi

@@ -0,0 +1,59 @@
+from _typeshed import Incomplete
+
+from pexpect import ExceptionPexpect, spawn
+
+class ExceptionPxssh(ExceptionPexpect): ...
+
+class pxssh(spawn):
+    name: str
+    UNIQUE_PROMPT: str
+    PROMPT: Incomplete
+    PROMPT_SET_SH: str
+    PROMPT_SET_CSH: str
+    SSH_OPTS: Incomplete
+    force_password: bool
+    debug_command_string: Incomplete
+    options: Incomplete
+    def __init__(
+        self,
+        timeout: int = 30,
+        maxread: int = 2000,
+        searchwindowsize: Incomplete | None = None,
+        logfile: Incomplete | None = None,
+        cwd: Incomplete | None = None,
+        env: Incomplete | None = None,
+        ignore_sighup: bool = True,
+        echo: bool = True,
+        options={},
+        encoding: Incomplete | None = None,
+        codec_errors: str = "strict",
+        debug_command_string: bool = False,
+        use_poll: bool = False,
+    ) -> None: ...
+    def levenshtein_distance(self, a, b): ...
+    def try_read_prompt(self, timeout_multiplier): ...
+    def sync_original_prompt(self, sync_multiplier: float = 1.0): ...
+    def login(
+        self,
+        server,
+        username: Incomplete | None = None,
+        password: str = "",
+        terminal_type: str = "ansi",
+        original_prompt: str = "[#$]",
+        login_timeout: int = 10,
+        port: Incomplete | None = None,
+        auto_prompt_reset: bool = True,
+        ssh_key: Incomplete | None = None,
+        quiet: bool = True,
+        sync_multiplier: int = 1,
+        check_local_ip: bool = True,
+        password_regex: str = "(?i)(?:password:)|(?:passphrase for key)",
+        ssh_tunnels={},
+        spawn_local_ssh: bool = True,
+        sync_original_prompt: bool = True,
+        ssh_config: Incomplete | None = None,
+        cmd: str = "ssh",
+    ): ...
+    def logout(self) -> None: ...
+    def prompt(self, timeout: int = -1): ...
+    def set_unique_prompt(self): ...

stubs/pexpect/pexpect/replwrap.pyi

@@ -0,0 +1,25 @@
+from _typeshed import Incomplete
+
+PY3: Incomplete
+basestring = str
+PEXPECT_PROMPT: str
+PEXPECT_CONTINUATION_PROMPT: str
+
+class REPLWrapper:
+    child: Incomplete
+    prompt: Incomplete
+    continuation_prompt: Incomplete
+    def __init__(
+        self,
+        cmd_or_spawn,
+        orig_prompt,
+        prompt_change,
+        new_prompt="[PEXPECT_PROMPT>",
+        continuation_prompt="[PEXPECT_PROMPT+",
+        extra_init_cmd: Incomplete | None = None,
+    ) -> None: ...
+    def set_prompt(self, orig_prompt, prompt_change) -> None: ...
+    def run_command(self, command, timeout: int = -1, async_: bool = False): ...
+
+def python(command: str = "python"): ...
+def bash(command: str = "bash"): ...

stubs/pexpect/pexpect/run.pyi

@@ -0,0 +1,24 @@
+from _typeshed import Incomplete
+
+def run(
+    command,
+    timeout: int = 30,
+    withexitstatus: bool = False,
+    events: Incomplete | None = None,
+    extra_args: Incomplete | None = None,
+    logfile: Incomplete | None = None,
+    cwd: Incomplete | None = None,
+    env: Incomplete | None = None,
+    **kwargs,
+): ...
+def runu(
+    command,
+    timeout: int = 30,
+    withexitstatus: bool = False,
+    events: Incomplete | None = None,
+    extra_args: Incomplete | None = None,
+    logfile: Incomplete | None = None,
+    cwd: Incomplete | None = None,
+    env: Incomplete | None = None,
+    **kwargs,
+): ...

stubs/pexpect/pexpect/screen.pyi

@@ -0,0 +1,80 @@
+from _typeshed import Incomplete
+
+NUL: int
+ENQ: int
+BEL: int
+BS: int
+HT: int
+LF: int
+VT: int
+FF: int
+CR: int
+SO: int
+SI: int
+XON: int
+XOFF: int
+CAN: int
+SUB: int
+ESC: int
+DEL: int
+SPACE: str
+PY3: Incomplete
+unicode = str
+
+def constrain(n, min, max): ...
+
+class screen:
+    rows: Incomplete
+    cols: Incomplete
+    encoding: Incomplete
+    encoding_errors: Incomplete
+    decoder: Incomplete
+    cur_r: int
+    cur_c: int
+    cur_saved_r: int
+    cur_saved_c: int
+    scroll_row_start: int
+    scroll_row_end: Incomplete
+    w: Incomplete
+    def __init__(self, r: int = 24, c: int = 80, encoding: str = "latin-1", encoding_errors: str = "replace") -> None: ...
+    def dump(self): ...
+    def pretty(self): ...
+    def fill(self, ch=" ") -> None: ...
+    def fill_region(self, rs, cs, re, ce, ch=" ") -> None: ...
+    def cr(self) -> None: ...
+    def lf(self) -> None: ...
+    def crlf(self) -> None: ...
+    def newline(self) -> None: ...
+    def put_abs(self, r, c, ch) -> None: ...
+    def put(self, ch) -> None: ...
+    def insert_abs(self, r, c, ch) -> None: ...
+    def insert(self, ch) -> None: ...
+    def get_abs(self, r, c): ...
+    def get(self) -> None: ...
+    def get_region(self, rs, cs, re, ce): ...
+    def cursor_constrain(self) -> None: ...
+    def cursor_home(self, r: int = 1, c: int = 1) -> None: ...
+    def cursor_back(self, count: int = 1) -> None: ...
+    def cursor_down(self, count: int = 1) -> None: ...
+    def cursor_forward(self, count: int = 1) -> None: ...
+    def cursor_up(self, count: int = 1) -> None: ...
+    def cursor_up_reverse(self) -> None: ...
+    def cursor_force_position(self, r, c) -> None: ...
+    def cursor_save(self) -> None: ...
+    def cursor_unsave(self) -> None: ...
+    def cursor_save_attrs(self) -> None: ...
+    def cursor_restore_attrs(self) -> None: ...
+    def scroll_constrain(self) -> None: ...
+    def scroll_screen(self) -> None: ...
+    def scroll_screen_rows(self, rs, re) -> None: ...
+    def scroll_down(self) -> None: ...
+    def scroll_up(self) -> None: ...
+    def erase_end_of_line(self) -> None: ...
+    def erase_start_of_line(self) -> None: ...
+    def erase_line(self) -> None: ...
+    def erase_down(self) -> None: ...
+    def erase_up(self) -> None: ...
+    def erase_screen(self) -> None: ...
+    def set_tab(self) -> None: ...
+    def clear_tab(self) -> None: ...
+    def clear_all_tabs(self) -> None: ...

stubs/pexpect/pexpect/spawnbase.pyi

@@ -0,0 +1,79 @@
+from _typeshed import Incomplete
+from collections.abc import Callable
+from re import Pattern
+from typing import AnyStr
+
+PY3: bool
+text_type: Callable[[Incomplete], Incomplete]
+
+class _NullCoder:
+    @staticmethod
+    def encode(b: str, final: bool = False): ...
+    @staticmethod
+    def decode(b: str, final: bool = False): ...
+
+class SpawnBase:
+    encoding: Incomplete
+    pid: Incomplete
+    flag_eof: bool
+    stdin: Incomplete
+    stdout: Incomplete
+    stderr: Incomplete
+    searcher: Incomplete
+    ignorecase: bool
+    before: Incomplete
+    after: Incomplete
+    match: Incomplete
+    match_index: Incomplete
+    terminated: bool
+    exitstatus: Incomplete
+    signalstatus: Incomplete
+    status: Incomplete
+    child_fd: int
+    timeout: Incomplete
+    delimiter: Incomplete
+    logfile: Incomplete
+    logfile_read: Incomplete
+    logfile_send: Incomplete
+    maxread: Incomplete
+    searchwindowsize: Incomplete
+    delaybeforesend: float
+    delayafterclose: float
+    delayafterterminate: float
+    delayafterread: float
+    softspace: bool
+    name: Incomplete
+    closed: bool
+    codec_errors: Incomplete
+    string_type: Incomplete
+    buffer_type: Incomplete
+    crlf: bytes
+    allowed_string_types: Incomplete
+    linesep: Incomplete
+    write_to_stdout: Incomplete
+    async_pw_transport: Incomplete
+    def __init__(
+        self,
+        timeout: int = 30,
+        maxread: int = 2000,
+        searchwindowsize: Incomplete | None = None,
+        logfile: Incomplete | None = None,
+        encoding: Incomplete | None = None,
+        codec_errors: str = "strict",
+    ) -> None: ...
+    buffer: Incomplete
+    def read_nonblocking(self, size: int = 1, timeout: int | None = None) -> bytes: ...
+    def compile_pattern_list(self, patterns) -> list[Pattern[AnyStr]]: ...
+    def expect(self, pattern, timeout: int = -1, searchwindowsize: int = -1, async_: bool = False, **kw) -> int: ...
+    def expect_list(self, pattern_list, timeout: int = -1, searchwindowsize: int = -1, async_: bool = False, **kw) -> int: ...
+    def expect_exact(self, pattern_list, timeout: int = -1, searchwindowsize: int = -1, async_: bool = False, **kw) -> int: ...
+    def expect_loop(self, searcher, timeout: int = -1, searchwindowsize: int = -1) -> int: ...
+    def read(self, size: int = -1) -> bytes: ...
+    def readline(self, size: int = -1) -> bytes: ...
+    def __iter__(self): ...
+    def readlines(self, sizehint: int = -1) -> list[str]: ...
+    def fileno(self): ...
+    def flush(self) -> None: ...
+    def isatty(self): ...
+    def __enter__(self): ...
+    def __exit__(self, etype, evalue, tb) -> None: ...

stubs/pexpect/pexpect/utils.pyi

@@ -0,0 +1,9 @@
+from _typeshed import Incomplete
+
+string_types: Incomplete
+
+def is_executable_file(path): ...
+def which(filename, env: Incomplete | None = None): ...
+def split_command_line(command_line): ...
+def select_ignore_interrupts(iwtd, owtd, ewtd, timeout: Incomplete | None = None): ...
+def poll_ignore_interrupts(fds, timeout: Incomplete | None = None): ...