Milvus Python 客户端测试框架使用指南：基于 PyMilvus 与 pytest 的完整测试体系解析

Milvus Python 客户端测试框架使用指南：基于 PyMilvus 与 pytest 的完整测试体系解析

【免费下载链接】milvusMilvus is a high-performance, cloud-native vector database built for scalable vector ANN search项目地址: https://gitcode.com/GitHub_Trending/mi/milvus

<output_article>

Milvus 测试框架使用指南：基于 pytest 与 PyMilvus 的 SDK 测试体系深入解析

导读

本文档系统讲解 Milvus 开源仓库中 tests/python_client 目录下基于 pytest 与 PyMilvus 构建的 SDK 测试框架。文章覆盖从 Milvus 服务部署、PyMilvus 测试环境搭建、用例执行到框架模块设计、测试代码编写规范的全链路内容，并结合仓库源码（wrapper 封装、check 检查模块、config 配置等）深入剖析其分层设计与调用原理。读完本文，你将掌握如何为 Milvus 任意部署形态编写、运行、调试 PyMilvus 自动化测试用例，并理解该框架"封装接口 + 统一校验 + pytest 管理"的核心工程模式。

一、框架简介：面向 PyMilvus 的 pytest 测试体系

该测试框架是一个基于pytest编写的PyMilvus测试框架，位于仓库 tests/python_client 目录。它服务于 Milvus 项目 Python 客户端的自动化回归测试，覆盖接口参数检查、功能操作验证、稳定性测试、性能测试（locust）、CDC 同步测试、滚动升级测试等众多场景。

从仓库目录结构看，testcases下既包含test_collection.py、test_partition.py、test_index.py、test_insert.py、test_query.py、test_delete.py、test_connection.py、test_alias.py、test_database.py等常规接口测试文件，也有test_e2e.py、test_concurrent.py、test_mix_scenes.py、test_issues.py等场景类测试，以及test_full_text_search.py、test_phrase_match.py、test_text_embedding_function_e2e.py等面向新特性（全文检索、文本嵌入函数）的专项用例。

二、快速开始：Milvus 部署方式选择

PyMilvus 测试框架支持任意部署形态下的 Milvus，仓库为测试准备了四种部署路径，可根据数据规模与调试需求选择：

注意：KinD 部署不适合性能或压力等有较大数据规模的场景。

2.1 KinD 一键部署与测试

KinD 部署提供一键安装：同时拉起最新的 Milvus 服务和测试客户端容器，非常适合开发/调试测试用例。步骤如下：

1. 准备环境：安装 Docker、Docker Compose、jq、kubectl、helm、kind（依赖清单见 tests/README.md）；
2. 进入脚本目录：tests/scripts/；
3. 新建 KinD 环境并自动执行 CI Regression 测试用例：

./e2e-k8s.sh

NOTE：默认参数下，KinD 环境将在执行完测试用例后被自动清理。

4. 如需保留 KinD 环境，使用--skip-cleanup参数：

./e2e-k8s.sh --skip-cleanup

5. 如不需要自动执行测试用例，并保留 KinD 环境（用于手动调试）：

./e2e-k8s.sh --skip-cleanup --skip-test --manual

NOTE：此模式下需要 login 到测试客户端的 container 进行手动执行或调试测试用例。

6. 查看更多脚本参数：

./e2e-k8s.sh --help

7. 导出集群日志（排查问题时非常有用）：

kind export logs .

三、PyMilvus 测试环境部署及用例执行

3.1 环境准备

推荐使用Python 3.12，与 Python client CI 运行环境保持一致。

NOTE：如选择 KinD 部署方式，以下步骤可以自动完成。

进入代码目录tests/python_client/，安装测试所需的 Python 包：

pip install -r requirements.txt

仓库的 requirements.txt 中关键依赖包括：

• pytest 生态：pytest==8.3.4、pytest-xdist==2.5.0（并行）、pytest-asyncio、pytest-timeout、pytest-repeat、pytest-rerunfailures（失败重跑）、pytest-html（HTML 报告）、pytest-cov、allure-pytest等；
• 客户端：pymilvus==3.1.0rc83（含bulk_writer扩展）、protobuf>=5.29.5；
• 数据处理：pandas、numpy、scikit-learn、h5py（benchmark）、pyarrow、fastparquet；
• 全文检索/文本处理：tantivy、bm25s、rjieba（与 Milvus 服务端 jieba-rs 版本对齐）、Unidecode；
• 对象存储/Bulk Insert 测试：minio、npy-append-array；
• 性能测试：locust==2.25.0（对应pytest.ini中-p no:locust的禁用说明）；
• K8s 操作：kubernetes==17.17.0（供 KinD 部署与 K8s 相关用例使用）。

3.2 日志配置

config目录下，测试的日志目录默认为/tmp/ci_logs/，可在启动测试用例之前通过环境变量修改存放路径：

export CI_LOG_PATH=/tmp/ci_logs/test/

日志分级写入不同文件（实现见 config/log_config.py）：

从源码看，log_config.py 通过get_env_variable()读取CI_LOG_PATH环境变量，缺失时回退到/tmp/ci_logs，并自动mkdir(parents=True, exist_ok=True)创建目录，同时生成 JSON 与 HTML 两种测试报告路径。

3.3 pytest 主配置

在主目录 pytest.ini 内可设置默认传递的参数。例如指定 Milvus 服务 IP 与测试报告输出：

addopts = --host *.*.*.* --html=/tmp/ci_logs/report.html

仓库实际默认配置为：

[pytest] addopts = -p no:locust -v log_format = [%(asctime)s - %(levelname)s - %(name)s]: %(message)s (%(filename)s:%(lineno)s) log_date_format = %Y-%m-%d %H:%M:%S markers = tags: custom tags for test cases CDC: CDC sync test cases (not run in regular e2e pipeline) filterwarnings = ignore::DeprecationWarning ignore::pymilvus.decorators.PyMilvusDeprecationWarning ignore:invalid escape sequence:SyntaxWarning asyncio_default_fixture_loop_scope = function timeout_method = thread

该文件还通过markers声明了自定义标记：tags（用例分级，如 L1）与CDC（CDC 同步用例，不纳入常规 e2e 流水线），并统一过滤了 Deprecation 类警告。

3.4 执行测试用例

进入testcases目录，执行命令与 pytest 框架标准命令一致：

python3 -W ignore -m pytest <选择的测试文件>

例如运行 partition 相关用例：python3 -W ignore -m pytest test_partition.py。-W ignore用于屏蔽 warning 输出，配合 pytest.ini 中的 filterwarnings 保持日志干净。

四、模块介绍与设计思路

4.1 工作目录及文件介绍

对照仓库实际结构：

• base/下封装了ApiConnectionsWrapper、ApiCollectionWrapper、ApiPartitionWrapper、ApiIndexWrapper、ApiUtilityWrapper、ApiDatabaseWrapper、ApiCollectionSchemaWrapper/ApiFieldSchemaWrapper、AsyncMilvusClientWrapper以及client_base.py、client_v2_base.py；
• check/下包含func_check.py（接口返回结果检查）与param_check.py（参数检查）；
• common/下包含common_type.py（CheckTasks、常量、默认参数）、common_func.py（通用方法）、common_params.py、constants.py、code_mapping.py以及各类数据生成器（text_generator.py、phrase_match_generator.py等）；
• config/下为 log_config.py；
• utils/下提供util_log.py（全局日志）、api_request.py（统一接口请求）、util_k8s.py、util_common.py、util_pymilvus.py、util_birdwatcher.py、util_fts.py、wrapper.py等工具。

4.2 主要设计思路

框架的分层设计是理解整个体系的关键：

• base/*_wrapper.py：封装被测接口，统一处理接口请求，提取接口请求的返回结果，传入check/func_check.py模块进行结果检查；
• check/func_check.py：编写各接口返回结果的检查方法，供测试用例调用；
• base/client_base.py：使用 pytest 框架，进行相应的 setup/teardown 方法处理；
• testcases目录下的测试文件，继承base/client_base.py里的TestcaseBase模块编写测试用例。用例里用到的通用参数和数据处理方法写入common模块供用例调用；
• config目录下加入全局配置，如日志路径；
• utils目录下实现全局方法，如全局可用的日志模块。

4.3 源码级调用链剖析

以ApiPartitionWrapper为例（见 base/partition_wrapper.py），其init_partition实现展示了 wrapper 层的标准模式：

def init_partition(self, collection, name, description="", check_task=None, check_items=None, **kwargs): """In order to distinguish the same name of partition""" func_name = sys._getframe().f_code.co_name response, is_succ = api_request([Partition, collection, name, description], **kwargs) self.partition = response if is_succ is True else None check_result = ResponseChecker(response, func_name, check_task, check_items, is_succ, **kwargs).run() return response, check_result

调用链为：

1. 通过utils/api_request.py中的api_request()统一发起 PyMilvus 接口调用；
2. 接口返回2 个值的 list：第一个是 PyMilvus 的接口返回结果，第二个是接口返回结果正常/异常的判断（True/False）；
3. 将结果、函数名、check_task、check_items一并交给ResponseChecker（定义于 check/func_check.py 第 51 行）执行检查并返回check_result。

TestcaseBase的 setup 方法中对被测类进行了统一初始化（见 base/client_base.py 的_setup_objects）：

self.connection_wrap = ApiConnectionsWrapper() self.utility_wrap = ApiUtilityWrapper() self.collection_wrap = ApiCollectionWrapper() self.partition_wrap = ApiPartitionWrapper() self.index_wrap = ApiIndexWrapper() self.collection_schema_wrap = ApiCollectionSchemaWrapper() self.field_schema_wrap = ApiFieldSchemaWrapper() self.database_wrap = ApiDatabaseWrapper() self.async_milvus_client_wrap = AsyncMilvusClientWrapper()

teardown_method则承担环境清理职责：删除测试创建的 collection、alias、资源组（resource group）、角色（role），移除连接并恢复默认连接配置，从而保证用例之间的隔离性。

五、代码添加：测试用例与框架工具的扩展

5.1 测试编码风格

test 文件：每一个 SDK 类对应一个 test 文件，Load 和 Search 单独对应一个 test 文件（仓库中确实存在独立的load/、search/目录以及test_utility.py、test_connection.py等文件）。

test 类：每一个 test 文件中分两个类：

• TestObjectParams（如TestPartitionParams）：Partition Interface 参数检查测试用例类。检查在不同输入参数条件下目标类/方法的表现，参数注意覆盖default、empty、none、datatype、maxsize 边界值等；
• TestObjectOperations（如TestPartitionOperations）：Partition Interface 针对不同 function 或操作的测试。检查在合法输入参数、与其他接口有一定交互的条件下，目标类/方法的返回和表现。

testcase 命名：

• TestObjectParams类：以 testcase输入参数区分命名，如test_partition_empty_name()表示验证空字符串作为 name 参数输入的表现；
• TestObjectOperations类：
  • 以 testcase操作步骤区分命名，如test_partition_drop_partition_twice()表示验证连续 drop 两次 partition 的表现；
  • 以 testcase验证点区分命名，如test_partition_maximum_partitions()表示验证创建 partition 的最大数量。

5.2 编码注意事项

（1）不能在测试用例文件中初始化 PyMilvus 对象

一般情况下，不在测试用例文件中直接添加日志代码；在测试用例中应直接调用封装好的方法或者属性。

当需要创建多个 partition 对象时，可调用self.init_partition_wrap()，该方法返回的结果就是新生成的 partition 对象；当无需创建多个对象时，直接使用self.partition_wrap即可：

# create partition - Call the default initialization method partition_w = self.init_partition_wrap() assert partition_w.is_empty

# create partition - Directly call the encapsulated object self.partition_wrap.init_partition(collection=collection_name, name=partition_name) assert self.partition_wrap.is_empty

（2）验证接口返回错误或异常

使用check_task=CheckTasks.err_res，并输入期望的错误码和错误信息。err_res是 common/common_type.py 中CheckTasks类（第 521 行起）定义的检查任务标识：

# create partition with collection is None self.partition_wrap.init_partition(collection=None, name=partition_name, check_task=CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: "'NoneType' object has no attribute"})

（3）验证接口返回正常返回值

使用check_task=CheckTasks.check_partition_property（对应 check/func_check.py 第 396 行的check_partition_property检查方法），可在CheckTasks中新建校验方法并在用例中调用，输入期望的结果供校验方法使用：

# create partition partition_w = self.init_partition_wrap(collection_w, partition_name, check_task=CheckTasks.check_partition_property, check_items={"name": partition_name, "description": description, "is_empty": True, "num_entities": 0})

5.3 测试用例添加完整示例

在base文件夹的 wrapper 文件底下找到封装好的同名被测接口（返回 2 值 list，可用作额外结果检查）；然后在testcases文件夹下找到被测接口对应的测试文件进行用例添加：

@pytest.mark.tags(CaseLabel.L1) @pytest.mark.parametrize("partition_name", [cf.gen_unique_str(prefix)]) def test_partition_dropped_collection(self, partition_name): """ target: verify create partition against a dropped collection method: 1. create collection1 2. drop collection1 3. create partition in collection1 expected: 1. raise exception """ # create collection collection_w = self.init_collection_wrap() # drop collection collection_w.drop() # create partition failed self.partition_wrap.init_partition( collection_w.collection, partition_name, check_task=CheckTasks.err_res, check_items={ ct.err_code: 1, ct.err_msg: "can't find collection"})

5.4 Tips：wrapper 参数约定

调用需要测试的接口时，应按照封装好的方法传入参数。以init_partition为例，除check_task、check_items两个参数外，其余参数与 PyMilvus 的接口参数一致：

def init_partition(self, collection, name, description="", check_task=None, check_items=None, **kwargs)

• check_task用来选择 check/func_check.py 文件中ResponseChecker检查类中对应的接口检查方法，可选择的方法在 common/common_type.py 文件的CheckTasks类中；
• check_items传入检查方法所需的特定内容，具体内容由实现的检查方法所决定；
• 默认不传这两个参数，则检查接口能正常返回请求结果（即默认只做"接口调用成功"的断言）。

TestcaseBase还提供了丰富的通用初始化方法以降低用例编写成本（见 base/client_base.py）：init_collection_wrap()（创建默认 schema 的 collection）、init_collection_general()（支持 binary/全部数据类型/稀疏向量/可空字段/默认值字段等组合）、insert_data_general()、init_resource_group()、init_user_with_privilege()（创建用户-角色-授权完整链路）等，均可直接复用。

5.5 框架功能添加

• 在utils目录下添加需要的全局方法或者工具（如util_log.py提供的全局test_log日志对象）；
• 可将相应的配置内容加入config目录下（如扩展 log_config.py 的配置项）。

六、总结

Milvus 的 Python 客户端测试框架通过wrapper 封装被测接口、ResponseChecker 统一结果校验、TestcaseBase 管理 pytest 生命周期、common/utils 沉淀通用能力的四层设计，将 PyMilvus 接口测试的编写成本降到最低：用例作者只需关注"输入参数 + 预期结果"，而无需关心连接管理、环境清理、日志与报告等基础设施。结合 KinD 一键部署脚本与 pytest 生态（xdist 并行、rerunfailures 重跑、html 报告），该框架可支撑从日常功能验证到 CI 回归、再到性能与稳定性测试的完整测试体系，是理解和扩展 Milvus SDK 测试的首选入口。 </output_article>

【免费下载链接】milvusMilvus is a high-performance, cloud-native vector database built for scalable vector ANN search项目地址: https://gitcode.com/GitHub_Trending/mi/milvus