cmds / lib: Fix more static checker findings

Fix more errors and warnings produced by "make check" as reported by CI and a pyright upgrade.

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-06-01 07:45:22 +02:00
commit 24928c6f5d
Signed by: Jan Lindemann
GPG key ID: 3750640C9E25DD61
16 changed files with 123 additions and 122 deletions

View file

@ -9,5 +9,5 @@ class CmdPkg(CmdBase): # export
super().__init__(parent, 'pkg', help = 'System package manager wrapper') super().__init__(parent, 'pkg', help = 'System package manager wrapper')
self.load_subcommands() self.load_subcommands()
def add_arguments(self, p: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(p) super().add_arguments(parser)

View file

@ -11,5 +11,5 @@ class CmdPlatform(CmdBase): # export
) )
self.load_subcommands() self.load_subcommands()
def add_arguments(self, p: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(p) super().add_arguments(parser)

View file

@ -16,5 +16,5 @@ class CmdPosix(CmdBase): # export
) )
self.load_subcommands() self.load_subcommands()
def add_arguments(self, p: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(p) super().add_arguments(parser)

View file

@ -15,8 +15,8 @@ class CmdProjects(CmdBase): # export
) )
self.load_subcommands() self.load_subcommands()
def add_arguments(self, p: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(p) super().add_arguments(parser)
async def _run(self, args): async def _run(self, args):
# Missing subcommand # Missing subcommand

View file

@ -9,5 +9,5 @@ class CmdSecrets(CmdBase): # export
super().__init__(parent, 'secrets', help = 'Manage package secrets') super().__init__(parent, 'secrets', help = 'Manage package secrets')
self.load_subcommands() self.load_subcommands()
def add_arguments(self, p: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(p) super().add_arguments(parser)

View file

@ -3,13 +3,14 @@ from argparse import ArgumentParser, Namespace
from .Cmd import Cmd, Parent from .Cmd import Cmd, Parent
from .lib.templates import tmpl_render from .lib.templates import tmpl_render
class CmdCreatePkgConfig(Cmd): # export class CmdCreatePkgConfig(Cmd): # export
def __init__(self, parent: Parent) -> None: def __init__(self, parent: Parent) -> None:
super().__init__(parent, super().__init__(
parent,
'create-pkg-config', 'create-pkg-config',
help='Generate a pkg-config file for a module') help = 'Generate a pkg-config file for a module'
)
@staticmethod @staticmethod
def __cleanup_requires(string: str) -> str: def __cleanup_requires(string: str) -> str:

View file

@ -3,7 +3,6 @@ import re
from argparse import ArgumentParser, Namespace from argparse import ArgumentParser, Namespace
from ...lib.base import Input
from ...lib.log import DEBUG, log from ...lib.log import DEBUG, log
from ...lib.Uri import Uri from ...lib.Uri import Uri
from ...lib.util import get_password, get_username, run_curl_into from ...lib.util import get_password, get_username, run_curl_into
@ -78,7 +77,7 @@ class CmdListRepos(Cmd): # export
print('\n'.join(result.stdout_str.splitlines())) print('\n'.join(result.stdout_str.splitlines()))
return return
case 'https': case 'https':
from jw.pkg.lib.base import InputMode from jw.pkg.lib.base import Input, InputMode
cmd_input: Input = InputMode.NonInteractive cmd_input: Input = InputMode.NonInteractive
if re.match(r'https://github.com', args.base_url): if re.match(r'https://github.com', args.base_url):

View file

@ -10,9 +10,9 @@ class CmdPythonpath(Cmd): # export
parent, 'pythonpath', help = 'Generate PYTHONPATH for given modules' parent, 'pythonpath', help = 'Generate PYTHONPATH for given modules'
) )
def add_arguments(self, p: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(p) super().add_arguments(parser)
p.add_argument('module', help = 'Modules', nargs = '*') parser.add_argument('module', help = 'Modules', nargs = '*')
async def _run(self, args: Namespace) -> None: async def _run(self, args: Namespace) -> None:
deps = self.app.get_project_refs( deps = self.app.get_project_refs(

View file

@ -61,7 +61,6 @@ class CmdRequiredOsPkg(Cmd): # export
vals = self.app.get_values(deps, ['pkg.requires.' + sec], [flavour]) vals = self.app.get_values(deps, ['pkg.requires.' + sec], [flavour])
if vals: if vals:
requires |= set(vals) requires |= set(vals)
if args.quote: out = [f'"{dep}"' for dep in requires] if args.quote else requires
out = [f'"{dep}"' for dep in requires]
# TODO: add all not in build tree as -devel # TODO: add all not in build tree as -devel
print(' '.join(out)) print(' '.join(out))

View file

@ -68,6 +68,7 @@ def pkg_relations(
if cur_pkg in visited or cur_pkg in ignore: if cur_pkg in visited or cur_pkg in ignore:
continue continue
for subsec in subsections: for subsec in subsections:
version: str | None = None
section = 'pkg.' + rel_type + '.' + subsec section = 'pkg.' + rel_type + '.' + subsec
visited.add(cur_pkg) visited.add(cur_pkg)
value = app.get_value(cur_pkg, section, flavour) value = app.get_value(cur_pkg, section, flavour)
@ -111,7 +112,8 @@ def pkg_relations(
raise Exception('Unknown version specifier in ' + spec) raise Exception('Unknown version specifier in ' + spec)
if len(dep) != 3 or not expand_semver_revision_range: if len(dep) != 3 or not expand_semver_revision_range:
expanded_deps = [dep] expanded_deps = [dep]
else: continue
assert version is not None
expanded_deps = [] expanded_deps = []
semver = re.split(r'[.-]', version) semver = re.split(r'[.-]', version)
if len(semver) != 4: if len(semver) != 4:

View file

@ -180,8 +180,8 @@ class App: # export
class NoopCompleter(BaseCompleter): class NoopCompleter(BaseCompleter):
def __call__(self, **kwargs): def __call__(self, *args, **kwargs):
return () return None
import argcomplete import argcomplete
@ -224,6 +224,7 @@ class App: # export
NOTICE, NOTICE,
f'Writing profile statistics to {self.__args.write_profile}' f'Writing profile statistics to {self.__args.write_profile}'
) )
assert self.__args.write_profile is not None, 'args.write_profile'
pr.dump_stats(self.__args.write_profile) pr.dump_stats(self.__args.write_profile)
if exit_status: if exit_status:

View file

@ -587,8 +587,8 @@ class ExecContext(Base):
cmd = ['mv', src, dst] cmd = ['mv', src, dst]
await self.run(cmd, cmd_input = InputMode.NonInteractive) await self.run(cmd, cmd_input = InputMode.NonInteractive)
async def _mkdir(self, name: str, mode: int) -> None: async def _mkdir(self, path: str, mode: int) -> None:
cmd = ['mkdir', name, '-m', self.__mode_str(mode)] cmd = ['mkdir', path, '-m', self.__mode_str(mode)]
await self.run(cmd, cmd_input = InputMode.NonInteractive) await self.run(cmd, cmd_input = InputMode.NonInteractive)
async def _mktemp(self, tmpl: str, directory: bool) -> str: async def _mktemp(self, tmpl: str, directory: bool) -> str:

View file

@ -177,10 +177,10 @@ class FileContext(abc.ABC):
) -> Result: ) -> Result:
mode_str = None if mode is None else oct(mode).replace('0o', '0') mode_str = None if mode is None else oct(mode).replace('0o', '0')
if self.__out_pipe is not None: if self.__out_pipe is not None:
result = await self.__out_pipe.run(content) content = (await self.__out_pipe.run(content)).stdout
return await self._put( return await self._put(
self._chroot(path), self._chroot(path),
result.stdout, content,
wd = wd, wd = wd,
throw = throw, throw = throw,
verbose = verbose, verbose = verbose,

View file

@ -84,16 +84,14 @@ class Result:
if err is None: if err is None:
return False return False
import re import re
return re.search(pattern, err) is not None return re.search(pattern, err) is not None
def __summarize(self, cmd: list[str] | None, wd: str | None = None) -> str: def __summarize(self, cmd: list[str] | None, wd: str | None = None) -> str:
from .util import pretty_cmd
if cmd is None: if cmd is None:
cmd = self.__cmd cmd = self.__cmd
call = '' call = ''
if cmd is not None: if cmd is not None:
from .util import pretty_cmd
if wd is None: if wd is None:
wd = self.__wd wd = self.__wd
call = f'"{pretty_cmd(cmd, wd)}" ' call = f'"{pretty_cmd(cmd, wd)}" '
@ -164,7 +162,6 @@ class StatResult(NamedTuple):
def from_os(cls, rhs: os.stat_result) -> StatResult: def from_os(cls, rhs: os.stat_result) -> StatResult:
import grp import grp
import pwd import pwd
return StatResult( return StatResult(
rhs.st_mode, rhs.st_mode,
pwd.getpwuid(rhs.st_uid).pw_name, pwd.getpwuid(rhs.st_uid).pw_name,

View file

@ -161,7 +161,6 @@ class Local(Base):
async def _erase(self, path: str) -> None: async def _erase(self, path: str) -> None:
if os.path.isdir(path): if os.path.isdir(path):
import shutil import shutil
shutil.rmtree(path) shutil.rmtree(path)
return return
os.unlink(path) os.unlink(path)
@ -169,8 +168,8 @@ class Local(Base):
async def _rename(self, src: str, dst: str) -> None: async def _rename(self, src: str, dst: str) -> None:
os.rename(src, dst) os.rename(src, dst)
async def _mkdir(self, name: str, mode: int) -> None: async def _mkdir(self, path: str, mode: int) -> None:
os.mkdir(name, mode) os.mkdir(path, mode)
async def _stat(self, path: str, follow_symlinks: bool) -> StatResult: async def _stat(self, path: str, follow_symlinks: bool) -> StatResult:
return StatResult.from_os(os.stat(path, follow_symlinks = follow_symlinks)) return StatResult.from_os(os.stat(path, follow_symlinks = follow_symlinks))

View file

@ -2,7 +2,14 @@ import datetime
import sys import sys
import syslog import syslog
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from typing import Final
# fmt: disable # don't conflate # fmt: disable # don't conflate
# yapf: disable # don't conflate
EMERG = int(syslog.LOG_EMERG) EMERG = int(syslog.LOG_EMERG)
ALERT = int(syslog.LOG_ALERT) ALERT = int(syslog.LOG_ALERT)
CRIT = int(syslog.LOG_CRIT) CRIT = int(syslog.LOG_CRIT)
@ -17,11 +24,8 @@ OFF = DEVEL + 1
_log_level = NOTICE _log_level = NOTICE
_last_tstamp = datetime.datetime.now() _last_tstamp = datetime.datetime.now()
_first_tstamp = _last_tstamp _first_tstamp = _last_tstamp
# fmt: enable
def _log_level_name_by_value(): _LOG_LEVEL_NAME_BY_VALUE: Final[dict[int, str]] = {
if _log_level_name_by_value.map is None:
_log_level_name_by_value.map = {
EMERG: 'EMERG', EMERG: 'EMERG',
ALERT: 'ALERT', ALERT: 'ALERT',
CRIT: 'CRIT', CRIT: 'CRIT',
@ -33,31 +37,30 @@ def _log_level_name_by_value():
DEVEL: 'DEVEL', DEVEL: 'DEVEL',
OFF: 'OFF', OFF: 'OFF',
} }
return _log_level_name_by_value.map # yapf: enable
# fmt: enable
_log_level_name_by_value.map: dict[int, str] | None = None # type: ignore _LOG_LEVEL_VALUE_BY_NAME: Final[dict[str, int]] = {
alias: value
def _log_level_value_by_name(): for value, name in _LOG_LEVEL_NAME_BY_VALUE.items()
if _log_level_value_by_name.map is None: for alias in (name, name.lower())
_log_level_value_by_name.map = {} }
for value, name in _log_level_name_by_value().items():
_log_level_value_by_name.map[name] = value
_log_level_value_by_name.map[name.lower()] = value
return _log_level_value_by_name.map
_log_level_value_by_name.map: dict[str, int] | None = None # type: ignore
def get_log_level_name(level: int) -> str: def get_log_level_name(level: int) -> str:
return _log_level_name_by_value()[level] return _LOG_LEVEL_NAME_BY_VALUE[level]
def parse_log_level(level: str | int) -> int: def parse_log_level(level: str | int) -> int:
try:
ret = int(level) def __int_level(level: int) -> int:
if ret >= 0 and ret <= DEVEL: if level >= 0 and level <= DEVEL:
return ret return level
except ValueError: raise Exception(f'Invalid log level number {level}')
return _log_level_value_by_name()[level]
raise Exception('Invalid log level ', level) if isinstance(level, int):
return __int_level(level)
if level in _LOG_LEVEL_VALUE_BY_NAME:
return _LOG_LEVEL_VALUE_BY_NAME[level]
return __int_level(int(level))
def set_log_level(level: str | int | None = None) -> int: def set_log_level(level: str | int | None = None) -> int:
global _log_level global _log_level