Remove invoke stubs (#10956)

This commit is contained in:
Nikita Sobolev
2023-10-31 13:31:31 +03:00
committed by GitHub
parent 40e0423a86
commit a5d616303d
25 changed files with 0 additions and 837 deletions

View File

@@ -43,7 +43,6 @@
"stubs/httplib2",
"stubs/humanfriendly",
"stubs/influxdb-client",
"stubs/invoke",
"stubs/jmespath",
"stubs/jsonschema",
"stubs/ldap3",

View File

@@ -1,3 +0,0 @@
# Metaclass differs:
invoke.parser.ParseMachine
invoke.parser.parser.ParseMachine

View File

@@ -1,17 +0,0 @@
from __future__ import annotations
from invoke import Context, task
# ===========================================
# This snippet is a regression test for #8936
# ===========================================
@task
def docker_build(context: Context) -> None:
pass
@task(docker_build)
def docker_push(context: Context) -> None:
pass

View File

@@ -1,7 +0,0 @@
version = "2.0.*"
upstream_repository = "https://github.com/pyinvoke/invoke"
obsolete_since = "2.1.2" # Released on 2023-05-15
partial_stub = true
[tool.stubtest]
ignore_missing_stub = true

View File

@@ -1,37 +0,0 @@
from typing import Any
from .collection import Collection as Collection
from .config import Config as Config
from .context import Context as Context, MockContext as MockContext
from .exceptions import (
AmbiguousEnvVar as AmbiguousEnvVar,
AuthFailure as AuthFailure,
CollectionNotFound as CollectionNotFound,
CommandTimedOut as CommandTimedOut,
Exit as Exit,
Failure as Failure,
ParseError as ParseError,
PlatformError as PlatformError,
ResponseNotAccepted as ResponseNotAccepted,
SubprocessPipeError as SubprocessPipeError,
ThreadException as ThreadException,
UncastableEnvVar as UncastableEnvVar,
UnexpectedExit as UnexpectedExit,
UnknownFileType as UnknownFileType,
UnpicklableConfigMember as UnpicklableConfigMember,
WatcherError as WatcherError,
)
from .executor import Executor as Executor
from .loader import FilesystemLoader as FilesystemLoader
from .parser import Argument as Argument, Parser as Parser, ParserContext as ParserContext, ParseResult as ParseResult
from .program import Program as Program
from .runners import Local as Local, Promise as Promise, Result as Result, Runner as Runner
from .tasks import Call as Call, Task as Task, call as call, task as task
from .terminals import pty_size as pty_size
from .watchers import FailingResponder as FailingResponder, Responder as Responder, StreamWatcher as StreamWatcher
__version_info__: tuple[int, int, int]
__version__: str
def run(command: str, **kwargs: Any) -> Result: ...
def sudo(command: str, **kwargs: Any) -> Result: ...

View File

@@ -1,24 +0,0 @@
from typing import Any
class Collection:
tasks: Any
collections: Any
default: str | None
name: str | None
loaded_from: Any
auto_dash_names: bool
def __init__(self, *args, **kwargs) -> None: ...
@classmethod
def from_module(cls, module, name=None, config=None, loaded_from=None, auto_dash_names=None): ...
def add_task(self, task, name=None, aliases=None, default=None) -> None: ...
def add_collection(self, coll, name=None, default=None) -> None: ...
def subcollection_from_path(self, path): ...
def task_with_config(self, name): ...
def to_contexts(self, ignore_unknown_help: bool | None = None): ...
def subtask_name(self, collection_name, task_name): ...
def transform(self, name): ...
@property
def task_names(self): ...
def configuration(self, taskpath=None): ...
def configure(self, options) -> None: ...
def serialized(self): ...

View File

@@ -1,11 +0,0 @@
from collections.abc import Iterable, Sequence
from typing import NoReturn
from ..collection import Collection
from ..parser import Parser, ParserContext, ParseResult
def complete(
names: Iterable[str], core: ParseResult, initial_context: ParserContext, collection: Collection, parser: Parser
) -> NoReturn: ...
def print_task_names(collection: Collection) -> None: ...
def print_completion_script(shell: str, names: Sequence[str]) -> None: ...

View File

@@ -1,60 +0,0 @@
from typing import Any, ClassVar
def load_source(name: str, path: str) -> dict[str, Any]: ...
class DataProxy:
@classmethod
def from_data(cls, data, root=None, keypath=()): ...
def __getattr__(self, key: str): ...
def __setattr__(self, key: str, value) -> None: ...
def __iter__(self): ...
def __eq__(self, other): ...
__hash__: ClassVar[None] # type: ignore[assignment]
def __len__(self) -> int: ...
def __setitem__(self, key, value) -> None: ...
def __getitem__(self, key): ...
def __contains__(self, key): ...
def __delitem__(self, key) -> None: ...
def __delattr__(self, name: str) -> None: ...
def clear(self) -> None: ...
def pop(self, *args): ...
def popitem(self): ...
def setdefault(self, *args): ...
def update(self, *args, **kwargs) -> None: ...
class Config(DataProxy):
prefix: str
file_prefix: Any
env_prefix: Any
@staticmethod
def global_defaults(): ...
def __init__(
self,
overrides=None,
defaults=None,
system_prefix=None,
user_prefix=None,
project_location=None,
runtime_path=None,
lazy: bool = False,
) -> None: ...
def load_base_conf_files(self) -> None: ...
def load_defaults(self, data, merge: bool = True) -> None: ...
def load_overrides(self, data, merge: bool = True) -> None: ...
def load_system(self, merge: bool = True) -> None: ...
def load_user(self, merge: bool = True) -> None: ...
def load_project(self, merge: bool = True) -> None: ...
def set_runtime_path(self, path) -> None: ...
def load_runtime(self, merge: bool = True) -> None: ...
def load_shell_env(self) -> None: ...
def load_collection(self, data, merge: bool = True) -> None: ...
def set_project_location(self, path) -> None: ...
def merge(self) -> None: ...
def clone(self, into=None): ...
class AmbiguousMergeError(ValueError): ...
def merge_dicts(base, updates): ...
def copy_dict(source): ...
def excise(dict_, keypath) -> None: ...
def obliterate(base, deletions) -> None: ...

View File

@@ -1,23 +0,0 @@
import pathlib
from contextlib import AbstractContextManager
from .config import Config, DataProxy
class Context(DataProxy):
def __init__(self, config: Config | None = None) -> None: ...
@property
def config(self) -> Config: ...
@config.setter
def config(self, value: Config) -> None: ...
def run(self, command: str, **kwargs): ...
def sudo(self, command: str, *, password: str = ..., user: str = ..., **kwargs): ...
def prefix(self, command: str) -> AbstractContextManager[None]: ...
@property
def cwd(self) -> str: ...
def cd(self, path: str | pathlib.Path) -> AbstractContextManager[None]: ...
class MockContext(Context):
def __init__(self, config: Config | None = None, **kwargs) -> None: ...
def run(self, command: str, *args, **kwargs): ...
def sudo(self, command: str, *args, **kwargs): ...
def set_result_for(self, attname, command, result) -> None: ...

View File

@@ -1,6 +0,0 @@
from typing import Any
class Environment:
data: dict[str, Any]
def __init__(self, config: dict[str, Any], prefix: str) -> None: ...
def load(self) -> dict[str, Any]: ...

View File

@@ -1,47 +0,0 @@
from typing import Any
class CollectionNotFound(Exception):
name: Any
start: Any
def __init__(self, name, start) -> None: ...
class Failure(Exception):
result: Any
reason: Any
def __init__(self, result, reason=None) -> None: ...
def streams_for_display(self): ...
class UnexpectedExit(Failure): ...
class CommandTimedOut(Failure):
timeout: Any
def __init__(self, result, timeout) -> None: ...
class AuthFailure(Failure):
result: Any
prompt: Any
def __init__(self, result, prompt) -> None: ...
class ParseError(Exception):
context: Any
def __init__(self, msg, context=None) -> None: ...
class Exit(Exception):
message: Any
def __init__(self, message=None, code=None) -> None: ...
@property
def code(self): ...
class PlatformError(Exception): ...
class AmbiguousEnvVar(Exception): ...
class UncastableEnvVar(Exception): ...
class UnknownFileType(Exception): ...
class UnpicklableConfigMember(Exception): ...
class ThreadException(Exception):
exceptions: Any
def __init__(self, exceptions) -> None: ...
class WatcherError(Exception): ...
class ResponseNotAccepted(WatcherError): ...
class SubprocessPipeError(Exception): ...

View File

@@ -1,17 +0,0 @@
from collections.abc import Iterable
from typing import Any
from .collection import Collection
from .config import Config
from .parser import ParserContext, ParseResult
from .tasks import Call, Task
class Executor:
collection: Collection
config: Config
core: ParseResult | None
def __init__(self, collection: Collection, config: Config | None = None, core: ParseResult | None = None) -> None: ...
def execute(self, *tasks: str | tuple[str, dict[str, Any]] | ParserContext) -> dict[Task[..., Any], Any]: ...
def normalize(self, tasks: Iterable[str | tuple[str, dict[str, Any]] | ParserContext]): ...
def dedupe(self, calls: Iterable[Call]) -> list[Call]: ...
def expand_calls(self, calls: Iterable[Call | Task[..., Any]]) -> list[Call]: ...

View File

@@ -1,15 +0,0 @@
from types import ModuleType
from typing import IO, Any
from . import Config
class Loader:
config: Config
def __init__(self, config: Config | None = None) -> None: ...
def find(self, name: str) -> tuple[str, IO[Any], str, tuple[str, str, int]]: ...
def load(self, name: str | None = None) -> tuple[ModuleType, str]: ...
class FilesystemLoader(Loader):
def __init__(self, start: str | None = None, **kwargs: Any) -> None: ...
@property
def start(self) -> str: ...

View File

@@ -1,3 +0,0 @@
from . import Program
program: Program

View File

@@ -1,3 +0,0 @@
from .argument import Argument as Argument
from .context import ParserContext as ParserContext, to_flag as to_flag, translate_underscores as translate_underscores
from .parser import *

View File

@@ -1,37 +0,0 @@
from typing import Any
class Argument:
names: Any
kind: Any
raw_value: Any
default: Any
help: Any
positional: Any
optional: Any
incrementable: Any
attr_name: Any
def __init__(
self,
name=None,
names=(),
kind=...,
default=None,
help=None,
positional: bool = False,
optional: bool = False,
incrementable: bool = False,
attr_name=None,
) -> None: ...
@property
def name(self): ...
@property
def nicknames(self): ...
@property
def takes_value(self): ...
@property
def value(self): ...
@value.setter
def value(self, arg) -> None: ...
def set_value(self, value, cast: bool = True): ...
@property
def got_value(self): ...

View File

@@ -1,24 +0,0 @@
from typing import Any
def translate_underscores(name: str) -> str: ...
def to_flag(name: str) -> str: ...
def sort_candidate(arg): ...
def flag_key(x): ...
class ParserContext:
args: Any
positional_args: Any
flags: Any
inverse_flags: Any
name: Any
aliases: Any
def __init__(self, name=None, aliases=(), args=()) -> None: ...
def add_arg(self, *args, **kwargs) -> None: ...
@property
def missing_positional_args(self): ...
@property
def as_kwargs(self): ...
def names_for(self, flag): ...
def help_for(self, flag): ...
def help_tuples(self): ...
def flag_names(self): ...

View File

@@ -1,42 +0,0 @@
from typing import Any
from .context import ParserContext
def is_flag(value: str) -> bool: ...
def is_long_flag(value: str) -> bool: ...
class Parser:
initial: Any
contexts: Any
ignore_unknown: Any
def __init__(self, contexts=(), initial=None, ignore_unknown: bool = False) -> None: ...
def parse_argv(self, argv): ...
class ParseMachine:
initial_state: str
def changing_state(self, from_, to) -> None: ...
ignore_unknown: Any
initial: Any
flag: Any
flag_got_value: bool
result: Any
contexts: Any
def __init__(self, initial, contexts, ignore_unknown) -> None: ...
@property
def waiting_for_flag_value(self): ...
def handle(self, token) -> None: ...
def store_only(self, token) -> None: ...
def complete_context(self) -> None: ...
context: Any
def switch_to_context(self, name) -> None: ...
def complete_flag(self) -> None: ...
def check_ambiguity(self, value): ...
def switch_to_flag(self, flag, inverse: bool = False) -> None: ...
def see_value(self, value) -> None: ...
def see_positional_arg(self, value) -> None: ...
def error(self, msg) -> None: ...
class ParseResult(list[ParserContext]):
remainder: str
unparsed: Any
def __init__(self, *args, **kwargs) -> None: ...

View File

@@ -1,71 +0,0 @@
from typing import Any
class Program:
def core_args(self): ...
def task_args(self): ...
leading_indent_width: int
leading_indent: str
indent_width: int
indent: str
col_padding: int
version: Any
namespace: Any
argv: Any
loader_class: Any
executor_class: Any
config_class: Any
def __init__(
self,
version=None,
namespace=None,
name=None,
binary=None,
loader_class=None,
executor_class=None,
config_class=None,
binary_names=None,
) -> None: ...
config: Any
def create_config(self) -> None: ...
def update_config(self, merge: bool = True) -> None: ...
def run(self, argv=None, exit: bool = True) -> None: ...
def parse_core(self, argv) -> None: ...
collection: Any
list_root: Any
list_depth: Any
list_format: str
scoped_collection: Any
def parse_collection(self) -> None: ...
def parse_cleanup(self) -> None: ...
def no_tasks_given(self) -> None: ...
def execute(self) -> None: ...
def normalize_argv(self, argv) -> None: ...
@property
def name(self): ...
@property
def called_as(self): ...
@property
def binary(self): ...
@property
def binary_names(self): ...
@property
def args(self): ...
@property
def initial_context(self): ...
def print_version(self) -> None: ...
def print_help(self) -> None: ...
core: Any
def parse_core_args(self) -> None: ...
def load_collection(self) -> None: ...
parser: Any
core_via_tasks: Any
tasks: Any
def parse_tasks(self) -> None: ...
def print_task_help(self, name) -> None: ...
def list_tasks(self) -> None: ...
def list_flat(self) -> None: ...
def list_nested(self) -> None: ...
def list_json(self) -> None: ...
def task_list_opener(self, extra: str = ""): ...
def display_with_columns(self, pairs, extra: str = "") -> None: ...
def print_columns(self, tuples) -> None: ...

View File

@@ -1,203 +0,0 @@
from collections.abc import Iterable, Mapping
from types import TracebackType
from typing import Any, TextIO, overload
from typing_extensions import Literal, TypeAlias
from .watchers import StreamWatcher
_Hide: TypeAlias = Literal[None, True, False, "out", "stdout", "err", "stderr", "both"]
class Runner:
read_chunk_size: int
input_sleep: float
context: Any
program_finished: Any
warned_about_pty_fallback: bool
watchers: Any
def __init__(self, context) -> None: ...
# If disown is True (default=False), returns None
@overload
def run(
self,
command: str,
*,
asynchronous: bool = ...,
disown: Literal[True],
dry: bool = ...,
echo: bool = ...,
echo_format: str = ...,
echo_stdin: bool | None = ...,
encoding: str = ...,
err_stream: TextIO | None = ...,
env: Mapping[str, str] = ...,
fallback: bool = ...,
hide: _Hide = ...,
in_stream: TextIO | None | bool = ...,
out_stream: TextIO | None = ...,
pty: bool = ...,
replace_env: bool = ...,
shell: str = ...,
timeout: float | None = ...,
warn: bool = ...,
watchers: Iterable[StreamWatcher] = ...,
) -> None: ...
# If disown is False (the default), and asynchronous is True (default=False) returns Promise
@overload
def run(
self,
command: str,
*,
asynchronous: Literal[True],
disown: Literal[False] = False,
dry: bool = ...,
echo: bool = ...,
echo_format: str = ...,
echo_stdin: bool | None = ...,
encoding: str = ...,
err_stream: TextIO | None = ...,
env: Mapping[str, str] = ...,
fallback: bool = ...,
hide: _Hide = ...,
in_stream: TextIO | None | bool = ...,
out_stream: TextIO | None = ...,
pty: bool = ...,
replace_env: bool = ...,
shell: str = ...,
timeout: float | None = ...,
warn: bool = ...,
watchers: Iterable[StreamWatcher] = ...,
) -> Promise: ...
# If disown and asynchronous are both False (the defaults), returns Result
@overload
def run(
self,
command: str,
*,
asynchronous: Literal[False] = False,
disown: Literal[False] = False,
dry: bool = ...,
echo: bool = ...,
echo_format: str = ...,
echo_stdin: bool | None = ...,
encoding: str = ...,
err_stream: TextIO | None = ...,
env: Mapping[str, str] = ...,
fallback: bool = ...,
hide: _Hide = ...,
in_stream: TextIO | None | bool = ...,
out_stream: TextIO | None = ...,
pty: bool = ...,
replace_env: bool = ...,
shell: str = ...,
timeout: float | None = ...,
warn: bool = ...,
watchers: Iterable[StreamWatcher] = ...,
) -> Result: ...
# Fallback overload: return Any
@overload
def run(
self,
command: str,
*,
asynchronous: bool,
disown: bool,
dry: bool = ...,
echo: bool = ...,
echo_format: str = ...,
echo_stdin: bool | None = ...,
encoding: str = ...,
err_stream: TextIO | None = ...,
env: Mapping[str, str] = ...,
fallback: bool = ...,
hide: _Hide = ...,
in_stream: TextIO | None | bool = ...,
out_stream: TextIO | None = ...,
pty: bool = ...,
replace_env: bool = ...,
shell: str = ...,
timeout: float | None = ...,
warn: bool = ...,
watchers: Iterable[StreamWatcher] = ...,
) -> Any: ...
def echo(self, command) -> None: ...
def make_promise(self): ...
def create_io_threads(self): ...
def generate_result(self, **kwargs): ...
def read_proc_output(self, reader) -> None: ...
def write_our_output(self, stream, string) -> None: ...
def handle_stdout(self, buffer_, hide, output) -> None: ...
def handle_stderr(self, buffer_, hide, output) -> None: ...
def read_our_stdin(self, input_): ...
def handle_stdin(self, input_, output, echo) -> None: ...
def should_echo_stdin(self, input_, output): ...
def respond(self, buffer_) -> None: ...
def generate_env(self, env, replace_env): ...
def should_use_pty(self, pty, fallback): ...
@property
def has_dead_threads(self): ...
def wait(self) -> None: ...
def write_proc_stdin(self, data) -> None: ...
def decode(self, data): ...
@property
def process_is_finished(self) -> None: ...
def start(self, command, shell, env) -> None: ...
def start_timer(self, timeout) -> None: ...
def read_proc_stdout(self, num_bytes) -> None: ...
def read_proc_stderr(self, num_bytes) -> None: ...
def close_proc_stdin(self) -> None: ...
def default_encoding(self): ...
def send_interrupt(self, interrupt) -> None: ...
def returncode(self) -> None: ...
def stop(self) -> None: ...
def kill(self) -> None: ...
@property
def timed_out(self): ...
class Local(Runner):
status: Any
def __init__(self, context) -> None: ...
def should_use_pty(self, pty: bool = False, fallback: bool = True): ...
process: Any
class Result:
stdout: str
stderr: str
encoding: str
command: str
shell: Any
env: dict[str, Any]
exited: int
pty: bool
hide: tuple[Literal["stdout", "stderr"], ...]
def __init__(
self,
stdout: str = "",
stderr: str = "",
encoding: str | None = None,
command: str = "",
shell: str = "",
env=None,
exited: int = 0,
pty: bool = False,
hide: tuple[Literal["stdout", "stderr"], ...] = (),
) -> None: ...
@property
def return_code(self) -> int: ...
def __bool__(self) -> bool: ...
@property
def ok(self) -> bool: ...
@property
def failed(self) -> bool: ...
def tail(self, stream: Literal["stderr", "stdout"], count: int = 10) -> str: ...
class Promise(Result):
runner: Any
def __init__(self, runner) -> None: ...
def join(self): ...
def __enter__(self): ...
def __exit__(
self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None
) -> None: ...
def normalize_hide(val, out_stream=None, err_stream=None): ...
def default_encoding() -> str: ...

View File

@@ -1,115 +0,0 @@
from _typeshed import Incomplete
from collections.abc import Callable, Iterable
from typing import Any, Generic, TypeVar, overload
from typing_extensions import ParamSpec, Self
from .config import Config
from .context import Context
from .parser import Argument
_P = ParamSpec("_P")
_R_co = TypeVar("_R_co", covariant=True)
_TaskT = TypeVar("_TaskT", bound=Task[..., Any])
class Task(Generic[_P, _R_co]):
body: Callable[_P, _R_co]
__doc__: str | None
__name__: str
__module__: str
aliases: tuple[str, ...]
is_default: bool
positional: Iterable[str]
optional: Iterable[str]
iterable: Iterable[str]
incrementable: Iterable[str]
auto_shortflags: bool
help: dict[str, str]
pre: Iterable[Task[..., Any] | Call]
post: Iterable[Task[..., Any] | Call]
times_called: int
autoprint: bool
def __init__(
self,
body: Callable[..., Any],
name: str | None = None,
aliases: tuple[str, ...] = (),
positional: Iterable[str] | None = None,
optional: Iterable[str] = (),
default: bool = False,
auto_shortflags: bool = True,
help: dict[str, str] | None = None,
pre: Iterable[Task[..., Any] | Call] | None = None,
post: Iterable[Task[..., Any] | Call] | None = None,
autoprint: bool = False,
iterable: Iterable[str] | None = None,
incrementable: Iterable[str] | None = None,
) -> None: ...
@property
def name(self): ...
def __eq__(self, other: Task[Incomplete, Incomplete]) -> bool: ... # type: ignore[override]
def __hash__(self) -> int: ...
def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R_co: ...
@property
def called(self) -> bool: ...
def argspec(self, body): ...
def fill_implicit_positionals(self, positional: Iterable[str] | None) -> Iterable[str]: ...
def arg_opts(self, name: str, default: Any, taken_names: Iterable[str]) -> dict[str, Any]: ...
def get_arguments(self, ignore_unknown_help: bool | None = None) -> list[Argument]: ...
@overload
def task( # type: ignore[misc]
*args: Task[..., Any] | Call,
name: str | None = ...,
aliases: tuple[str, ...] = ...,
positional: Iterable[str] | None = ...,
optional: Iterable[str] = ...,
default: bool = ...,
auto_shortflags: bool = ...,
help: dict[str, str] | None = ...,
pre: list[Task[..., Any] | Call] | None = ...,
post: list[Task[..., Any] | Call] | None = ...,
autoprint: bool = ...,
iterable: Iterable[str] | None = ...,
incrementable: Iterable[str] | None = ...,
) -> Callable[[Callable[_P, _R_co]], Task[_P, _R_co]]: ...
@overload
def task(
*args: Task[..., Any] | Call,
name: str | None = ...,
aliases: tuple[str, ...] = ...,
positional: Iterable[str] | None = ...,
optional: Iterable[str] = ...,
default: bool = ...,
auto_shortflags: bool = ...,
help: dict[str, str] | None = ...,
pre: list[Task[..., Any] | Call] | None = ...,
post: list[Task[..., Any] | Call] | None = ...,
autoprint: bool = ...,
iterable: Iterable[str] | None = ...,
incrementable: Iterable[str] | None = ...,
klass: type[_TaskT],
) -> Callable[[Callable[..., Any]], _TaskT]: ...
@overload
def task(__func: Callable[_P, _R_co]) -> Task[_P, _R_co]: ...
class Call:
task: Task[..., Any]
called_as: str | None
args: tuple[Any, ...]
kwargs: dict[str, Any]
def __init__(
self,
task: Task[..., Any],
called_as: str | None = None,
args: tuple[Any, ...] | None = None,
kwargs: dict[str, Any] | None = None,
) -> None: ...
def __getattr__(self, name: str) -> Any: ...
def __deepcopy__(self, memo: Any) -> Self: ...
def __eq__(self, other: Call) -> bool: ... # type: ignore[override]
def make_context(self, config: Config) -> Context: ...
def clone_data(self): ...
# TODO use overload
def clone(self, into: type[Call] | None = None, with_: dict[str, Any] | None = None) -> Call: ...
def call(task: Task[..., Any], *args: Any, **kwargs: Any) -> Call: ...

View File

@@ -1,12 +0,0 @@
from contextlib import AbstractContextManager
from io import TextIOWrapper
from typing import Any
WINDOWS: bool
def pty_size() -> tuple[int, int]: ...
def stdin_is_foregrounded_tty(stream: Any) -> bool: ...
def cbreak_already_set(stream: TextIOWrapper) -> bool: ...
def character_buffered(stream: TextIOWrapper) -> AbstractContextManager[None]: ...
def ready_for_reading(input_: TextIOWrapper) -> bool: ...
def bytes_to_read(input_: TextIOWrapper) -> int: ...

View File

@@ -1,39 +0,0 @@
import threading
from collections.abc import Callable, Iterable, Mapping
from contextlib import AbstractContextManager
from logging import Logger
from types import TracebackType
from typing import Any, NamedTuple
LOG_FORMAT: str
def enable_logging() -> None: ...
log: Logger
def task_name_sort_key(name: str) -> tuple[list[str], str]: ...
def cd(where: str) -> AbstractContextManager[None]: ...
def has_fileno(stream) -> bool: ...
def isatty(stream) -> bool: ...
def helpline(obj: Callable[..., object]) -> str | None: ...
class ExceptionHandlingThread(threading.Thread):
def __init__(
self,
*,
group: None = None,
target: Callable[..., object] | None = ...,
name: str | None = ...,
args: Iterable[Any] = ...,
kwargs: Mapping[str, Any] | None = ...,
daemon: bool | None = ...,
) -> None: ...
def exception(self) -> ExceptionWrapper | None: ...
@property
def is_dead(self) -> bool: ...
class ExceptionWrapper(NamedTuple):
kwargs: Any
type: type[BaseException]
value: BaseException
traceback: TracebackType

View File

@@ -1,20 +0,0 @@
import threading
from collections.abc import Iterable
class StreamWatcher(threading.local):
def submit(self, stream) -> Iterable[str]: ...
class Responder(StreamWatcher):
pattern: str
response: str
index: int
def __init__(self, pattern: str, response: str) -> None: ...
def pattern_matches(self, stream: str, pattern: str, index_attr: str) -> Iterable[str]: ...
def submit(self, stream: str) -> Iterable[str]: ...
class FailingResponder(Responder):
sentinel: str
failure_index: int
tried: bool
def __init__(self, pattern: str, response: str, sentinel: str) -> None: ...
def submit(self, stream: str) -> Iterable[str]: ...