在软件测试自动化中，我们常常需要远程操作服务器 —— 列如部署测试环境、查看日志、执行脚本、传输测试文件等。传统方式靠手动SSH连接，效率低且易出错，而Paramiko作为Python中最常用的SSH协议实现库，能完美解决这些问题：通过代码实现 SSH 远程登录、命令执行、文件传输，轻松集成到自动化测试流程中。

本文将聚焦Paramiko在软件测试自动化中的核心应用，从环境搭建、重点函数解析到完整实战案例，帮你快速掌握 “代码操控服务器” 的技能，让测试自动化覆盖更全场景。

Python 自动化测试神器：Paramiko实现SSH远程操作

一、Paramiko 基础：环境搭建与核心价值

在讲具体用法前，先快速搞定环境搭建，明确Paramiko能解决哪些测试痛点。

1. 环境搭建（2 分钟上手）

Paramiko是第三方库，需通过pip安装

# 安装核心库（适用于
Python 3.6+）
pip install paramiko

安装完成后，在 Python 脚本中导入即可使用：

import paramiko

2. 核心价值（测试自动化中必用的 3 个场景）

远程执行命令：自动化部署测试服务（如启动 / 停止被测应用、重启数据库）、查看服务器状态（CPU / 内存使用率）；
文件上传下载：将本地测试脚本、测试数据上传到服务器，或把服务器上的测试日志、执行结果下载到本地分析；
持久化连接：保持SSH长连接，批量执行多组测试命令，避免频繁登录耗时。

二、重点函数详解（参数 + 案例，测试核心必掌握）

Paramiko的核心是SSHClient（SSH 连接客户端）和 SFTPClient（文件传输客户端），以下5个重点函数覆盖了 90%的测试场景，必须吃透。

1.paramiko.SSHClient()：创建SSH客户端对象

功能：初始化一个SSH客户端实例，是后续所有SSH操作的 “入口”，需搭配其他方法实现连接、认证、命令执行。

关键关联方法（必用）

创建客户端后，需调用以下方法完成连接流程：

set_missing_host_key_policy(policy)：设置未知主机密钥策略（测试环境常用 “自动接受”，避免首次连接报错）；
connect(hostname, port, username, password, **kwargs)：建立SSH连接（核心参数见下文）。

参数说明（connect方法核心参数）

参数名	类型	是否必填	说明	示例
hostname	str	是	远程服务器 IP 或域名	“192.168.1.100”
port	int	否	SSH 端口，默认 22	22（默认可省略）
username	str	是	登录服务器的用户名	“test_user”
password	str	否	登录密码（与 pkey 二选一，推荐密钥登录更安全）	“test_123456”
pkey	paramiko.RSAKey	否	私钥对象（密钥登录时使用，替代密码）	见下文 “密钥登录案例”
timeout	float	否	连接超时时间（秒），避免无限等待	10.0

案例 1：密码登录服务器（快速测试场景）

import paramiko

def ssh_password_login():
    # 1. 创建 SSH 客户端对象
    ssh_client = paramiko.SSHClient()
    
    try:
        # 2. 设置未知主机密钥策略（测试环境用 AutoAddPolicy 自动接受）
        ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        
        # 3. 连接服务器（密码登录）
        ssh_client.connect(
            hostname="192.168.1.100",  # 服务器IP
            port=22,                   # 端口
            username="test_user",      # 用户名
            password="test_123456",    # 密码
            timeout=10                 # 超时时间10秒
        )
        print("✅ 服务器登录成功！")
        
    except Exception as e:
        print(f"❌ 登录失败：{str(e)}")
    finally:
        # 4. 关闭连接（无论成功失败都要关闭，避免资源泄露）
        if ssh_client.get_transport() and ssh_client.get_transport().is_active():
            ssh_client.close()
            print(" 连接已关闭")

# 执行登录
ssh_password_login()

案例 2：密钥登录服务器（生产 / 正式测试环境，更安全）

密码登录存在泄露风险，实际测试环境更推荐RSA密钥登录，需先在本地生成密钥对（id_rsa私钥、id_rsa.pub公钥），并将公钥上传到服务器 ~/.ssh/authorized_keys中。

import paramiko
from paramiko.rsakey import RSAKey

def ssh_key_login():
    ssh_client = paramiko.SSHClient()
    
    try:
        ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        
        # 加载本地私钥（私钥路径一般为 ~/.ssh/id_rsa，Windows 为 C:Users用户名.sshid_rsa）
        private_key = RSAKey.from_private_key_file(
            filename="~/.ssh/id_rsa",  # 私钥文件路径
            password="key_password"    # 私钥密码（若生成时设置了，无则省略）
        )
        
        # 密钥登录（用 pkey 参数替代 password）
        ssh_client.connect(
            hostname="192.168.1.100",
            username="test_user",
            pkey=private_key,  # 私钥对象
            timeout=10
        )
        print("✅ 密钥登录服务器成功！")
        
    except Exception as e:
        print(f"❌ 登录失败：{str(e)}")
    finally:
        if ssh_client.get_transport() and ssh_client.get_transport().is_active():
            ssh_client.close()
            print(" 连接已关闭")

# 执行密钥登录
ssh_key_login()

2.SSHClient.exec_command(command, **kwargs)：远程执行命令

功能：在已连接的服务器上执行 Shell 命令（如 ls、ps、启动服务等），返回命令的标准输入、标准输出、标准错误流，用于获取执行结果和错误信息。

参数说明

参数名	类型	是否必填	说明	示例
command	str	是	要执行的 Shell 命令	`”ps -ef
timeout	float	否	命令执行超时时间（秒），避免命令卡住	30.0
get_pty	bool	否	是否请求伪终端（执行需要交互的命令时设为 True，如 top）	False（默认，非交互命令）

返回值说明

执行后返回一个元组 (stdin, stdout, stderr)：

stdin：命令的标准输入流（需输入内容时用，如密码确认）；
stdout：命令的标准输出流（命令执行成功的结果）；
stderr：命令的标准错误流（命令执行失败的错误信息）。

案例：自动化查看服务器进程 + 日志（测试常用场景）

import paramiko

def ssh_execute_command():
    ssh_client = paramiko.SSHClient()
    
    try:
        ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh_client.connect(
            hostname="192.168.1.100",
            username="test_user",
            password="test_123456",
            timeout=10
        )
        
        # 场景1：查看被测应用（假设名为 test_app）的进程状态
        print("=== 查看 test_app 进程 ===")
        command1 = "ps -ef | grep test_app | grep -v grep"  # 过滤掉 grep 自身进程
        stdin1, stdout1, stderr1 = ssh_client.exec_command(command1, timeout=10)
        
        # 读取输出（stdout 是字节流，需解码为字符串）
        process_result = stdout1.read().decode("utf-8").strip()
        error1 = stderr1.read().decode("utf-8").strip()
        
        if process_result:
            print(f"进程存在：
{process_result}")
        else:
            print(f"进程不存在，错误信息：{error1}")
        
        # 场景2：查看被测应用的最新10行日志（日志路径假设为 /var/log/test_app.log）
        print("
=== 查看 test_app 最新10行日志 ===")
        command2 = "tail -n 10 /var/log/test_app.log"
        stdin2, stdout2, stderr2 = ssh_client.exec_command(command2, timeout=15)
        
        log_result = stdout2.read().decode("utf-8").strip()
        error2 = stderr2.read().decode("utf-8").strip()
        
        if log_result:
            print(f"日志内容：
{log_result}")
        else:
            print(f"读取日志失败，错误信息：{error2}")
        
    except Exception as e:
        print(f"❌ 执行命令失败：{str(e)}")
    finally:
        if ssh_client.get_transport() and ssh_client.get_transport().is_active():
            ssh_client.close()
            print("
 连接已关闭")

# 执行命令
ssh_execute_command()

3.SSHClient.open_sftp()：创建SFTP客户端

功能：在SSH连接基础上，创建SFTP（SSH File Transfer Protocol）客户端对象，用于实现本地与服务器之间的文件上传（put）、下载（get）、删除（remove）等操作 —— 这是测试自动化中传输测试文件的核心方法。

返回值

返回paramiko.SFTPClient 对象，该对象提供了文件传输的所有核心方法（put、get 等）。

案例：创建SFTP客户端（后续文件操作的前置步骤）

import paramiko

def create_sftp_client():
    ssh_client = paramiko.SSHClient()
    sftp_client = None  # 初始化 SFTP 客户端
    
    try:
        # 1. 先建立 SSH 连接
        ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh_client.connect(
            hostname="192.168.1.100",
            username="test_user",
            password="test_123456",
            timeout=10
        )
        
        # 2. 创建 SFTP 客户端（基于已有的 SSH 连接，无需重新认证）
        sftp_client = ssh_client.open_sftp()
        print("✅ SFTP 客户端创建成功！")
        
        # 后续可通过 sftp_client 执行上传、下载等操作
        # ...
        
    except Exception as e:
        print(f"❌ 操作失败：{str(e)}")
    finally:
        # 先关闭 SFTP 客户端，再关闭 SSH 连接
        if sftp_client:
            sftp_client.close()
            print(" SFTP 客户端已关闭")
        if ssh_client.get_transport() and ssh_client.get_transport().is_active():
            ssh_client.close()
            print(" SSH 连接已关闭")

# 创建 SFTP 客户端
create_sftp_client()

4.SFTPClient.put(localpath, remotepath, **kwargs)：上传文件到服务器

功能：将本地文件上传到远程服务器指定路径，测试中常用于上传测试脚本、测试数据（如 CSV 用例文件）、配置文件等。

参数说明

参数名	类型	是否必填	说明	示例
localpath	str	是	本地文件的绝对路径或相对路径	“ ./test_cases/test_data.csv”（本地测试数据）
remotepath	str	是	服务器上的目标路径（需包含文件名）	“ /opt/test_data/test_data.csv”（服务器路径）
confirm	bool	否	上传后是否验证文件大小（确保上传完整），默认 True	True（推荐开启）
callback	callable	否	上传进度回调函数（可选，用于显示上传进度）	见下文 “进度显示案例”

案例 1：上传测试数据文件到服务器

import paramiko

def sftp_upload_file():
    ssh_client = paramiko.SSHClient()
    sftp_client = None
    
    try:
        # 1. 建立 SSH 连接
        ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh_client.connect(
            hostname="192.168.1.100",
            username="test_user",
            password="test_123456",
            timeout=10
        )
        
        # 2. 创建 SFTP 客户端
        sftp_client = ssh_client.open_sftp()
        
        # 3. 上传本地测试数据文件
        local_file = "./test_cases/test_data.csv"  # 本地文件路径
        remote_file = "/opt/test_data/test_data.csv"  # 服务器目标路径
        
        # 执行上传
        sftp_client.put(localpath=local_file, remotepath=remote_file, confirm=True)
        print(f"✅ 文件上传成功！
本地路径：{local_file}
服务器路径：{remote_file}")
        
        # 可选：验证上传后的文件是否存在
        try:
            sftp_client.stat(remote_file)  # 获取服务器文件信息
            print("✅ 验证：服务器文件存在")
        except FileNotFoundError:
            print("❌ 验证：服务器文件不存在，上传失败")
        
    except Exception as e:
        print(f"❌ 上传失败：{str(e)}")
    finally:
        if sftp_client:
            sftp_client.close()
        if ssh_client.get_transport() and ssh_client.get_transport().is_active():
            ssh_client.close()
            print(" 连接已关闭")

# 执行上传
sftp_upload_file()

案例 2：上传进度显示（优化体验）

大文件上传时，可通过callback参数显示进度，便于观察上传状态：

import paramiko
import os

def upload_progress(transferred, total):
    """上传进度回调函数：显示已上传百分比"""
    progress = (transferred / total) * 100
    print(f"
上传进度：{progress:.1f}%（{transferred}/{total} 字节）", end="")

def sftp_upload_large_file():
    ssh_client = paramiko.SSHClient()
    sftp_client = None
    
    try:
        ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh_client.connect(
            hostname="192.168.1.100",
            username="test_user",
            password="test_123456",
            timeout=10
        )
        
        sftp_client = ssh_client.open_sftp()
        
        # 本地大文件（如测试用的安装包）
        local_file = "./test_packages/test_app_v1.0.tar.gz"
        remote_file = "/opt/packages/test_app_v1.0.tar.gz"
        
        # 获取本地文件大小（用于进度计算）
        local_file_size = os.path.getsize(local_file)
        print(f"开始上传大文件（大小：{local_file_size/1024/1024:.2f} MB）")
        
        # 带进度上传
        sftp_client.put(
            localpath=local_file,
            remotepath=remote_file,
            confirm=True,
            callback=lambda t, s: upload_progress(t, local_file_size)  # 进度回调
        )
        print("
✅ 大文件上传成功！")
        
    except Exception as e:
        print(f"
❌ 上传失败：{str(e)}")
    finally:
        if sftp_client:
            sftp_client.close()
        if ssh_client.get_transport() and ssh_client.get_transport().is_active():
            ssh_client.close()

# 执行大文件上传
sftp_upload_large_file()

5.SFTPClient.get(remotepath, localpath, **kwargs)：从服务器下载文件

功能：将服务器上的文件下载到本地指定路径，测试中常用于下载测试日志、执行结果报告、错误截图等，方便本地分析。

参数说明

参数名	类型	是否必填	说明	示例
remotepath	str	是	服务器上文件的路径（需包含文件名）	“ /var/log/test_app_error.log”（服务器日志）
localpath	str	是	本地目标路径（需包含文件名）	“ ./test_logs/test_app_error.log”（本地保存路径）
confirm	bool	否	下载后是否验证文件大小，默认 True	True（推荐开启）
callback	callable	否	下载进度回调函数（可选）	类似上传进度函数

案例：下载服务器测试日志到本地

import paramiko
import os
import datetime

def download_progress(transferred, total):
    """下载进度回调函数"""
    progress = (transferred / total) * 100
    print(f"
下载进度：{progress:.1f}%（{transferred}/{total} 字节）", end="")

def sftp_download_file():
    ssh_client = paramiko.SSHClient()
    sftp_client = None
    
    try:
        ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh_client.connect(
            hostname="192.168.1.100",
            username="test_user",
            password="test_123456",
            timeout=10
        )
        
        sftp_client = ssh_client.open_sftp()
        
        # 服务器日志路径 + 本地保存路径
        remote_log = "/var/log/test_app/test_app_run.log"  # 服务器日志
        local_log_dir = "./test_logs"  # 本地日志目录（不存在则创建）
        local_log = f"{local_log_dir}/test_app_run_{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}.log"  # 带时间戳避免重复
        
        # 创建本地目录（若不存在）
        if not os.path.exists(local_log_dir):
            os.makedirs(local_log_dir)
        
        # 获取服务器文件大小
        remote_file_attr = sftp_client.stat(remote_log)
        remote_file_size = remote_file_attr.st_size
        print(f"开始下载日志文件（大小：{remote_file_size/1024:.2f} KB）")
        
        # 带进度下载
        sftp_client.get(
            remotepath=remote_log,
            localpath=local_log,
            confirm=True,
            callback=lambda t, s: download_progress(t, remote_file_size)
        )
        print(f"
✅ 日志下载成功！本地路径：{local_log}")
        
    except Exception as e:
        print(f"
❌ 下载失败：{str(e)}")
    finally:
        if sftp_client:
            sftp_client.close()
        if ssh_client.get_transport() and ssh_client.get_transport().is_active():
            ssh_client.close()

# 执行下载
sftp_download_file()

三、常用补充函数（覆盖高频测试场景）

除了上述重点函数，Paramiko 还有一批实用工具函数，能高效解决测试中的边缘场景。

1.SFTPClient.listdir(remotepath)：列出服务器目录下的文件

功能：获取服务器指定目录下的所有文件 / 文件夹名称，测试中常用于验证 “上传的文件是否在目标目录”“日志目录是否生成新文件”。

# 承接上文 SFTP 客户端案例，在 sftp_client 创建后添加：
remote_dir = "/opt/test_data"
file_list = sftp_client.listdir(remote_dir)
print(f"服务器 {remote_dir} 目录下的文件：")
for file in file_list:
    print(f"- {file}")

2.SFTPClient.stat(remotepath)：获取服务器文件属性

功能：获取文件的大小、创建时间、修改时间等属性，用于验证文件完整性（如上传后大小是否匹配）、判断文件是否更新。

# 承接上文，获取文件属性：
remote_file = "/opt/test_data/test_data.csv"
file_attr = sftp_client.stat(remote_file)
print(f"文件大小：{file_attr.st_size} 字节")
print(f"最后修改时间：{datetime.datetime.fromtimestamp(file_attr.st_mtime)}")

3.SFTPClient.remove(remotepath)：删除服务器文件

功能：删除服务器上的指定文件，测试中常用于 “清理测试残留文件”（如每次执行测试前删除旧日志、旧数据）。

# 承接上文，删除旧测试数据：
old_file = "/opt/test_data/old_test_data.csv"
try:
    sftp_client.remove(old_file)
    print(f"✅ 成功删除旧文件：{old_file}")
except FileNotFoundError:
    print(f"⚠️  旧文件不存在，无需删除：{old_file}")

4.SSHClient.exec_command()执行交互命令（进阶）

部分命令需要交互（如sudo执行需要输入密码），可通过stdin流输入内容，测试中常用于 “需要权限的操作”（如重启服务、安装依赖）。

import paramiko

def ssh_execute_interactive_command():
    ssh_client = paramiko.SSHClient()
    
    try:
        ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh_client.connect(
            hostname="192.168.1.100",
            username="test_user",
            password="test_123456",
            timeout=10
        )
        
        # 执行 sudo 命令（重启被测应用，需要输入密码）
        command = "sudo systemctl restart test_app"
        stdin, stdout, stderr = ssh_client.exec_command(
            command,
            get_pty=True,  # 交互命令必须开启伪终端
            timeout=20
        )
        
        # 向 stdin 输入 sudo 密码（注意末尾加换行符 
，模拟回车）
        sudo_password = "test_123456"
        stdin.write(f"{sudo_password}
")
        stdin.flush()  # 确保输入被发送
        
        # 读取结果
        result = stdout.read().decode("utf-8").strip()
        error = stderr.read().decode("utf-8").strip()
        
        if not error:
            print(f"✅ 执行 sudo 命令成功：{command}")
            print(f"结果：{result}")
        else:
            print(f"❌ 执行 sudo 命令失败：{error}")
        
    except Exception as e:
        print(f"❌ 操作失败：{str(e)}")
    finally:
        if ssh_client.get_transport() and ssh_client.get_transport().is_active():
            ssh_client.close()

# 执行交互命令
ssh_execute_interactive_command()

四、实战案例：完整的测试环境自动化部署流程

结合上述函数，我们搭建一个 “测试环境自动化部署” 的完整脚本，覆盖 “上传安装包→执行安装命令→启动服务→验证服务状态→下载日志” 全流程，这是测试自动化中超级典型的场景。

测试场景需求

本地准备被测应用安装包 test_app_v1.0.tar.gz；
上传安装包到服务器 /opt/packages 目录；
解压安装包到 /opt/test_app 目录；
启动应用服务（sudo systemctl start test_app）；
验证服务是否启动成功（查看进程）；
下载应用启动日志到本地 /test_logs 目录。

完整代码

import paramiko
import os
import datetime

def upload_progress(transferred, total):
    progress = (transferred / total) * 100
    print(f"
上传进度：{progress:.1f}%", end="")

def deploy_test_env():
    # 配置信息（可单独放在配置文件中，避免硬编码）
    config = {
        "hostname": "192.168.1.100",
        "username": "test_user",
        "password": "test_123456",
        "sudo_password": "test_123456",
        "local_package": "./test_packages/test_app_v1.0.tar.gz",
        "remote_package_dir": "/opt/packages",
        "remote_install_dir": "/opt/test_app",
        "remote_log_path": "/var/log/test_app/start.log",
        "local_log_dir": "./test_logs"
    }
    
    ssh_client = paramiko.SSHClient()
    sftp_client = None
    
    try:
        # 1. 初始化 SSH 连接
        print("=== 1. 建立服务器连接 ===")
        ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh_client.connect(
            hostname=config["hostname"],
            username=config["username"],
            password=config["password"],
            timeout=15
        )
        print("✅ 服务器连接成功")
        
        # 2. 创建 SFTP 客户端
        sftp_client = ssh_client.open_sftp()
        
        # 3. 上传安装包
        print("
=== 2. 上传应用安装包 ===")
        # 确保服务器包目录存在
        try:
            sftp_client.stat(config["remote_package_dir"])
        except FileNotFoundError:
            ssh_client.exec_command(f"mkdir -p {config['remote_package_dir']}")
            print(f"⚠️  服务器包目录不存在，已创建：{config['remote_package_dir']}")
        
        # 上传文件
        remote_package_path = f"{config['remote_package_dir']}/{os.path.basename(config['local_package'])}"
        local_package_size = os.path.getsize(config["local_package"])
        print(f"开始上传：{os.path.basename(config['local_package'])}（{local_package_size/1024/1024:.2f} MB）")
        sftp_client.put(
            localpath=config["local_package"],
            remotepath=remote_package_path,
            confirm=True,
            callback=lambda t, s: upload_progress(t, local_package_size)
        )
        print(f"
✅ 安装包上传成功，服务器路径：{remote_package_path}")
        
        # 4. 解压安装包
        print("
=== 3. 解压安装包 ===")
        # 先删除旧安装目录（避免残留）
        ssh_client.exec_command(f"rm -rf {config['remote_install_dir']}")
        # 创建新安装目录并解压
        unzip_command = f"mkdir -p {config['remote_install_dir']} && tar -zxvf {remote_package_path} -C {config['remote_install_dir']}"
        stdin, stdout, stderr = ssh_client.exec_command(unzip_command, timeout=30)
        error = stderr.read().decode("utf-8").strip()
        if not error:
            print("✅ 安装包解压成功")
        else:
            print(f"❌ 解压失败：{error}")
            return
        
        # 5. 启动应用服务（sudo 命令）
        print("
=== 4. 启动应用服务 ===")
        start_command = f"sudo systemctl start test_app"
        stdin, stdout, stderr = ssh_client.exec_command(start_command, get_pty=True, timeout=20)
        stdin.write(f"{config['sudo_password']}
")
        stdin.flush()
        error = stderr.read().decode("utf-8").strip()
        if not error:
            print("✅ 应用服务启动成功")
        else:
            print(f"❌ 启动失败：{error}")
            return
        
        # 6. 验证服务状态
        print("
=== 5. 验证服务状态 ===")
        check_command = "ps -ef | grep test_app | grep -v grep"
        stdin, stdout, stderr = ssh_client.exec_command(check_command, timeout=10)
        process = stdout.read().decode("utf-8").strip()
        if process:
            print(f"✅ 服务运行正常，进程信息：
{process}")
        else:
            print("❌ 服务未启动，进程不存在")
            return
        
        # 7. 下载启动日志
        print("
=== 6. 下载启动日志 ===")
        # 确保本地日志目录存在
        if not os.path.exists(config["local_log_dir"]):
            os.makedirs(config["local_log_dir"])
        # 本地日志路径（带时间戳）
        local_log_path = f"{config['local_log_dir']}/start_log_{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}.log"
        # 下载日志
        remote_log_attr = sftp_client.stat(config["remote_log_path"])
        remote_log_size = remote_log_attr.st_size
        print(f"开始下载日志（大小：{remote_log_size/1024:.2f} KB）")
        sftp_client.get(
            remotepath=config["remote_log_path"],
            localpath=local_log_path,
            confirm=True
        )
        print(f"✅ 日志下载成功，本地路径：{local_log_path}")
        
        print("
 测试环境部署完成，所有步骤执行成功！")
        
    except Exception as e:
        print(f"
❌ 部署流程失败：{str(e)}")
    finally:
        # 清理资源
        if sftp_client:
            sftp_client.close()
            print("
 SFTP 客户端已关闭")
        if ssh_client.get_transport() and ssh_client.get_transport().is_active():
            ssh_client.close()
            print(" SSH 连接已关闭")

# 执行部署
if __name__ == "__main__":
    deploy_test_env()

代码亮点

配置聚焦管理：将服务器信息、路径等配置单独存放，便于修改和维护；
异常处理完善：每个步骤都有错误捕获，失败时及时返回并提示缘由；
自动化校验：上传后验证目录、启动后验证进程、下载后保存日志，确保每步可靠；
用户体验优化：上传进度显示、步骤编号，清晰跟踪执行状态。

五、避坑指南：Paramiko 测试自动化常见问题

连接超时 / 拒绝连接：检查服务器 IP、端口是否正确（默认 22，若修改过需对应调整）；确认服务器 SSH 服务已启动（systemctl status sshd）；检查防火墙是否开放 SSH 端口（测试环境可临时关闭防火墙 systemctl stop firewalld）。
密钥登录失败：验证私钥路径是否正确（Windows 路径注意转义，如 C:Userstest.sshid_rsa）；检查私钥权限（Linux 下私钥文件权限需为 600，chmod 600 ~/.ssh/id_rsa）；确认公钥已添加到服务器 ~/.ssh/authorized_keys（可手动执行 ssh-copy-id 用户名@服务器IP 同步）。
执行命令无结果 / 报错 “Permission denied”：命令需要权限时，开启 get_pty=True 并通过 stdin 输入 sudo 密码；检查服务器路径是否存在（如 /opt/test_data 需先创建 mkdir -p）；避免执行长时间阻塞的命令（如 top），若需执行需设置合理 timeout。
文件上传 / 下载后大小不一致：开启 confirm=True（默认开启），让 Paramiko 自动验证文件大小；大文件传输时避免中断网络，可通过 callback 监控进度，中断后重新执行。
ssh连接使用完毕后记得通过close函数关闭，减少服务器端的连接量。