janware/jw-pkg
1
Code Pull requests Activity Actions

cmds / lib: Fix more static checker findings

Browse source 
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
Download patch file Download diff file Expand all files Collapse all files

5
src/python/jw/pkg/lib/App.py
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:

4
src/python/jw/pkg/lib/ExecContext.py
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:

4
src/python/jw/pkg/lib/FileContext.py
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,

5
src/python/jw/pkg/lib/base.py
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,

5
src/python/jw/pkg/lib/ec/Local.py
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))

99
src/python/jw/pkg/lib/log.py
View file

@ -2,62 +2,65 @@ import datetime
import sys
import syslog
# fmt: disable # don't conflate
EMERG = int(syslog.LOG_EMERG)
ALERT = int(syslog.LOG_ALERT)
CRIT = int(syslog.LOG_CRIT)
ERR = int(syslog.LOG_ERR)
WARNING = int(syslog.LOG_WARNING)
NOTICE = int(syslog.LOG_NOTICE)
INFO = int(syslog.LOG_INFO)
DEBUG = int(syslog.LOG_DEBUG)
DEVEL = int(syslog.LOG_DEBUG + 1)
OFF = DEVEL + 1
from typing import TYPE_CHECKING
_log_level = NOTICE
_last_tstamp = datetime.datetime.now()
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)
ERR = int(syslog.LOG_ERR)
WARNING = int(syslog.LOG_WARNING)
NOTICE = int(syslog.LOG_NOTICE)
INFO = int(syslog.LOG_INFO)
DEBUG = int(syslog.LOG_DEBUG)
DEVEL = int(syslog.LOG_DEBUG + 1)
OFF = DEVEL + 1
_log_level = NOTICE
_last_tstamp = datetime.datetime.now()
_first_tstamp = _last_tstamp
_LOG_LEVEL_NAME_BY_VALUE: Final[dict[int, str]] = {
EMERG: 'EMERG',
ALERT: 'ALERT',
CRIT: 'CRIT',
ERR: 'ERR',
WARNING: 'WARNING',
NOTICE: 'NOTICE',
INFO: 'INFO',
DEBUG: 'DEBUG',
DEVEL: 'DEVEL',
OFF: 'OFF',
}
# yapf: enable
# fmt: enable
def _log_level_name_by_value():
if _log_level_name_by_value.map is None:
_log_level_name_by_value.map = {
EMERG: 'EMERG',
ALERT: 'ALERT',
CRIT: 'CRIT',
ERR: 'ERR',
WARNING: 'WARNING',
NOTICE: 'NOTICE',
INFO: 'INFO',
DEBUG: 'DEBUG',
DEVEL: 'DEVEL',
OFF: 'OFF',
}
return _log_level_name_by_value.map
_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