当前位置:首页 >> 脚本专栏

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()