Cortex E2E 测试框架实战指南：从依赖安装到全量端到端测试运行-开发者社区

后端
云原生
模型推理服务
MLOps
人工智能

【免费下载链接】cortex

Production 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-zero）
- utils.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.ini：minversion = 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_PATH=<cortex_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`	跳过 Inferentia（AWS 推理芯片）相关测试	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_TIMEOUT=120 CORTEX_TEST_BATCH_DEPLOY_TIMEOUT=60 CORTEX_TEST_BATCH_JOB_TIMEOUT=120 CORTEX_TEST_BATCH_S3_PATH=s3://<s3_bucket>/test/jobs

结合 test/e2e/tests/conftest.py 的pytest_configure实现，完整的环境变量清单与默认值如下（时间单位均为秒）：

环境变量	默认值	说明
`CORTEX_TEST_REALTIME_DEPLOY_TIMEOUT`	`320`	RealtimeAPI 部署就绪等待超时
`CORTEX_TEST_BATCH_DEPLOY_TIMEOUT`	`150`	BatchAPI 部署就绪等待超时
`CORTEX_TEST_BATCH_JOB_TIMEOUT`	`200`	Batch 作业执行完成等待超时
`CORTEX_TEST_BATCH_S3_PATH`	无	Batch 作业结果写入的 S3 路径（优先级高于`--s3-path`吗？见下方说明）
`CORTEX_TEST_ASYNC_DEPLOY_TIMEOUT`	`320`	AsyncAPI 部署就绪等待超时
`CORTEX_TEST_ASYNC_WORKLOAD_TIMEOUT`	`200`	Async 工作负载完成轮询上限（同时作为结果轮询的 poll_retries）
`CORTEX_TEST_TASK_DEPLOY_TIMEOUT`	`75`	TaskAPI 部署就绪等待超时
`CORTEX_TEST_TASK_JOB_TIMEOUT`	`200`	Task 作业完成等待超时

关于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、状态码超时60s；
Async 压测：总计10**3次请求、副本数20、并发10、提交超时120s、工作负载超时120s；
Batch 压测：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参数化了三个示例 API（resnet50 图像分类、素数生成、文本生成），并对 resnet50 附加路径v1/models/resnet50:predict验证自定义路径转发。执行流程（test_realtime_api）：

读取test/apis/realtime/<api>/cortex_cpu.yaml得到 API 规格，可选注入node_groups；
若存在expectations.yaml则解析期望（parse_expectations）；
client.deploy()部署后，通过apis_ready轮询，要求requested == ready == up_to_date且requested >= 1；
发送sample.json中的 payload（POST 或 GET），断言 HTTP 200；
若配置了 expectations，调用assert_response_expectations校验响应文本/JSON 或 JSON Schema；
无论成败，finally中删除 API 清理现场；失败时 best-effort 打印client.get_api信息并流式输出日志。

GPU 用例使用cortex_gpu.yaml并在--skip-gpus时 skip；ARM 用例使用cortex_cpu_arm64.yaml且method="GET"。

AsyncAPI 用例

test_async.py使用async/text-generator（含 GPU 版本），执行流程（test_async_api）：

部署后等apis_ready；
POST 提交工作负载，断言响应包含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_size=2与dest_s3_dir（来自 S3 路径配置）提交批量预测，失败可重试（retry_attempts=5）；
取得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 注入sleep=1.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_to=0），随后：

第一次请求断言响应头x-cortex-origin == "activator"（请求先被 activator 接住，触发冷启动拉起副本）；
后续请求在 60 秒内等待x-cortex-origin == "api"（请求已直接转发到 API 副本）。

这条用例直接验证了 scale-to-zero 场景下 activator 与 API 之间的冷启动切换。

压测与长期稳定性用例

test_load.py：包含 realtime（realtime/prime-generator，10 万请求）、async（async/text-generator，千级请求并核对每个响应 id）、batch（batch/sum，依赖sample_generator.py动态生成样本，用 boto3 分页统计 S3 结果对象数量，校验成功批次数为ceil(items_per_job / batch_size)且batches_in_queue == 0）；
test_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 与 status），Batch/Task 额外打印作业状态，并后台线程流式输出 API/作业日志（复用cortex logs ... --random-pod，见 test/e2e/e2e/utils.py 的stream_api_logs/stream_job_logs），据此可快速定位部署失败或推理错误。
结果断言：期望文件expectations.yaml支持content_type（text/json/binary）与expected（精确值）或json_schema（JSON 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）灵活编排出一套覆盖功能、扩缩容、压测与长期稳定性的完整回归方案。