pytest-xdist分布式测试原理浅析

╰半夏微凉° 2024-03-23 19:47 56阅读 0赞

**目录**

pytest-xdist执行流程：

pytest-xdist 模块结构：

pytest-xdist分布式测试原理：

pytest-xdist源码浅读：

--------------------

### pytest-xdist执行流程： ###

1.  解析命令行参数：pytest-xdist 会解析命令行参数，获取用户指定的分发模式、进程数、主机列表等信息。
2.  加载测试用例：pytest-xdist 会加载所有的 pytest 测试用例，包括在当前目录和子目录下的所有测试文件和测试函数。
3.  分发测试用例：根据用户指定的分发模式，pytest-xdist 会将测试用例分发到多个进程或主机上执行。如果是分发到多个进程，pytest-xdist 会创建多个子进程，每个子进程都会执行一部分测试用例。如果是分发到多个主机，pytest-xdist 会在每个主机上启动一个 Python 进程，然后将测试用例分发给每个 Python 进程执行。
4.  执行测试用例：在每个进程或主机上，pytest-xdist 会执行分配给它的测试用例。每个进程或主机上的测试执行都是独立的，它们之间没有任何数据共享或通信。
5.  汇总测试结果：在所有进程或主机上的测试执行完成后，pytest-xdist 会将所有测试结果汇总到主进程中，并输出测试报告。在汇总测试结果的过程中，pytest-xdist 还会根据用户指定的选项合并相同的测试结果，比如合并多个进程或主机上的相同测试用例结果。
6.  清理资源：在所有测试结果都汇总完成后，pytest-xdist 会清理所有的资源，包括关闭分配给每个进程或主机的 Python 进程、删除临时文件等。

需要注意的是，pytest-xdist 的执行流程是一个异步的过程，不同进程或主机上的测试执行是并行的，它们之间没有任何阻塞或等待

####  ####

### pytest-xdist 模块结构： ###

1.  `xdist` 模块：这是 pytest-xdist 的主模块，包含了分发测试用例的主要逻辑。在 `xdist` 模块中，主要包含了以下几个子模块：
    
     *  `looponfail`: 在测试用例执行失败时自动重试的逻辑。
     *  `loadfile`: 加载分布式测试配置文件的逻辑。
     *  `rsync`: 在多个主机之间同步文件的逻辑。
     *  `newhooks`: 扩展 pytest 的钩子函数以支持分布式测试的逻辑。
2.  `testing` 模块：这是 pytest-xdist 的测试模块，包含了对 pytest-xdist 的单元测试和集成测试。
3.  `docs` 模块：这是 pytest-xdist 的文档模块，包含了 pytest-xdist 的文档说明和示例代码。

### pytest-xdist分布式测试原理： ###

pytest-xdist 的核心原理是使用 py.execnet 这个 Python 库，它是一个用于远程执行 Python 代码的库。pytest-xdist 利用 py.execnet 提供的功能，将测试用例分发到多个进程或主机上执行，然后将结果汇总返回给主进程。

具体来说，pytest-xdist 在执行 pytest 测试用例时，会根据用户指定的分发模式（如 --numprocesses 或者 --tx），将测试用例分发到多个进程或者多个主机上。对于分发到多个进程的情况，pytest-xdist 会创建多个子进程，每个子进程都会执行一部分测试用例。对于分发到多个主机的情况，pytest-xdist 利用 py.execnet 在每个主机上启动一个 Python 进程，然后将测试用例分发给每个 Python 进程执行。

在测试用例执行完成后，pytest-xdist 会将所有测试结果汇总到主进程中，并输出测试报告。此外，pytest-xdist 还提供了一些功能，如在多个进程或主机之间共享数据、控制测试用例的执行顺序等。

总的来说，pytest-xdist 利用 py.execnet 提供的远程执行 Python 代码的功能，将 pytest 测试用例分发到多个进程或主机上执行，从而实现了测试用例的并行执行，提高了测试效率。

### pytest-xdist源码浅读： ###

**解析命令行参数**

pytest-xdist 首先会解析命令行参数，从而获取用户指定的分发模式、进程数、主机列表等信息。这个过程是通过 `pytest_addoption` 钩子函数来实现的，它会在 pytest 启动时被调用，从而向 pytest 注册新的命令行选项。这里是相关的源码：

def pytest_addoption(parser):
        group = parser.getgroup("xdist", "distributed and subprocess testing")
        group._addoption(
            "-n",
            "--numprocesses",
            dest="numprocesses",
            type=int,
            default=None,
            help="shortcut for '--dist=load --tx=NUM*popen//python=python%s' (default: %default)" % sys.version_info[0],
        )
        group._addoption(
            "--tx",
            dest="tx",
            metavar="xspec",
            help="addrs[:spec] of test exec environments to use, "
            "see \"xdist help spec\".  (type \"xdist help spec\" for details)",
        )
        # ...

**加载测试用例**

pytest-xdist 加载 pytest 测试用例的过程和普通的 pytest 测试用例加载过程相同，它会递归地查找当前目录及其子目录下的所有 `test_*.py` 文件和 `*_test.py` 文件，以及所有以 `test_` 或者 `test` 开头的测试函数。这个过程是通过 `pytest_collection_modifyitems` 钩子函数来实现的，它会在 pytest 执行测试用例前被调用，从而修改 pytest 收集到的测试用例列表。这里是相关的源码：

def pytest_collection_modifyitems(items):
        config = items[0].session.config
        if config.option.numprocesses and not config.option.dist:
            config.option.dist = "load"
        # ...

**分发测试用例**

pytest-xdist 根据用户指定的分发模式，将测试用例分发到多个进程或主机上。这个过程是通过 `pytest_runtestloop` 函数来实现的，它是 pytest 执行测试用例的入口函数。在 `pytest_runtestloop` 函数中，pytest-xdist 根据用户指定的分发模式，创建多个子进程或者多个主机，然后将测试用例分发给每个子进程或主机执行。这里是相关的源码：

def pytest_runtestloop(session):
        # ...
        if session.config.option.numprocesses:
            from .dist import load
            return load(session)
        elif session.config.option.dist == "load":
            from .dist import load
            return load(session)
        elif session.config.option.dist == "each":
            from .dist import each
            return each(session)
        # ...

其中，`load` 函数是用于分发测试用例到多个进程的，`each` 函数是用于分发测试用例到多个主机的。这里是 `load` 函数的相关源码：

def load(session):
        numprocesses = session.config.option.numprocesses
        # ...
        if numprocesses <= 0:
            raise ValueError("number of processes must be greater than 0")
        config = session.config
        if config.option.capture == "no":
            config.option.capture = "fd"
        # ...
        else:
            from . import (  # noqa: F401
                box,
                worker,
            )
            with box.Box(config, numprocesses=numprocesses) as box:
                box.makegateways()
                gwlist = box.gwlist()
                result = box.invoke_gateways(gwlist, "start_worker", numprocesses=numprocesses, **box._kwds)
        # ...

在 `load` 函数中，首先会获取用户指定的进程数，然后根据进程数创建一个 `Box` 对象。`Box` 对象是 pytest-xdist 中的一个重要概念，它表示一个运行环境，用于管理多个子进程或主机。在 `Box` 对象创建完成后，pytest-xdist 会调用 `Box` 对象的 `makegateways` 方法，用于创建与子进程或主机的通信通道。然后，pytest-xdist 会调用 `Box` 对象的 `invoke_gateways` 方法，用于在所有子进程或主机上启动测试用例执行。在 `invoke_gateways` 方法中，pytest-xdist 会将要执行的测试用例发送给每个子进程或主机，然后等待所有子进程或主机执行完成。

**执行测试用例**

在每个子进程或主机上，pytest-xdist 会执行分配给它的测试用例。具体来说，pytest-xdist 会在每个子进程或主机上启动一个 Python 进程，然后在该进程中执行测试用例。这个过程是通过 `pytest_runtest_protocol` 函数来实现的，它是 pytest 执行单个测试用例的函数。在 `pytest_runtest_protocol` 函数中，pytest-xdist 会将要执行的测试用例通过 `execnet` 模块发送给子进程或主机，然后等待执行结果。这里是相关的源码：

def pytest_runtest_protocol(item, nextitem):
        # ...
        if config.option.numprocesses:
            from .dist import worker
            return worker.worker_runtest(item=item, nextitem=nextitem)
        elif config.option.dist == "load":
            from .dist import worker
            return worker.worker_runtest(item=item, nextitem=nextitem)
        elif config.option.dist == "each":
            return runtestprotocol(item, nextitem=nextitem)
        # ...

在分发测试用例到多个主机时，pytest-xdist 会将测试用例通过 SSH 协议发送到每个主机，然后在每个主机上启动一个 Python 进程，并在该进程中执行测试用例。这个过程是通过 `ssh_run` 函数来实现的，它是 pytest-xdist 中的一个辅助函数，用于执行远程命令。这里是相关的源码：

def ssh_run(host, command, capture=True):
        # ...
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(hostname=host, username=username, password=password, port=port)
        with ssh:
            stdin, stdout, stderr = ssh.exec_command(command)
            if capture:
                out = stdout.read().decode()
                err = stderr.read().decode()
                return out, err
            else:
                return None, None

在 `ssh_run` 函数中，pytest-xdist 使用 Paramiko 模块建立 SSH 连接，然后通过 SSH 协议发送测试用例和执行命令。这个过程中，所有的输入输出都通过 SSH 协议进行传输。

**汇总测试结果**

在所有子进程或主机上的测试执行完成后，pytest-xdist 会将所有测试结果汇总到主进程中，并输出测试报告。这个过程是通过 `pytest_terminal_summary` 钩子函数来实现的，它会在 pytest 执行完成后被调用，从而输出测试报告。这里是相关的源码：

def pytest_terminal_summary(terminalreporter):
        session = terminalreporter.config.session
        if session.testsfailed and session.config.option.looponfail:
            terminalreporter.write("re-running failed tests...\n")
            return pytest_runtestloop(session)
        # ...

在 `pytest_terminal_summary` 函数中，pytest-xdist 会检查测试结果，然后输出测试报告。如果测试用例执行失败，并且用户指定了 `--looponfail` 参数，pytest-xdist 会自动重试执行测试用例。

**清理资源**

在所有测试结果都汇总完成后，pytest-xdist 会清理所有的资源，包括关闭分配给每个子进程或主机的 Python 进程、删除临时文件等。这个过程是通过 `pytest_unconfigure` 钩子函数来实现的，它会在 pytest 执行完成后被调用，从而清理 pytest-xdist 使用的所有资源。这里是相关的源码：

def pytest_unconfigure(config):
        # ...
        if hasattr(config, "_xdist_worker_collection"):

pytest-xdist 参数浅解：

> 1.  `-n`: 指定分发模式，可以是一个数字，表示分发到多少个进程；也可以是一个字符串，表示分发到多少个主机（如 `-n 4` 表示分发到 4 个进程，`-n 4 --hosts=host1,host2,host3,host4` 表示分发到 4 个主机）。
> 2.  `--numprocesses`: 指定分发到多少个进程执行测试用例。
> 3.  `--tx`: 指定分发到多少个主机执行测试用例，格式为 `popen//ssh:user@host:port`。
> 4.  `--max-worker-restart`: 指定在某个子进程或主机上测试用例执行失败时的最大重试次数。
> 5.  `--rsyncdir`: 指定用于同步文件的目录，该目录下的所有文件会被同步到所有子进程或主机上。
> 6.  `--rsyncignore`: 指定需要忽略同步的文件或目录的规则。
> 7.  `--boxed`: 指定在子进程或主机中使用进程隔离（process isolation）模式执行测试用例。
> 8.  `--capture`: 指定在子进程或主机中使用的输出捕获模式，可以是 `fd`、`sys` 或者 `no`。
> 9.  `--ignore`: 指定需要忽略的测试文件或目录。
> 10. `--looponfail`: 指定在测试用例执行失败时自动重试的次数。

--------------------

**以下是我收集到的比较好的学习教程资源，虽然不是什么很值钱的东西，如果你刚好需要，可以评论区，留言【777】直接拿走就好了**

![c1653dd19f35495583cef3678b57b6b0.png][]

![775632765b6744fbbecbd8fdd097d1ab.png][]

**各位想获取资料的朋友请****点赞 + 评论 + 收藏****，三连！**

**三连**之后我会在评论区挨个私信发给你们~

[c1653dd19f35495583cef3678b57b6b0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/fb5935fe0d0a4c7697c49deaab04397e.png
[775632765b6744fbbecbd8fdd097d1ab.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/cc241aece15e4ece8740ca84797d369e.png