8图片这样的网站怎么做,网站等保测评怎么做,视频专用客户端app,站群软件哪个好目录
pytest-xdist执行流程#xff1a;
pytest-xdist 模块结构#xff1a;
pytest-xdist分布式测试原理#xff1a;
pytest-xdist源码浅读#xff1a; pytest-xdist执行流程#xff1a; 解析命令行参数#xff1a;pytest-xdist 会解析命令行参数#xff0c;获取用户…目录
pytest-xdist执行流程
pytest-xdist 模块结构
pytest-xdist分布式测试原理
pytest-xdist源码浅读 pytest-xdist执行流程 解析命令行参数pytest-xdist 会解析命令行参数获取用户指定的分发模式、进程数、主机列表等信息。 加载测试用例pytest-xdist 会加载所有的 pytest 测试用例包括在当前目录和子目录下的所有测试文件和测试函数。 分发测试用例根据用户指定的分发模式pytest-xdist 会将测试用例分发到多个进程或主机上执行。如果是分发到多个进程pytest-xdist 会创建多个子进程每个子进程都会执行一部分测试用例。如果是分发到多个主机pytest-xdist 会在每个主机上启动一个 Python 进程然后将测试用例分发给每个 Python 进程执行。 执行测试用例在每个进程或主机上pytest-xdist 会执行分配给它的测试用例。每个进程或主机上的测试执行都是独立的它们之间没有任何数据共享或通信。 汇总测试结果在所有进程或主机上的测试执行完成后pytest-xdist 会将所有测试结果汇总到主进程中并输出测试报告。在汇总测试结果的过程中pytest-xdist 还会根据用户指定的选项合并相同的测试结果比如合并多个进程或主机上的相同测试用例结果。 清理资源在所有测试结果都汇总完成后pytest-xdist 会清理所有的资源包括关闭分配给每个进程或主机的 Python 进程、删除临时文件等。
需要注意的是pytest-xdist 的执行流程是一个异步的过程不同进程或主机上的测试执行是并行的它们之间没有任何阻塞或等待 pytest-xdist 模块结构 xdist 模块这是 pytest-xdist 的主模块包含了分发测试用例的主要逻辑。在 xdist 模块中主要包含了以下几个子模块 looponfail: 在测试用例执行失败时自动重试的逻辑。loadfile: 加载分布式测试配置文件的逻辑。rsync: 在多个主机之间同步文件的逻辑。newhooks: 扩展 pytest 的钩子函数以支持分布式测试的逻辑。 testing 模块这是 pytest-xdist 的测试模块包含了对 pytest-xdist 的单元测试和集成测试。 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,destnumprocesses,typeint,defaultNone,helpshortcut for --distload --txNUM*popen//pythonpython%s (default: %default) % sys.version_info[0],)group._addoption(--tx,desttx,metavarxspec,helpaddrs[: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.configif 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 loadreturn load(session)elif session.config.option.dist load:from .dist import loadreturn load(session)elif session.config.option.dist each:from .dist import eachreturn 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.configif config.option.capture no:config.option.capture fd# ...else:from . import ( # noqa: F401box,worker,)with box.Box(config, numprocessesnumprocesses) as box:box.makegateways()gwlist box.gwlist()result box.invoke_gateways(gwlist, start_worker, numprocessesnumprocesses, **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 workerreturn worker.worker_runtest(itemitem, nextitemnextitem)elif config.option.dist load:from .dist import workerreturn worker.worker_runtest(itemitem, nextitemnextitem)elif config.option.dist each:return runtestprotocol(item, nextitemnextitem)# ... 在分发测试用例到多个主机时pytest-xdist 会将测试用例通过 SSH 协议发送到每个主机然后在每个主机上启动一个 Python 进程并在该进程中执行测试用例。这个过程是通过 ssh_run 函数来实现的它是 pytest-xdist 中的一个辅助函数用于执行远程命令。这里是相关的源码 def ssh_run(host, command, captureTrue):# ...ssh paramiko.SSHClient()ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())ssh.connect(hostnamehost, usernameusername, passwordpassword, portport)with ssh:stdin, stdout, stderr ssh.exec_command(command)if capture:out stdout.read().decode()err stderr.read().decode()return out, errelse: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.sessionif 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 参数浅解 -n: 指定分发模式可以是一个数字表示分发到多少个进程也可以是一个字符串表示分发到多少个主机如 -n 4 表示分发到 4 个进程-n 4 --hostshost1,host2,host3,host4 表示分发到 4 个主机。 --numprocesses: 指定分发到多少个进程执行测试用例。 --tx: 指定分发到多少个主机执行测试用例格式为 popen//ssh:userhost:port。 --max-worker-restart: 指定在某个子进程或主机上测试用例执行失败时的最大重试次数。 --rsyncdir: 指定用于同步文件的目录该目录下的所有文件会被同步到所有子进程或主机上。 --rsyncignore: 指定需要忽略同步的文件或目录的规则。 --boxed: 指定在子进程或主机中使用进程隔离process isolation模式执行测试用例。 --capture: 指定在子进程或主机中使用的输出捕获模式可以是 fd、sys 或者 no。 --ignore: 指定需要忽略的测试文件或目录。 --looponfail: 指定在测试用例执行失败时自动重试的次数。 以下是我收集到的比较好的学习教程资源虽然不是什么很值钱的东西如果你刚好需要可以评论区留言【777】直接拿走就好了 各位想获取资料的朋友请点赞 评论 收藏三连
三连之后我会在评论区挨个私信发给你们~
