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')
self.load_subcommands()
def add_arguments(self, p: ArgumentParser) -> None:
super().add_arguments(p)
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -3,7 +3,6 @@ import re
from argparse import ArgumentParser, Namespace
from ...lib.base import Input
from ...lib.log import DEBUG, log
from ...lib.Uri import Uri
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()))
return
case 'https':
from jw.pkg.lib.base import InputMode
from jw.pkg.lib.base import Input, InputMode
cmd_input: Input = InputMode.NonInteractive
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'
)
def add_arguments(self, p: ArgumentParser) -> None:
super().add_arguments(p)
p.add_argument('module', help = 'Modules', nargs = '*')
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('module', help = 'Modules', nargs = '*')
async def _run(self, args: Namespace) -> None:
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])
if vals:
requires |= set(vals)
if args.quote:
out = [f'"{dep}"' for dep in requires]
out = [f'"{dep}"' for dep in requires] if args.quote else requires
# TODO: add all not in build tree as -devel
print(' '.join(out))

View file

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

View file

@ -180,8 +180,8 @@ class App: # export
class NoopCompleter(BaseCompleter):
def __call__(self, **kwargs):
return ()
def __call__(self, *args, **kwargs):
return None
import argcomplete
@ -224,6 +224,7 @@ class App: # export
NOTICE,
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)
if exit_status:

View file

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

View file

@ -177,10 +177,10 @@ class FileContext(abc.ABC):
) -> Result:
mode_str = None if mode is None else oct(mode).replace('0o', '0')
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(
self._chroot(path),
result.stdout,
content,
wd = wd,
throw = throw,
verbose = verbose,

View file

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

View file

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

View file

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