How to use _cleanup_master_ssh method in autotest

Best Python code snippet using autotest_python

connection.py

Source:connection.py

...87                if (not os.path.exists(socket_path) or88                        self._master_ssh_proc.poll() is not None):89                    self.log.debug('Master ssh connection to %s is down.',90                                   self._settings.hostname)91                    self._cleanup_master_ssh()92            if self._master_ssh_proc is None:93                # Create a shared socket in a temp location.94                self._master_ssh_tempdir = tempfile.mkdtemp(95                    prefix='ssh-master')96                # Setup flags and options for running the master ssh97                # -N: Do not execute a remote command.98                # ControlMaster: Spawn a master connection.99                # ControlPath: The master connection socket path.100                extra_flags = {'-N': None}101                extra_options = {102                    'ControlMaster': True,103                    'ControlPath': self.socket_path,104                    'BatchMode': True105                }106                # Construct the command and start it.107                master_cmd = self._formatter.format_ssh_local_command(108                    self._settings,109                    extra_flags=extra_flags,110                    extra_options=extra_options)111                self.log.info('Starting master ssh connection.')112                self._master_ssh_proc = job.run_async(master_cmd)113                end_time = time.time() + timeout_seconds114                while time.time() < end_time:115                    if os.path.exists(self.socket_path):116                        break117                    time.sleep(.2)118                else:119                    self._cleanup_master_ssh()120                    raise Error('Master ssh connection timed out.')121    def run(self,122            command,123            timeout=3600,124            ignore_status=False,125            env=None,126            io_encoding='utf-8',127            attempts=2):128        """Runs a remote command over ssh.129        Will ssh to a remote host and run a command. This method will130        block until the remote command is finished.131        Args:132            command: The command to execute over ssh. Can be either a string133                     or a list.134            timeout: number seconds to wait for command to finish.135            ignore_status: bool True to ignore the exit code of the remote136                           subprocess.  Note that if you do ignore status codes,137                           you should handle non-zero exit codes explicitly.138            env: dict environment variables to setup on the remote host.139            io_encoding: str unicode encoding of command output.140            attempts: Number of attempts before giving up on command failures.141        Returns:142            A job.Result containing the results of the ssh command.143        Raises:144            job.TimeoutError: When the remote command took to long to execute.145            Error: When the ssh connection failed to be created.146            CommandError: Ssh worked, but the command had an error executing.147        """148        if attempts == 0:149            return None150        if env is None:151            env = {}152        try:153            self.setup_master_ssh(self._settings.connect_timeout)154        except Error:155            self.log.warning('Failed to create master ssh connection, using '156                             'normal ssh connection.')157        extra_options = {'BatchMode': True}158        if self._master_ssh_proc:159            extra_options['ControlPath'] = self.socket_path160        identifier = str(uuid.uuid4())161        full_command = 'echo "CONNECTED: %s"; %s' % (identifier, command)162        terminal_command = self._formatter.format_command(163            full_command, env, self._settings, extra_options=extra_options)164        dns_retry_count = 2165        while True:166            result = job.run(167                terminal_command, ignore_status=True, timeout=timeout)168            output = result.stdout169            # Check for a connected message to prevent false negatives.170            valid_connection = re.search(171                '^CONNECTED: %s' % identifier, output, flags=re.MULTILINE)172            if valid_connection:173                # Remove the first line that contains the connect message.174                line_index = output.find('\n')175                real_output = output[line_index + 1:].encode(result._encoding)176                result = job.Result(177                    command=result.command,178                    stdout=real_output,179                    stderr=result._raw_stderr,180                    exit_status=result.exit_status,181                    duration=result.duration,182                    did_timeout=result.did_timeout,183                    encoding=result._encoding)184                if result.exit_status and not ignore_status:185                    raise job.Error(result)186                return result187            error_string = result.stderr188            had_dns_failure = (result.exit_status == 255 and re.search(189                r'^ssh: .*: Name or service not known',190                error_string,191                flags=re.MULTILINE))192            if had_dns_failure:193                dns_retry_count -= 1194                if not dns_retry_count:195                    raise Error('DNS failed to find host.', result)196                self.log.debug('Failed to connect to host, retrying...')197            else:198                break199        had_timeout = re.search(200            r'^ssh: connect to host .* port .*: '201            r'Connection timed out\r$',202            error_string,203            flags=re.MULTILINE)204        if had_timeout:205            raise Error('Ssh timed out.', result)206        permission_denied = 'Permission denied' in error_string207        if permission_denied:208            raise Error('Permission denied.', result)209        unknown_host = re.search(210            r'ssh: Could not resolve hostname .*: '211            r'Name or service not known',212            error_string,213            flags=re.MULTILINE)214        if unknown_host:215            raise Error('Unknown host.', result)216        self.log.error('An unknown error has occurred. Job result: %s' % result)217        ping_output = job.run(218            'ping %s -c 3 -w 1' % self._settings.hostname, ignore_status=True)219        self.log.error('Ping result: %s' % ping_output)220        if attempts > 1:221            self._cleanup_master_ssh()222            self.run(command, timeout, ignore_status, env, io_encoding,223                     attempts - 1)224        raise Error('The job failed for unknown reasons.', result)225    def run_async(self, command, env=None):226        """Starts up a background command over ssh.227        Will ssh to a remote host and startup a command. This method will228        block until there is confirmation that the remote command has started.229        Args:230            command: The command to execute over ssh. Can be either a string231                     or a list.232            env: A dictonary of environment variables to setup on the remote233                 host.234        Returns:235            The result of the command to launch the background job.236        Raises:237            CmdTimeoutError: When the remote command took to long to execute.238            SshTimeoutError: When the connection took to long to established.239            SshPermissionDeniedError: When permission is not allowed on the240                                      remote host.241        """242        command = '(%s) < /dev/null > /dev/null 2>&1 & echo -n $!' % command243        result = self.run(command, env=env)244        return result245    def close(self):246        """Clean up open connections to remote host."""247        self._cleanup_master_ssh()248        while self._tunnels:249            self.close_ssh_tunnel(self._tunnels[0].local_port)250    def _cleanup_master_ssh(self):251        """252        Release all resources (process, temporary directory) used by an active253        master SSH connection.254        """255        # If a master SSH connection is running, kill it.256        if self._master_ssh_proc is not None:257            self.log.debug('Nuking master_ssh_job.')258            self._master_ssh_proc.kill()259            self._master_ssh_proc.wait()260            self._master_ssh_proc = None261        # Remove the temporary directory for the master SSH socket.262        if self._master_ssh_tempdir is not None:263            self.log.debug('Cleaning master_ssh_tempdir.')264            shutil.rmtree(self._master_ssh_tempdir)...

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.