diff --git a/pmb/helpers/apk.py b/pmb/helpers/apk.py index 29bf1c81..858ad975 100644 --- a/pmb/helpers/apk.py +++ b/pmb/helpers/apk.py @@ -102,7 +102,7 @@ def _prepare_fifo() -> Path: return fifo -def _create_command_with_progress(command, fifo): +def _create_command_with_progress(command: list[str], fifo: Path) -> list[str]: """Build a full apk command from a subcommand, set up to redirect progress into a fifo. :param command: apk subcommand in list form @@ -132,7 +132,7 @@ def _compute_progress(line: str) -> float: return cur / tot if tot > 0 else 0 -def _apk_with_progress(command: Sequence[str]) -> None: +def _apk_with_progress(command: list[str]) -> None: """Run an apk subcommand while printing a progress bar to STDOUT. :param command: apk subcommand in list form diff --git a/pmb/helpers/apk_static.py b/pmb/helpers/apk_static.py index 6bc4115d..c31fc4cc 100644 --- a/pmb/helpers/apk_static.py +++ b/pmb/helpers/apk_static.py @@ -55,7 +55,7 @@ def read_signature_info(tar: tarfile.TarFile) -> tuple[str, str]: return (sigfilename, sigkey_path) -def extract_temp(tar, sigfilename): +def extract_temp(tar: tarfile.TarFile, sigfilename: str) -> dict[str, dict]: """ Extract apk.static and signature as temporary files. """ @@ -64,19 +64,25 @@ def extract_temp(tar, sigfilename): "sig": {"filename": sigfilename, "temp_path": None}, } for ftype in ret.keys(): - member = tar.getmember(ret[ftype]["filename"]) + filename = ret[ftype]["filename"] + if filename is None: + raise AssertionError + member = tar.getmember(filename) fd, path = tempfile.mkstemp(ftype, "pmbootstrap") handle = open(fd, "wb") ret[ftype]["temp_path"] = path - shutil.copyfileobj(tar.extractfile(member), handle) + extracted_file = tar.extractfile(member) + if extracted_file is None: + raise AssertionError + shutil.copyfileobj(extracted_file, handle) logging.debug(f"extracted: {path}") handle.close() return ret -def verify_signature(files, sigkey_path): +def verify_signature(files: dict[str, dict], sigkey_path: str) -> None: """ Verify the signature with openssl. diff --git a/pmb/helpers/pmaports.py b/pmb/helpers/pmaports.py index d2621927..f95fcb73 100644 --- a/pmb/helpers/pmaports.py +++ b/pmb/helpers/pmaports.py @@ -188,8 +188,31 @@ def show_pkg_not_found_systemd_hint(package: str, with_extra_repos: WithExtraRep ) +@overload +def find( + package: str, + must_exist: Literal[True] = ..., + subpackages: bool = ..., + with_extra_repos: WithExtraRepos = ..., +) -> Path: ... + + +@overload +def find( + package: str, + must_exist: bool = ..., + subpackages: bool = ..., + with_extra_repos: WithExtraRepos = ..., +) -> Path | None: ... + + @Cache("package", "subpackages", "with_extra_repos") -def find(package, must_exist=True, subpackages=True, with_extra_repos="default"): +def find( + package: str, + must_exist: bool = True, + subpackages: bool = True, + with_extra_repos: WithExtraRepos = "default", +) -> Path | None: """Find the directory in pmaports that provides a package or subpackage. If you want the parsed APKBUILD instead, use pmb.helpers.pmaports.get(). diff --git a/pmb/helpers/run_core.py b/pmb/helpers/run_core.py index ab8a8d94..7f0ab696 100644 --- a/pmb/helpers/run_core.py +++ b/pmb/helpers/run_core.py @@ -2,7 +2,7 @@ # SPDX-License-Identifier: GPL-3.0-or-later import fcntl from pmb.core.context import get_context -from pmb.types import PathString, Env, RunOutputType +from pmb.types import PathString, Env, RunOutputType, RunReturnType from pmb.helpers import logging import os from pathlib import Path @@ -72,7 +72,9 @@ def sanity_checks( raise RuntimeError("Can't use output_return with output: " + output) -def background(cmd: str, working_dir: PathString | None = None) -> subprocess.Popen: +def background( + cmd: PathString | Sequence[PathString], working_dir: PathString | None = None +) -> subprocess.Popen: """Run a subprocess in background and redirect its output to the log.""" ret = subprocess.Popen( cmd, stdout=pmb.helpers.logging.logfd, stderr=pmb.helpers.logging.logfd, cwd=working_dir @@ -81,7 +83,9 @@ def background(cmd: str, working_dir: PathString | None = None) -> subprocess.Po return ret -def pipe(cmd: str, working_dir: PathString | None = None) -> subprocess.Popen: +def pipe( + cmd: PathString | Sequence[PathString], working_dir: PathString | None = None +) -> subprocess.Popen: """Run a subprocess in background and redirect its output to a pipe.""" ret = subprocess.Popen( cmd, @@ -114,6 +118,16 @@ def pipe_read( ) -> None: ... +@overload +def pipe_read( + process: subprocess.Popen, + output_to_stdout: bool = ..., + output_log: bool = ..., + output_return: bool = ..., + output_return_buffer: list[bytes] | None = ..., +) -> None: ... + + def pipe_read( process: subprocess.Popen, output_to_stdout: bool = False, @@ -195,15 +209,15 @@ def kill_command(pid: int, sudo: bool) -> None: def foreground_pipe( - cmd, - working_dir=None, - output_to_stdout=False, - output_return=False, - output_log=True, - output_timeout=True, - sudo=False, - stdin=None, -): + cmd: PathString | Sequence[PathString], + working_dir: Path | None = None, + output_to_stdout: bool = False, + output_return: bool = False, + output_log: bool = True, + output_timeout: bool = True, + sudo: bool = False, + stdin: int | None = None, +) -> tuple[int, str]: """Run a subprocess in foreground with redirected output. Optionally kill it after being silent for too long. @@ -270,7 +284,9 @@ def foreground_pipe( return (process.returncode, b"".join(output_buffer).decode("utf-8")) -def foreground_tui(cmd: str, working_dir: PathString | None = None) -> int: +def foreground_tui( + cmd: PathString | Sequence[PathString], working_dir: PathString | None = None +) -> int: """Run a subprocess in foreground without redirecting any of its output. This is the only way text-based user interfaces (ncurses programs like @@ -342,15 +358,15 @@ def add_proxy_env_vars(env: Env) -> None: def core( - log_message, - cmd, - working_dir=None, - output="log", - output_return=False, - check=None, - sudo=False, - disable_timeout=False, -): + log_message: str, + cmd: Sequence[PathString], + working_dir: Path | None = None, + output: RunOutputType = "log", + output_return: bool = False, + check: bool | None = None, + sudo: bool = False, + disable_timeout: bool = False, +) -> RunReturnType: """Run a command and create a log entry. This is a low level function not meant to be used directly. Use one of the diff --git a/pmb/parse/arguments.py b/pmb/parse/arguments.py index f9bb2789..4abbe311 100644 --- a/pmb/parse/arguments.py +++ b/pmb/parse/arguments.py @@ -1,10 +1,11 @@ # Copyright 2023 Oliver Smith # SPDX-License-Identifier: GPL-3.0-or-later import argparse +from collections.abc import Sequence import os from pathlib import Path import sys -from typing import cast +from typing import Any, cast from pmb.core.arch import Arch from pmb.core import Config @@ -45,10 +46,16 @@ def toggle_other_boolean_flags( """ class SetOtherDestinationsAction(argparse.Action): - def __init__(self, option_strings, dest, **kwargs): + def __init__(self, option_strings: list[str], dest: str, **kwargs: Any) -> None: super().__init__(option_strings, dest, nargs=0, const=value, default=value, **kwargs) - def __call__(self, parser, namespace, values, option_string=None): + def __call__( + self, + parser: argparse.ArgumentParser, + namespace: argparse.Namespace, + values: str | Sequence[Any] | None, + option_string: str | None = None, + ) -> None: for destination in other_destinations: setattr(namespace, destination, value) @@ -720,7 +727,7 @@ def arguments_kconfig(subparser: argparse._SubParsersAction) -> None: add_kernel_arg(migrate, nargs=1) -def arguments_repo_bootstrap(subparser): +def arguments_repo_bootstrap(subparser: argparse._SubParsersAction) -> argparse.ArgumentParser: arch_choices = Arch.supported() ret = subparser.add_parser("repo_bootstrap") @@ -792,16 +799,26 @@ def arguments_ci(subparser: argparse._SubParsersAction) -> argparse.ArgumentPars return ret -def package_completer(prefix, action, parser=None, parsed_args=None): +def package_completer( + prefix: str, + action: str, + parser: argparse.ArgumentParser | None = None, + parsed_args: list[str] | None = None, +) -> set[str]: packages = set( package for package in pmb.helpers.pmaports.get_list() if package.startswith(prefix) ) return packages -def kernel_completer(prefix, action, parser=None, parsed_args=None): +def kernel_completer( + prefix: str, + action: str, + parser: argparse.ArgumentParser | None = None, + parsed_args: list[str] | None = None, +) -> list[str]: """:returns: matched linux-* packages, with linux-* prefix and without""" - ret = [] + ret: list[str] = [] # Full package name, starting with "linux-" if len("linux-") < len(prefix) and prefix.startswith("linux-") or "linux-".startswith(prefix): @@ -814,7 +831,9 @@ def kernel_completer(prefix, action, parser=None, parsed_args=None): return ret -def add_packages_arg(subparser, name="packages", *args, **kwargs): +def add_packages_arg( + subparser: argparse.ArgumentParser, name: str = "packages", *args: str, **kwargs: Any +) -> None: arg = subparser.add_argument(name, *args, **kwargs) if "argcomplete" in sys.modules: arg.completer = package_completer # type: ignore[attr-defined] diff --git a/pmb/parse/version.py b/pmb/parse/version.py index 44a6a08e..ea8a59f6 100644 --- a/pmb/parse/version.py +++ b/pmb/parse/version.py @@ -129,7 +129,7 @@ def parse_suffix(rest: str) -> tuple[str, int, bool]: return (rest, 0, True) -def get_token(previous, rest): +def get_token(previous: str, rest: str) -> tuple[str, int, str]: """ This function does three things: * get the next token