janware/jw-pkg
1
Code Pull requests Activity Actions

jw.pkg: Fix "make check" static code check fallout

Browse source 
The previous commits have put rules for linting and formatting via ruff, yapf, mypy and pyright into place. They are checked with the make check target, and this commit adds the fixes for the target to succeed.
It does some refactoring where type checking dug up dirty bits, and also adds lots of churn in the Python code. To a good deal, that's owed to mere formatting changes. It would have been better to seperate those from syntax and refactoring fixes into multiple commits, so that the interesting changes don't drown in the formatting nose. However, that would have been a lot of additional work only to be thrown away by later commits, hence this commit has a big diff in one piece. The size of the diff is regrettable but hopefully a one-off: What it buys is automatic format checking for CI and predictble formats for smaller diffs in the future.
Rules that "make check" enforces are, in the following order
- Syntax checkers:
      - ruff check .
      - mypy .
      - pyright
- Format check:
     - yapf --diff --recursive .
The refactoring includes:
  - Turn the Result class into a more elaborate object, capable of
    doing more heavy lifting around stderr and stdout decoding,
    summarizing outcome, and matching error strings.
    Aside from fixing broken type checks, this also removes lots of
    boilerplate calling code which is currently used for handling
    possible call outcome scenarios. Trying to access an inexistent,
    decoded string should raise a meaningful exception by itself now,
    which removes lots of code with case distinctions.
- Fix Cmd type hierarchy:
    - Add the AbstractCmd class above Cmd. This is necessary because
      the checker rightfully complains it can't instantiate a Cmd
      instance where constructor arguments were needed. They never
      were, but the type used at the instantiating code's location in
      jw.pkg.App so claims.
    - Lots of sub- and sub-subcommands are derived from the base
      class of the invoking command. That provides some properties
      shared across the ancestor hierarchy of a command, but is
      semantically unsound. Fix that by introducing jw.pkg.BaseCmd
      class as a place to provide basic helpers shared across all
      commands used in a jw.pkg.App's context, and derive all command
      classes from that afresh. The parent command is still reachable
      via a common parent property.
Formatting changes are conforming to PEP-8, mostly, with minor tweaks. All in all they include the following changes.
- Remove # -*- coding: utf-8 -*-
    The line was needed by Python 2 which is not supported anylonger.
    For Python 3, the default encoding is UTF-8, anyway.
  - Allow to run "make py-format" without having it produce any
    changes. It's basically "yapf --in-place --recursive ." with some
    code style settings, see conf/topdir/pyproject.toml. The settings
    may be debatable. I've had custom tweaks in place on that target,
    too, but then again, IDEs would have more hassle to integrate
    that.
- Introduce a 88 character line length limit
  - One import per line, reshuffle them semantically, see
    [tool.isort] in pyproject.toml.
- Hide imports needed for type-checking only behind
      if TYPE_CHECKING
  - Spaces around assignments accounts for much churn. Having having
    no spaces in inline parameter list assignments and default
    parameter values would arguably be more compact where it's
    useful. On the other hand, I have not found a code formatter
    which allows spaces around assignments in parameter lists broken
    into one per line and that's often better than a wall of text.
  - Add two spaces before # export, as this seems to be mandated by
    PEP-8
- Use single quotes by default
Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-05-27 07:16:05 +02:00
commit 6db73873e7
Signed by: Jan Lindemann
GPG key ID: 3750640C9E25DD61
97 changed files with 3229 additions and 1893 deletions
Download patch file Download diff file Expand all files Collapse all files

463
src/python/jw/pkg/lib/ExecContext.py
View file

@ -1,42 +1,43 @@
# -*- coding: utf-8 -*-
from __future__ import annotations
import abc, re, sys, errno
from enum import Enum, auto
from typing import NamedTuple, TYPE_CHECKING
from decimal import Decimal, ROUND_FLOOR
import abc
import errno
import sys
from decimal import ROUND_FLOOR, Decimal
from typing import TYPE_CHECKING, NamedTuple
if TYPE_CHECKING:
from typing import Self, Type
from typing import Type
from types import TracebackType
from .log import *
from .base import Input, InputMode, Result, StatResult
from .FileContext import FileContext as Base
from .log import DEBUG, ERR, NOTICE, log
_US = "\x1f" # unlikely to appear in numeric output
_US = '\x1f' # unlikely to appear in numeric output
_BILLION = Decimal(1_000_000_000)
def _looks_like_option_error(stderr: str) -> bool:
def _looks_like_option_error(stderr: str | None) -> bool:
if stderr is None:
return False
s = stderr.lower()
return any(
needle in s
for needle in (
"unrecognized option",
"illegal option",
"unknown option",
"invalid option",
"option requires an argument",
)
needle in s for needle in (
'unrecognized option',
'illegal option',
'unknown option',
'invalid option',
'option requires an argument', )
)
def _raise_stat_error(path: str, stderr: str, returncode: int) -> None:
msg = (stderr or "").strip() or f"stat exited with status {returncode}"
def _raise_stat_error(path: str, result: Result) -> None:
stderr = result.stderr_str_or_none or f'stat exited with status {result.status}'
msg = stderr.strip()
lower = msg.lower()
if "no such file" in lower:
if 'no such file' in lower:
raise FileNotFoundError(errno.ENOENT, msg, path)
if "permission denied" in lower or "operation not permitted" in lower:
if 'permission denied' in lower or 'operation not permitted' in lower:
raise PermissionError(errno.EACCES, msg, path)
raise OSError(errno.EIO, msg, path)
@ -46,14 +47,14 @@ def _parse_epoch(value: str) -> tuple[int, float, int]:
(integer seconds for tuple slot, float seconds for attribute, ns for *_ns)
"""
dec = Decimal(value.strip())
sec = int(dec.to_integral_value(rounding=ROUND_FLOOR))
ns = int((dec * _BILLION).to_integral_value(rounding=ROUND_FLOOR))
sec = int(dec.to_integral_value(rounding = ROUND_FLOOR))
ns = int((dec * _BILLION).to_integral_value(rounding = ROUND_FLOOR))
return sec, float(dec), ns
def _build_stat_result(fields: list[str], mode_base: int) -> StatResult:
if len(fields) != 13:
raise ValueError(
f"unexpected stat output: expected 13 fields, got {len(fields)}: {fields!r}"
f'unexpected stat output: expected 13 fields, got {len(fields)}: {fields!r}'
)
(
@ -73,9 +74,9 @@ def _build_stat_result(fields: list[str], mode_base: int) -> StatResult:
) = fields
st_mode = int(mode_s, mode_base)
st_ino = int(ino_s)
st_dev = int(dev_s)
st_nlink = int(nlink_s)
# st_ino = int(ino_s)
# st_dev = int(dev_s)
# st_nlink = int(nlink_s)
st_uid = uid_s
st_gid = gid_s
st_size = int(size_s)
@ -84,9 +85,9 @@ def _build_stat_result(fields: list[str], mode_base: int) -> StatResult:
st_mtime_i, st_mtime_f, st_mtime_ns = _parse_epoch(mtime_s)
st_ctime_i, st_ctime_f, st_ctime_ns = _parse_epoch(ctime_s)
st_blksize = int(blksize_s)
st_blocks = int(blocks_s)
st_rdev = int(rdev_s)
# st_blksize = int(blksize_s)
# st_blocks = int(blocks_s)
# st_rdev = int(rdev_s)
return StatResult(
mode = st_mode,
@ -103,38 +104,38 @@ class ExecContext(Base):
class CallContext:
def __init__(
self,
parent: ExecContext,
title: str|None,
cmd: list[str],
cmd_input: Input,
mod_env: dict[str, str]|None,
wd: str|None,
log_prefix: str,
throw: bool,
verbose: bool,
) -> None:
self,
parent: ExecContext,
title: str | None,
cmd: list[str],
cmd_input: Input,
mod_env: dict[str, str] | None,
wd: str | None,
log_prefix: str,
throw: bool,
verbose: bool | None,
) -> None:
self.__cmd = cmd
self.__wd = wd
self.__log_prefix = log_prefix
self.__parent = parent
self.__title = title
self.__pretty_cmd: str|None = None
self.__delim = title if title is not None else f'---- {parent.uri}: Running {self.pretty_cmd} -'
self.__pretty_cmd: str | None = None
self.__delim = (
title if title is not None else
f'---- {parent.uri}: Running {self.pretty_cmd} -'
)
delim_len = 120
self.__delim += '-' * max(0, delim_len - len(self.__delim))
self.__mod_env = {'LC_ALL': 'C'} if mod_env is None else mod_env
self.__cmd_input: bytes | None = None
# -- At the end of this dance, interactive needs to be either True
# or False
interactive: bool|None = None
if not isinstance(cmd_input, InputMode):
interactive = False
self.__cmd_input = (
cmd_input if isinstance(cmd_input, bytes) else
cmd_input.encode(sys.stdout.encoding or "utf-8")
)
else:
interactive: bool | None = None
cmd_input_bytes: None | bytes
if isinstance(cmd_input, InputMode):
cmd_input_bytes = None
match cmd_input:
case InputMode.Interactive:
interactive = True
@ -148,25 +149,34 @@ class ExecContext(Base):
interactive = parent.interactive
if interactive is None:
interactive = sys.stdin.isatty()
self.__cmd_input = None
assert interactive in [ True, False ], f'Invalid: interactive = {invalid}'
else:
interactive = False
if cmd_input is None:
cmd_input_bytes = None
elif isinstance(cmd_input, str):
cmd_input_bytes = cmd_input.encode(sys.stdout.encoding or 'utf-8')
else:
cmd_input_bytes = cmd_input
self.__cmd_input = cmd_input_bytes
assert interactive in [True, False], f'Invalid: interactive = {interactive}'
self.__interactive = interactive
self.__cmd_input = cmd_input if not isinstance(cmd_input, InputMode) else None
self.__throw = throw
self.__verbose = verbose if verbose is not None else parent.verbose_default
def __enter__(self) -> CallContext:
self.log_delim(start=True)
def __enter__(self) -> ExecContext.CallContext:
self.log_delim(start = True)
return self
def __exit__(
self,
exc_type: Type[BaseException]|None,
exc_value: BaseException|None,
traceback: TracebackType|None
exc_type: Type[BaseException] | None,
exc_value: BaseException | None,
traceback: TracebackType | None,
) -> bool:
self.log_delim(start=False)
self.log_delim(start = False)
return True
@property
def log_prefix(self) -> str:
@ -181,7 +191,7 @@ class ExecContext(Base):
return self.__verbose
@property
def cmd_input(self) -> bytes|None:
def cmd_input(self) -> bytes | None:
return self.__cmd_input
@property
@ -193,7 +203,7 @@ class ExecContext(Base):
return self.__throw
@property
def wd(self) -> str|None:
def wd(self) -> str | None:
return self.__wd
@property
@ -204,16 +214,17 @@ class ExecContext(Base):
def pretty_cmd(self) -> str:
if self.__pretty_cmd is None:
from .util import pretty_cmd
self.__pretty_cmd = pretty_cmd(self.__cmd, self.__wd)
return self.__pretty_cmd
def log(prio: int, *args, **kwargs) -> None:
def log(self, prio: int, *args, **kwargs) -> None:
log(prio, self.__log_prefix, *args, **kwargs)
def log_delim(self, start: bool) -> None:
if not self.__verbose:
return None
if self.__interactive: # Don't log footer in interative mode
if self.__interactive: # Don't log footer in interative mode
if start:
log(NOTICE, self.__delim)
return
@ -223,12 +234,9 @@ class ExecContext(Base):
def check_exit_code(self, result: Result) -> None:
if result.status == 0:
return
if (self.__throw or self.__verbose):
msg = f'Command exited with status {result.status}: {self.pretty_cmd}'
if result.stderr:
msg += ': ' + result.decode().stderr.strip()
if self.__throw or self.__verbose:
if self.__throw:
raise RuntimeError(msg)
raise RuntimeError(result.summary)
def exception(self, result: Result, e: Exception) -> Result:
log(ERR, self.__log_prefix, f'Failed to run {self.pretty_cmd}')
@ -243,15 +251,35 @@ class ExecContext(Base):
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
@classmethod
def create(cls, *args, **kwargs) -> ExecContext:
ret = super().create(*args, **kwargs)
if not isinstance(ret, cls):
raise TypeError(f'Expected {cls.__name__}, got {type(ret).__name__}')
return ret
@abc.abstractmethod
async def _run(
self,
cmd: list[str],
wd: str | None,
verbose: bool,
cmd_input: bytes | None,
mod_env: dict[str, str] | None,
interactive: bool,
log_prefix: str,
) -> Result:
raise NotImplementedError('Called pure virtual method _run()')
async def run(
self,
cmd: list[str],
wd: str|None = None,
wd: str | None = None,
throw: bool = True,
verbose: bool|None = None,
verbose: bool | None = None,
cmd_input: Input = InputMode.OptInteractive,
mod_env: dict[str, str]|None = None,
title: str = None
mod_env: dict[str, str] | None = None,
title: str | None = None,
) -> Result:
"""
Run a command asynchronously and return its output
@ -262,13 +290,16 @@ class ExecContext(Base):
throw: Raise an exception on non-zero exit status if True
verbose: Emit log output while the command runs
cmd_input:
- "InputMode.OptInteractive" -> Let --interactive govern how to handle interactivity (default)
- "InputMode.OptInteractive" -> Let --interactive govern how to handle
interactivity (default)
- "InputMode.Interactive" -> Inherit terminal stdin
- "InputMode.Auto" -> Inherit terminal stdin if it is a TTY
- "InputMode.NonInteractive" -> stdin from /dev/null
- None -> Alias for InputMode.NonInteractive
- otherwise -> Feed cmd_input to stdin
mod_env: Change set to command's environment. key: val adds a variable, key: None removes it
mod_env: Change set to command's environment:
- key: val adds a variable,
- key: None removes it
Returns:
A Result instance
@ -285,8 +316,17 @@ class ExecContext(Base):
try:
ret = Result(None, None, 1)
with self.CallContext(self, title=title, cmd=cmd, cmd_input=cmd_input, mod_env=mod_env, wd=wd,
log_prefix='|', throw=throw, verbose=verbose) as cc:
with self.CallContext(
self,
title = title,
cmd = cmd,
cmd_input = cmd_input,
mod_env = mod_env,
wd = wd,
log_prefix = '|',
throw = throw,
verbose = verbose,
) as cc:
try:
ret = await self._run(
cmd = cc.cmd,
@ -295,7 +335,7 @@ class ExecContext(Base):
cmd_input = cc.cmd_input,
mod_env = cc.mod_env,
interactive = cc.interactive,
log_prefix = cc.log_prefix
log_prefix = cc.log_prefix,
)
except Exception as e:
return cc.exception(ret, e)
@ -308,11 +348,11 @@ class ExecContext(Base):
async def _sudo(
self,
cmd: list[str],
opts: list[str]|None,
wd: str|None,
mod_env_sudo: dict[str, str]|None,
mod_env_cmd: dict[str, str]|None,
cmd_input: bytes|None,
opts: list[str] | None,
wd: str | None,
mod_env_sudo: dict[str, str] | None,
mod_env_cmd: dict[str, str] | None,
cmd_input: bytes | None,
verbose: bool,
interactive: bool,
log_prefix: str,
@ -320,20 +360,22 @@ class ExecContext(Base):
def __check_equal_values(d1: dict[str, str], d2: dict[str, str]) -> None:
for key, val in d1.items():
if not d2.get(key, None) in [None, val]:
raise ValueError(f'Outer and inner environments differ at least for {key}: "{val}" != "{d2.get(key)}"')
if d2.get(key, None) not in [None, val]:
raise ValueError(
'Outer and inner environments differ at least for '
f'{key}: "{val}" != "{d2.get(key)}"'
)
fw_cmd: list[str] = []
fw_env: dict[str, str] = {}
if opts is None:
opts = {}
opts = []
if mod_env_cmd:
fw_env.update(mod_env_cmd)
if self.username != 'root':
if mod_env_sudo and mod_env_cmd:
__check_equal_values(mod_env_sudo, mod_env_cmd)
__check_equal_values(mod_env_cmd, mod_env_sudo)
@ -345,7 +387,7 @@ class ExecContext(Base):
fw_cmd.append('--preserve-env=' + ','.join(mod_env_cmd.keys()))
if wd is not None:
opts.extend('-D', wd)
opts.extend(['-D', wd])
wd = None
fw_cmd.extend(opts)
@ -361,20 +403,20 @@ class ExecContext(Base):
verbose = verbose,
cmd_input = cmd_input,
interactive = interactive,
log_prefix = log_prefix
log_prefix = log_prefix,
)
async def sudo(
self,
cmd: list[str],
opts: list[str]|None = None,
wd: str|None = None,
mod_env_sudo: dict[str, str]|None = None,
mod_env_cmd: dict[str, str]|None = None,
opts: list[str] | None = None,
wd: str | None = None,
mod_env_sudo: dict[str, str] | None = None,
mod_env_cmd: dict[str, str] | None = None,
throw: bool = True,
verbose: bool|None = None,
verbose: bool | None = None,
cmd_input: Input = InputMode.OptInteractive,
title: str = None
title: str | None = None,
) -> Result:
# Note that in the calls to the wrapped method, cmd_input == None can
@ -382,9 +424,17 @@ class ExecContext(Base):
assert cmd_input is not None, 'Invalid: cmd_input is None'
ret = Result(None, None, 1)
with self.CallContext(self, title=title, cmd=cmd, cmd_input=cmd_input,
mod_env=mod_env_cmd, wd=wd,
log_prefix='|', throw=throw, verbose=verbose) as cc:
with self.CallContext(
self,
title = title,
cmd = cmd,
cmd_input = cmd_input,
mod_env = mod_env_cmd,
wd = wd,
log_prefix = '|',
throw = throw,
verbose = verbose,
) as cc:
try:
ret = await self._sudo(
cmd = cc.cmd,
@ -416,19 +466,22 @@ class ExecContext(Base):
)
async def _get(
self,
path: str,
wd: str|None,
throw: bool,
verbose: bool|None,
title: str
self, path: str, wd: str | None, throw: bool, verbose: bool | None, title: str
) -> Result:
ret = Result(None, None, 1)
if wd is not None:
path = wd + '/' + path
with self.CallContext(self, title=title, cmd=['cat', path],
cmd_input=InputMode.NonInteractive, wd=None, mod_env=None,
log_prefix='|', throw=throw, verbose=verbose) as cc:
with self.CallContext(
self,
title = title,
cmd = ['cat', path],
cmd_input = InputMode.NonInteractive,
wd = None,
mod_env = None,
log_prefix = '|',
throw = throw,
verbose = verbose,
) as cc:
try:
ret = await self._run(
cmd = cc.cmd,
@ -437,12 +490,12 @@ class ExecContext(Base):
cmd_input = cc.cmd_input,
mod_env = cc.mod_env,
interactive = cc.interactive,
log_prefix = cc.log_prefix
log_prefix = cc.log_prefix,
)
except Exception as e:
return cc.exception(ret, e)
if ret.status != 0 and ret.stderr.decode().find('No such file') != -1:
raise FileNotFoundError(ret.stderr)
if ret.matches_error('No such file'):
raise FileNotFoundError(ret.summarize(cc.cmd, wd = cc.wd))
cc.check_exit_code(ret)
return ret
@ -450,138 +503,176 @@ class ExecContext(Base):
self,
path: str,
content: bytes,
wd: str|None,
wd: str | None,
throw: bool,
verbose: bool|None,
verbose: bool | None,
title: str,
owner: str|None,
group: str|None,
mode: str|None,
owner: str | None,
group: str | None,
mode: str | None,
atomic: bool,
) -> Result:
from .util import pretty_cmd
async def __run(cmd: list[str], cmd_input: Input=InputMode.NonInteractive, **kwargs) -> Result:
return await self.run(cmd, cmd_input=cmd_input, **kwargs)
async def __run(
cmd: list[str],
cmd_input: Input = InputMode.NonInteractive,
**kwargs
) -> Result:
return await self.run(cmd, cmd_input = cmd_input, **kwargs)
ret = Result(None, None, 1)
try:
class RemoteCmd(NamedTuple):
cmd: list[str]
cmd_input: Input = InputMode.NonInteractive
if wd is not None:
path = wd + '/' + path
cmds: list[dict[str, str|list[str]|bool]] = []
out = (await __run(['mktemp', path + '.XXXXXX'])).stdout.decode().strip() if atomic else path
cmds.append({'cmd': ['tee', out], 'cmd_input': content})
cmds: list[RemoteCmd] = []
stdout = (await __run(['mktemp', path + '.XXXXXX'])).stdout_str
if stdout is None:
raise Exception(f'Failed to create tmp-directory on {self.root}')
out = stdout.strip() if atomic else path
cmds.append(RemoteCmd(
cmd = ['tee', out],
cmd_input = content,
))
if owner is not None and group is not None:
cmds.append({'cmd': ['chown', f'{owner}:{group}', out]})
cmds.append(RemoteCmd(
cmd = ['chown', f'{owner}:{group}', out],
))
elif owner is not None:
cmds.append({'cmd': ['chown', owner, out]})
cmds.append(RemoteCmd(
cmd = ['chown', owner, out],
))
elif group is not None:
cmds.append({'cmd': ['chgrp', group, out]})
cmds.append(RemoteCmd(
cmd = ['chgrp', group, out],
))
if mode is not None:
cmds.append({'cmd': ['chmod', mode, out]})
cmds.append(RemoteCmd(
cmd = ['chmod', mode, out],
))
if atomic:
cmds.append({'cmd': ['mv', out, path]})
cmds.append(RemoteCmd(
cmd = ['mv', out, path],
))
await self.open()
try:
for cmd in cmds:
log(DEBUG, f'{self.log_name}: Running {pretty_cmd(cmd['cmd'], wd)}')
ret = await __run(**cmd)
log(DEBUG, f'{self.log_name}: Running {pretty_cmd(cmd.cmd, wd)}')
ret = await __run(cmd.cmd)
return ret
finally:
await self.close()
except:
except Exception as e:
msg = f'Failed to get {path} from {self.root} ({str(e)})'
if throw:
raise
return cc.exception(ret, e)
raise Exception(msg)
log(ERR, msg)
return ret
async def _unlink(self, path: str) -> None:
cmd = ['rm', '-f', path]
await self.run(cmd, cmd_input=InputMode.NonInteractive)
await self.run(cmd, cmd_input = InputMode.NonInteractive)
async def _erase(self, path: str) -> None:
cmd = ['rm', '-rf', path]
await self.run(cmd, cmd_input=InputMode.NonInteractive)
await self.run(cmd, cmd_input = InputMode.NonInteractive)
async def _rename(self, src: str, dst: str) -> None:
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:
cmd = ['mkdir', name, '-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:
cmd = ['mktemp']
if directory:
cmd.append('-d')
cmd.append(tmpl)
result = await self.run(cmd, cmd_input=InputMode.NonInteractive)
return result.stdout.strip().decode()
result = await self.run(cmd, cmd_input = InputMode.NonInteractive, throw = True)
if result.status != 0 or result.stdout is None:
raise Exception(
f'Failed to create temporary file on {self.root}: {result.summary}'
)
return result.stdout_str
async def _stat(self, path: str, follow_symlinks: bool) -> StatResult:
async def __stat(opts: list[str]) -> str:
mod_env = {
'LC_ALL': 'C'
}
async def __stat(opts: list[str]) -> Result:
mod_env = {'LC_ALL': 'C'}
cmd = ['stat']
if follow_symlinks:
cmd.append('-L')
cmd.extend(opts)
cmd.append(path)
return (await self.run(cmd, mod_env=mod_env, throw=False,
cmd_input=InputMode.NonInteractive)).decode()
return await self.run(
cmd,
mod_env = mod_env,
throw = False,
cmd_input = InputMode.NonInteractive
)
# GNU coreutils stat
gnu_format = _US.join([
"%f", # st_mode in hex
"%i", # st_ino
"%d", # st_dev
"%h", # st_nlink
"%U", # st_uid
"%G", # st_gid
"%s", # st_size
"%.9X", # st_atime
"%.9Y", # st_mtime
"%.9Z", # st_ctime
"%o", # st_blksize hint
"%b", # st_blocks
"%r", # st_rdev
])
gnu_format = _US.join(
[
'%f', # st_mode in hex
'%i', # st_ino
'%d', # st_dev
'%h', # st_nlink
'%U', # st_uid
'%G', # st_gid
'%s', # st_size
'%.9X', # st_atime
'%.9Y', # st_mtime
'%.9Z', # st_ctime
'%o', # st_blksize hint
'%b', # st_blocks
'%r', # st_rdev
]
)
result = await __stat(['--printf', gnu_format])
if result.status == 0:
return _build_stat_result(result.stdout.split(_US), mode_base=16)
if result.status == 0 and result.stdout is not None:
return _build_stat_result(result.stdout_str.split(_US), mode_base = 16)
if not _looks_like_option_error(result.stderr):
if not _looks_like_option_error(result.stderr_str_or_none):
# log(DEBUG, f'GNU stat attempt failed on "{path}" ({str(e)})')
_raise_stat_error(path, result.stderr, result.status)
_raise_stat_error(path, result)
# BSD / macOS / OpenBSD / NetBSD stat
bsd_format = _US.join([
"%p", # st_mode in octal
"%i", # st_ino
"%d", # st_dev
"%l", # st_nlink
"%U", # st_uid
"%G", # st_gid
"%z", # st_size
"%.9Fa", # st_atime
"%.9Fm", # st_mtime
"%.9Fc", # st_ctime
"%k", # st_blksize
"%b", # st_blocks
"%r", # st_rdev
])
bsd_format = _US.join(
[
'%p', # st_mode in octal
'%i', # st_ino
'%d', # st_dev
'%l', # st_nlink
'%U', # st_uid
'%G', # st_gid
'%z', # st_size
'%.9Fa', # st_atime
'%.9Fm', # st_mtime
'%.9Fc', # st_ctime
'%k', # st_blksize
'%b', # st_blocks
'%r', # st_rdev
]
)
result = await __stat(['-n', '-f', bst_format])
if proc.returncode == 0:
return _build_stat_result(proc.stdout.rstrip('\n').split(_US), mode_base=8)
_raise_stat_error(path, result.stderr, result.status)
result = await __stat(['-n', '-f', bsd_format])
stdout = result.stdout_str_or_none
if result.status != 0 or stdout is None:
_raise_stat_error(path, result)
assert stdout is not None # Just there to pacify the linter
return _build_stat_result(stdout.rstrip('\n').split(_US), mode_base = 8)
async def _chown(self, path: str, owner: str|None, group: str|None) -> None:
async def _chown(self, path: str, owner: str | None, group: str | None) -> None:
if owner is None and group is None:
raise ValueError(f'Tried to chown("{path}") without owner and group')
if group is None:
@ -590,7 +681,11 @@ class ExecContext(Base):
ownership = ':' + group
else:
ownership = owner + ':' + group
await self.run(['chown', ownership, path], cmd_input=InputMode.NonInteractive)
assert ownership is not None # Impossible, just there to calm the linter
await self.run(['chown', ownership, path], cmd_input = InputMode.NonInteractive)
async def _chmod(self, path: str, mode: int) -> None:
await self.run(['chmod', self.__mode_str(mode), path], cmd_input=InputMode.NonInteractive)
await self.run(
['chmod', self.__mode_str(mode), path],
cmd_input = InputMode.NonInteractive
)