Implementation of Python Using paramiko to Connect Remote Server to Execute Shell Command
- 2021-09-16 07:33:57
- OfStack
Demand
In
自动化测试
In the scenario, it is sometimes necessary to obtain some data of the remote server in the code, or to execute some query commands, such as obtaining Linux system version number\ obtaining CPU and memory occupation, etc. This chapter records the method of connecting the server with SSH of paramiko module under 1
1. Install the paramiko library first
pip3 install paramiko
2. Code
#!/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(" Connect {} Failure , Please check the configuration or try again ".format(self._cfg.host))
ssh_cli.close()
return ssh_cli
def execute_cmd(self, cmd):
"""
:param cmd: Single command
:return: Output information from the server
"""
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: Command list
:return: List of output information from the server
"""
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()