clan-cli: SSH now completely refactored to use dataclasses everywhere
This commit is contained in:
@@ -4,13 +4,12 @@ import logging
|
|||||||
import math
|
import math
|
||||||
import os
|
import os
|
||||||
import shlex
|
import shlex
|
||||||
import subprocess
|
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from shlex import quote
|
from shlex import quote
|
||||||
from typing import IO, Any
|
from typing import IO, Any
|
||||||
|
|
||||||
from clan_cli.cmd import Log, MsgColor
|
from clan_cli.cmd import CmdOut, Log, MsgColor
|
||||||
from clan_cli.cmd import run as local_run
|
from clan_cli.cmd import run as local_run
|
||||||
from clan_cli.colors import AnsiColor
|
from clan_cli.colors import AnsiColor
|
||||||
from clan_cli.ssh.host_key import HostKeyCheck
|
from clan_cli.ssh.host_key import HostKeyCheck
|
||||||
@@ -66,7 +65,7 @@ class Host:
|
|||||||
msg_color: MsgColor | None = None,
|
msg_color: MsgColor | None = None,
|
||||||
shell: bool = False,
|
shell: bool = False,
|
||||||
timeout: float = math.inf,
|
timeout: float = math.inf,
|
||||||
) -> subprocess.CompletedProcess[str]:
|
) -> CmdOut:
|
||||||
res = local_run(
|
res = local_run(
|
||||||
cmd,
|
cmd,
|
||||||
shell=shell,
|
shell=shell,
|
||||||
@@ -83,12 +82,7 @@ class Host:
|
|||||||
msg_color=msg_color,
|
msg_color=msg_color,
|
||||||
needs_user_terminal=needs_user_terminal,
|
needs_user_terminal=needs_user_terminal,
|
||||||
)
|
)
|
||||||
return subprocess.CompletedProcess(
|
return res
|
||||||
args=res.command_list,
|
|
||||||
returncode=res.returncode,
|
|
||||||
stdout=res.stdout,
|
|
||||||
stderr=res.stderr,
|
|
||||||
)
|
|
||||||
|
|
||||||
def run_local(
|
def run_local(
|
||||||
self,
|
self,
|
||||||
@@ -102,7 +96,7 @@ class Host:
|
|||||||
shell: bool = False,
|
shell: bool = False,
|
||||||
needs_user_terminal: bool = False,
|
needs_user_terminal: bool = False,
|
||||||
log: Log = Log.BOTH,
|
log: Log = Log.BOTH,
|
||||||
) -> subprocess.CompletedProcess[str]:
|
) -> CmdOut:
|
||||||
"""
|
"""
|
||||||
Command to run locally for the host
|
Command to run locally for the host
|
||||||
"""
|
"""
|
||||||
@@ -146,7 +140,7 @@ class Host:
|
|||||||
msg_color: MsgColor | None = None,
|
msg_color: MsgColor | None = None,
|
||||||
shell: bool = False,
|
shell: bool = False,
|
||||||
log: Log = Log.BOTH,
|
log: Log = Log.BOTH,
|
||||||
) -> subprocess.CompletedProcess[str]:
|
) -> CmdOut:
|
||||||
"""
|
"""
|
||||||
Command to run on the host via ssh
|
Command to run on the host via ssh
|
||||||
"""
|
"""
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
import subprocess
|
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from typing import Generic
|
from typing import Generic
|
||||||
|
|
||||||
|
from clan_cli.errors import CmdOut
|
||||||
from clan_cli.ssh import T
|
from clan_cli.ssh import T
|
||||||
from clan_cli.ssh.host import Host
|
from clan_cli.ssh.host import Host
|
||||||
|
|
||||||
@@ -30,4 +30,4 @@ class HostResult(Generic[T]):
|
|||||||
return self._result
|
return self._result
|
||||||
|
|
||||||
|
|
||||||
Results = list[HostResult[subprocess.CompletedProcess[str]]]
|
Results = list[HostResult[CmdOut]]
|
||||||
|
|||||||
@@ -1,8 +1,6 @@
|
|||||||
import subprocess
|
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from clan_cli.cmd import Log
|
from clan_cli.cmd import Log
|
||||||
from clan_cli.errors import ClanError
|
from clan_cli.errors import ClanError, CmdOut
|
||||||
from clan_cli.ssh.host import Host
|
from clan_cli.ssh.host import Host
|
||||||
from clan_cli.ssh.host_group import HostGroup
|
from clan_cli.ssh.host_group import HostGroup
|
||||||
from clan_cli.ssh.host_key import HostKeyCheck
|
from clan_cli.ssh.host_key import HostKeyCheck
|
||||||
@@ -74,7 +72,7 @@ def test_run_exception(host_group: HostGroup) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def test_run_function_exception(host_group: HostGroup) -> None:
|
def test_run_function_exception(host_group: HostGroup) -> None:
|
||||||
def some_func(h: Host) -> subprocess.CompletedProcess[str]:
|
def some_func(h: Host) -> CmdOut:
|
||||||
return h.run_local(["exit 1"], shell=True)
|
return h.run_local(["exit 1"], shell=True)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
|||||||
Reference in New Issue
Block a user