refactor: Include exit status in send_cmd and improve decoding logic

Extend `send_cmd` to return `exit_status` along with `stdout` and
`stderr`. Adjust handling to ensure proper decoding of byte outputs to
strings.

Signed-off-by: Sven Sager <akira@narux.de>
This commit is contained in:
2026-08-07 06:26:48 +02:00
parent 393ec93649
commit 7eda0ac20c
2 changed files with 11 additions and 7 deletions
+1 -1
View File
@@ -383,7 +383,7 @@ class ConnectionManager(QtCore.QThread):
# Check for Unix socket on remote system # Check for Unix socket on remote system
try: try:
stdout, stderr = ssh_tunnel_server.send_cmd("cat /etc/revpipyload/revpipyload.conf") stdout, stderr, exit_code = ssh_tunnel_server.send_cmd("cat /etc/revpipyload/revpipyload.conf")
if stdout: if stdout:
config = ConfigParser() config = ConfigParser()
config.read_string(stdout) config.read_string(stdout)
+10 -6
View File
@@ -187,13 +187,13 @@ class SSHLocalTunnel:
except Exception: except Exception:
return True return True
def send_cmd(self, cmd: str, timeout: float = None) -> Union[Tuple[str, str], Tuple[None, None]]: def send_cmd(self, cmd: str, timeout: float = None) -> Union[Tuple[str, str, int], Tuple[None, None, None]]:
""" """
Send simple command to ssh host. Send simple command to ssh host.
:param cmd: Shell command to execute on remote host :param cmd: Shell command to execute on remote host
:param timeout: Timeout for execution :param timeout: Timeout for execution
:return: Tuple with stdout and stderr :return: Tuple with stdout, stderr, exit status
""" """
if not self.connected: if not self.connected:
raise RuntimeError("Not connected") raise RuntimeError("Not connected")
@@ -201,15 +201,19 @@ class SSHLocalTunnel:
# Running async command from sync context # Running async command from sync context
async def _exec(): async def _exec():
result = await self._conn.run(cmd, timeout=timeout) result = await self._conn.run(cmd, timeout=timeout)
return result.stdout, result.stderr return result.stdout, result.stderr, result.exit_status
try: try:
future = asyncio.run_coroutine_threadsafe(_exec(), self._loop) future = asyncio.run_coroutine_threadsafe(_exec(), self._loop)
stdout, stderr = future.result(timeout=timeout) stdout, stderr, exit_status = future.result(timeout=timeout)
return stdout, stderr if type(stdout) is bytes:
stdout = stdout.decode(errors="ignore")
if type(stderr) is bytes:
stderr = stderr.decode(errors="ignore")
return stdout, stderr, exit_status
except Exception as e: except Exception as e:
log.error(e) log.error(e)
return None, None return None, None, None
@property @property
def connected(self): def connected(self):