Unlock 30% off on Manual Testing Annual Plans this Holiday Season.
Copied to Clipboard!
How to use _close_jump_boxes method in lisa

Best Python code snippet using lisa_python
shell.py
Source:shell.py
...185                f"failed to connect SSH "186                f"[{self._connection_info.address}:{self._connection_info.port}], "187                f"{identifier.__class__.__name__}: {identifier}"188            )189        self._close_jump_boxes()190        # Some windows doesn't end the text stream, so read first line only.191        # it's  enough to detect os.192        stdout_content = stdout.readline()193        stdout.close()194        if stdout_content and "Windows" in stdout_content:195            self.is_posix = False196            shell_type = WindowsShellType()197        else:198            self.is_posix = True199            shell_type = spur.ssh.ShellTypes.sh200        sock = self._establish_jump_boxes(201            address=self._connection_info.address,202            port=self._connection_info.port,203        )204        spur_kwargs = {205            "hostname": self._connection_info.address,206            "port": self._connection_info.port,207            "username": self._connection_info.username,208            "password": self._connection_info.password,209            "private_key_file": self._connection_info.private_key_file,210            "missing_host_key": spur.ssh.MissingHostKey.accept,211            # There are too many servers in cloud, and they may reuse the same212            # IP in different time. If so, there is host key conflict. So do not213            # load host keys to avoid this kind of error.214            "load_system_host_keys": False,215            "sock": sock,216        }217        spur_ssh_shell = spur.SshShell(shell_type=shell_type, **spur_kwargs)218        sftp = spurplus.sftp.ReconnectingSFTP(219            sftp_opener=spur_ssh_shell._open_sftp_client220        )221        self._inner_shell = spurplus.SshShell(spur_ssh_shell=spur_ssh_shell, sftp=sftp)222    def close(self) -> None:223        if self._inner_shell:224            self._inner_shell.close()225            # after closed, can be reconnect226            self._inner_shell = None227        self._is_initialized = False228        self._close_jump_boxes()229    @property230    def is_connected(self) -> bool:231        is_inner_shell_ready = False232        if self._inner_shell:233            is_inner_shell_ready = True234        return is_inner_shell_ready235    def spawn(236        self,237        command: Sequence[str],238        update_env: Optional[Mapping[str, str]] = None,239        store_pid: bool = False,240        cwd: Optional[Union[str, Path]] = None,241        stdout: Any = None,242        stderr: Any = None,243        encoding: str = "utf-8",244        use_pty: bool = True,245        allow_error: bool = True,246    ) -> spur.ssh.SshProcess:247        self.initialize()248        assert self._inner_shell249        try:250            process: spur.ssh.SshProcess = _spawn_ssh_process(251                self._inner_shell,252                command=command,253                update_env=update_env,254                store_pid=store_pid,255                cwd=cwd,256                stdout=stdout,257                stderr=stderr,258                encoding=encoding,259                use_pty=use_pty,260                allow_error=allow_error,261            )262        except FunctionTimedOut:263            raise LisaException(264                f"The remote node is timeout on execute {command}. "265                f"It may be caused by paramiko/spur not support the shell of node."266            )267        return process268    def mkdir(269        self,270        path: PurePath,271        mode: int = 0o777,272        parents: bool = True,273        exist_ok: bool = False,274    ) -> None:275        """Create the directory(ies), if they do not already exist.276        Inputs:277            path: directory path. (Absolute. Use a PurePosixPath, if the278                                   target node is a Posix one, because LISA279                                   might be ran from Windows)280            mode: directory creation mode (Posix targets only)281            parents: make parent directories as needed282            exist_ok: return with no error if target already present283        """284        path_str = self._purepath_to_str(path)285        self.initialize()286        assert self._inner_shell287        try:288            self._inner_shell.mkdir(289                path_str, mode=mode, parents=parents, exist_ok=exist_ok290            )291        except PermissionError:292            self._inner_shell.run(command=["sudo", "mkdir", "-p", path_str])293        except SSHException as identifier:294            # no sftp, try commands295            if "Channel closed." in str(identifier):296                assert isinstance(path_str, str)297                self.spawn(command=["mkdir", "-p", path_str])298    def exists(self, path: PurePath) -> bool:299        """Check if a target directory/file exist300        Inputs:301            path: target path. (Absolute. Use a PurePosixPath, if the302                                target node is a Posix one, because LISA303                                might be ran from Windows)304        Outputs:305            bool: True if present, False otherwise306        """307        self.initialize()308        assert self._inner_shell309        path_str = self._purepath_to_str(path)310        return cast(bool, self._inner_shell.exists(path_str))311    def remove(self, path: PurePath, recursive: bool = False) -> None:312        """Remove a target directory/file313        Inputs:314            path: target path. (Absolute. Use a PurePosixPath, if the315                                target node is a Posix one, because LISA316                                might be ran from Windows)317            recursive: whether to remove recursively, if target is a directory318                       (will fail if that's the case and this flag is off)319        """320        self.initialize()321        assert self._inner_shell322        path_str = self._purepath_to_str(path)323        try:324            self._inner_shell.remove(path_str, recursive)325        except PermissionError:326            self._inner_shell.run(command=["sudo", "rm", path_str])327        except SSHException as identifier:328            # no sftp, try commands329            if "Channel closed." in str(identifier):330                assert isinstance(path_str, str)331                self.spawn(command=["rm", path_str])332    def chmod(self, path: PurePath, mode: int) -> None:333        """Change the file mode bits of each given file according to mode (Posix targets only)334        Inputs:335            path: target path. (Absolute. Use a PurePosixPath, if the336                                target node is a Posix one, because LISA337                                might be ran from Windows)338            mode: numerical chmod mode entry339        """340        self.initialize()341        assert self._inner_shell342        path_str = self._purepath_to_str(path)343        self._inner_shell.chmod(path_str, mode)344    def stat(self, path: PurePath) -> os.stat_result:345        """Display file/directory status.346        Inputs:347            path: target path. (Absolute. Use a PurePosixPath, if the348                                target node is a Posix one, because LISA349                                might be ran from Windows)350        Outputs:351            os.stat_result: The status structure/class352        """353        self.initialize()354        assert self._inner_shell355        path_str = self._purepath_to_str(path)356        sftp_attributes: paramiko.SFTPAttributes = self._inner_shell.stat(path_str)357        result = os.stat_result(358            (359                # st_mode360                sftp_attributes.st_mode if sftp_attributes.st_mode is not None else 0,361                # st_ino362                0,363                # st_dev364                0,365                # st_nlink366                0,367                # st_uid368                sftp_attributes.st_uid if sftp_attributes.st_uid is not None else 0,369                # st_gid370                sftp_attributes.st_gid if sftp_attributes.st_gid is not None else 0,371                # st_size372                sftp_attributes.st_size if sftp_attributes.st_size is not None else 0,373                # st_atime374                sftp_attributes.st_atime if sftp_attributes.st_atime is not None else 0,375                # st_mtime376                sftp_attributes.st_mtime if sftp_attributes.st_mtime is not None else 0,377                # st_ctime378                0,379            )380        )381        return result382    def is_dir(self, path: PurePath) -> bool:383        """Check if given path is a directory384        Inputs:385            path: target path. (Absolute. Use a PurePosixPath, if the386                                target node is a Posix one, because LISA387                                might be ran from Windows)388        Outputs:389            bool: True if it is a directory, False otherwise390        """391        self.initialize()392        assert self._inner_shell393        path_str = self._purepath_to_str(path)394        return cast(bool, self._inner_shell.is_dir(path_str))395    def is_symlink(self, path: PurePath) -> bool:396        """Check if given path is a symlink397        Inputs:398            path: target path. (Absolute. Use a PurePosixPath, if the399                                target node is a Posix one, because LISA400                                might be ran from Windows)401        Outputs:402            bool: True if it is a symlink, False otherwise403        """404        self.initialize()405        assert self._inner_shell406        path_str = self._purepath_to_str(path)407        return cast(bool, self._inner_shell.is_symlink(path_str))408    def symlink(self, source: PurePath, destination: PurePath) -> None:409        """Create a symbolic link from source to destination, in the target node410        Inputs:411            source: source path. (Absolute. Use a PurePosixPath, if the412                                 target node is a Posix one, because LISA413                                 might be ran from Windows)414            destination: destination path. (Absolute. Use a PurePosixPath, if the415                                            target node is a Posix one, because LISA416                                            might be ran from Windows)417        """418        self.initialize()419        assert self._inner_shell420        source_str = self._purepath_to_str(source)421        destination_str = self._purepath_to_str(destination)422        self._inner_shell.symlink(source_str, destination_str)423    def copy(self, local_path: PurePath, node_path: PurePath) -> None:424        """Upload local file to target node425        Inputs:426            local_path: local path. (Absolute. Use a PurePosixPath, if the427                                     target node is a Posix one, because LISA428                                     might be ran from Windows)429            node_path: target path. (Absolute. Use a PurePosixPath, if the430                                     target node is a Posix one, because LISA431                                     might be ran from Windows)432        """433        self.mkdir(node_path.parent, parents=True, exist_ok=True)434        self.initialize()435        assert self._inner_shell436        local_path_str = self._purepath_to_str(local_path)437        node_path_str = self._purepath_to_str(node_path)438        self._inner_shell.put(439            local_path_str,440            node_path_str,441            create_directories=True,442            consistent=self.is_posix,443        )444    def copy_back(self, node_path: PurePath, local_path: PurePath) -> None:445        """Download target node's file to local node446        Inputs:447            local_path: local path. (Absolute. Use a PurePosixPath, if the448                                     target node is a Posix one, because LISA449                                     might be ran from Windows)450            node_path: target path. (Absolute. Use a PurePosixPath, if the451                                     target node is a Posix one, because LISA452                                     might be ran from Windows)453        """454        self.initialize()455        assert self._inner_shell456        node_path_str = self._purepath_to_str(node_path)457        local_path_str = self._purepath_to_str(local_path)458        self._inner_shell.get(459            node_path_str,460            local_path_str,461            consistent=self.is_posix,462        )463    def _purepath_to_str(464        self, path: Union[Path, PurePath, str]465    ) -> Union[Path, PurePath, str]:466        """467        spurplus doesn't support pure path, so it needs to convert.468        """469        if isinstance(path, PurePath):470            path = str(path)471        return path472    def _establish_jump_boxes(self, address: str, port: int) -> Any:473        jump_boxes_runbook = development.get_jump_boxes()474        sock: Any = None475        is_trace_enabled = development.is_trace_enabled()476        if is_trace_enabled:477            jb_logger = _get_jump_box_logger()478            jb_logger.debug(f"proxy sock: {sock}")479        for index, runbook in enumerate(jump_boxes_runbook):480            if is_trace_enabled:481                jb_logger.debug(f"creating connection from source: {runbook} ")482            client = paramiko.SSHClient()483            client.set_missing_host_key_policy(paramiko.MissingHostKeyPolicy())484            client.connect(485                hostname=runbook.address,486                port=runbook.port,487                username=runbook.username,488                password=runbook.password,489                key_filename=runbook.private_key_file,490                banner_timeout=10,491                sock=sock,492            )493            if index < len(jump_boxes_runbook) - 1:494                next_hop = jump_boxes_runbook[index + 1]495                dest_address = (496                    next_hop.private_address497                    if next_hop.private_address498                    else next_hop.address499                )500                dest_port = (501                    next_hop.private_port if next_hop.private_port else next_hop.port502                )503            else:504                dest_address = address505                dest_port = port506            if is_trace_enabled:507                jb_logger.debug(f"next hop: {dest_address}:{dest_port}")508            sock = self._open_jump_box_channel(509                client,510                src_address=runbook.address,511                src_port=runbook.port,512                dest_address=dest_address,513                dest_port=dest_port,514            )515            self._jump_boxes.append(client)516        return sock517    def _open_jump_box_channel(518        self,519        client: paramiko.SSHClient,520        src_address: str,521        src_port: int,522        dest_address: str,523        dest_port: int,524    ) -> Any:525        transport = client.get_transport()526        assert transport527        sock = transport.open_channel(528            kind="direct-tcpip",529            src_addr=(src_address, src_port),530            dest_addr=(dest_address, dest_port),531        )532        return sock533    def _close_jump_boxes(self) -> None:534        for index in reversed(range(len(self._jump_boxes))):535            self._jump_boxes[index].close()536            self._jump_boxes[index] = None537        self._jump_boxes.clear()538        self._jump_box_sock = None539class LocalShell(InitializableMixin):540    def __init__(self) -> None:541        super().__init__()542        self.is_remote = False543        self._inner_shell = spur.LocalShell()544    def _initialize(self, *args: Any, **kwargs: Any) -> None:545        if "win32" == sys.platform:546            self.is_posix = False547        else:...
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.