mirror of
https://github.com/kernelkit/infix.git
synced 2026-07-27 19:23:02 +02:00
Tobias Waldekranz
e65625e32f
Support the familiar subprocess.run API, and runsh from netns.py. This should only be used for debugging, or in the rare instances when we want to test functionality that is not available via NETCONF.
47 lines
1.3 KiB
Python
47 lines
1.3 KiB
Python
import subprocess
|
|
|
|
from dataclasses import dataclass
|
|
|
|
@dataclass
|
|
class Location:
|
|
host: str
|
|
password: str
|
|
username: str = "admin"
|
|
port: int = 22
|
|
|
|
class Device(object):
|
|
def __init__(self, location: Location):
|
|
self.location = location
|
|
|
|
def _mangle_subprocess_args(self, args, kwargs):
|
|
if not args:
|
|
return
|
|
|
|
args = list(args)
|
|
if type(args[0]) == str:
|
|
if kwargs.get("shell"):
|
|
args[0] = ["/bin/sh", "-c", args[0]]
|
|
kwargs["shell"] = False
|
|
else:
|
|
args[0] = [args[0]]
|
|
|
|
args[0] = [ "ssh",
|
|
"-oStrictHostKeyChecking no",
|
|
"-oUserKnownHostsFile /dev/null",
|
|
"-oLogLevel QUIET",
|
|
f"-l{self.location.username}",
|
|
self.location.host ] + args[0]
|
|
|
|
if self.location.password:
|
|
args[0] = [ "sshpass" , f"-p{self.location.password}" ] + args[0]
|
|
|
|
return args, kwargs
|
|
|
|
def run(self, *args, **kwargs):
|
|
args, kwargs = self._mangle_subprocess_args(args, kwargs)
|
|
return subprocess.run(*args, **kwargs)
|
|
|
|
def runsh(self, script, *args, **kwargs):
|
|
return self.run("/bin/sh", text=True, input=script,
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, *args, **kwargs)
|