lib.ExecContext,Local: Remove callback default params

Remove defaults from protected callback function parameters. They
have to be decided by the base class's public API.

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-03-06 17:01:05 +01:00
commit 1325222fbd
2 changed files with 28 additions and 15 deletions

View file

@ -11,8 +11,15 @@ class Result(NamedTuple):
class ExecContext(abc.ABC):
def __init__(self, interactive: bool=True):
def __init__(self, interactive: bool=True, verbose_default=False):
self.__interactive = interactive
self.__verbose_default = verbose_default
assert verbose_default is not None
def _verbose(self, verbose: bool|None):
if verbose is not None:
return verbose
return self.__verbose_default
@property
def interactive(self):
@ -56,11 +63,15 @@ class ExecContext(abc.ABC):
stdout: stderr each as a string/bytes or None
In PTY mode stderr is always None because PTY merges stdout/stderr.
"""
if verbose is None:
verbose = self.__verbose_default
return await self._run(
args=args,
wd=wd,
throw=throw,
verbose=verbose,
verbose=self._verbose(verbose),
cmd_input=cmd_input,
env=env,
title=title,
@ -68,8 +79,8 @@ class ExecContext(abc.ABC):
)
@abc.abstractmethod
async def _sudo(self, cmd: list[str], mod_env: dict[str, str] = {}, opts: list[str]=[], verbose=True) -> Result:
async def _sudo(self, cmd: list[str], mod_env: dict[str, str], opts: list[str], verbose: bool) -> Result:
pass
async def sudo(self, cmd: list[str], mod_env: dict[str, str] = {}, opts: list[str]=[], verbose=True) -> Result:
return await self._sudo(cmd, mod_env, opts, verbose)
async def sudo(self, cmd: list[str], mod_env: dict[str, str] = {}, opts: list[str]=[], verbose: bool|None=None) -> Result:
return await self._sudo(cmd, mod_env, opts, self._verbose(verbose))

View file

@ -14,14 +14,14 @@ class Local(Base):
async def _run(
self,
args: list[str],
wd: str|None = None,
throw: bool = True,
verbose: bool = False,
cmd_input: str|None = None,
env: dict[str, str]|None = None,
title: str=None,
output_encoding: str|None = None, # None => unchanged; "bytes" => return raw bytes
) -> tuple[str|bytes|None, str|bytes|None]:
wd: str|None,
throw: bool,
verbose: bool,
cmd_input: str|None,
env: dict[str, str]|None,
title: str,
output_encoding: str|None, # None => unchanged; "bytes" => return raw bytes
) -> Result:
want_bytes = (output_encoding == "bytes")
@ -197,7 +197,7 @@ class Local(Base):
if verbose and not interactive:
log(NOTICE, '`' + delim + ' <')
async def _sudo(self, cmd: list[str], mod_env: dict[str, str] = {}, opts: list[str]=[], verbose=True) -> Result:
async def _sudo(self, cmd: list[str], mod_env: dict[str, str], opts: list[str], verbose: bool) -> Result:
env: dict[str, str]|None = None
cmd_input: str|None = None
if mod_env:
@ -212,4 +212,6 @@ class Local(Base):
cmdline.extend(cmd)
if self.interactive:
cmd_input = "mode:interactive"
return await self._run(cmdline, throw=True, verbose=verbose, env=env, cmd_input=cmd_input)
# Need to call the base class function, because _run() needs more
# parameters than we have values for
return await self.run(cmdline, throw=True, verbose=verbose, env=env, cmd_input=cmd_input)