1
0
mirror of https://github.com/ARM-software/devlib.git synced 2025-01-31 02:00:45 +00:00

target/{host,ssh}: Align push/pull with cp/mv behaviour

When pushing or pulling a folder, replicate the mv/cp/scp/adb behaviour,
which is:
    * splitting the destination into (existing, new) components
    * if {new} component is empty, set it to the basename of the source.
    * mkdir {new} if necessary
    * merge the hierarchies of {src} and {existing}/{new}
This commit is contained in:
douglas-raillard-arm 2020-06-18 12:25:17 +01:00 committed by Marc Bonnici
parent f23fbd22b6
commit 8695344969
2 changed files with 54 additions and 22 deletions

View File

@ -64,19 +64,34 @@ class LocalConnection(ConnectionBase):
self.unrooted = unrooted self.unrooted = unrooted
self.password = password self.password = password
def push(self, sources, dest, timeout=None, as_root=False): # pylint: disable=unused-argument
self.logger.debug('copying {} to {}'.format(sources, dest)) def _copy_path(self, source, dest):
for source in sources: self.logger.debug('copying {} to {}'.format(source, dest))
if os.path.isdir(source):
# Behave similarly as cp, scp, adb push, etc. by creating a new
# folder instead of merging hierarchies
if os.path.exists(dest):
dest = os.path.join(dest, os.path.basename(os.path.normpath(src)))
# Use distutils copy_tree since it behaves the same as
# shutils.copytree except that it won't fail if some folders
# already exist.
#
# Mirror the behavior of all other targets which only copy the
# content without metadata
copy_tree(source, dest, preserve_mode=False, preserve_times=False)
else:
shutil.copy(source, dest) shutil.copy(source, dest)
def pull(self, sources, dest, timeout=None, as_root=False): # pylint: disable=unused-argument def _copy_paths(self, sources, dest):
for source in sources: for source in sources:
self.logger.debug('copying {} to {}'.format(source, dest)) self._copy_path(source, dest)
if os.path.isdir(source):
# Use distutils to allow copying into an existing directory structure. def push(self, sources, dest, timeout=None, as_root=False): # pylint: disable=unused-argument
copy_tree(source, dest) self._copy_paths(sources, dest)
else:
shutil.copy(source, dest) def pull(self, sources, dest, timeout=None, as_root=False): # pylint: disable=unused-argument
self._copy_paths(sources, dest)
# pylint: disable=unused-argument # pylint: disable=unused-argument
def execute(self, command, timeout=None, check_exit_code=True, def execute(self, command, timeout=None, check_exit_code=True,

View File

@ -451,28 +451,45 @@ class SshConnection(SshConnectionBase):
try: try:
sftp.put(src, dst) sftp.put(src, dst)
# Maybe the dst was a folder # Maybe the dst was a folder
except OSError as e: except OSError as orig_excep:
logger.debug('sftp transfer error: {}'.format(repr(e))) # If dst was an existing folder, we add the src basename to create
# This might fail if the folder already exists # a new destination for the file as cp would do
with contextlib.suppress(IOError):
sftp.mkdir(dst)
new_dst = os.path.join( new_dst = os.path.join(
dst, dst,
os.path.basename(src), os.path.basename(src),
) )
logger.debug('Trying: {} -> {}'.format(src, new_dst)) logger.debug('Trying: {} -> {}'.format(src, new_dst))
sftp.put(src, new_dst) try:
sftp.put(src, new_dst)
# This still failed, which either means:
# * There are some missing folders in the dirnames
# * Something else SFTP-related is wrong
except OSError as e:
# Raise the original exception, as it is closer to what the
# user asked in the first place
raise orig_excep
@classmethod
def _path_exists(cls, sftp, path):
try:
sftp.lstat(path)
except FileNotFoundError:
return False
else:
return True
@classmethod @classmethod
def _push_folder(cls, sftp, src, dst): def _push_folder(cls, sftp, src, dst):
# Behave like the "mv" command or adb push: a new folder is created # Behave like the "mv" command or adb push: a new folder is created
# inside the destination folder, rather than merging the trees. # inside the destination folder, rather than merging the trees, but
dst = os.path.join( # only if the destination already exists. Otherwise, it is use as-is as
dst, # the new hierarchy name.
os.path.basename(os.path.normpath(src)), if cls._path_exists(sftp, dst):
) dst = os.path.join(
dst,
os.path.basename(os.path.normpath(src)),
)
return cls._push_folder_internal(sftp, src, dst) return cls._push_folder_internal(sftp, src, dst)
@classmethod @classmethod