Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 24 additions & 9 deletions src/local_ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,10 +40,10 @@ class LocalOperations(OsOperations):

# TODO: make it read-only
conn_params: ConnectionParams
host: str
ssh_key: typing.Optional[str]
_host: str
_ssh_key: typing.Optional[str]
remote: bool
username: str
_username: typing.Optional[str]

def __init__(self, conn_params=None):
super().__init__()
Expand All @@ -55,10 +55,10 @@ def __init__(self, conn_params=None):
conn_params = ConnectionParams()

self.conn_params = conn_params
self.host = conn_params.host
self.ssh_key = None
self._host = conn_params.host
self._ssh_key = None
self.remote = False
self.username = conn_params.username or getpass.getuser()
self._username = conn_params.username or getpass.getuser()

@staticmethod
def get_single_instance() -> OsOperations:
Expand All @@ -76,16 +76,31 @@ def get_single_instance() -> OsOperations:
assert type(__class__.sm_single_instance) is __class__
return __class__.sm_single_instance

@property
def host(self) -> str:
assert type(self._host) is str
return self._host

@property
def ssh_key(self) -> typing.Optional[str]:
assert self._ssh_key is None or type(self._ssh_key) is str
return self._ssh_key

@property
def username(self) -> typing.Optional[str]:
assert self._username is None or type(self._username) is str
return self._username

def get_platform(self) -> str:
return str(sys.platform)

def create_clone(self) -> LocalOperations:
clone = __class__(__class__.sm_dummy_conn_params)
clone.conn_params = copy.copy(self.conn_params)
clone.host = self.host
clone.ssh_key = self.ssh_key
clone._host = self._host
clone._ssh_key = self._ssh_key
clone.remote = self.remote
clone.username = self.username
clone._username = self._username
return clone

@staticmethod
Expand Down
12 changes: 12 additions & 0 deletions src/os_ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,18 @@ class OsOperations:
def __init__(self):
pass

@property
def host(self) -> str:
raise NotImplementedError()

@property
def ssh_key(self) -> typing.Optional[str]:
raise NotImplementedError()

@property
def username(self) -> typing.Optional[str]:
raise NotImplementedError()

def get_platform(self) -> str:
raise NotImplementedError()

Expand Down
39 changes: 27 additions & 12 deletions src/remote_ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,12 +52,12 @@ class RemoteOperations(OsOperations):
sm_dummy_conn_params = ConnectionParams()

conn_params: ConnectionParams
host: str
_host: str
port: int
ssh_key: str
_ssh_key: typing.Optional[str]
ssh_args: list
remote: bool
username: str
_username: typing.Optional[str]
ssh_dest: str

def __init__(self, conn_params: ConnectionParams):
Expand All @@ -70,33 +70,48 @@ def __init__(self, conn_params: ConnectionParams):
return

self.conn_params = conn_params
self.host = conn_params.host
self._host = conn_params.host
self.port = conn_params.port
self.ssh_key = conn_params.ssh_key
self._ssh_key = conn_params.ssh_key
self.ssh_args = []
if self.ssh_key:
self.ssh_args += ["-i", self.ssh_key]
if self._ssh_key:
self.ssh_args += ["-i", self._ssh_key]
if self.port:
self.ssh_args += ["-p", self.port]
self.remote = True
self.username = conn_params.username or getpass.getuser()
self.ssh_dest = f"{self.username}@{self.host}" if conn_params.username else self.host
self._username = conn_params.username or getpass.getuser()
self.ssh_dest = f"{self._username}@{self._host}" if conn_params.username else self._host

def __enter__(self):
return self

@property
def host(self) -> str:
assert type(self._host) is str
return self._host

@property
def ssh_key(self) -> typing.Optional[str]:
assert self._ssh_key is None or type(self._ssh_key) is str
return self._ssh_key

@property
def username(self) -> typing.Optional[str]:
assert self._username is None or type(self._username) is str
return self._username

def get_platform(self) -> str:
return "linux"

def create_clone(self) -> RemoteOperations:
clone = __class__(__class__.sm_dummy_conn_params)
clone.conn_params = copy.copy(self.conn_params)
clone.host = self.host
clone._host = self._host
clone.port = self.port
clone.ssh_key = self.ssh_key
clone._ssh_key = self._ssh_key
clone.ssh_args = copy.copy(self.ssh_args)
clone.remote = self.remote
clone.username = self.username
clone._username = self._username
clone.ssh_dest = self.ssh_dest
return clone

Expand Down