Cortex E2E 测试框架实战指南：从依赖安装到全量端到端测试运行

后端云原生模型推理服务MLOps人工智能【免费下载链接】cortexProduction infrastructure for machine learning at scale项目地址https://gitcode.com/gh_mirrors/co/cortex点击查看免费下载导读本文基于 Cortex 仓库Production infrastructure for machine learning at scale自带的端到端测试框架展开围绕 test/e2e/README.md 讲解如何在本机安装 e2e 测试包、配置 Python 客户端指向开发版 CLI、在既有集群或新建集群上运行 pytest 端到端测试以及如何通过命令行开关与.env环境变量精确控制测试范围与超时行为。读完本文你将掌握这套测试框架的完整运行流程、各参数与超时配置的底层含义结合 test/e2e/tests/conftest.py 等源码佐证并了解 RealtimeAPI、AsyncAPI、BatchAPI、TaskAPI、自动扩缩容、压测与长期稳定性等测试用例各自验证了什么。框架概览e2e 测试包的结构Cortex 的端到端测试并不是一个简单的 pytest 脚本集合而是一个独立安装的 Python 包其目录结构如下均位于test/e2e/下e2e/测试核心库包含测试用例实现与工具函数tests.py各工作负载类型的通用测试流程realtime / batch / async / task / autoscaling / load / long-running / scale-to-zeroutils.py轮询、并发请求、日志流式输出等底层工具cluster.py通过 CLI 创建/删除测试集群expectations.py响应断言与 expectations 文件解析generator.py动态加载样本生成器供批量压测构造请求样本exceptions.py自定义异常类型tests/按 AWS 环境组织的 pytest 用例入口通过参数化把具体示例 API 与e2e.tests中的通用流程绑定setup.py包定义声明了requests、jsonschema、pytest、python-dotenv、pyyaml、boto3以及cortexPython 客户端等依赖pytest.iniminversion 6.0默认addopts -s -v -r sxf从结构看框架刻意把「测试流程逻辑」与「被测示例 API」解耦e2e/tests.py中每个test_*函数只接受 API 名称、超时等参数而tests/aws/下的用例负责读取 config、决定用哪份 YAML 配置cortex_cpu.yaml/cortex_gpu.yaml/cortex_cpu_arm64.yaml/cortex_scale_to_zero.yaml并调用对应流程。这种设计使得新增一个示例 API 只需要在test/apis/下放好 YAML、sample.json和expectations.yaml再在用例入口中登记即可。第一步安装 e2e 测试包在项目根目录执行pip install -e test/e2e以 editable 模式安装且该步骤只需执行一次代码改动后无需重装。setup.py会通过dependency_links指向python/client下的 cortex 客户端源码cortex_client_dir root.parent.parent / python / client若该目录不存在会直接抛出ModuleNotFoundError。注意如果你此前已安装过 cortex 客户端官方 README 建议先执行下面两条命令再安装 e2e 包pip3 uninstall cortex pip3 install -e python/client/原因是 e2e 包依赖cortex客户端需要确保使用的是仓库内的开发版客户端python/client/cortex/而非 PyPI 上的正式版。第二步配置测试所用的 CLI 与目标集群让 Python 客户端使用开发版 CLI 二进制运行测试前需要通过环境变量指定 CLI 二进制路径让 Python 客户端使用你本机编译的 Cortex CLI 而不是全局安装的版本export CORTEX_CLI_PATHcortex_repo_path/bin/cortex其中cortex_repo_path替换为当前仓库的实际路径。仓库的dev/build_cli.sh会把 CLI 编译产出到bin/目录。如果不设置该变量cortex客户端会回落到 PATH 中可找到的cortex命令。两种测试目标集群模式测试可以在两种集群环境下运行通过互斥的两个参数选择test/e2e/tests/conftest.py 中会校验--env与--config不能同时给出否则抛出ValueError模式一复用已有集群推荐用于日常开发验证pytest test/e2e/tests --env env_name--env指定 Cortex environment 名称pytest fixture 会通过cx.client(env_name)创建客户端直接连接既有集群测试结束后不创建也不删除任何集群资源。模式二临时新建集群测试专用跑完自动销毁pytest test/e2e/tests --config cluster.yaml--config指向集群配置文件如manager/manifests/ami.json配套的 cluster config YAML。此时 test/e2e/tests/aws/conftest.py 中的pytest_configure会先调用e2e.create_cluster(cluster_config)底层执行cortex cluster up cluster.yaml -y --configure-env cluster_name而pytest_unconfigure会在全部测试结束后调用e2e.delete_cluster(cluster_config)执行cortex cluster down -y --config cluster.yaml即「为测试而生、跑完即删」的隔离模式避免污染长期集群。若--env和--config均未提供fixture 会直接pytest.skip提示必须二选一。BatchAPI 测试需要 S3 路径BatchAPI 测试包括test_batch.py与test_load.py中的批量用例会把批次预测结果写入 S3因此必须提供测试用 S3 桶pytest test/e2e/tests --config cluster.yaml --s3-path s3://s3_bucket/test/jobs更推荐通过环境变量CORTEX_TEST_BATCH_S3_PATH定义该桶见下文「配置」小节。若未提供test_batch.py会 skip 掉批量用例并提示需要--s3-path或对应环境变量。第三步常用运行开关测试入口定义了以下布尔开关用于按需裁剪测试范围全部定义于 test/e2e/tests/conftest.py 的pytest_addoption开关作用关联用例--skip-gpus跳过 GPU 相关测试realtime/async/batch 的cortex_gpu.yaml用例test_realtime.py、test_async.py、test_batch.py--skip-infs跳过 InferentiaAWS 推理芯片相关测试Inferentia 用例--skip-autoscaling跳过自动扩缩容测试test_autoscaling.py--skip-load跳过压测用例realtime / async / batch 三种负载test_load.py--skip-long-running跳过长期稳定性测试test_long_running.py例如在无 GPU 的 CPU 集群上完整跑一遍功能测试pytest test/e2e/tests --env dev --skip-gpus --skip-infs另外还有两个非布尔选项值得了解--local-operator开启后 BatchAPI / TaskAPI 用例会通过本地 operator 地址http://localhost:8888/batch/api、http://localhost:8888/tasks/api发请求见 test/e2e/e2e/utils.py 中request_batch_prediction/request_task的local_operator分支适用于本地开发 operator 场景。--arm-nodegroups/--x86-nodegroups以逗号分隔指定 ARM / x86 节点组配合cortex_cpu_arm64.yaml的 ARM 用例使用。第四步通过环境变量或 .env 文件配置测试行为测试框架支持通过环境变量或项目根目录下的.env文件python-dotenv自动加载见 test/e2e/tests/conftest.py 中的load_dotenv(.env)配置测试行为。README 给出了.env文件示例# .env file CORTEX_TEST_REALTIME_DEPLOY_TIMEOUT120 CORTEX_TEST_BATCH_DEPLOY_TIMEOUT60 CORTEX_TEST_BATCH_JOB_TIMEOUT120 CORTEX_TEST_BATCH_S3_PATHs3://s3_bucket/test/jobs结合 test/e2e/tests/conftest.py 的pytest_configure实现完整的环境变量清单与默认值如下时间单位均为秒环境变量默认值说明CORTEX_TEST_REALTIME_DEPLOY_TIMEOUT320RealtimeAPI 部署就绪等待超时CORTEX_TEST_BATCH_DEPLOY_TIMEOUT150BatchAPI 部署就绪等待超时CORTEX_TEST_BATCH_JOB_TIMEOUT200Batch 作业执行完成等待超时CORTEX_TEST_BATCH_S3_PATH无Batch 作业结果写入的 S3 路径优先级高于--s3-path吗见下方说明CORTEX_TEST_ASYNC_DEPLOY_TIMEOUT320AsyncAPI 部署就绪等待超时CORTEX_TEST_ASYNC_WORKLOAD_TIMEOUT200Async 工作负载完成轮询上限同时作为结果轮询的 poll_retriesCORTEX_TEST_TASK_DEPLOY_TIMEOUT75TaskAPI 部署就绪等待超时CORTEX_TEST_TASK_JOB_TIMEOUT200Task 作业完成等待超时关于CORTEX_TEST_BATCH_S3_PATH与--s3-path的优先级从 test/e2e/tests/conftest.py 源码可见逻辑为s3_path os.environ.get(CORTEX_TEST_BATCH_S3_PATH) s3_path config.getoption(--s3-path) if not s3_path else s3_path即环境变量优先只有当环境变量未设置时才回退到--s3-path命令行参数。因此更推荐「定义在.env中」这一方式——README 也明确说明这是更便捷的做法。除上述环境变量外pytest_configure还内置了压测与长期测试的默认规模参数load_test_config/long_running_test_config例如Realtime 压测总计10**5次请求、目标副本数50、并发50、状态码超时60sAsync 压测总计10**3次请求、副本数20、并发10、提交超时120s、工作负载超时120sBatch 压测10个作业、每作业10个 worker、每作业10**5个样本、batch_size20、超时300s长期测试持续运行5 * 24 * 3600秒5 天状态码超时60s。这些默认值展示了该框架既能做轻量功能回归也能承担大规模压测与长时间稳定性验证。测试覆盖全景每类用例在验证什么框架的用例入口在 test/e2e/tests/aws/具体执行逻辑在 test/e2e/e2e/tests.py。下面按工作负载类型拆解。RealtimeAPI 用例test_realtime.py参数化了三个示例 APIresnet50 图像分类、素数生成、文本生成并对 resnet50 附加路径v1/models/resnet50:predict验证自定义路径转发。执行流程test_realtime_api读取test/apis/realtime/api/cortex_cpu.yaml得到 API 规格可选注入node_groups若存在expectations.yaml则解析期望parse_expectationsclient.deploy()部署后通过apis_ready轮询要求requested ready up_to_date且requested 1发送sample.json中的 payloadPOST 或 GET断言 HTTP 200若配置了 expectations调用assert_response_expectations校验响应文本/JSON 或 JSON Schema无论成败finally中删除 API 清理现场失败时 best-effort 打印client.get_api信息并流式输出日志。GPU 用例使用cortex_gpu.yaml并在--skip-gpus时 skipARM 用例使用cortex_cpu_arm64.yaml且methodGET。AsyncAPI 用例test_async.py使用async/text-generator含 GPU 版本执行流程test_async_api部署后等apis_readyPOST 提交工作负载断言响应包含id以poll_retries默认取async_workload_timeout为上限轮询GET endpoint/request_id直到status completed断言结果 JSON 包含id、status、result、timestamp四个键且id与请求一致、timestamp/result非空若配置了 expectations对result内容执行 JSON Schema 校验。这相当于一条完整的「提交 → 轮询 → 校验结果」异步推理链路回归。BatchAPI 用例test_batch.py使用batch/image-classifier-alexnet流程test_batch_api部署后等endpoint_ready向 endpoint POST 空请求期望收到 400 即视为就绪携带sample.json的 item_list、batch_size2与dest_s3_dir来自 S3 路径配置提交批量预测失败可重试retry_attempts5取得job_id后轮询job_done直到作业状态为succeeded失败时打印 API 信息与client.get_job的作业状态并流式输出作业日志。TaskAPI 用例test_task.py使用task/iris-classifier-trainer流程test_task_api与 Batch 类似部署 →endpoint_ready→request_task提交任务 → 轮询job_done直至成功。自动扩缩容用例test_autoscaling.py使用主 APIrealtime/sleep加一个 dummy APIrealtime/prime-generator向主 API 注入sleep1.0查询参数以控制请求时长。test_autoscaling的核心思路通过autoscaling_test_config[max_replicas]默认 20设定目标副本数并将并发请求数设为max_replicas 1以确保能撑满副本部署时覆盖autoscaling配置max_replicas与downscale_stabilization_period: 1m依据 autoscaler 的max_upscale_factor/max_downscale_factor与upscale_stabilization_period/downscale_stabilization_period估算测试总超时并预留 2 倍余量给镜像下载和节点扩容用threading.Event控制请求流先并发打满副本断言requested到达max_replicas再停止请求验证自动缩回 1 副本全程通过check_futures_healthy监控请求线程健康超时则断言失败。这是对 autoscaler「扩上去、缩下来」闭环能力的真实负载验证。缩容到零用例test_scale_to_zero.py使用realtime/hello-world的cortex_scale_to_zero.yaml。流程部署后允许requested 0就绪greater_or_equal_to0随后第一次请求断言响应头x-cortex-origin activator请求先被 activator 接住触发冷启动拉起副本后续请求在 60 秒内等待x-cortex-origin api请求已直接转发到 API 副本。这条用例直接验证了 scale-to-zero 场景下 activator 与 API 之间的冷启动切换。压测与长期稳定性用例test_load.py包含 realtimerealtime/prime-generator10 万请求、asyncasync/text-generator千级请求并核对每个响应 id、batchbatch/sum依赖sample_generator.py动态生成样本用 boto3 分页统计 S3 结果对象数量校验成功批次数为ceil(items_per_job / batch_size)且batches_in_queue 0test_long_running.py对realtime/text-generator持续time_to_run默认 5 天周期性 POST逐次断言 200 与期望内容。batch 压测中对sample_generator.py的约束必须有且只能有一个无参generate_sample函数由 test/e2e/e2e/generator.py 的load_generator在运行期校验不满足会抛GeneratorValidationException。排查与调试建议先跑最小集合首次在既有集群上建议--skip-gpus --skip-infs --skip-autoscaling --skip-load --skip-long-running只保留核心功能用例realtime / async / batch / task快速确认环境连通性。Batch 用例被 skip优先检查是否设置了CORTEX_TEST_BATCH_S3_PATH或--s3-path以及该 S3 桶是否存在、是否有写权限。部署超时CORTEX_TEST_REALTIME_DEPLOY_TIMEOUT默认 320s若镜像较大或节点扩容较慢可调大集群配置不足时建议先扩容节点组。失败诊断每个用例失败时都会 best-effort 打印client.get_api()的完整 JSON含 spec 与 statusBatch/Task 额外打印作业状态并后台线程流式输出 API/作业日志复用cortex logs ... --random-pod见 test/e2e/e2e/utils.py 的stream_api_logs/stream_job_logs据此可快速定位部署失败或推理错误。结果断言期望文件expectations.yaml支持content_typetext/json/binary与expected精确值或json_schemaJSON Schema 校验两者互斥并支持 grpc 字段用于 gRPC 服务校验详见 test/e2e/e2e/expectations.py非法配置会抛ExpectationsValidationException。结语Cortex 的 e2e 测试框架把「安装依赖 → 指定 CLI → 选择目标集群 → 裁剪范围 → 配置超时」串成了一条可直接落地的 CI 或本地验证流水线。理解 test/e2e/tests/conftest.py 中每个开关与超时默认值、test/e2e/e2e/tests.py 中各工作负载的断言逻辑以及 test/e2e/e2e/utils.py 中的轮询与并发工具你就能根据自己的集群规模与硬件条件CPU / GPU / Inferentia / ARM灵活编排出一套覆盖功能、扩缩容、压测与长期稳定性的完整回归方案。赞分享后端云原生模型推理服务MLOps人工智能【免费下载链接】cortexProduction infrastructure for machine learning at scale项目地址https://gitcode.com/gh_mirrors/co/cortex点击查看免费下载相关推荐Composio E2E 测试体系实战基于 Docker 的多运行时端到端测试框架解析Composio E2E 测试体系实战基于 Docker 的多运行时端到端测试框架解析 导读 ts/e2e tests/ 是 Composio 仓库中专为 人工智能AI Agent工具调用MCP 服务MCP ClientsReth e2e-testsuite 框架实战指南编写可运行的端到端区块链测试Reth e2e testsuite 框架实战指南编写可运行的端到端区块链测试 本文以 crates/e2e test utils/src/testsuite区块链Intern测试框架运行指南从基础到云端测试Intern测试框架运行指南从基础到云端测试 还在为JavaScript项目的测试覆盖率而烦恼吗面对复杂的浏览器兼容性测试束手无策Intern测试框架为你开发工具上一篇ESP-IDF RTC 功耗模式测试应用解析用 rtc_power_modes 测量深睡与浅睡子模式功耗下一篇Agent Zero 插件开发实战指南从最小本地插件到社区发布创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考