milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[skip e2e]Update utility for alias (#13661) 

Signed-off-by: zhuwenxing <wenxing.zhu@zilliz.com>
pull/13670/head
zhuwenxing 2021-12-17 20:46:15 +08:00 committed by GitHub
parent 1ca08d9138
commit bc53995d83
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 32 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
tests/python_client/base/utility_wrapper.py
View File

@ -121,6 +121,38 @@ class ApiUtilityWrapper:
timeout=timeout, using=using).run() timeout=timeout, using=using).run()
return res, check_result return res, check_result
def create_alias(self, collection_name, alias, timeout=None, using="default", check_task=None, check_items=None):
timeout = TIMEOUT if timeout is None else timeout
func_name = sys._getframe().f_code.co_name
res, is_succ = api_request([self.ut.create_alias, collection_name, alias, timeout, using])
check_result = ResponseChecker(res, func_name, check_task, check_items, is_succ,
timeout=timeout, using=using).run()
return res, check_result
def drop_alias(self, alias, timeout=None, using="default", check_task=None, check_items=None):
timeout = TIMEOUT if timeout is None else timeout
func_name = sys._getframe().f_code.co_name
res, is_succ = api_request([self.ut.drop_alias, alias, timeout, using])
check_result = ResponseChecker(res, func_name, check_task, check_items, is_succ,
timeout=timeout, using=using).run()
return res, check_result
def alter_alias(self, collection_name, alias, timeout=None, using="default", check_task=None, check_items=None):
timeout = TIMEOUT if timeout is None else timeout
func_name = sys._getframe().f_code.co_name
res, is_succ = api_request([self.ut.alter_alias, collection_name, alias, timeout, using])
check_result = ResponseChecker(res, func_name, check_task, check_items, is_succ,
timeout=timeout, using=using).run()
return res, check_result
def list_aliases(self, collection_name, timeout=None, using="default", check_task=None, check_items=None):
timeout = TIMEOUT if timeout is None else timeout
func_name = sys._getframe().f_code.co_name
res, is_succ = api_request([self.ut.list_aliases, collection_name, timeout, using])
check_result = ResponseChecker(res, func_name, check_task, check_items, is_succ,
timeout=timeout, using=using).run()
return res, check_result
def mkts_from_datetime(self, d_time=None, milliseconds=0., delta=None): def mkts_from_datetime(self, d_time=None, milliseconds=0., delta=None):
d_time = datetime.now() if d_time is None else d_time d_time = datetime.now() if d_time is None else d_time
res, _ = api_request([self.ut.mkts_from_datetime, d_time, milliseconds, delta]) res, _ = api_request([self.ut.mkts_from_datetime, d_time, milliseconds, delta])