janware/jw-pkg
1
mirror of ssh://git.janware.com/janware/proj/jw-pkg synced 2026-04-28 13:55:24 +02:00
Code Activity Actions

lib.ExecContext.run(): Push code up into base class

Browse source 
Take implementation burden from the derived classes _run() callback
by moving the respective code into the run() wrapper methods of the
base class.

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-03-18 10:22:21 +01:00
commit 52dd3b8f21
3 changed files with 103 additions and 129 deletions
Download patch file Download diff file Expand all files Collapse all files

46
src/python/jw/pkg/lib/SSHClient.py
View file

@ -4,7 +4,7 @@ from typing import Any
import os, abc, shlex, sys
from .util import run_cmd
from .util import run_cmd, pretty_cmd
from .log import *
from .ExecContext import ExecContext, Result
from urllib.parse import urlparse
@ -27,27 +27,24 @@ class SSHClient(ExecContext):
pass
async def _run(
self,
args: list[str],
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:
self,
args: list[str],
wd: str|None,
verbose: bool,
cmd_input: str|None,
env: dict[str, str]|None,
interactive: bool,
log_prefix: str
) -> Result:
def __log(prio, *args, verbose=verbose):
log(prio, log_prefix, *args)
if wd is not None:
args = ['cd', wd, '&&', *args]
if verbose:
log(WARNING, f'Verbose SSH commands are not yet implemented')
interactive = (
cmd_input == "mode:interactive"
or (cmd_input == "mode:auto" and sys.stdin.isatty())
)
__log(WARNING, f'Verbose SSH commands are not yet implemented')
if interactive:
raise NotImplementedError('Interactive SSH is not yet implemented')
@ -55,20 +52,7 @@ class SSHClient(ExecContext):
if env is not None:
raise NotImplementedError('Passing an environment to SSH commands is not yet implemented')
stdout_b, stderr_b, status = await self._run_ssh(args, cmd_input=cmd_input)
if throw and status:
raise Exception(f'SSH command returned error {status}')
if output_encoding == 'bytes':
return stdout_b, stderr_b, status
if output_encoding is None:
output_encoding = sys.stdout.encoding or "utf-8"
stdout_s = stdout_b.decode(output_encoding, errors="replace") if stdout_b is not None else None
stderr_s = stderr_b.decode(output_encoding, errors="replace") if stderr_b is not None else None
return stdout_s, stderr_s, status
return await self._run_ssh(args, cmd_input=cmd_input)
async def _sudo(self, cmd: list[str], mod_env: dict[str, str], opts: list[str], *args, **kwargs) -> Result:
if self.username != 'root':