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

How to use _make_ssh_cmd_default method in autotest

Best Python code snippet using autotest_python

abstract_ssh.py

Source:abstract_ssh.py

...16from caliper.server.hosts import remote_host17from caliper.client.shared.settings import settings18enable_master_ssh = settings.get_value('SERVER', 'enable_master_ssh',19                                        type=bool, default=False)20def _make_ssh_cmd_default(user="root", port=22, opts='',21                            hosts_file='/dev/null', connect_timeout=30,22                            alive_interval=300):23    base_command = ("/usr/bin/ssh -a -x %s -o StrictHostKeyChecking=no "24                    "-o UserKnownHostsFile=%s -o BatchMode=yes "25                    "-o ConnectTimeout=%d -o ServerAliveInterval=%d "26                    "-l %s -p %d")27    assert isinstance(connect_timeout, (int, long))28    assert connect_timeout > 029    if isinstance(port, str):30        port = int(port)31    return base_command % (opts, hosts_file,32                            connect_timeout, alive_interval, user, port)33class AbstractSSHHost(remote_host.RemoteHost):34    def _initialize(self, hostname, user="root", port=22, password="", *args,35                    **dargs):36        super(AbstractSSHHost, self)._initialize(hostname=hostname, *args,37                **dargs)38        self.ip = socket.getaddrinfo(self.hostname, None)[0][4][0]39        self.user = user40        self.port = port41        self.password = password42        self._use_rsync = None43        self.known_hosts_file = tempfile.mkstemp()[1]44        """45        master SSH connection backgroud job, socket temp directory and socket46        control path option. if master-ssh is enabled, these fields will be47        initialized by start_master_ssh when a new SSH connection is initialed48        """49        self.master_ssh_job = None50        self.master_ssh_tempdir = None51        self.master_ssh_option = ''52    def use_rsync(self):53        if self._use_rsync is not None:54            return self._use_rsync55        # check if the rsync is available on the remote host. If it's not,56        # don't try to use it for any future file transfers.57        self._use_rsync = self._check_rsync()58        if not self._use_rsync:59            logging.warn("rsync not available on remote host %s -- disabled",60                            self.hostname)61        return self._use_rsync62    def _check_rsync(self):63        try:64            self.run("rsync --version", stdout_tee=None, stderr_tee=None)65        except error.ServRunError:66            return False67        return True68    def _make_rsync_cmd(self, sources, dest, delete_dest, preserve_symlinks):69        """70        given a list of file paths, encodes it as a single remote path,71        in the style used by rsync and scp72        """73        ssh_cmd = _make_ssh_cmd_default(user=self.user, port=self.port,74                                        opts=self.master_ssh_option,75                                        hosts_file=self.known_hosts_file)76        if delete_dest:77            delete_flag = "--delete"78        else:79            delete_flag = ""80        if preserve_symlinks:81            symlink_flag = ""82        else:83            symlink_flag = "-L"84        command = "rsync %s %s --timeout=1800 --rsh='%s' -az %s %s"85        return command % (symlink_flag, delete_flag, ssh_cmd,86                        " ".join(sources), dest)87    def _make_rsync_compatible_globs(self, path, is_local):88        """89        given an rsync-style path, returns a list of globbed paths that will90        hopefully provide equivalent behaviour for scp. Does not support the91        full range of rsync pattern matching behaviour, only that exposed in92        the get/send_file interface.93        """94        if len(path) == 0 or path[-1] != "/":95            return [path]96        if is_local:97            def glob_matches_files(path, pattern):98                return len(glob.glob(path + pattern)) > 099        else:100            def glob_matches_files(path, pattern):101                result = self.run("ls \"%s\"%s" %102                            (utils.sh_escape(path), pattern),103                            stdout_tee=None, ignore_status=True)104                return result.exit_status == 0105        patterns = ["*", ".[!.]*"]106        patterns = [p for p in patterns if glob_matches_files(path, p)]107        # convert them into a set of paths suitable for the commandline108        if is_local:109            return ["\"%s\"%s" % (utils.sh_escape(path), pattern)110                    for pattern in patterns]111        else:112            return [utils.scp_remote_escape(path) + pattern113                    for pattern in patterns]114    def _make_rsync_compatible_source(self, source, is_local):115        """116        Applies the same logic as _make_rsync_ccompatible_globs, but applies if117        to an entire list of sources, producing a new list of sources,118        properly quoted.119        """120        return sum((self._make_rsync_compatible_globs(path, is_local)121                    for path in source), [])122    def _encode_remote_paths(self, paths, escape=True):123        """124        Given a list of file paths, encodes it as a single remote path, in the125        style used by rsync and scp.126        """127        if escape:128            paths = [utils.scp_remote_escape(path) for path in paths]129        return '%s@%s:"%s"' % (self.user, self.hostname, " ".join(paths))130    def _make_ssh_cmd(self, cmd):131        base_cmd = _make_ssh_cmd_default(user=self.user, port=self.port,132                                    opts=self.master_ssh_option,133                                    hosts_file=self.known_hosts_file)134        return '%s %s "%s"' % (base_cmd, self.hostname, utils.sh_escape(cmd))135    def _make_scp_cmd(self, sources, dest):136        """137        Given a list of source paths and a destination path, produces the138        appropriate scp command for encoding it. Remote paths must be139        pre-encoded.140        """141        command = ("scp -rq %s -o StrictHostKeyChecking=no -o "142                    "UserKnownHostsFile=%s -P %d %s '%s'")143        return command % (self.master_ssh_option, self.known_hosts_file,144                            self.port, " ".join(sources), dest)145    def _set_umask_perms(self, dest):...

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.