基于 pytest 编写的 PyMilvus 的测试框架。
测试代码: https://github.com/milvus-io/milvus/tree/master/tests/python_client
Milvus 支持4种部署方式,请根据需求选择部署方式,PyMilvus 支持任意部署下的 Milvus。
KinD 部署
KinD部署提供一键安装部署:最新的Milvus服务和测试客户端。KinD部署非常适合开发/调试测试用例,功能验证等对数据规模要求不大的场景,但并不适合性能或压力等有较大数据规模的场景。
准备环境
脚本安装
进入代码目录 */milvus/tests/scripts/
新建KinD环境,并自动执行CI Regression测试用例:
./e2e-k8s.sh
NOTE: 默认参数下KinD环境将在执行完测试用例后被自动清理
./e2e-k8s.sh --skip-cleanup
如不需要自动执行测试用例,并保留KinD环境:
./e2e-k8s.sh --skip-cleanup --skip-test --manual
NOTE: 需要login到测试客户端的container进行手动执行或调试测试用例
./e2e-k8s.sh --help
导出集群日志
kind export logs .
推荐使用 Python 3(>= 3.8) ,与 PyMilvus 支持的 Python 版本保持一致。
NOTE: 如选择KinD部署方式,以下步骤可以自动完成
安装测试所需的 Python 包,进入代码 */milvus/tests/python_client/
目录,执行命令:
pip install -r requirements.txt
config
目录下,测试的日志目录默认为:/tmp/ci_logs/
,可在启动测试用例之前添加环境变量来修改log的存放路径:export CI_LOG_PATH=/tmp/ci_logs/test/
Log Level | Log File |
---|---|
Debug | ci_test_log.debug |
Info | ci_test_log.log |
Error | ci_test_log.err |
在主目录 pytest.ini
文件内可设置默认传递的参数,如下例中 ip 为所需要设置的 milvus 服务的ip地址,*.html
为测试生成的 report
:
addopts = --host *.*.*.* --html=/tmp/ci_logs/report.html
testcases
目录,命令与 pytest 框架的执行命令一致,运行如下命令执行测试用例:python3 -W ignore -m pytest <选择的测试文件>
模块调用关系图
可参考添加新的测试用例或框架工具。
test 文件:每一个 SDK 类对应一个 test 文件,Load 和 Search 单独对应一个 test 文件
test类:每一个 test 文件中分两个类
TestObjectParams :
TestObjectOperations:
testcase 命名
不能在测试用例文件中初始化 PyMilvus 对象
在测试用例中,应直接调用封装好的方法或者属性,如下所示:
NOTE: 如当需要创建多个 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
验证接口返回错误或异常
输入期望的错误码和错误信息
# 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"})
验证接口返回正常返回值
输入期望的结果,供校验方法使用
# 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})
在 testcases 文件夹下找到被测接口相应的测试文件,进行用例添加。如下所示,全部测试用例可直接参考 testcases 目录下的所有 test 文件:
@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"})
Tips
用例注释分为三个部分:目标,测试方法及期望结果,依此说明相应内容
在 base/client_base.py 文件 Base 类的 setup 方法中对被测试的类进行了初始化,如下图所示:
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()
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 传入检查方法所需的特定内容,具体内容由实现的检查方法所决定。
默认不传这两个参数,则检查接口能正常返回请求结果。