设备重启后如何在Teardown中调用Session级SSH连接Fixture重连？

方案1：给SSH客户端增加自动重连机制 #

修改你的SshClient类，添加连接状态检查和自动重连逻辑，执行命令时如果发现连接已断开，会自动重建：

class SshClient:
    def __init__(self, ip, port, user, passwd):
        self.ip = ip
        self.port = port
        self.user = user
        self.passwd = passwd
        self.ssh_instance = None

    def connect(self):
        # 原有的连接实现（示例用paramiko）
        import paramiko
        self.ssh_instance = paramiko.SSHClient()
        self.ssh_instance.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        self.ssh_instance.connect(
            self.ip,
            port=self.port,
            username=self.user,
            password=self.passwd
        )

    def _is_connection_active(self):
        try:
            transport = self.ssh_instance.get_transport()
            return transport is not None and transport.is_active()
        except Exception:
            return False

    def exec_cmd(self, cmd):
        # 执行命令前检查连接状态，失效则重连
        if not self._is_connection_active():
            self.connect()
        # 原有的命令执行逻辑
        stdin, stdout, stderr = self.ssh_instance.exec_command(cmd)
        return stdout.read().decode(), stderr.read().decode()

修改后，原Session级Fixture无需改动，Teardown阶段调用exec_cmd时会自动重建连接。

方案2：改用连接工厂Fixture #

将Session级Fixture改为返回一个工厂函数，每次调用时都会检查并返回有效连接：

修改conftest.py中的Fixture #

@pytest.fixture(scope="session")
def ssh_connection_factory(vars, logger):
    ssh_client = None

    def get_valid_connection():
        nonlocal ssh_client
        # 检查连接是否存在且有效，否则重新创建
        if ssh_client is None or not ssh_client._is_connection_active():
            ssh_client = SshClient(
                ip=vars["ip"],
                port=vars["port"],
                user=vars["username"],
                passwd=vars["pw"]
            )
            ssh_client.connect()
        return ssh_client

    yield get_valid_connection
    # 会话结束后关闭连接
    if ssh_client is not None:
        ssh_client.ssh_instance.close()

在测试脚本中使用工厂函数 #

@pytest.fixture(scope="class", autouse=True)
def setup_teardown(self, ssh_connection_factory):
    # Setup阶段获取连接
    ssh = ssh_connection_factory()
    ssh.exec_cmd(f"mv {file} {newfile}")
    self.restartDevice()
    
    yield
    
    # Teardown阶段重新获取有效连接
    ssh = ssh_connection_factory()
    ssh.exec_cmd(f"mv {newfile} {file}")
    self.restartDevice()

方案3：Teardown阶段手动重连（简单但不够健壮） #

如果不想修改SshClient类，可以在Teardown阶段手动调用connect方法重建连接：

@pytest.fixture(scope="class", autouse=True)
def setup_teardown(self, get_ssh_connection):
    self.ssh = get_ssh_connection
    self.ssh.exec_cmd(f"mv {file} {newfile}")
    self.restartDevice()
    
    yield
    
    # 手动重新建立连接
    self.ssh.connect()
    self.ssh.exec_cmd(f"mv {newfile} {file}")
    self.restartDevice()

注意：这个方案可能存在连接残留状态的问题，比如原连接的传输通道未完全清理，不如前两个方案可靠。

内容的提问来源于stack exchange，提问作者panchanan