How to use _get_proxy_client_info method in tempest

Best Python code snippet using tempest_python

ssh.py

Source:ssh.py

...107                    LOG.exception("Failed to establish authenticated ssh"108                                  " connection to %s@%s after %d attempts. "109                                  "Proxy client: %s",110                                  self.username, self.host, attempts,111                                  self._get_proxy_client_info())112                    raise exceptions.SSHTimeout(host=self.host,113                                                user=self.username,114                                                password=self.password)115                bsleep += backoff116                attempts += 1117                LOG.warning("Failed to establish authenticated ssh"118                            " connection to %s@%s (%s). Number attempts: %s."119                            " Retry after %d seconds.",120                            self.username, self.host, e, attempts, bsleep)121                time.sleep(bsleep)122    def _is_timed_out(self, start_time):123        return (time.time() - self.timeout) > start_time124    @staticmethod125    def _can_system_poll():126        return hasattr(select, 'poll')127    def exec_command(self, cmd, encoding="utf-8"):128        """Execute the specified command on the server129        Note that this method is reading whole command outputs to memory, thus130        shouldn't be used for large outputs.131        :param str cmd: Command to run at remote server.132        :param str encoding: Encoding for result from paramiko.133                             Result will not be decoded if None.134        :returns: data read from standard output of the command.135        :raises: SSHExecCommandFailed if command returns nonzero136                 status. The exception contains command status stderr content.137        :raises: TimeoutException if cmd doesn't end when timeout expires.138        """139        ssh = self._get_ssh_connection()140        transport = ssh.get_transport()141        with transport.open_session() as channel:142            channel.fileno()  # Register event pipe143            channel.exec_command(cmd)144            channel.shutdown_write()145            # If the executing host is linux-based, poll the channel146            if self._can_system_poll():147                out_data_chunks = []148                err_data_chunks = []149                poll = select.poll()150                poll.register(channel, select.POLLIN)151                start_time = time.time()152                while True:153                    ready = poll.poll(self.channel_timeout)154                    if not any(ready):155                        if not self._is_timed_out(start_time):156                            continue157                        raise exceptions.TimeoutException(158                            "Command: '{0}' executed on host '{1}'.".format(159                                cmd, self.host))160                    if not ready[0]:  # If there is nothing to read.161                        continue162                    out_chunk = err_chunk = None163                    if channel.recv_ready():164                        out_chunk = channel.recv(self.buf_size)165                        out_data_chunks += out_chunk,166                    if channel.recv_stderr_ready():167                        err_chunk = channel.recv_stderr(self.buf_size)168                        err_data_chunks += err_chunk,169                    if not err_chunk and not out_chunk:170                        break171                out_data = b''.join(out_data_chunks)172                err_data = b''.join(err_data_chunks)173            # Just read from the channels174            else:175                out_file = channel.makefile('rb', self.buf_size)176                err_file = channel.makefile_stderr('rb', self.buf_size)177                out_data = out_file.read()178                err_data = err_file.read()179            if encoding:180                out_data = out_data.decode(encoding)181                err_data = err_data.decode(encoding)182            exit_status = channel.recv_exit_status()183        ssh.close()184        if 0 != exit_status:185            raise exceptions.SSHExecCommandFailed(186                command=cmd, exit_status=exit_status,187                stderr=err_data, stdout=out_data)188        return out_data189    def test_connection_auth(self):190        """Raises an exception when we can not connect to server via ssh."""191        connection = self._get_ssh_connection()192        connection.close()193    def _get_proxy_channel(self):194        conn = self.proxy_client._get_ssh_connection()195        # Keep a reference to avoid g/c196        # https://github.com/paramiko/paramiko/issues/440197        self._proxy_conn = conn198        transport = conn.get_transport()199        chan = transport.open_session()200        cmd = 'nc %s %s' % (self.host, self.port)201        chan.exec_command(cmd)202        return chan203    def _get_proxy_client_info(self):204        if not self.proxy_client:205            return 'no proxy client'206        nested_pclient = self.proxy_client._get_proxy_client_info()207        return ('%(username)s@%(host)s:%(port)s, nested proxy client: '208                '%(nested_pclient)s' % {'username': self.proxy_client.username,209                                        'host': self.proxy_client.host,210                                        'port': self.proxy_client.port,...

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.