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

news2024/11/24 15:57:29

目录

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】直接拿走就好了

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

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

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/849774.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

Vue3_对响应式对象解构赋值之后失去响应性——toRefs()

Vue3_对响应式对象解构赋值之后失去响应性——toRefs()

官网toRefs() :响应式 API&#xff1a;工具函数 | Vue.js toRefs 在调用时只会为源对象上可以枚举的属性创建 ref。如果要为可能还不存在的属性创建 ref&#xff0c;请改用 toRef。 setup(){const state reactive({name:"张三"age:14})const stateAsToRefs toRef…
阅读更多...
SpringBoot 热部署

SpringBoot 热部署

文章目录 前言一、spring-boot-devtools添加热部署框架支持settings 开启项目自动编译开启运行中热部署使用Debug启动 二、IDEA 自带 HowSwap 功能设置 Spring Boot 启动类等待项目启动完成点击热加载按钮存在的问题 三、JRebel 插件【推荐】安装插件使用插件 前言 在日常开发…
阅读更多...
linux学习——Redis基础

linux学习——Redis基础

目录 一、noSQL 类型 特点及应用场景 二、Redis 三、安装方式 编译安装 rpm安装 四、目录结构 /etc/redis.conf 五、Redis命令 六、本地登录和远程登录 本地登录 远程登录 七、数据库操作 帮助信息 库操作 数据操作 八、Redis持久化 一、RDB类型 二、AOF模式 一…
阅读更多...
小白到运维工程师自学之路 第六十九集 （构建Docker容器监控系统：Cadvisor +Prometheus+Grafana）

小白到运维工程师自学之路 第六十九集 （构建Docker容器监控系统：Cadvisor +Prometheus+Grafana）

一、概述 Prometheus产品简介 Prometheus是一个最初在SoundCloud上构建的开源系统监视和警报工具包。自2012年成立以来&#xff0c;很多公司和组织都采用了Prometheus&#xff0c;该项目拥有非常活跃的开发者和用户社区。 它现在是一个独立的开源项目&#xff0c;可以独立于任…
阅读更多...
由于找不到msvcp140.dll无法继续执行代码多种解决方法

由于找不到msvcp140.dll无法继续执行代码多种解决方法

msvcp140.dll的作用是提供C程序运行所需的一些基本函数和类库&#xff0c;包括字符串处理、数学计算、文件操作、内存管理等功能。它为C程序员提供了一些常用的工具和函数&#xff0c;使得他们可以更方便地开发和调试程序。 当你的计算机缺少msvcp140.dll文件时&#xff0c;可能…
阅读更多...
怎么把几秒的视频变成gif？视频转gif图片教程分享

怎么把几秒的视频变成gif？视频转gif图片教程分享

无论是一段有趣的宠物视频、搞笑的表演片段&#xff0c;还是喜欢的电影或电视剧片段&#xff0c;通过gif形式分享&#xff0c;能够更好地吸引用户的关注、传递情感&#xff0c;并在社交媒体上引发互动与共鸣&#xff0c;那么如何才能将一段几秒钟的视频转gif图片&#xff08;ht…
阅读更多...
VoxWeekly｜The Sandbox 生态周报｜20230731

VoxWeekly｜The Sandbox 生态周报｜20230731

欢迎来到由 The Sandbox 发布的《VoxWeekly》。我们会在每周发布&#xff0c;对上一周 The Sandbox 生态系统所发生的事情进行总结。 如果你喜欢我们内容&#xff0c;欢迎与朋友和家人分享。请订阅我们的 Medium 、关注我们的 Twitter&#xff0c;并加入 Discord 社区&#xf…
阅读更多...
Grafana技术文档--基本安装-docker安装并挂载数据卷-《十分钟搭建》

Grafana技术文档--基本安装-docker安装并挂载数据卷-《十分钟搭建》

阿丹&#xff1a; Prometheus技术文档--基本安装-docker安装并挂载数据卷-《十分钟搭建》_一单成的博客-CSDN博客 在正确安装了Prometheus之后开始使用并安装Grafana作为Prometheus的仪表盘。 一、拉取镜像 搜索可拉取版本 docker search Grafana拉取镜像 docker pull gra…
阅读更多...
活动发布报名平台型公众号开源版开发

活动发布报名平台型公众号开源版开发

活动发布报名平台型公众号开源版开发 后台管理、手机端自由发布活动&#xff01; 为个人、企业或主办方举办各类活动提供一个发布推广与活动报名平台&#xff0c;主办方可以在平台进行活动发布&#xff0c;用户可以免费注册并灵活使用该系统的发布、报名管理、核销等功能。 功能…
阅读更多...
Matlab绘图 图例legend 太长，怎么减小指示线的长度

Matlab绘图 图例legend 太长，怎么减小指示线的长度

来源 绘图时&#xff0c;稍微减小文字已经不能正常放下图例&#xff0c;想通过调整图例指示线段长度缩减整个图例长度。 方法一 参考matlab官方论坛 leg legend(Plot1,Plot2,...); leg.ItemTokenSize [x1,x2]; By default x130 and x218 so put larger or smaller number…
阅读更多...
AI Chat 设计模式：13. 代理模式

AI Chat 设计模式：13. 代理模式

本文是该系列的第十三篇&#xff0c;采用问答式的方式展开&#xff0c;和前面的文章有一些不同&#xff0c;我不再进行提问了&#xff0c;改为由 GPT 1 号提问&#xff0c;GPT 2 号作答&#xff0c;每一节的小标题是我从 GPT 1 号的提问中总结出来的。我现在是完完全全的旁观者…
阅读更多...
[保研/考研机试] 猫狗收容所 C++实现

[保研/考研机试] 猫狗收容所 C++实现

题目描述&#xff1a; 输入&#xff1a; 第一个是n&#xff0c;它代表操作序列的次数。接下来是n行&#xff0c;每行有两个值m和t&#xff0c;分别代表题目中操作的两个元素。 输出&#xff1a; 按顺序输出收养动物的序列&#xff0c;编号之间以空格间隔。 源代码&#xff…
阅读更多...
服务器之LNMP

服务器之LNMP

lnmp的构成 L&#xff1a;linux系统,操作系统。 N&#xff1a;nginx网站服务&#xff0c;前端,提供前端的静态页面服务。同时具有代理,转发的作用。 转发&#xff1a;主要是转发后端请求。转发到PHP。nginx没有处理动态资源的功能,他有可以支持转发动态请求的模块。 M&…
阅读更多...
Python socket详解，全网最全教程

Python socket详解，全网最全教程

前言 嗨喽&#xff0c;大家好呀~这里是爱看美女的茜茜呐 Socket编程 1.基本概念 1.什么是客户端/服务器架构&#xff1f;服务器就是一系列硬件或软件&#xff0c;为一个或多个客户端&#xff08;服务的用户&#xff09;提供所需的“服务”。 它存在唯一目的就是等待客户端的…
阅读更多...
14_基于Flink将pulsar数据写入到HBase

14_基于Flink将pulsar数据写入到HBase

3.7.基于Flink将数据写入到HBase 3.7.1.编写Flink完成数据写入到Hbase操作, 完成数据备份, 便于后续进行即席查询和离线分析 3.7.1.1.HBase基本介绍 hbase是基于Google发布bigTable论文产生一款软件, 是一款noSQL型数据, 不支持SQL. 不支持join的操作, 没有表关系, 不支持事…
阅读更多...
15_基于Flink将pulsar数据写入到ClickHouse

15_基于Flink将pulsar数据写入到ClickHouse

3.8.基于Flink将数据写入到ClickHouse 编写Flink完成数据写入到ClickHouse操作, 后续基于CK完成指标统计操作 3.8.1.ClickHouse基本介绍 ClickHouse 是俄罗斯的Yandex于2016年开源的列式存储数据库&#xff08;DBMS&#xff09;&#xff0c;使用C语言编写&#xff0c;主要用…
阅读更多...
vue手把手教学封装分页器

vue手把手教学封装分页器

1.vue中前台 <template><div><h6>"start":{{ pageStartEnd.start }},"当前页"&#xff1a;{{ pagenow }}"end":{{ pageStartEnd.end }}</h6><!-- 如果点击上一页按钮&#xff0c;当前页减去1&#xff0c;并且如果当…
阅读更多...
两个多选框(select)之间值的左右上下移动

两个多选框(select)之间值的左右上下移动

<!DOCTYPE html> <html> <head><meta charset"utf-8"><title>两个多选框(select)之间值的左右上下移动</title> </head> <script src"https://cdn.bootcss.com/jquery/3.3.1/jquery.js"></script>&…
阅读更多...
vue基础知识二：你对SPA单页面的理解，它的优缺点分别是什么？如何实现SPA应用呢

vue基础知识二：你对SPA单页面的理解，它的优缺点分别是什么？如何实现SPA应用呢

一、什么是SPA SPA&#xff08;single-page application&#xff09;&#xff0c;翻译过来就是单页应用SPA是一种网络应用程序或网站的模型&#xff0c;它通过动态重写当前页面来与用户交互&#xff0c;这种方法避免了页面之间切换打断用户体验在单页应用中&#xff0c;所有必…
阅读更多...
【AutoLayout案例1-按钮居中显示 Objective-C语言】

【AutoLayout案例1-按钮居中显示 Objective-C语言】

一、按钮居中显示 1.接下来，我们就用这个autoLayout，自动布局，给大家写一个，实现几个案例，给大家看一下 那么，首先，第一个，大家注意， 当我们使用autoLayout，自动布局的时候，我们新建一个项目， 这个新建的项目，里面有一个控制器，这个控制器，是不是默认，是四四…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023