Python使用paramiko连接远程服务器执行Shell命令的实现_Python

Python使用paramiko连接远程服务器执行Shell命令的实现

2021-09-14 00:29木法星人 Python

这篇文章主要介绍了Python使用paramiko连接远程服务器执行Shell命令的实现,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下

需求

在自动化测试场景里, 有时需要在代码里获取远程服务器的某些数据, 或执行一些查询命令,如获取Linux系统版本号 \ 获取CPU及内存的占用等, 本章记录一下使用paramiko模块SSH连接服务器的方法

1. 先安装paramiko库

				?

									pip3 install paramiko

2. 代码

				?

									#!/usr/bin/env python

									# coding=utf-8

									"""

									# :author: Terry Li

									# :url: https://blog.csdn.net/qq_42183962

									# :copyright: © 2020-present Terry Li

									# :motto: I believe that the God rewards the diligent.

									"""

									import paramiko

									class cfg:

									    host = "192.168.2.2"

									    user = "root"

									    password = "123456"

									class sshChannel:

									    def __init__(self, cfg_obj, timeout_s=5, port=22):

									        self._cfg = cfg_obj

									        self.ssh_connect_timeout = timeout_s

									        self.port = port

									        self.ssh = self.connect_server()

									    def connect_server(self):

									        ssh_cli = paramiko.SSHClient()

									        key = paramiko.AutoAddPolicy()

									        ssh_cli.set_missing_host_key_policy(key)

									        try:

									            ssh_cli.connect(self._cfg.host, port=self.port, username=self._cfg.user, password=self._cfg.password,

									                            timeout=self.ssh_connect_timeout)

									        except paramiko.ssh_exception.SSHException:

									            print("连接{}失败, 请检查配置或重试".format(self._cfg.host))

									            ssh_cli.close()

									        return ssh_cli

									    def execute_cmd(self, cmd):

									        """

									        :param cmd: 单个命令

									        :return: 服务器的输出信息

									        """

									        stdin, stdout, stderr = self.ssh.exec_command(cmd)

									        self.ssh.close()

									        return stdout.read().decode('utf-8')

									    def execute_cmd_list(self, cmd_list):

									        """

									        :param cmd: 命令列表

									        :return: 服务器的输出信息的列表

									        """

									        out_list = list(map(self.execute_cmd, cmd_list))

									        return out_list

									    def test_get_sys_version(self):

									        sys_version = self.execute_cmd("lsb_release -rd")

									        print(sys_version)

									    def test_get_sys_disk_free_and_memory_free(self):

									        sys_info = self.execute_cmd_list(["df -h -BG /", "free -m"])

									        print(sys_info)

									if __name__ == '__main__':

									    server = sshChannel(cfg)

									    server.test_get_sys_version()

									    server.test_get_sys_disk_free_and_memory_free()