exo 如何用 pytest cluster/instance marker 编写多机集群集成测试【免费下载链接】exoRun frontier AI locally.项目地址: https://gitcode.com/GitHub_Trending/exo8/exoexo 的多机集群集成测试位于仓库的tests/目录整套机制由两个自定义 pytest marker 驱动pytest.mark.cluster声明测试需要的机器数量和网络拓扑pytest.mark.instance声明要部署的模型与并行方式。你只需要写好带 marker 的测试函数并注入sessionfixture集群部署、模型实例放置和清理全部由 fixture 自动完成适合新增 2 节点、4 节点推理测试或节点断连恢复类测试。前提是环境中可用ecoCLI 管理一批 Mac 主机EcoSession通过子进程调用eco命令部署集群测试命令使用uv run pytest执行。框架的三个组成部分写测试前先了解各模块职责方便定位问题tests/conftest.py注册cluster与instance两个 marker定义--hosts命令行选项和按测试注入的sessionfixture并在测试失败时把集群日志附加到 pytest 报告。tests/framework.pyClusterSpec/InstanceSpec数据类、marker 解析函数和Session封装chat、multi_turn、disconnect_node等方法默认模型常量DEFAULT_MODEL mlx-community/Llama-3.2-1B-Instruct-4bit。tools/src/exo_tools/cluster.py 中的EcoSession对ecoCLI 的包装eco --json start --deploy、eco stop、eco logs等tools/src/exo_tools/harness.py 中的Sharding、Comm枚举和place_instance、cleanup_all_instances等实例生命周期工具。声明 cluster 与 instance markermarker 签名在 tests/conftest.py 的pytest_configure中注册cluster(countN, thunderboltThunderbolt|None, min_memoryGB, chipPATTERN)声明集群需求。instance(model_id, shardingSharding, commComm, min_nodesN)声明实例放置model_id是必需的位置参数缺失时解析会抛出ValueError。参数取值来自仓库中的枚举定义参数枚举/取值说明thunderboltThunderbolt.A2Aall-to-all对应eco --tb-a2a、Thunderbolt.RINGring 拓扑对应eco --tb-ring选择带指定 Thunderbolt 拓扑的主机shardingSharding.PIPELINE值Pipeline层跨节点切分、Sharding.TENSOR值Tensor缺省为PIPELINEcommComm.RING值MlxRing网络 ring all-reduce、Comm.JACCL值MlxJacclRDMA over Thunderbolt缺省为RINGchipChip枚举M1到M4_ULTRA按芯片型号约束min_nodes整数缺省为1注意pytest.mark.instance是可选的。如果测试没有打 instance markerfixture 不会放置任何实例此时再调用session.chat()会抛出RuntimeError(No instance placed; add pytest.mark.instance to the test)。编写一个两节点测试以仓库中 tests/test_2node.py 的真实测试为例测试 2 节点 tensor 切分 JACCL 通信的推理import pytest from exo_tools.cluster import Thunderbolt from exo_tools.harness import Comm, Sharding from .framework import DEFAULT_MODEL pytest.mark.cluster(count2, thunderboltThunderbolt.A2A) pytest.mark.instance( DEFAULT_MODEL, shardingSharding.TENSOR, commComm.JACCL, min_nodes2 ) def test_2node_jaccl(session): resp session.chat(Say hello in one sentence.) assert len(resp) 0结构上就是三件事clustermarker 说明要 2 台 A2A 拓扑的主机instancemarker 说明放置哪个模型、用什么切分和通信方式、至少占几台节点测试体里通过session.chat(prompt, max_tokens100)发送请求并断言返回非空。session.multi_turn(messages)可用于多轮对话session.chat_raw则返回完整的ChatCompletionResponse。单节点测试只需要去掉 Thunderbolt 约束见 tests/test_1node.pypytest.mark.cluster(count1) pytest.mark.instance(DEFAULT_MODEL) def test_place_instance_and_chat(session): resp session.chat(Say hello in one sentence.) assert len(resp) 0tests/test_4node.py 展示了 4 节点版本count4min_nodes4其余写法一致。运行测试仓库根 pyproject.toml 的[tool.pytest.ini_options]中addopts -m not slow --ignoretests --ignoretmp默认的uv run pytest不会收集多机集成测试需要显式指向tests/。各测试文件 docstring 中给出的运行方式# 单个测试文件 uv run pytest tests/test_2node.py -v # 整个集成测试套件 uv run pytest tests/ -v # 显式指定主机覆盖基于约束的自动预留 uv run pytest tests/ -v --hosts s2,s4,s9,s10--hosts传入逗号分隔的主机名列表。不传时fixture 按clustermarker 的count、thunderbolt、chip、min_memory约束通过eco start --deploy自动选择并预留主机传入时则直接取列表的前count台。pytest 报告头部会显示本次会话的eco user形如test-加 8 位十六进制后缀以及 hosts 覆盖信息可据此确认选机是否符合预期。另外EcoSession.start_deploy默认通过 rsync 从本地源码部署设置环境变量EXO_REF或传ref参数则改为从 GitHub 分支/tag 部署文档注明这是给 CI 用的方式。session fixture 自动完成的编排sessionfixture 是函数级作用域每个测试独立注入流程如下逻辑见 tests/conftest.py 的sessionfixture读取测试上的cluster和instancemarker解析为ClusterSpec与InstanceSpec查找集群缓存集群按ClusterSpec缓存相同 cluster marker 的测试共享同一部署只有第一次出现的 spec 才真正调用eco start --deploy ... --wait若声明了 instance marker调用place_instance放置模型并等待就绪返回instance_id测试结束后清理本测试放置的实例cleanup_all_instances如果测试期间有节点被断开_stopped_hosts非空则使该 spec 的缓存失效并停止集群让下一个测试重新部署一套干净环境会话结束时session 级 autouse fixture 会停止所有缓存的集群EcoSession还注册了 atexit 与 SIGTERM/SIGHUP 处理器兜底执行eco stop无需手动停机。验证结果与失败时查看集群日志成功条件就是 pytest 用例行本身通过示例测试断言session.chat()返回非空字符串说明实例已放置就绪且推理链路正常。失败时的关键机制是 tests/conftest.py 的pytest_runtest_makereport当使用sessionfixture 的测试在 call 阶段失败时fixture 会通过eco logs拉取集群主机最近 200 行日志以 JSON 形式附加到测试报告的Cluster Logs段落中直接用于定位是部署、放置还是推理环节出了问题。涉及节点状态变化时用session.wait_ready(expected_nodes, timeout60)等待集群收敛它轮询/state直到nodeIdentities和nodeMemory的条目数都恰好等于预期节点数才返回超时抛出TimeoutError。可选节点断连/重连的恢复测试tests/test_resilience.py 展示了一个更复杂的场景2 节点推理 → 断开一个节点 → 在剩余节点上放置 1 节点实例验证推理 → 重新连接节点并等待集群重组 → 再次放置 2 节点实例。核心片段# --- 断开第 2 个节点并等待集群收敛 --- session.disconnect_node(1) session.wait_ready(60) cleanup_all_instances(session.client) # --- 在剩余节点上放置 1 节点实例 --- place_instance(session.client, DEFAULT_MODEL, min_nodes1) session.instance_spec InstanceSpec(model_idDEFAULT_MODEL, min_nodes1) resp session.chat(Hello) assert len(resp) 0 # --- 重新连回节点恢复 2 节点推理 --- cleanup_all_instances(session.client) session.reconnect_node(1) session.wait_ready(60) place_instance(session.client, DEFAULT_MODEL, min_nodes2) session.instance_spec InstanceSpec(model_idDEFAULT_MODEL, min_nodes2) resp session.chat(Hello again) assert len(resp) 0disconnect_node(index)在指定节点上执行eco stop --keep保留预留reconnect_node(index)用原 namespace 重新启动该节点。因为这类测试会弄脏集群fixture teardown 检测到_stopped_hosts后会停止整个集群后续同 spec 的测试将重新部署而不是继续复用残留状态。限制整套测试必须运行在eco可管理的真实 Mac 主机环境中且JACCL通信依赖 Thunderbolt 上的 RDMAcommComm.JACCL的测试应配合thunderboltThunderbolt.A2A之类的拓扑约束声明。模型默认使用DEFAULT_MODELLlama-3.2-1B-Instruct-4bit换成其他模型时把完整 model id 作为instancemarker 的位置参数传入即可放置时模型需要能下载到集群节点。集群由eco自动预留和释放不要在测试里手动管理主机生命周期否则会与 fixture 的缓存和清理逻辑冲突。【免费下载链接】exoRun frontier AI locally.项目地址: https://gitcode.com/GitHub_Trending/exo8/exo创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 exo 如何用 pytest cluster/instance marker 编写多机集群集成测试？
exo 如何用 pytest cluster/instance marker 编写多机集群集成测试【免费下载链接】exoRun frontier AI locally.项目地址: https://gitcode.com/GitHub_Trending/exo8/exoexo 的多机集群集成测试位于仓库的tests/目录整套机制由两个自定义 pytest marker 驱动pytest.mark.cluster声明测试需要的机器数量和网络拓扑pytest.mark.instance声明要部署的模型与并行方式。你只需要写好带 marker 的测试函数并注入sessionfixture集群部署、模型实例放置和清理全部由 fixture 自动完成适合新增 2 节点、4 节点推理测试或节点断连恢复类测试。前提是环境中可用ecoCLI 管理一批 Mac 主机EcoSession通过子进程调用eco命令部署集群测试命令使用uv run pytest执行。框架的三个组成部分写测试前先了解各模块职责方便定位问题tests/conftest.py注册cluster与instance两个 marker定义--hosts命令行选项和按测试注入的sessionfixture并在测试失败时把集群日志附加到 pytest 报告。tests/framework.pyClusterSpec/InstanceSpec数据类、marker 解析函数和Session封装chat、multi_turn、disconnect_node等方法默认模型常量DEFAULT_MODEL mlx-community/Llama-3.2-1B-Instruct-4bit。tools/src/exo_tools/cluster.py 中的EcoSession对ecoCLI 的包装eco --json start --deploy、eco stop、eco logs等tools/src/exo_tools/harness.py 中的Sharding、Comm枚举和place_instance、cleanup_all_instances等实例生命周期工具。声明 cluster 与 instance markermarker 签名在 tests/conftest.py 的pytest_configure中注册cluster(countN, thunderboltThunderbolt|None, min_memoryGB, chipPATTERN)声明集群需求。instance(model_id, shardingSharding, commComm, min_nodesN)声明实例放置model_id是必需的位置参数缺失时解析会抛出ValueError。参数取值来自仓库中的枚举定义参数枚举/取值说明thunderboltThunderbolt.A2Aall-to-all对应eco --tb-a2a、Thunderbolt.RINGring 拓扑对应eco --tb-ring选择带指定 Thunderbolt 拓扑的主机shardingSharding.PIPELINE值Pipeline层跨节点切分、Sharding.TENSOR值Tensor缺省为PIPELINEcommComm.RING值MlxRing网络 ring all-reduce、Comm.JACCL值MlxJacclRDMA over Thunderbolt缺省为RINGchipChip枚举M1到M4_ULTRA按芯片型号约束min_nodes整数缺省为1注意pytest.mark.instance是可选的。如果测试没有打 instance markerfixture 不会放置任何实例此时再调用session.chat()会抛出RuntimeError(No instance placed; add pytest.mark.instance to the test)。编写一个两节点测试以仓库中 tests/test_2node.py 的真实测试为例测试 2 节点 tensor 切分 JACCL 通信的推理import pytest from exo_tools.cluster import Thunderbolt from exo_tools.harness import Comm, Sharding from .framework import DEFAULT_MODEL pytest.mark.cluster(count2, thunderboltThunderbolt.A2A) pytest.mark.instance( DEFAULT_MODEL, shardingSharding.TENSOR, commComm.JACCL, min_nodes2 ) def test_2node_jaccl(session): resp session.chat(Say hello in one sentence.) assert len(resp) 0结构上就是三件事clustermarker 说明要 2 台 A2A 拓扑的主机instancemarker 说明放置哪个模型、用什么切分和通信方式、至少占几台节点测试体里通过session.chat(prompt, max_tokens100)发送请求并断言返回非空。session.multi_turn(messages)可用于多轮对话session.chat_raw则返回完整的ChatCompletionResponse。单节点测试只需要去掉 Thunderbolt 约束见 tests/test_1node.pypytest.mark.cluster(count1) pytest.mark.instance(DEFAULT_MODEL) def test_place_instance_and_chat(session): resp session.chat(Say hello in one sentence.) assert len(resp) 0tests/test_4node.py 展示了 4 节点版本count4min_nodes4其余写法一致。运行测试仓库根 pyproject.toml 的[tool.pytest.ini_options]中addopts -m not slow --ignoretests --ignoretmp默认的uv run pytest不会收集多机集成测试需要显式指向tests/。各测试文件 docstring 中给出的运行方式# 单个测试文件 uv run pytest tests/test_2node.py -v # 整个集成测试套件 uv run pytest tests/ -v # 显式指定主机覆盖基于约束的自动预留 uv run pytest tests/ -v --hosts s2,s4,s9,s10--hosts传入逗号分隔的主机名列表。不传时fixture 按clustermarker 的count、thunderbolt、chip、min_memory约束通过eco start --deploy自动选择并预留主机传入时则直接取列表的前count台。pytest 报告头部会显示本次会话的eco user形如test-加 8 位十六进制后缀以及 hosts 覆盖信息可据此确认选机是否符合预期。另外EcoSession.start_deploy默认通过 rsync 从本地源码部署设置环境变量EXO_REF或传ref参数则改为从 GitHub 分支/tag 部署文档注明这是给 CI 用的方式。session fixture 自动完成的编排sessionfixture 是函数级作用域每个测试独立注入流程如下逻辑见 tests/conftest.py 的sessionfixture读取测试上的cluster和instancemarker解析为ClusterSpec与InstanceSpec查找集群缓存集群按ClusterSpec缓存相同 cluster marker 的测试共享同一部署只有第一次出现的 spec 才真正调用eco start --deploy ... --wait若声明了 instance marker调用place_instance放置模型并等待就绪返回instance_id测试结束后清理本测试放置的实例cleanup_all_instances如果测试期间有节点被断开_stopped_hosts非空则使该 spec 的缓存失效并停止集群让下一个测试重新部署一套干净环境会话结束时session 级 autouse fixture 会停止所有缓存的集群EcoSession还注册了 atexit 与 SIGTERM/SIGHUP 处理器兜底执行eco stop无需手动停机。验证结果与失败时查看集群日志成功条件就是 pytest 用例行本身通过示例测试断言session.chat()返回非空字符串说明实例已放置就绪且推理链路正常。失败时的关键机制是 tests/conftest.py 的pytest_runtest_makereport当使用sessionfixture 的测试在 call 阶段失败时fixture 会通过eco logs拉取集群主机最近 200 行日志以 JSON 形式附加到测试报告的Cluster Logs段落中直接用于定位是部署、放置还是推理环节出了问题。涉及节点状态变化时用session.wait_ready(expected_nodes, timeout60)等待集群收敛它轮询/state直到nodeIdentities和nodeMemory的条目数都恰好等于预期节点数才返回超时抛出TimeoutError。可选节点断连/重连的恢复测试tests/test_resilience.py 展示了一个更复杂的场景2 节点推理 → 断开一个节点 → 在剩余节点上放置 1 节点实例验证推理 → 重新连接节点并等待集群重组 → 再次放置 2 节点实例。核心片段# --- 断开第 2 个节点并等待集群收敛 --- session.disconnect_node(1) session.wait_ready(60) cleanup_all_instances(session.client) # --- 在剩余节点上放置 1 节点实例 --- place_instance(session.client, DEFAULT_MODEL, min_nodes1) session.instance_spec InstanceSpec(model_idDEFAULT_MODEL, min_nodes1) resp session.chat(Hello) assert len(resp) 0 # --- 重新连回节点恢复 2 节点推理 --- cleanup_all_instances(session.client) session.reconnect_node(1) session.wait_ready(60) place_instance(session.client, DEFAULT_MODEL, min_nodes2) session.instance_spec InstanceSpec(model_idDEFAULT_MODEL, min_nodes2) resp session.chat(Hello again) assert len(resp) 0disconnect_node(index)在指定节点上执行eco stop --keep保留预留reconnect_node(index)用原 namespace 重新启动该节点。因为这类测试会弄脏集群fixture teardown 检测到_stopped_hosts后会停止整个集群后续同 spec 的测试将重新部署而不是继续复用残留状态。限制整套测试必须运行在eco可管理的真实 Mac 主机环境中且JACCL通信依赖 Thunderbolt 上的 RDMAcommComm.JACCL的测试应配合thunderboltThunderbolt.A2A之类的拓扑约束声明。模型默认使用DEFAULT_MODELLlama-3.2-1B-Instruct-4bit换成其他模型时把完整 model id 作为instancemarker 的位置参数传入即可放置时模型需要能下载到集群节点。集群由eco自动预留和释放不要在测试里手动管理主机生命周期否则会与 fixture 的缓存和清理逻辑冲突。【免费下载链接】exoRun frontier AI locally.项目地址: https://gitcode.com/GitHub_Trending/exo8/exo创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考