mirror of
https://gitee.com/milvus-io/milvus.git
synced 2026-02-02 01:06:41 +08:00
6f6a1bfd82
* [skip ci]fix test case bug Signed-off-by: godchen0212 <qingxiang.chen@zilliz.com> * cancel change config Signed-off-by: godchen0212 <qingxiang.chen@zilliz.com> * fix Cmd req bug and change config test case Signed-off-by: godchen0212 <qingxiang.chen@zilliz.com> * format code Signed-off-by: godchen0212 <qingxiang.chen@zilliz.com> * change test case in config and throw inmutable params problem Signed-off-by: godchen0212 <qingxiang.chen@zilliz.com> * format code Signed-off-by: godchen0212 <qingxiang.chen@zilliz.com>
86 lines
2.8 KiB
Python
86 lines
2.8 KiB
Python
import pdb
|
|
import pytest
|
|
import logging
|
|
import itertools
|
|
from time import sleep
|
|
from multiprocessing import Process
|
|
from utils import *
|
|
|
|
collection_id = "load_collection"
|
|
nb = 6000
|
|
default_fields = gen_default_fields()
|
|
entities = gen_entities(nb)
|
|
field_name = "float_vector"
|
|
|
|
|
|
class TestLoadCollection:
|
|
|
|
"""
|
|
******************************************************************
|
|
The following cases are used to test `load_collection` function
|
|
******************************************************************
|
|
"""
|
|
@pytest.fixture(
|
|
scope="function",
|
|
params=gen_simple_index()
|
|
)
|
|
def get_simple_index(self, request, connect):
|
|
if str(connect._cmd("mode")) == "CPU":
|
|
if request.param["index_type"] in index_cpu_not_support():
|
|
pytest.skip("sq8h not support in cpu mode")
|
|
return request.param
|
|
|
|
def test_load_collection_after_index(self, connect, collection, get_simple_index):
|
|
'''
|
|
target: test load collection, after index created
|
|
method: insert and create index, load collection with correct params
|
|
expected: describe raise exception
|
|
'''
|
|
connect.insert(collection, entities)
|
|
connect.flush([collection])
|
|
logging.getLogger().info(get_simple_index)
|
|
connect.create_index(collection, field_name, get_simple_index)
|
|
connect.load_collection(collection)
|
|
|
|
def load_empty_collection(self, connect, collection):
|
|
'''
|
|
target: test load collection
|
|
method: no entities in collection, load collection with correct params
|
|
expected: load success
|
|
'''
|
|
connect.load_collection(collection)
|
|
|
|
@pytest.mark.level(1)
|
|
def test_load_collection_dis_connect(self, dis_connect, collection):
|
|
'''
|
|
target: test load collection, without connection
|
|
method: load collection with correct params, with a disconnected instance
|
|
expected: load raise exception
|
|
'''
|
|
with pytest.raises(Exception) as e:
|
|
dis_connect.load_collection(collection)
|
|
|
|
@pytest.mark.level(2)
|
|
def test_load_collection_not_existed(self, connect, collection):
|
|
collection_name = gen_unique_str(collection_id)
|
|
with pytest.raises(Exception) as e:
|
|
connect.load_collection(collection_name)
|
|
|
|
|
|
class TestLoadCollectionInvalid(object):
|
|
"""
|
|
Test load collection with invalid params
|
|
"""
|
|
@pytest.fixture(
|
|
scope="function",
|
|
params=gen_invalid_strs()
|
|
)
|
|
def get_collection_name(self, request):
|
|
yield request.param
|
|
|
|
@pytest.mark.level(2)
|
|
def test_load_collection_with_invalid_collectionname(self, connect, get_collection_name):
|
|
collection_name = get_collection_name
|
|
with pytest.raises(Exception) as e:
|
|
connect.has_collection(collection_name)
|