如何实现Pytest多主机并行、单主机内测试串行执行

实现逻辑对齐需求 #

• 会话初始化阶段统一创建所有目标主机SSH连接，存入全局连接池
• 用例收集完成后按归属主机分组，每组分配独立工作线程
• 单线程内严格按用例收集顺序串行执行归属当前主机的用例，避免同主机多连接并发操作的冲突
• 会话销毁阶段统一关闭连接池内所有SSH连接，无资源泄漏
• 修复自定义插件日志线程名显示异常、执行流程不符合pytest原生逻辑的问题

完整实现代码 #

公共模块 shared.py #

负责SSH连接池管理、线程上下文绑定：

import threading
import paramiko
from typing import Dict

SSH_POOL: Dict[str, paramiko.SSHClient] = {}
# 线程本地存储，记录当前线程绑定的主机信息
THREAD_CTX = threading.local()

def init_ssh_pool(host_configs: list) -> None:
    for cfg in host_configs:
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        client.connect(
            hostname=cfg["ip"],
            port=cfg.get("port", 22),
            username=cfg["user"],
            password=cfg["pwd"],
            timeout=10
        )
        SSH_POOL[cfg["host_tag"]] = client

def destroy_ssh_pool() -> None:
    for client in SSH_POOL.values():
        try:
            client.close()
        except Exception:
            pass
    SSH_POOL.clear()

def get_current_ssh() -> paramiko.SSHClient:
    host_tag = getattr(THREAD_CTX, "bind_host", None)
    if not host_tag:
        raise RuntimeError("当前线程未绑定目标主机")
    return SSH_POOL[host_tag]

pytest配置文件 conftest.py #

负责钩子注册、用例分组、日志修复：

import pytest
import threading
from _pytest.nodes import Item
from shared import init_ssh_pool, destroy_ssh_pool, THREAD_CTX

# 目标主机配置，可根据实际场景改成从yaml/命令行参数读取
HOST_CONFIGS = [
    {"host_tag": "node1", "ip": "192.168.1.11", "user": "root", "pwd": "test123"},
    {"host_tag": "node2", "ip": "192.168.1.12", "user": "root", "pwd": "test123"},
]

def pytest_configure(config):
    config.addinivalue_line(
        "markers", "target_host(tag): 标记用例归属的主机标签"
    )
    # 会话初始化建连
    init_ssh_pool(HOST_CONFIGS)

def pytest_unconfigure(config):
    # 会话结束销毁连接
    destroy_ssh_pool()

def pytest_collection_modifyitems(config, items: list[Item]):
    # 按主机标签对用例分组
    case_group = {h["host_tag"]: [] for h in HOST_CONFIGS}
    for item in items:
        mark = item.get_closest_marker("target_host")
        if not mark:
            case_group.setdefault("common", []).append(item)
            continue
        tag = mark.args[0]
        if tag not in case_group:
            raise ValueError(f"用例{item.nodeid}配置的主机标签{tag}不存在")
        case_group[tag].append(item)
    config.case_group = case_group

# 修复日志线程名、主机信息显示问题
class LogContextFilter:
    def filter(self, record):
        record.threadName = threading.current_thread().name
        record.bind_host = getattr(THREAD_CTX, "bind_host", "main")
        return True

@pytest.hookimpl
def pytest_configure_logging(config, log_plugin):
    for handler in log_plugin.handlers:
        handler.addFilter(LogContextFilter())

自定义调度器 host_parallel_runner.py #

负责接管pytest执行流程，按主机分线程执行：

import threading
from typing import Dict, List
from _pytest.nodes import Item
from _pytest.runner import runtestprotocol
from shared import THREAD_CTX

class HostParallelRunner:
    def __init__(self, config, case_group: Dict[str, List[Item]]):
        self.config = config
        self.case_group = case_group
        self.threads = []

    def _run_single_host_cases(self, host_tag: str, cases: List[Item]):
        # 绑定当前线程的主机上下文，修改线程名
        THREAD_CTX.bind_host = host_tag
        threading.current_thread().name = f"worker-{host_tag}"
        # 按收集顺序串行执行当前主机下所有用例
        for item in cases:
            # 复用pytest原生执行逻辑，兼容fixture、断言、报告生成全流程
            runtestprotocol(item, nextitem=None, log=False)

    def run(self):
        # 启动所有主机工作线程
        for host_tag, cases in self.case_group.items():
            if host_tag == "common":
                continue
            t = threading.Thread(
                target=self._run_single_host_cases,
                args=(host_tag, cases),
                daemon=False
            )
            self.threads.append(t)
            t.start()
        # 等待所有工作线程执行完成
        for t in self.threads:
            t.join()
        # 主线程执行无主机标记的公共用例
        if "common" in self.case_group:
            self._run_single_host_cases("common", self.case_group["common"])

# 替换pytest默认执行循环
def pytest_runtestloop(session):
    if not hasattr(session.config, "case_group"):
        return False
    runner = HostParallelRunner(session.config, session.config.case_group)
    runner.run()
    return True

测试用例示例 test_cases.py #

import pytest
import logging
from shared import get_current_ssh

log = logging.getLogger(__name__)

@pytest.mark.target_host("node1")
def test_node1_cpu_load():
    ssh = get_current_ssh()
    _, stdout, _ = ssh.exec_command("uptime")
    res = stdout.read().decode()
    log.info(f"node1负载信息: {res.strip()}")
    assert "load average" in res

@pytest.mark.target_host("node1")
def test_node1_disk_usage():
    ssh = get_current_ssh()
    _, stdout, _ = ssh.exec_command("df -h /")
    res = stdout.read().decode()
    log.info(f"node1根分区使用情况: {res.strip()}")
    assert "%" in res

@pytest.mark.target_host("node2")
def test_node2_memory():
    ssh = get_current_ssh()
    _, stdout, _ = ssh.exec_command("free -h")
    res = stdout.read().decode()
    log.info(f"node2内存使用情况: {res.strip()}")
    assert "Mem" in res

关键问题修复说明 #

• 日志线程名始终显示MainThread：通过自定义日志过滤器，在日志输出时动态读取当前线程的真实名称，同时追加绑定的主机标签，日志格式配置为%(asctime)s [%(threadName)s] [%(bind_host)s] %(levelname)s %(message)s即可正常显示
• 执行逻辑不符合预期：不修改pytest内部用例执行逻辑，仅接管最外层的调度循环，单条用例执行完全复用原生runtestprotocol方法，所有fixture、参数化、跳过、失败逻辑和原生pytest行为完全一致
• 连接管理问题：SSH连接生命周期完全绑定pytest会话的初始化/销毁钩子，不会出现执行中途连接意外释放、结束后连接泄漏的问题

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