import paramiko import openpyxl # 读取Excel中的账号和密码信息 wb = openpyxl.load_workbook('/root/hosts.xlsx') sheet = wb.active # 连接至Linux主机 # values_only=True ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) for row in sheet.iter_rows(min_row=2,values_only=True): username = row[0] password = row[1] host = row[2] ssh.connect(hostname=host, username=username, password=password, port=22) # 执行脚本 stdin, stdout, stderr = ssh.exec_command(''' #!/bin/bash NAME=$(uname -n) if [ -x "$(command -v mysql)" ]; then echo "MySQL is installed" fi if [ -x "$(command -v oracle)" ]; then echo "Oracle is installed" fi if [ -x "$(command -v postgres)" ]; then echo "PostgreSQL is installed" fi if [ -x "$(command -v mongo)" ]; then echo "Mongo is installed" fi if [ -x "$(command -v redis-cli)" ]; then echo "Redis is installed" fi if [ ! -x "$(command -v mysql)" ] && [ ! -x "$(command -v oracle)" ] && [ ! -x "$(command -v postgres)" ] && [ ! -x "$(command -v mongo)" ] && [ ! -x "$(command -v redis-cli)" ]; then echo "No supported database is installed" ''') # 收集查询结果并反馈 output = stdout.read().decode() print(host+"\n"+output) # 关闭连接 ssh.close() 这个python脚本无法检测Oracle，但是里面的shell脚本单独执行可以检测Oracle

这可能是由于在Python脚本中使用了双引号，导致Oracle命令被解析成了变量。可以尝试将双引号改成单引号或者使用转义符号来解决这个问题，例如：

if [ -x "$(command -v 'oracle')" ]; then

或者

if [ -x "$(command -v \"oracle\")" ]; then

相关问题

检查代码中的错误 class ADB_SHELL: def get_ip(self): # 执行adb shell命令并输出结果 subprocess.check_output( ['adb', 'shell', 'udhcpc'] ) subprocess.check_output( ['adb', 'shell', 'udhcpc -i eth1'] ) self.conf = subprocess.check_output( ['adb', 'shell', 'ifconfig'] ).decode() # conf = str(ip).split(r'\r\r\n') # tmp = conf.replace( "\r\r\n", "\n" ) # print( tmp) self.ip = re.findall( r'addr:(.*?) Bcast', str( self.conf ) ) print(self.ip) for self.i in selfip : speed = subprocess.check_output((['adb', 'shell', f'iperf3 -B {self.i} -c 192.168.102.105'])).decode() print(speed.replace("\r\r\n", "\n")) for i in range(5): write_data = subprocess.check_output(['adb', 'shell', 'time dd if=/dev/zero of=/data/test.data bs=128k count=1024']).decode() print(write_data.replace("\r\r\n", "\n")) read_data = subprocess.check_output(['adb', 'shell', 'time dd if=/data/test.data of=/dev/null bs=128k count=1024']).decode() print(read_data.replace("\r\r\n", "\n")) ls = subprocess.check_output( ['adb', 'shell', 'ls /data'] ).decode() print( ls.replace( '\r', ' ' ) ) dl = subprocess.check_output( ['adb', 'shell', 'rm /data/test.data'] ).decode() l = subprocess.check_output(['adb', 'shell', 'ls /data']).decode() print( l.replace( '\r', ' ' ) ) subprocess.check_output( (['adb', 'shell', f'iperf3 -s']) ) def get_ssh(self): ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy) for i in self.ip: ssh.connect(hostname='192.168.102.105',port=22,username='root',password='xiayi123456',timeout=30) stdin,stdout,stderr = ssh.exec_command(f'iperf3 -c {i}') print(stdout.read()) # ssh.close() if __name__ == '__main__': # get_ip() # get_ssh() A = ADB_SHELL t1 = threading.Thread(target=A.get_ssh) t2 = threading.Thread(target=A.get_ip) t1.start() t2.start() t1.join() t2.join()

在第12行代码中，将 `selfip` 改为 `self.ip`。此外，第28行代码中的 `ssh.set_missing_host_key_policy` 应该是 `ssh.set_missing_host_key_policy()`。最后，第39行代码中的 `A = ADB_SHELL` 应该是 `A = ADB_SHELL()`，因为 ADB_SHELL 是一个类，需要实例化才能调用其方法。

import paramiko import pytest import psutil from common.abs_common import log @pytest.fixture(scope="module") def ssh_client(): print('-----ssh远程连接-------') ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect('10.67.6.56', username='root', password='1') yield ssh ssh.close() print("获取PCIe设备的bus number，带宽和速度") def test_network_interface(ssh_client): stdin, stdout, stderr = ssh_client.exec_command("lspci|grep -i 'non' | cut -f1 -d ' '") interface = stdout.read().decode() stdin, stdout, stderr = ssh_client.exec_command(f"lspci -s {interface} -vvv |grep -i Lnkcap | cut -f2 -d ','") widthcap = stdout.read().decode().strip() outputwidthcap = stdout.read().decode("utf-8") stdin, stdout, stderr = ssh_client.exec_command(f"lspci -s {interface} -vvv |grep -i Lnksta | cut -f2 -d ','") widthsta = stdout.read().decode().strip() outputwidthsta = stdout.read().decode("utf-8") stdin, stdout, stderr = ssh_client.exec_command(f"lspci -s {interface} -vvv |grep -i Lnkcap | cut -f3 -d ','") speedcap = stdout.read().decode().strip() outputspeedcap = stdout.read().decode("utf-8") stdin, stdout, stderr = ssh_client.exec_command(f"lspci -s {interface} -vvv |grep -i Lnksta | cut -f3 -d ','") speedsta = stdout.read().decode().strip() outputspeedsta = stdout.read().decode("utf-8") print("比较lspci中PCIe设备的带宽和速度Lnkcap和Lnksta") assert widthcap == widthsta assert speedcap == speedsta print(outputwidthcap,outputwidthsta,outputspeedcap,outputspeedsta) if __name__ == '__main__': pytest.main(['-o log_cli=True --html=./test_get_nic_width.html -s'])

这段代码是用于测试获取PCIe设备的带宽和速度，其中使用了paramiko库建立远程SSH连接，执行了一些命令来获取相关信息。在test_network_interface函数中，首先使用lspci命令获取PCIe设备的bus number，然后通过grep和cut命令获取Lnkcap和Lnksta信息，分别表示PCIe设备的带宽和速度，最后通过assert语句比较这两个值是否相等。如果测试通过，会输出相关信息，并生成一个HTML格式的测试报告。