Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _encode_remote_paths method in autotest

Best Python code snippet using autotest_python

abstract_ssh.py

Source:abstract_ssh.py

...28        self.ip = socket.getaddrinfo(self.hostname, None)[0][4][0]29        self.user = user30        self.port = port31        self.password = password32    def _encode_remote_paths(self, paths):33        """ Given a list of file paths, encodes it as a single remote path, in34        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"...

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.