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/milvusoutput_articleMilvus 测试框架使用指南基于 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仓库为测试准备了四种部署路径可根据数据规模与调试需求选择部署方式适用场景源码编译部署需要验证最新代码特性本地开发调试Docker Compose 部署单机 / 分布式快速拉起服务进行功能验证Kubernetes 部署单机 / 分布式Helm集群环境下的测试KinD 部署开发/调试测试用例、功能验证等对数据规模要求不大的场景注意KinD 部署不适合性能或压力等有较大数据规模的场景。2.1 KinD 一键部署与测试KinD 部署提供一键安装同时拉起最新的 Milvus 服务和测试客户端容器非常适合开发/调试测试用例。步骤如下准备环境安装 Docker、Docker Compose、jq、kubectl、helm、kind依赖清单见 tests/README.md进入脚本目录tests/scripts/新建 KinD 环境并自动执行 CI Regression 测试用例./e2e-k8s.shNOTE默认参数下KinD 环境将在执行完测试用例后被自动清理。如需保留 KinD 环境使用--skip-cleanup参数./e2e-k8s.sh --skip-cleanup如不需要自动执行测试用例并保留 KinD 环境用于手动调试./e2e-k8s.sh --skip-cleanup --skip-test --manualNOTE此模式下需要 login 到测试客户端的 container 进行手动执行或调试测试用例。查看更多脚本参数./e2e-k8s.sh --help导出集群日志排查问题时非常有用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 生态pytest8.3.4、pytest-xdist2.5.0并行、pytest-asyncio、pytest-timeout、pytest-repeat、pytest-rerunfailures失败重跑、pytest-htmlHTML 报告、pytest-cov、allure-pytest等客户端pymilvus3.1.0rc83含bulk_writer扩展、protobuf5.29.5数据处理pandas、numpy、scikit-learn、h5pybenchmark、pyarrow、fastparquet全文检索/文本处理tantivy、bm25s、rjieba与 Milvus 服务端 jieba-rs 版本对齐、Unidecode对象存储/Bulk Insert 测试minio、npy-append-array性能测试locust2.25.0对应pytest.ini中-p no:locust的禁用说明K8s 操作kubernetes17.17.0供 KinD 部署与 K8s 相关用例使用。3.2 日志配置config目录下测试的日志目录默认为/tmp/ci_logs/可在启动测试用例之前通过环境变量修改存放路径export CI_LOG_PATH/tmp/ci_logs/test/日志分级写入不同文件实现见 config/log_config.pyLog LevelLog FileDebugci_test_log.debugInfoci_test_log.logErrorci_test_log.err从源码看log_config.py 通过get_env_variable()读取CI_LOG_PATH环境变量缺失时回退到/tmp/ci_logs并自动mkdir(parentsTrue, exist_okTrue)创建目录同时生成 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与CDCCDC 同步用例不纳入常规 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放置已封装好的 PyMilvus 模块文件以及 pytest 框架的 setup/teardown 处理check接口返回结果的检查模块common测试用例通用的方法和参数config基础配置内容testcases存放测试用例脚本utils通用程序如全局日志类、环境检查方法等requirements执行测试文件所依赖的 Python 包即 requirements.txtconftest.py编写装饰器函数或自己实现的本地插件作用范围为该文件存放的目录及其子目录pytest.inipytest 的主配置文件对照仓库实际结构base/下封装了ApiConnectionsWrapper、ApiCollectionWrapper、ApiPartitionWrapper、ApiIndexWrapper、ApiUtilityWrapper、ApiDatabaseWrapper、ApiCollectionSchemaWrapper/ApiFieldSchemaWrapper、AsyncMilvusClientWrapper以及client_base.py、client_v2_base.pycheck/下包含func_check.py接口返回结果检查与param_check.py参数检查common/下包含common_type.pyCheckTasks、常量、默认参数、common_func.py通用方法、common_params.py、constants.py、code_mapping.py以及各类数据生成器text_generator.py、phrase_match_generator.py等config/下为 log_config.pyutils/下提供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_taskNone, check_itemsNone, **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调用链为通过utils/api_request.py中的api_request()统一发起 PyMilvus 接口调用接口返回2 个值的 list第一个是 PyMilvus 的接口返回结果第二个是接口返回结果正常/异常的判断True/False将结果、函数名、check_task、check_items一并交给ResponseChecker定义于 check/func_check.py 第 51 行执行检查并返回check_result。TestcaseBase的 setup 方法中对被测类进行了统一初始化见 base/client_base.py 的_setup_objectsself.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如TestPartitionParamsPartition Interface 参数检查测试用例类。检查在不同输入参数条件下目标类/方法的表现参数注意覆盖default、empty、none、datatype、maxsize 边界值等TestObjectOperations如TestPartitionOperationsPartition 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(collectioncollection_name, namepartition_name) assert self.partition_wrap.is_empty2验证接口返回错误或异常使用check_taskCheckTasks.err_res并输入期望的错误码和错误信息。err_res是 common/common_type.py 中CheckTasks类第 521 行起定义的检查任务标识# create partition with collection is None self.partition_wrap.init_partition(collectionNone, namepartition_name, check_taskCheckTasks.err_res, check_items{ct.err_code: 1, ct.err_msg: NoneType object has no attribute})3验证接口返回正常返回值使用check_taskCheckTasks.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_taskCheckTasks.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_taskCheckTasks.err_res, check_items{ ct.err_code: 1, ct.err_msg: cant find collection})5.4 Tipswrapper 参数约定调用需要测试的接口时应按照封装好的方法传入参数。以init_partition为例除check_task、check_items两个参数外其余参数与 PyMilvus 的接口参数一致def init_partition(self, collection, name, description, check_taskNone, check_itemsNone, **kwargs)check_task用来选择 check/func_check.py 文件中ResponseChecker检查类中对应的接口检查方法可选择的方法在 common/common_type.py 文件的CheckTasks类中check_items传入检查方法所需的特定内容具体内容由实现的检查方法所决定默认不传这两个参数则检查接口能正常返回请求结果即默认只做接口调用成功的断言。TestcaseBase还提供了丰富的通用初始化方法以降低用例编写成本见 base/client_base.pyinit_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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考