github/opencas-test-framework
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #6 from katlapinka/kasiat/paramiko-copy

Browse Source 
Add paramiko remote copy method
This commit is contained in:
Daniel Madej 2024-09-04 07:28:09 +02:00 committed by GitHub
commit 96a9605fe7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 19 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
connection/base_executor.py
View File

@ -19,6 +19,9 @@ class BaseExecutor:
dut_to_controller): dut_to_controller):
raise NotImplementedError() raise NotImplementedError()
def _copy(self, src, dst, dut_to_controller: bool):
raise NotImplementedError()
def rsync_to(self, src, dst, delete=False, symlinks=False, checksum=False, exclude_list=[], def rsync_to(self, src, dst, delete=False, symlinks=False, checksum=False, exclude_list=[],
timeout: timedelta = timedelta(seconds=90)): timeout: timedelta = timedelta(seconds=90)):
return self._rsync(src, dst, delete, symlinks, checksum, exclude_list, timeout, False) return self._rsync(src, dst, delete, symlinks, checksum, exclude_list, timeout, False)
@ -27,6 +30,12 @@ class BaseExecutor:
timeout: timedelta = timedelta(seconds=90)): timeout: timedelta = timedelta(seconds=90)):
return self._rsync(src, dst, delete, symlinks, checksum, exclude_list, timeout, True) return self._rsync(src, dst, delete, symlinks, checksum, exclude_list, timeout, True)
def copy_to(self, src, dst):
return self._copy(src, dst, True)
def copy_from(self, src, dst):
return self._copy(src, dst, False)
def is_remote(self): def is_remote(self):
return False return False

10
connection/ssh_executor.py
View File

@ -161,6 +161,16 @@ class SshExecutor(BaseExecutor):
if completed_process.returncode: if completed_process.returncode:
raise Exception(f"rsync failed:\n{completed_process}") raise Exception(f"rsync failed:\n{completed_process}")
def _copy(self, src, dst, dut_to_controller: bool):
sftp = self.ssh.open_sftp()
if dut_to_controller:
sftp.put(src, dst)
else:
sftp.get(src, dst)
sftp.close()
def is_remote(self): def is_remote(self):
return True return True