KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _make_rsync_cmd method in autotest

Best Python code snippet using autotest_python

abstract_ssh.py

Source:abstract_ssh.py

...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 = self.run("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            utils.run(rsync)161        except error.CmdError, e:162            logging.warn("warning: rsync failed with: %s", e)163            logging.info("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                    utils.run(scp)175                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            utils.run(rsync)217        except error.CmdError, e:218            logging.warn("Command rsync failed with: %s", e)219            logging.info("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 = self.run("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                    self.run(cmd)228            local_sources = self._make_rsync_compatible_source(source, True)...

transporter.py

Source:transporter.py

...154        finally:155            self.stderr = process.stderr.read().decode()156            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 regex.search(filename):221            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 = subprocess.run(cmd, 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__}")...

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.