furyhawk/milvus
1
0
Fork 0
mirror of https://github.com/milvus-io/milvus.git synced 2026-07-21 02:05:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

test: align snapshot client tests with latest pymilvus (#49762)

Browse Source 
## Summary
- Align snapshot wrapper and test call signatures with the latest
PyMilvus snapshot API.
- Update Python client test dependency to `pymilvus==3.1.0rc5`.
- Make snapshot test cleanup safe under `pytest -n 6` by isolating
cleanup lists and tracking RBAC users/roles per test.
- Skip the `CreateSnapshot`/`DropCollection` race case and link it to
#49761.
- Remove stale xfail marks for snapshot RBAC privilege group cases now
covered by the fixed server behavior.

## Test plan
- `.venv/bin/python -m py_compile base/client_base.py
base/client_v2_base.py milvus_client/test_milvus_client_snapshot.py`
- `git diff --check`
- `.venv/bin/python -m pytest -q
milvus_client/test_milvus_client_snapshot.py --dist loadgroup -n 6
--host 10.100.36.198 --port 19530 --tb=short`
  - `119 passed, 1 skipped in 293.48s`
- Cleanup audit after the full run:
  - `collections []`
  - `users ['root']`
  - `roles ['admin', 'public']`
  - `databases ['default']`

---------

Signed-off-by: zhuwenxing <wenxing.zhu@zilliz.com>
This commit is contained in:
zhuwenxing
2026-05-15 18:06:28 +08:00
committed by GitHub
parent 6b18f5a4d7
commit ebed483bb4
5 changed files with 1005 additions and 912 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/python_client/base/client_base.py
+230 -136
View File
@@ -1,28 +1,27 @@
import sys
from typing import Dict, List
from pymilvus import DefaultConfig
from base.database_wrapper import ApiDatabaseWrapper
from pymilvus import DefaultConfig
sys.path.append("..")
from base.connections_wrapper import ApiConnectionsWrapper
from base.collection_wrapper import ApiCollectionWrapper
from base.partition_wrapper import ApiPartitionWrapper
from base.index_wrapper import ApiIndexWrapper
from base.utility_wrapper import ApiUtilityWrapper
from base.schema_wrapper import ApiCollectionSchemaWrapper, ApiFieldSchemaWrapper
import pymilvus
from base.async_milvus_client_wrapper import AsyncMilvusClientWrapper
from utils.util_log import test_log as log
from base.collection_wrapper import ApiCollectionWrapper
from base.connections_wrapper import ApiConnectionsWrapper
from base.index_wrapper import ApiIndexWrapper
from base.partition_wrapper import ApiPartitionWrapper
from base.schema_wrapper import ApiCollectionSchemaWrapper, ApiFieldSchemaWrapper
from base.utility_wrapper import ApiUtilityWrapper
from common import common_func as cf
from common import common_type as ct
from common.common_params import IndexPrams
from pymilvus import ResourceGroupInfo, DataType, utility, MilvusClient
import pymilvus
from pymilvus import DataType, MilvusClient, ResourceGroupInfo, utility
from utils.util_log import test_log as log
class Base:
""" Initialize class object """
"""Initialize class object"""
connection_wrap = None
collection_wrap = None
partition_wrap = None
@@ -35,6 +34,7 @@ class Base:
resource_group_list = []
async_milvus_client_wrap = None
skip_connection = False
skip_global_role_cleanup = False
def setup_class(self):
log.info("[setup_class] Start setup class...")
@@ -45,7 +45,7 @@ class Base:
def setup_method(self, method):
log.info(("*" * 35) + " setup " + ("*" * 35))
log.info(f"pymilvus version: {pymilvus.__version__}")
log.info("[setup_method] Start setup test case %s." % method.__name__)
log.info(f"[setup_method] Start setup test case {method.__name__}.")
self._setup_objects()
def _setup_objects(self):
@@ -61,7 +61,7 @@ class Base:
def teardown_method(self, method):
log.info(("*" * 35) + " teardown " + ("*" * 35))
log.info("[teardown_method] Start teardown test case %s..." % method.__name__)
log.info(f"[teardown_method] Start teardown test case {method.__name__}...")
self._teardown_objects()
def _teardown_objects(self):
@@ -74,7 +74,11 @@ class Base:
if cf.param_info.param_token:
token = cf.param_info.param_token
else:
token = f"{cf.param_info.param_user}:{cf.param_info.param_password}" if cf.param_info.param_user and cf.param_info.param_password else None
token = (
f"{cf.param_info.param_user}:{cf.param_info.param_password}"
if cf.param_info.param_user and cf.param_info.param_password
else None
)
try:
""" Drop collection before disconnect """
@@ -104,34 +108,36 @@ class Base:
for rg_name in self.resource_group_list:
if rg_name is not None and rg_name in rgs_list:
rg = self.utility_wrap.describe_resource_group(
name=rg_name, check_task=ct.CheckTasks.check_nothing)[0]
name=rg_name, check_task=ct.CheckTasks.check_nothing
)[0]
if isinstance(rg, ResourceGroupInfo):
if rg.num_available_node > 0:
self.utility_wrap.transfer_node(source=rg_name,
target=ct.default_resource_group_name,
num_node=rg.num_available_node)
self.utility_wrap.transfer_node(
source=rg_name, target=ct.default_resource_group_name, num_node=rg.num_available_node
)
self.utility_wrap.drop_resource_group(rg_name, check_task=ct.CheckTasks.check_nothing)
except Exception as e:
log.debug(str(e))
try:
""" Drop roles before disconnect """
if not self.connection_wrap.has_connection(alias=DefaultConfig.DEFAULT_USING)[0]:
if token:
self.connection_wrap.connect(alias=DefaultConfig.DEFAULT_USING, uri=uri, token=token)
else:
self.connection_wrap.connect(alias=DefaultConfig.DEFAULT_USING, uri=uri)
if not self.skip_global_role_cleanup:
try:
""" Drop roles before disconnect """
if not self.connection_wrap.has_connection(alias=DefaultConfig.DEFAULT_USING)[0]:
if token:
self.connection_wrap.connect(alias=DefaultConfig.DEFAULT_USING, uri=uri, token=token)
else:
self.connection_wrap.connect(alias=DefaultConfig.DEFAULT_USING, uri=uri)
role_list = self.utility_wrap.list_roles(False)[0]
for role in role_list.groups:
role_name = role.role_name
if role_name not in ["admin", "public"]:
each_role = self.utility_wrap.init_role(name=role_name)[0]
each_role.drop()
role_list = self.utility_wrap.list_roles(False)[0]
for role in role_list.groups:
role_name = role.role_name
if role_name not in ["admin", "public"]:
each_role = self.utility_wrap.init_role(name=role_name)[0]
each_role.drop()
except Exception as e:
log.debug(str(e))
except Exception as e:
log.debug(str(e))
try:
""" Delete connection and reset configuration"""
@@ -140,8 +146,9 @@ class Base:
self.connection_wrap.remove_connection(i[0])
# because the connection is in singleton mode, it needs to be restored to the original state after teardown
self.connection_wrap.add_connection(default={"host": DefaultConfig.DEFAULT_HOST,
"port": DefaultConfig.DEFAULT_PORT})
self.connection_wrap.add_connection(
default={"host": DefaultConfig.DEFAULT_HOST, "port": DefaultConfig.DEFAULT_PORT}
)
except Exception as e:
log.debug(str(e))
@@ -151,10 +158,11 @@ class TestcaseBase(Base):
Additional methods;
Public methods that can be used for test cases.
"""
client = None
def _connect(self, enable_milvus_client_api=False):
""" Add a connection and create the connect """
"""Add a connection and create the connect"""
if self.skip_connection:
return None
@@ -167,7 +175,11 @@ class TestcaseBase(Base):
if cf.param_info.param_token:
token = cf.param_info.param_token
else:
token = f"{cf.param_info.param_user}:{cf.param_info.param_password}" if cf.param_info.param_user and cf.param_info.param_password else None
token = (
f"{cf.param_info.param_user}:{cf.param_info.param_password}"
if cf.param_info.param_user and cf.param_info.param_password
else None
)
if enable_milvus_client_api:
self.connection_wrap.connect(alias=DefaultConfig.DEFAULT_USING, uri=uri, token=token)
@@ -175,13 +187,11 @@ class TestcaseBase(Base):
self.client = MilvusClient(uri=uri, token=token)
else:
if token:
res, is_succ = self.connection_wrap.connect(alias=DefaultConfig.DEFAULT_USING,
uri=uri,
token=token,
secure=cf.param_info.param_secure)
res, is_succ = self.connection_wrap.connect(
alias=DefaultConfig.DEFAULT_USING, uri=uri, token=token, secure=cf.param_info.param_secure
)
else:
res, is_succ = self.connection_wrap.connect(alias=DefaultConfig.DEFAULT_USING,
uri=uri)
res, is_succ = self.connection_wrap.connect(alias=DefaultConfig.DEFAULT_USING, uri=uri)
self.client = MilvusClient(uri=uri, token=token)
server_version = utility.get_server_version()
@@ -194,15 +204,11 @@ class TestcaseBase(Base):
else:
uri = "http://" + cf.param_info.param_host + ":" + str(cf.param_info.param_port)
client = MilvusClient(
uri = uri,
token = cf.param_info.param_token
)
client = MilvusClient(uri=uri, token=cf.param_info.param_token)
res = client.run_analyzer(text, analyzer_params, with_detail=True, with_hash=True)
tokens = [r['token'] for r in res.tokens]
tokens = [r["token"] for r in res.tokens]
return tokens
# def init_async_milvus_client(self):
# uri = cf.param_info.param_uri or f"http://{cf.param_info.param_host}:{cf.param_info.param_port}"
# kwargs = {
@@ -213,16 +219,26 @@ class TestcaseBase(Base):
# }
# self.async_milvus_client_wrap.init_async_client(**kwargs)
def init_collection_wrap(self, name=None, schema=None, check_task=None, check_items=None,
enable_dynamic_field=False, with_json=True, **kwargs):
def init_collection_wrap(
self,
name=None,
schema=None,
check_task=None,
check_items=None,
enable_dynamic_field=False,
with_json=True,
**kwargs,
):
name = cf.gen_collection_name_by_testcase_name(2) if name is None else name
schema = cf.gen_default_collection_schema(enable_dynamic_field=enable_dynamic_field, with_json=with_json) \
if schema is None else schema
schema = (
cf.gen_default_collection_schema(enable_dynamic_field=enable_dynamic_field, with_json=with_json)
if schema is None
else schema
)
if not self.connection_wrap.has_connection(alias=DefaultConfig.DEFAULT_USING)[0]:
self._connect()
collection_w = ApiCollectionWrapper()
collection_w.init_collection(name=name, schema=schema, check_task=check_task,
check_items=check_items, **kwargs)
collection_w.init_collection(name=name, schema=schema, check_task=check_task, check_items=check_items, **kwargs)
self.tear_down_collection_names.append(name)
return collection_w
@@ -235,25 +251,36 @@ class TestcaseBase(Base):
assert collection_w.num_entities == ct.default_nb
return collection_w, df
def init_partition_wrap(self, collection_wrap=None, name=None, description=None,
check_task=None, check_items=None, **kwargs):
def init_partition_wrap(
self, collection_wrap=None, name=None, description=None, check_task=None, check_items=None, **kwargs
):
name = cf.gen_unique_str("partition_") if name is None else name
description = cf.gen_unique_str("partition_des_") if description is None else description
collection_wrap = self.init_collection_wrap() if collection_wrap is None else collection_wrap
partition_wrap = ApiPartitionWrapper()
partition_wrap.init_partition(collection_wrap.collection, name, description,
check_task=check_task, check_items=check_items,
**kwargs)
partition_wrap.init_partition(
collection_wrap.collection, name, description, check_task=check_task, check_items=check_items, **kwargs
)
return partition_wrap
def insert_data_general(self, prefix="test", insert_data=False, nb=ct.default_nb,
partition_num=0, is_binary=False, is_all_data_type=False,
auto_id=False, dim=ct.default_dim,
primary_field=ct.default_int64_field_name, is_flush=True, name=None,
enable_dynamic_field=False, with_json=True, **kwargs):
"""
"""
def insert_data_general(
self,
prefix="test",
insert_data=False,
nb=ct.default_nb,
partition_num=0,
is_binary=False,
is_all_data_type=False,
auto_id=False,
dim=ct.default_dim,
primary_field=ct.default_int64_field_name,
is_flush=True,
name=None,
enable_dynamic_field=False,
with_json=True,
**kwargs,
):
""" """
self._connect()
collection_name = cf.gen_unique_str(prefix)
if name is not None:
@@ -263,38 +290,70 @@ class TestcaseBase(Base):
insert_ids = []
time_stamp = 0
# 1 create collection
default_schema = cf.gen_default_collection_schema(auto_id=auto_id, dim=dim, primary_field=primary_field,
enable_dynamic_field=enable_dynamic_field,
with_json=with_json)
default_schema = cf.gen_default_collection_schema(
auto_id=auto_id,
dim=dim,
primary_field=primary_field,
enable_dynamic_field=enable_dynamic_field,
with_json=with_json,
)
if is_binary:
default_schema = cf.gen_default_binary_collection_schema(auto_id=auto_id, dim=dim,
primary_field=primary_field)
default_schema = cf.gen_default_binary_collection_schema(
auto_id=auto_id, dim=dim, primary_field=primary_field
)
if is_all_data_type:
default_schema = cf.gen_collection_schema_all_datatype(auto_id=auto_id, dim=dim,
primary_field=primary_field,
enable_dynamic_field=enable_dynamic_field,
with_json=with_json)
default_schema = cf.gen_collection_schema_all_datatype(
auto_id=auto_id,
dim=dim,
primary_field=primary_field,
enable_dynamic_field=enable_dynamic_field,
with_json=with_json,
)
log.info("insert_data_general: collection creation")
collection_w = self.init_collection_wrap(name=collection_name, schema=default_schema, **kwargs)
pre_entities = collection_w.num_entities
if insert_data:
collection_w, vectors, binary_raw_vectors, insert_ids, time_stamp = \
cf.insert_data(collection_w, nb, is_binary, is_all_data_type, auto_id=auto_id, dim=dim,
enable_dynamic_field=enable_dynamic_field,
with_json=with_json)
collection_w, vectors, binary_raw_vectors, insert_ids, time_stamp = cf.insert_data(
collection_w,
nb,
is_binary,
is_all_data_type,
auto_id=auto_id,
dim=dim,
enable_dynamic_field=enable_dynamic_field,
with_json=with_json,
)
if is_flush:
collection_w.flush()
assert collection_w.num_entities == nb + pre_entities
return collection_w, vectors, binary_raw_vectors, insert_ids, time_stamp
def init_collection_general(self, prefix="test", insert_data=False, nb=ct.default_nb,
partition_num=0, is_binary=False, is_all_data_type=False,
auto_id=False, dim=ct.default_dim, is_index=True,
primary_field=ct.default_int64_field_name, is_flush=True, name=None,
enable_dynamic_field=False, with_json=True, random_primary_key=False,
multiple_dim_array=[], is_partition_key=None, vector_data_type=DataType.FLOAT_VECTOR,
nullable_fields={}, default_value_fields={}, language=None, **kwargs):
def init_collection_general(
self,
prefix="test",
insert_data=False,
nb=ct.default_nb,
partition_num=0,
is_binary=False,
is_all_data_type=False,
auto_id=False,
dim=ct.default_dim,
is_index=True,
primary_field=ct.default_int64_field_name,
is_flush=True,
name=None,
enable_dynamic_field=False,
with_json=True,
random_primary_key=False,
multiple_dim_array=[],
is_partition_key=None,
vector_data_type=DataType.FLOAT_VECTOR,
nullable_fields={},
default_value_fields={},
language=None,
**kwargs,
):
"""
target: create specified collections
method: 1. create collections (binary/non-binary, default/all data type, auto_id or not)
@@ -323,33 +382,47 @@ class TestcaseBase(Base):
insert_ids = []
time_stamp = 0
# 1 create collection
default_schema = cf.gen_default_collection_schema(auto_id=auto_id, dim=dim, primary_field=primary_field,
enable_dynamic_field=enable_dynamic_field,
with_json=with_json, multiple_dim_array=multiple_dim_array,
is_partition_key=is_partition_key,
vector_data_type=vector_data_type,
nullable_fields=nullable_fields,
default_value_fields=default_value_fields)
default_schema = cf.gen_default_collection_schema(
auto_id=auto_id,
dim=dim,
primary_field=primary_field,
enable_dynamic_field=enable_dynamic_field,
with_json=with_json,
multiple_dim_array=multiple_dim_array,
is_partition_key=is_partition_key,
vector_data_type=vector_data_type,
nullable_fields=nullable_fields,
default_value_fields=default_value_fields,
)
if is_binary:
default_schema = cf.gen_default_binary_collection_schema(auto_id=auto_id, dim=dim,
primary_field=primary_field,
nullable_fields=nullable_fields,
default_value_fields=default_value_fields)
default_schema = cf.gen_default_binary_collection_schema(
auto_id=auto_id,
dim=dim,
primary_field=primary_field,
nullable_fields=nullable_fields,
default_value_fields=default_value_fields,
)
if vector_data_type == DataType.SPARSE_FLOAT_VECTOR:
default_schema = cf.gen_default_sparse_schema(auto_id=auto_id, primary_field=primary_field,
enable_dynamic_field=enable_dynamic_field,
with_json=with_json,
multiple_dim_array=multiple_dim_array,
nullable_fields=nullable_fields,
default_value_fields=default_value_fields)
default_schema = cf.gen_default_sparse_schema(
auto_id=auto_id,
primary_field=primary_field,
enable_dynamic_field=enable_dynamic_field,
with_json=with_json,
multiple_dim_array=multiple_dim_array,
nullable_fields=nullable_fields,
default_value_fields=default_value_fields,
)
if is_all_data_type:
default_schema = cf.gen_collection_schema_all_datatype(auto_id=auto_id, dim=dim,
primary_field=primary_field,
enable_dynamic_field=enable_dynamic_field,
with_json=with_json,
multiple_dim_array=multiple_dim_array,
nullable_fields=nullable_fields,
default_value_fields=default_value_fields)
default_schema = cf.gen_collection_schema_all_datatype(
auto_id=auto_id,
dim=dim,
primary_field=primary_field,
enable_dynamic_field=enable_dynamic_field,
with_json=with_json,
multiple_dim_array=multiple_dim_array,
nullable_fields=nullable_fields,
default_value_fields=default_value_fields,
)
log.info("init_collection_general: collection creation")
collection_w = self.init_collection_wrap(name=collection_name, schema=default_schema, **kwargs)
vector_name_list = cf.extract_vector_field_name_list(collection_w)
@@ -358,12 +431,22 @@ class TestcaseBase(Base):
cf.gen_partitions(collection_w, partition_num)
# 3 insert data if specified
if insert_data:
collection_w, vectors, binary_raw_vectors, insert_ids, time_stamp = \
cf.insert_data(collection_w, nb, is_binary, is_all_data_type, auto_id=auto_id,
dim=dim, enable_dynamic_field=enable_dynamic_field, with_json=with_json,
random_primary_key=random_primary_key, multiple_dim_array=multiple_dim_array,
primary_field=primary_field, vector_data_type=vector_data_type,
nullable_fields=nullable_fields, language=language)
collection_w, vectors, binary_raw_vectors, insert_ids, time_stamp = cf.insert_data(
collection_w,
nb,
is_binary,
is_all_data_type,
auto_id=auto_id,
dim=dim,
enable_dynamic_field=enable_dynamic_field,
with_json=with_json,
random_primary_key=random_primary_key,
multiple_dim_array=multiple_dim_array,
primary_field=primary_field,
vector_data_type=vector_data_type,
nullable_fields=nullable_fields,
language=language,
)
if is_flush:
assert collection_w.is_empty is False
assert collection_w.num_entities == nb
@@ -391,7 +474,7 @@ class TestcaseBase(Base):
return collection_w, vectors, binary_raw_vectors, insert_ids, time_stamp
def insert_entities_into_two_partitions_in_half(self, half, prefix='query'):
def insert_entities_into_two_partitions_in_half(self, half, prefix="query"):
"""
insert default entities into two partitions(partition_w and _default) in half(int64 and float fields values)
:param half: half of nb
@@ -413,8 +496,9 @@ class TestcaseBase(Base):
collection_w.load(partition_names=[partition_w.name, "_default"])
return collection_w, partition_w, df_partition, df_default
def collection_insert_multi_segments_one_shard(self, collection_prefix, num_of_segment=2, nb_of_segment=1,
is_dup=True):
def collection_insert_multi_segments_one_shard(
self, collection_prefix, num_of_segment=2, nb_of_segment=1, is_dup=True
):
"""
init collection with one shard, insert data into two segments on one shard (they can be merged)
:param collection_prefix: collection name prefix
@@ -437,9 +521,9 @@ class TestcaseBase(Base):
if not self.connection_wrap.has_connection(alias=DefaultConfig.DEFAULT_USING)[0]:
self._connect()
utility_w = ApiUtilityWrapper()
res, check_result = utility_w.create_resource_group(name=name, using=using, timeout=timeout,
check_task=check_task,
check_items=check_items, **kwargs)
res, check_result = utility_w.create_resource_group(
name=name, using=using, timeout=timeout, check_task=check_task, check_items=check_items, **kwargs
)
if res is None and check_result:
self.resource_group_list.append(name)
return res, check_result
@@ -470,23 +554,25 @@ class TestcaseBase(Base):
self.utility_wrap.create_role()
# grant privilege to the role
self.utility_wrap.role_grant(object=privilege_object, object_name=object_name, privilege=privilege,
db_name=db_name)
self.utility_wrap.role_grant(
object=privilege_object, object_name=object_name, privilege=privilege, db_name=db_name
)
# bind the role to the user
self.utility_wrap.role_add_user(tmp_user)
return tmp_user, tmp_pwd, tmp_role
def build_multi_index(self, index_params: Dict[str, IndexPrams], collection_obj: ApiCollectionWrapper = None):
def build_multi_index(self, index_params: dict[str, IndexPrams], collection_obj: ApiCollectionWrapper = None):
collection_obj = collection_obj or self.collection_wrap
for k, v in index_params.items():
collection_obj.create_index(field_name=k, index_params=v.to_dict, index_name=k)
log.info(f"[TestcaseBase] Build all indexes done: {list(index_params.keys())}")
return collection_obj
def drop_multi_index(self, index_names: List[str], collection_obj: ApiCollectionWrapper = None,
check_task=None, check_items=None):
def drop_multi_index(
self, index_names: list[str], collection_obj: ApiCollectionWrapper = None, check_task=None, check_items=None
):
collection_obj = collection_obj or self.collection_wrap
for n in index_names:
collection_obj.drop_index(index_name=n, check_task=check_task, check_items=check_items)
@@ -496,15 +582,23 @@ class TestcaseBase(Base):
def show_indexes(self, collection_obj: ApiCollectionWrapper = None):
collection_obj = collection_obj or self.collection_wrap
indexes = {n.field_name: n.params for n in self.collection_wrap.indexes}
log.info("[TestcaseBase] Collection: `{0}` index: {1}".format(collection_obj.name, indexes))
log.info(f"[TestcaseBase] Collection: `{collection_obj.name}` index: {indexes}")
return indexes
""" Property """
@property
def all_scalar_fields(self):
dtypes = [DataType.INT8, DataType.INT16, DataType.INT32, DataType.INT64, DataType.VARCHAR, DataType.BOOL,
DataType.FLOAT, DataType.DOUBLE]
dtypes = [
DataType.INT8,
DataType.INT16,
DataType.INT32,
DataType.INT64,
DataType.VARCHAR,
DataType.BOOL,
DataType.FLOAT,
DataType.DOUBLE,
]
dtype_names = [f"{n.name}" for n in dtypes] + [f"ARRAY_{n.name}" for n in dtypes] + [DataType.JSON.name]
return dtype_names
@@ -546,8 +640,8 @@ class TestCaseClassBase(TestcaseBase):
def setup_method(self, method):
log.info(" setup ".center(80, "*"))
log.info("[setup_method] Start setup test case %s." % method.__name__)
log.info(f"[setup_method] Start setup test case {method.__name__}.")
def teardown_method(self, method):
log.info(" teardown ".center(80, "*"))
log.info("[teardown_method] Start teardown test case %s..." % method.__name__)
log.info(f"[teardown_method] Start teardown test case {method.__name__}...")
tests/python_client/base/client_v2_base.py
+70 -30
View File
@@ -1618,9 +1618,11 @@ class TestMilvusClientV2Base(Base):
def create_snapshot(
self,
client,
collection_name,
snapshot_name,
collection_name,
db_name="",
description="",
compaction_protection_seconds=0,
timeout=None,
check_task=None,
check_items=None,
@@ -1628,16 +1630,19 @@ class TestMilvusClientV2Base(Base):
):
"""Create a snapshot for a collection.
Note: wrapper keeps ``collection_name`` before ``snapshot_name`` for test
readability, but the SDK signature is ``create_snapshot(snapshot_name,
collection_name, ...)`` so we forward positionals in SDK order.
Signature follows ``MilvusClient.create_snapshot``; check arguments are
wrapper-only and appended after the SDK arguments.
"""
timeout = TIMEOUT if timeout is None else timeout
kwargs.update({"timeout": timeout})
func_name = sys._getframe().f_code.co_name
res, check = api_request(
[client.create_snapshot, snapshot_name, collection_name], description=description, **kwargs
[client.create_snapshot, snapshot_name, collection_name],
db_name=db_name,
description=description,
compaction_protection_seconds=compaction_protection_seconds,
**kwargs,
)
check_result = ResponseChecker(
res,
@@ -1647,19 +1652,30 @@ class TestMilvusClientV2Base(Base):
check,
snapshot_name=snapshot_name,
collection_name=collection_name,
db_name=db_name,
description=description,
compaction_protection_seconds=compaction_protection_seconds,
**kwargs,
).run()
return res, check_result
@trace()
def drop_snapshot(
self, client, snapshot_name, collection_name, timeout=None, check_task=None, check_items=None, **kwargs
self,
client,
snapshot_name,
collection_name,
db_name="",
timeout=None,
check_task=None,
check_items=None,
**kwargs,
):
timeout = TIMEOUT if timeout is None else timeout
kwargs.update({"timeout": timeout})
func_name = sys._getframe().f_code.co_name
res, check = api_request([client.drop_snapshot, snapshot_name, collection_name], **kwargs)
res, check = api_request([client.drop_snapshot, snapshot_name, collection_name], db_name=db_name, **kwargs)
check_result = ResponseChecker(
res,
func_name,
@@ -1668,31 +1684,42 @@ class TestMilvusClientV2Base(Base):
check,
snapshot_name=snapshot_name,
collection_name=collection_name,
db_name=db_name,
**kwargs,
).run()
return res, check_result
@trace()
def list_snapshots(self, client, collection_name="", timeout=None, check_task=None, check_items=None, **kwargs):
timeout = TIMEOUT if timeout is None else timeout
kwargs.update({"timeout": timeout})
func_name = sys._getframe().f_code.co_name
res, check = api_request([client.list_snapshots, collection_name], **kwargs)
check_result = ResponseChecker(
res, func_name, check_task, check_items, check, collection_name=collection_name, **kwargs
).run()
return res, check_result
@trace()
def describe_snapshot(
self, client, snapshot_name, collection_name, timeout=None, check_task=None, check_items=None, **kwargs
def list_snapshots(
self, client, collection_name="", db_name="", timeout=None, check_task=None, check_items=None, **kwargs
):
timeout = TIMEOUT if timeout is None else timeout
kwargs.update({"timeout": timeout})
func_name = sys._getframe().f_code.co_name
res, check = api_request([client.describe_snapshot, snapshot_name, collection_name], **kwargs)
res, check = api_request([client.list_snapshots, collection_name], db_name=db_name, **kwargs)
check_result = ResponseChecker(
res, func_name, check_task, check_items, check, collection_name=collection_name, db_name=db_name, **kwargs
).run()
return res, check_result
@trace()
def describe_snapshot(
self,
client,
snapshot_name,
collection_name,
db_name="",
timeout=None,
check_task=None,
check_items=None,
**kwargs,
):
timeout = TIMEOUT if timeout is None else timeout
kwargs.update({"timeout": timeout})
func_name = sys._getframe().f_code.co_name
res, check = api_request([client.describe_snapshot, snapshot_name, collection_name], db_name=db_name, **kwargs)
check_result = ResponseChecker(
res,
func_name,
@@ -1701,6 +1728,7 @@ class TestMilvusClientV2Base(Base):
check,
snapshot_name=snapshot_name,
collection_name=collection_name,
db_name=db_name,
**kwargs,
).run()
return res, check_result
@@ -1710,8 +1738,10 @@ class TestMilvusClientV2Base(Base):
self,
client,
snapshot_name,
source_collection_name,
target_collection_name,
source_collection_name="",
source_db_name="",
target_db_name="",
timeout=None,
check_task=None,
check_items=None,
@@ -1719,15 +1749,18 @@ class TestMilvusClientV2Base(Base):
):
"""Restore a snapshot into a new collection.
SDK positional order is ``(snapshot_name, source_collection_name,
target_collection_name)`` and ``source_collection_name`` is now required.
Signature follows ``MilvusClient.restore_snapshot``; check arguments are
wrapper-only and appended after the SDK arguments.
"""
timeout = TIMEOUT if timeout is None else timeout
kwargs.update({"timeout": timeout})
func_name = sys._getframe().f_code.co_name
res, check = api_request(
[client.restore_snapshot, snapshot_name, source_collection_name, target_collection_name], **kwargs
[client.restore_snapshot, snapshot_name, source_collection_name, target_collection_name],
source_db_name=source_db_name,
target_db_name=target_db_name,
**kwargs,
)
check_result = ResponseChecker(
res,
@@ -1738,6 +1771,8 @@ class TestMilvusClientV2Base(Base):
snapshot_name=snapshot_name,
target_collection_name=target_collection_name,
source_collection_name=source_collection_name,
source_db_name=source_db_name,
target_db_name=target_db_name,
**kwargs,
).run()
return res, check_result
@@ -1754,15 +1789,15 @@ class TestMilvusClientV2Base(Base):
@trace()
def list_restore_snapshot_jobs(
self, client, collection_name="", timeout=None, check_task=None, check_items=None, **kwargs
self, client, collection_name="", db_name="", timeout=None, check_task=None, check_items=None, **kwargs
):
timeout = TIMEOUT if timeout is None else timeout
kwargs.update({"timeout": timeout})
func_name = sys._getframe().f_code.co_name
res, check = api_request([client.list_restore_snapshot_jobs, collection_name], **kwargs)
res, check = api_request([client.list_restore_snapshot_jobs, collection_name], db_name=db_name, **kwargs)
check_result = ResponseChecker(
res, func_name, check_task, check_items, check, collection_name=collection_name, **kwargs
res, func_name, check_task, check_items, check, collection_name=collection_name, db_name=db_name, **kwargs
).run()
return res, check_result
@@ -1772,6 +1807,7 @@ class TestMilvusClientV2Base(Base):
client,
snapshot_name,
collection_name,
db_name="",
ttl_seconds=0,
timeout=None,
check_task=None,
@@ -1788,7 +1824,10 @@ class TestMilvusClientV2Base(Base):
func_name = sys._getframe().f_code.co_name
res, check = api_request(
[client.pin_snapshot_data, snapshot_name, collection_name], ttl_seconds=ttl_seconds, **kwargs
[client.pin_snapshot_data, snapshot_name, collection_name],
db_name=db_name,
ttl_seconds=ttl_seconds,
**kwargs,
)
check_result = ResponseChecker(
res,
@@ -1798,6 +1837,7 @@ class TestMilvusClientV2Base(Base):
check,
snapshot_name=snapshot_name,
collection_name=collection_name,
db_name=db_name,
ttl_seconds=ttl_seconds,
**kwargs,
).run()
tests/python_client/milvus_client/test_milvus_client_snapshot.py
+701 -744
View File
File diff suppressed because it is too large Load Diff
tests/python_client/requirements.txt
+2 -2
View File
@@ -23,8 +23,8 @@ pytest-sugar==0.9.5
pytest-random-order
# pymilvus
pymilvus==2.7.0rc214
pymilvus[bulk_writer]==2.7.0rc214
pymilvus==3.1.0rc8
pymilvus[bulk_writer]==3.1.0rc8
# for protobuf
protobuf>=5.29.5
tests/python_client/testcases/test_bulk_insert.py
+2
View File
@@ -415,6 +415,7 @@ class TestBulkInsertNullableVector(TestcaseBaseBulkInsert):
null_ids = []
non_null_ids = []
writer_config = {"nullkey": "null"} if file_type == BulkFileType.CSV else None
with RemoteBulkWriter(
schema=schema,
remote_path="bulk_data",
@@ -425,6 +426,7 @@ class TestBulkInsertNullableVector(TestcaseBaseBulkInsert):
secret_key="minioadmin",
),
file_type=file_type,
config=writer_config,
chunk_size=2048 if file_type == BulkFileType.PARQUET else 1024 * 1024 * 1024,
) as remote_writer:
for i in range(entities):