How to use _make_rsync_cmd method in autotest

Best Python code snippet using autotest_python Github


Full Screen

...34 the style used by rsync and scp. """35 escaped_paths = [utils.scp_remote_escape(path) for path in paths]36 return '%s@%s:"%s"' % (self.user, self.hostname,37 " ".join(escaped_paths))38 def _make_rsync_cmd(self, sources, dest, delete_dest, preserve_symlinks):39 """ Given a list of source paths and a destination path, produces the40 appropriate rsync command for copying them. Remote paths must be41 pre-encoded. """42 ssh_cmd = make_ssh_command(self.user, self.port)43 if delete_dest:44 delete_flag = "--delete"45 else:46 delete_flag = ""47 if preserve_symlinks:48 symlink_flag = ""49 else:50 symlink_flag = "-L"51 command = "rsync %s %s --timeout=1800 --rsh='%s' -az %s %s"52 return command % (symlink_flag, delete_flag, ssh_cmd,53 " ".join(sources), dest)54 def _make_scp_cmd(self, sources, dest):55 """ Given a list of source paths and a destination path, produces the56 appropriate scp command for encoding it. Remote paths must be57 pre-encoded. """58 command = "scp -rpq -P %d %s '%s'"59 return command % (self.port, " ".join(sources), dest)60 def _make_rsync_compatible_globs(self, path, is_local):61 """ Given an rsync-style path, returns a list of globbed paths62 that will hopefully provide equivalent behaviour for scp. Does not63 support the full range of rsync pattern matching behaviour, only that64 exposed in the get/send_file interface (trailing slashes).65 The is_local param is flag indicating if the paths should be66 interpreted as local or remote paths. """67 # non-trailing slash paths should just work68 if len(path) == 0 or path[-1] != "/":69 return [path]70 # make a function to test if a pattern matches any files71 if is_local:72 def glob_matches_files(path):73 return len(glob.glob(path)) > 074 else:75 def glob_matches_files(path):76 result ="ls \"%s\"" % utils.sh_escape(path),77 ignore_status=True)78 return result.exit_status == 079 # take a set of globs that cover all files, and see which are needed80 patterns = ["*", ".[!.]*"]81 patterns = [p for p in patterns if glob_matches_files(path + p)]82 # convert them into a set of paths suitable for the commandline83 path = utils.sh_escape(path)84 if is_local:85 return ["\"%s\"%s" % (path, pattern) for pattern in patterns]86 else:87 return ["\"%s\"" % (path + pattern) for pattern in patterns]88 def _make_rsync_compatible_source(self, source, is_local):89 """ Applies the same logic as _make_rsync_compatible_globs, but90 applies it to an entire list of sources, producing a new list of91 sources, properly quoted. """92 return sum((self._make_rsync_compatible_globs(path, is_local)93 for path in source), [])94 def _set_umask_perms(self, dest):95 """Given a destination file/dir (recursively) set the permissions on96 all the files and directories to the max allowed by running umask."""97 # now this looks strange but I haven't found a way in Python to _just_98 # get the umask, apparently the only option is to try to set it99 umask = os.umask(0)100 os.umask(umask)101 max_privs = 0777 & ~umask102 def set_file_privs(filename):103 file_stat = os.stat(filename)104 file_privs = max_privs105 # if the original file permissions do not have at least one106 # executable bit then do not set it anywhere107 if not file_stat.st_mode & 0111:108 file_privs &= ~0111109 os.chmod(filename, file_privs)110 # try a bottom-up walk so changes on directory permissions won't cut111 # our access to the files/directories inside it112 for root, dirs, files in os.walk(dest, topdown=False):113 # when setting the privileges we emulate the chmod "X" behaviour114 # that sets to execute only if it is a directory or any of the115 # owner/group/other already has execute right116 for dirname in dirs:117 os.chmod(os.path.join(root, dirname), max_privs)118 for filename in files:119 set_file_privs(os.path.join(root, filename))120 # now set privs for the dest itself121 if os.path.isdir(dest):122 os.chmod(dest, max_privs)123 else:124 set_file_privs(dest)125 def get_file(self, source, dest, delete_dest=False, preserve_perm=True,126 preserve_symlinks=False):127 """128 Copy files from the remote host to a local path.129 Directories will be copied recursively.130 If a source component is a directory with a trailing slash,131 the content of the directory will be copied, otherwise, the132 directory itself and its content will be copied. This133 behavior is similar to that of the program 'rsync'.134 Args:135 source: either136 1) a single file or directory, as a string137 2) a list of one or more (possibly mixed)138 files or directories139 dest: a file or a directory (if source contains a140 directory or more than one element, you must141 supply a directory dest)142 delete_dest: if this is true, the command will also clear143 out any old files at dest that are not in the144 source145 preserve_perm: tells get_file() to try to preserve the sources146 permissions on files and dirs147 preserve_symlinks: try to preserve symlinks instead of148 transforming them into files/dirs on copy149 Raises:150 AutoservRunError: the scp command failed151 """152 if isinstance(source, basestring):153 source = [source]154 dest = os.path.abspath(dest)155 try:156 remote_source = self._encode_remote_paths(source)157 local_dest = utils.sh_escape(dest)158 rsync = self._make_rsync_cmd([remote_source], local_dest,159 delete_dest, preserve_symlinks)160 except error.CmdError, e:162 logging.warn("warning: rsync failed with: %s", e)163"attempting to copy with scp instead")164 # scp has no equivalent to --delete, just drop the entire dest dir165 if delete_dest and os.path.isdir(dest):166 shutil.rmtree(dest)167 os.mkdir(dest)168 remote_source = self._make_rsync_compatible_source(source, False)169 if remote_source:170 remote_source = self._encode_remote_paths(remote_source)171 local_dest = utils.sh_escape(dest)172 scp = self._make_scp_cmd([remote_source], local_dest)173 try:174 except error.CmdError, e:176 raise error.AutoservRunError(e.args[0], e.args[1])177 if not preserve_perm:178 # we have no way to tell scp to not try to preserve the179 # permissions so set them after copy instead.180 # for rsync we could use "--no-p --chmod=ugo=rwX" but those181 # options are only in very recent rsync versions182 self._set_umask_perms(dest)183 def send_file(self, source, dest, delete_dest=False,184 preserve_symlinks=False):185 """186 Copy files from a local path to the remote host.187 Directories will be copied recursively.188 If a source component is a directory with a trailing slash,189 the content of the directory will be copied, otherwise, the190 directory itself and its content will be copied. This191 behavior is similar to that of the program 'rsync'.192 Args:193 source: either194 1) a single file or directory, as a string195 2) a list of one or more (possibly mixed)196 files or directories197 dest: a file or a directory (if source contains a198 directory or more than one element, you must199 supply a directory dest)200 delete_dest: if this is true, the command will also clear201 out any old files at dest that are not in the202 source203 preserve_symlinks: controls if symlinks on the source will be204 copied as such on the destination or transformed into the205 referenced file/directory206 Raises:207 AutoservRunError: the scp command failed208 """209 if isinstance(source, basestring):210 source = [source]211 remote_dest = self._encode_remote_paths([dest])212 try:213 local_sources = [utils.sh_escape(path) for path in source]214 rsync = self._make_rsync_cmd(local_sources, remote_dest,215 delete_dest, preserve_symlinks)216 except error.CmdError, e:218 logging.warn("Command rsync failed with: %s", e)219"Attempting to copy with scp instead")220 # scp has no equivalent to --delete, just drop the entire dest dir221 if delete_dest:222 is_dir ="ls -d %s/" % dest,223 ignore_status=True).exit_status == 0224 if is_dir:225 cmd = "rm -rf %s && mkdir %s"226 cmd %= (remote_dest, remote_dest)227 local_sources = self._make_rsync_compatible_source(source, True)...

Full Screen

Full Screen Github


Full Screen

...154 finally:155 self.stderr = logger.debug(self.stderr)157 process.wait()158def _make_rsync_cmd(*options, remote_shell=None):159 """creates the command list for running rsync commands"""160 cmd = [RSYNC_EXECUTABLE]161 if remote_shell:162 if '"' in remote_shell:163 raise ValueError("double quote character in the remote_shell?")164 cmd.extend(["-e", remote_shell])165 for option in options:166 if not option.startswith("-"):167 raise ValueError(f"option '{option}' does not start with '-'")168 cmd.append(option)169 return cmd170def _make_remote_shell_option(171 target, local_ssh_cmd=SSH_EXECUTABLE, tunnel_ssh_cmd="ssh"172):173 return f"{local_ssh_cmd} -A {target} {tunnel_ssh_cmd}"174def _make_remote_path(remote, path):175 return f'{remote}:"{os.fspath(path)}"'176def list_files_on_remote(177 path,178 *,179 target,180 tunnel=None,181 recursive=True,182 long=False,183 regex=None,184 image_id_json=None,185):186 """list files on the remote"""187 options = ["-avz", "--list-only"]188 if not recursive:189 options.append("--no-recursive")190 remote_shell = _make_remote_shell_option(tunnel) if tunnel else None191 remote_location = _make_remote_path(target, path)192 cmd = _make_rsync_cmd(*options, remote_shell=remote_shell)193 cmd.append(remote_location)194 if regex is not None:195 regex = re.compile(regex)196 if image_id_json is not None:197 _image_id_map = defaultdict(set)198 for record in image_id_json:199 iid = tuple(record["image_id"])200 _image_id_map[len(iid)].add(iid)201 image_id_json = dict(sorted(_image_id_map.items()))202 cmd_iter = _CommandIter(cmd)203 it = iter(cmd_iter)204 try:205 line0 = next(it)206 except StopIteration:207 print(cmd_iter.stderr, file=sys.stderr)208 raise RuntimeError(209 "rsync command failed with return_code:", cmd_iter.return_code210 )211 status_msgs = {"receiving file list ... done", "receiving incremental file list"}212 if line0 not in status_msgs:213 raise RuntimeError(f"received: '{line0!r}'")214 # parse files215 for line in it:216 try:217 permission, size, date, mtime, filename = line.split(maxsplit=4)218 except ValueError:219 break # we reached the end of the file list220 if regex and not continue222 if image_id_json:223 fn_parts = Path(filename).parts224 for length, image_ids in image_id_json.items():225 if fn_parts[-length:] in image_ids:226 break227 else:228 continue229 if long:230 print(line)231 else:232 print(filename)233 # parse summary234 _, _ = it # ignore the two summary lines for now235 if cmd_iter.return_code != 0:236 raise RuntimeError(237 "rsync command failed with return_code:", cmd_iter.return_code238 )239def _make_path(path, *, base):240 if base is None:241 base = "/"242 return os.path.join(base, path)243def pull_files_from_remote(*, local_path, remote_base_path, files, target, tunnel=None):244 """pull files from the remote"""245 options = [246 "-avz",247 "--ignore-existing",248 "--partial",249 "--progress",250 f"--files-from={files}",251 ]252 remote_shell = _make_remote_shell_option(tunnel) if tunnel else None253 remote_location = _make_remote_path(target, remote_base_path)254 cmd = _make_rsync_cmd(*options, remote_shell=remote_shell)255 cmd.append(remote_location)256 cmd.append(local_path)257 proc =, env=os.environ)258 if proc.returncode != 0:259 raise RuntimeError("rsync command failed with return_code:", proc.returncode)260# -- commands ---------------------------------------------------------261def main(argv=None):262 global parser263 args = parser.parse_args(argv)264 if args.cmd is None:265 if args.version:266 # noinspection PyProtectedMember267 from pado import __version__268 print(f"{__version__}")...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:


You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?