test_utility.py

import threading
import time

import pytest
from pymilvus import DefaultConfig
from pymilvus.exceptions import MilvusException
from base.client_base import TestcaseBase
from base.utility_wrapper import ApiUtilityWrapper
from utils.util_log import test_log as log
from common import common_func as cf
from common import common_type as ct
from common.common_type import CaseLabel, CheckTasks
from common.milvus_sys import MilvusSys
from pymilvus.grpc_gen.common_pb2 import SegmentState
import random

prefix = "utility"
default_schema = cf.gen_default_collection_schema()
default_int64_field_name = ct.default_int64_field_name
default_field_name = ct.default_float_vec_field_name
default_index_params = {"index_type": "IVF_SQ8", "metric_type": "L2", "params": {"nlist": 64}}
default_dim = ct.default_dim
default_nb = ct.default_nb
num_loaded_entities = "num_loaded_entities"
num_total_entities = "num_total_entities"
loading_progress = "loading_progress"
num_loaded_partitions = "num_loaded_partitions"
not_loaded_partitions = "not_loaded_partitions"
exp_name = "name"


class TestUtilityParams(TestcaseBase):
    """ Test case of index interface """

    @pytest.fixture(scope="function", params=ct.get_invalid_strs)
    def get_invalid_metric_type(self, request):
        if request.param == [] or request.param == "":
            pytest.skip("metric empty is valid for distance calculation")
        if isinstance(request.param, str):
            pytest.skip("string is valid type for metric")
        yield request.param

    @pytest.fixture(scope="function", params=ct.get_invalid_strs)
    def get_invalid_metric_value(self, request):
        if request.param == [] or request.param == "":
            pytest.skip("metric empty is valid for distance calculation")
        if not isinstance(request.param, str):
            pytest.skip("Skip invalid type for metric")
        yield request.param

    @pytest.fixture(scope="function", params=["JACCARD", "Superstructure", "Substructure"])
    def get_not_support_metric(self, request):
        yield request.param

    @pytest.fixture(scope="function", params=["metric_type", "metric"])
    def get_support_metric_field(self, request):
        yield request.param

    @pytest.fixture(scope="function", params=ct.get_invalid_strs)
    def get_invalid_partition_names(self, request):
        if isinstance(request.param, list):
            if len(request.param) == 0:
                pytest.skip("empty is valid for partition")
        if request.param is None:
            pytest.skip("None is valid for partition")
        yield request.param

    """
    ******************************************************************
    #  The followings are invalid cases
    ******************************************************************
    """

    @pytest.mark.tags(CaseLabel.L2)
    def test_has_collection_name_invalid(self, get_invalid_collection_name):
        """
        target: test has_collection with error collection name
        method: input invalid name
        expected: raise exception
        """
        self._connect()
        c_name = get_invalid_collection_name
        if isinstance(c_name, str) and c_name:
            self.utility_wrap.has_collection(
                c_name,
                check_task=CheckTasks.err_res,
                check_items={ct.err_code: 1, ct.err_msg: "Invalid collection name"})
        # elif not isinstance(c_name, str): self.utility_wrap.has_collection(c_name, check_task=CheckTasks.err_res,
        # check_items={ct.err_code: 1, ct.err_msg: "illegal"})

    @pytest.mark.tags(CaseLabel.L2)
    def test_has_partition_collection_name_invalid(self, get_invalid_collection_name):
        """
        target: test has_partition with error collection name
        method: input invalid name
        expected: raise exception
        """
        self._connect()
        c_name = get_invalid_collection_name
        p_name = cf.gen_unique_str(prefix)
        if isinstance(c_name, str) and c_name:
            self.utility_wrap.has_partition(
                c_name, p_name,
                check_task=CheckTasks.err_res,
                check_items={ct.err_code: 1, ct.err_msg: "Invalid"})

    @pytest.mark.tags(CaseLabel.L2)
    def test_has_partition_name_invalid(self, get_invalid_partition_name):
        """
        target: test has_partition with error partition name
        method: input invalid name
        expected: raise exception
        """
        self._connect()
        ut = ApiUtilityWrapper()
        c_name = cf.gen_unique_str(prefix)
        p_name = get_invalid_partition_name
        if isinstance(p_name, str) and p_name:
            ex, _ = ut.has_partition(
                c_name, p_name,
                check_task=CheckTasks.err_res,
                check_items={ct.err_code: 1, ct.err_msg: "Invalid"})

    @pytest.mark.tags(CaseLabel.L2)
    def test_drop_collection_name_invalid(self, get_invalid_collection_name):
        self._connect()
        error = f'`collection_name` value {get_invalid_collection_name} is illegal'
        self.utility_wrap.drop_collection(get_invalid_collection_name, check_task=CheckTasks.err_res,
                                          check_items={ct.err_code: 1, ct.err_msg: error})

    # TODO: enable
    @pytest.mark.tags(CaseLabel.L2)
    def test_list_collections_using_invalid(self):
        """
        target: test list_collections with invalid using
        method: input invalid name
        expected: raise exception
        """
        self._connect()
        using = "empty"
        ut = ApiUtilityWrapper()
        ex, _ = ut.list_collections(using=using, check_task=CheckTasks.err_res,
                                    check_items={ct.err_code: 0, ct.err_msg: "should create connect"})

    @pytest.mark.tags(CaseLabel.L1)
    def test_index_process_invalid_name(self, get_invalid_collection_name):
        """
        target: test building_process
        method: input invalid name
        expected: raise exception
        """
        pass
        # self._connect() c_name = get_invalid_collection_name ut = ApiUtilityWrapper() if isinstance(c_name,
        # str) and c_name: ex, _ = ut.index_building_progress(c_name, check_items={ct.err_code: 1, ct.err_msg:
        # "Invalid collection name"})

    # TODO: not support index name
    @pytest.mark.tags(CaseLabel.L1)
    def _test_index_process_invalid_index_name(self, get_invalid_index_name):
        """
        target: test building_process
        method: input invalid index name
        expected: raise exception
        """
        self._connect()
        c_name = cf.gen_unique_str(prefix)
        index_name = get_invalid_index_name
        ut = ApiUtilityWrapper()
        ex, _ = ut.index_building_progress(c_name, index_name)
        log.error(str(ex))
        assert "invalid" or "illegal" in str(ex)

    @pytest.mark.tags(CaseLabel.L2)
    def test_wait_index_invalid_name(self, get_invalid_collection_name):
        """
        target: test wait_index
        method: input invalid name
        expected: raise exception
        """
        pass
        # self._connect()
        # c_name = get_invalid_collection_name
        # ut = ApiUtilityWrapper()
        # if isinstance(c_name, str) and c_name:
        #     ex, _ = ut.wait_for_index_building_complete(c_name,
        #                                                 check_items={ct.err_code: 1,
        #                                                              ct.err_msg: "Invalid collection name"})

    @pytest.mark.tags(CaseLabel.L1)
    def _test_wait_index_invalid_index_name(self, get_invalid_index_name):
        """
        target: test wait_index
        method: input invalid index name
        expected: raise exception
        """
        self._connect()
        c_name = cf.gen_unique_str(prefix)
        index_name = get_invalid_index_name
        ut = ApiUtilityWrapper()
        ex, _ = ut.wait_for_index_building_complete(c_name, index_name)
        log.error(str(ex))
        assert "invalid" or "illegal" in str(ex)

    @pytest.mark.tags(CaseLabel.L2)
    @pytest.mark.parametrize("invalid_c_name", ["12-s", "12 s", "(mn)", "中文", "%$#"])
    def test_loading_progress_invalid_collection_name(self, invalid_c_name):
        """
        target: test loading progress with invalid collection name
        method: input invalid collection name
        expected: raise exception
        """
        self._connect()
        c_name = cf.gen_unique_str(prefix)
        df = cf.gen_default_dataframe_data()
        self.collection_wrap.construct_from_dataframe(c_name, df, primary_field=ct.default_int64_field_name)
        self.collection_wrap.load()
        error = {ct.err_code: 1, ct.err_msg: "Invalid collection name: {}".format(invalid_c_name)}
        self.utility_wrap.loading_progress(invalid_c_name, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L2)
    def test_loading_progress_not_existed_collection_name(self):
        """
        target: test loading progress with invalid collection name
        method: input invalid collection name
        expected: raise exception
        """
        self._connect()
        c_name = cf.gen_unique_str(prefix)
        df = cf.gen_default_dataframe_data()
        self.collection_wrap.construct_from_dataframe(c_name, df, primary_field=ct.default_int64_field_name)
        self.collection_wrap.load()
        error = {ct.err_code: 1, ct.err_msg: "describe collection failed: can't find collection"}
        self.utility_wrap.loading_progress("not_existed_name", check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L2)
    def test_loading_progress_invalid_partition_names(self, get_invalid_partition_names):
        """
        target: test loading progress with invalid partition names
        method: input invalid partition names
        expected: raise an exception
        """
        collection_w = self.init_collection_general(prefix)[0]
        partition_names = get_invalid_partition_names
        err_msg = {ct.err_code: 0, ct.err_msg: "`partition_name_array` value {} is illegal".format(partition_names)}
        collection_w.load()
        self.utility_wrap.loading_progress(collection_w.name, partition_names,
                                           check_task=CheckTasks.err_res, check_items=err_msg)

    @pytest.mark.tags(CaseLabel.L1)
    @pytest.mark.parametrize("partition_names", [[ct.default_tag], [ct.default_partition_name, ct.default_tag]])
    def test_loading_progress_not_existed_partitions(self, partition_names):
        """
        target: test loading progress with not existed partitions
        method: input all or part not existed partition names
        expected: raise exception
        """
        collection_w = self.init_collection_general(prefix)[0]
        log.debug(collection_w.num_entities)
        collection_w.load()
        err_msg = {ct.err_code: -1, ct.err_msg: f"Partitions not exist: [{ct.default_tag}]"}
        self.utility_wrap.loading_progress(collection_w.name, partition_names,
                                           check_task=CheckTasks.err_res, check_items=err_msg)

    @pytest.mark.tags(CaseLabel.L2)
    def test_wait_for_loading_collection_not_existed(self):
        """
        target: test wait for loading
        method: input collection not created before
        expected: raise exception
        """
        self._connect()
        c_name = cf.gen_unique_str(prefix)
        self.utility_wrap.wait_for_loading_complete(
            c_name,
            check_task=CheckTasks.err_res,
            check_items={ct.err_code: 1, ct.err_msg: "can't find collection"})

    @pytest.mark.tags(CaseLabel.L2)
    def test_wait_for_loading_partition_not_existed(self):
        """
        target: test wait for loading
        method: input partition not created before
        expected: raise exception
        """
        self._connect()
        collection_w = self.init_collection_wrap()
        self.utility_wrap.wait_for_loading_complete(
            collection_w.name, partition_names=[ct.default_tag],
            check_task=CheckTasks.err_res,
            check_items={ct.err_code: 1, ct.err_msg: f'partitionID of partitionName:{ct.default_tag} can not be find'})

    @pytest.mark.tags(CaseLabel.L2)
    def test_drop_collection_not_existed(self):
        """
        target: test drop an not existed collection
        method: drop a not created collection
        expected: raise exception
        """
        self._connect()
        c_name = cf.gen_unique_str(prefix)

        # error = {ct.err_code: 1, ct.err_msg: f"DescribeCollection failed: can't find collection: {c_name}"}
        # self.utility_wrap.drop_collection(c_name, check_task=CheckTasks.err_res, check_items=error)

        # @longjiquan: dropping collection should be idempotent.
        self.utility_wrap.drop_collection(c_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_left_vector_invalid_type(self, get_invalid_vector_dict):
        """
        target: test calculated distance with invalid vectors
        method: input invalid vectors type
        expected: raise exception
        """
        self._connect()
        invalid_vector = get_invalid_vector_dict
        if not isinstance(invalid_vector, dict):
            self.utility_wrap.calc_distance(invalid_vector, invalid_vector,
                                            check_task=CheckTasks.err_res,
                                            check_items={"err_code": 1,
                                                         "err_msg": "vectors_left value {} "
                                                                    "is illegal".format(invalid_vector)})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_left_vector_invalid_value(self, get_invalid_vector_dict):
        """
        target: test calculated distance with invalid vectors
        method: input invalid vectors value
        expected: raise exception
        """
        self._connect()
        invalid_vector = get_invalid_vector_dict
        if isinstance(invalid_vector, dict):
            self.utility_wrap.calc_distance(invalid_vector, invalid_vector,
                                            check_task=CheckTasks.err_res,
                                            check_items={"err_code": 1,
                                                         "err_msg": "vectors_left value {} "
                                                                    "is illegal".format(invalid_vector)})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_right_vector_invalid_type(self, get_invalid_vector_dict):
        """
        target: test calculated distance with invalid vectors
        method: input invalid vectors type
        expected: raise exception
        """
        self._connect()
        invalid_vector = get_invalid_vector_dict
        vector = cf.gen_vectors(default_nb, default_dim)
        op_l = {"float_vectors": vector}
        if not isinstance(invalid_vector, dict):
            self.utility_wrap.calc_distance(op_l, invalid_vector,
                                            check_task=CheckTasks.err_res,
                                            check_items={"err_code": 1,
                                                         "err_msg": "vectors_right value {} "
                                                                    "is illegal".format(invalid_vector)})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_right_vector_invalid_value(self, get_invalid_vector_dict):
        """
        target: test calculated distance with invalid vectors
        method: input invalid vectors value
        expected: raise exception
        """
        self._connect()
        invalid_vector = get_invalid_vector_dict
        vector = cf.gen_vectors(default_nb, default_dim)
        op_l = {"float_vectors": vector}
        if isinstance(invalid_vector, dict):
            self.utility_wrap.calc_distance(op_l, invalid_vector,
                                            check_task=CheckTasks.err_res,
                                            check_items={"err_code": 1,
                                                         "err_msg": "vectors_right value {} "
                                                                    "is illegal".format(invalid_vector)})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_invalid_metric_type(self, get_support_metric_field, get_invalid_metric_type):
        """
        target: test calculated distance with invalid metric
        method: input invalid metric
        expected: raise exception
        """
        self._connect()
        vectors_l = cf.gen_vectors(default_nb, default_dim)
        vectors_r = cf.gen_vectors(default_nb, default_dim)
        op_l = {"float_vectors": vectors_l}
        op_r = {"float_vectors": vectors_r}
        metric_field = get_support_metric_field
        metric = get_invalid_metric_type
        params = {metric_field: metric}
        self.utility_wrap.calc_distance(op_l, op_r, params,
                                        check_task=CheckTasks.err_res,
                                        check_items={"err_code": 1,
                                                     "err_msg": "params value {{'metric': {}}} "
                                                                "is illegal".format(metric)})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_invalid_metric_value(self, get_support_metric_field, get_invalid_metric_value):
        """
        target: test calculated distance with invalid metric
        method: input invalid metric
        expected: raise exception
        """
        self._connect()
        vectors_l = cf.gen_vectors(default_nb, default_dim)
        vectors_r = cf.gen_vectors(default_nb, default_dim)
        op_l = {"float_vectors": vectors_l}
        op_r = {"float_vectors": vectors_r}
        metric_field = get_support_metric_field
        metric = get_invalid_metric_value
        params = {metric_field: metric}
        self.utility_wrap.calc_distance(op_l, op_r, params,
                                        check_task=CheckTasks.err_res,
                                        check_items={"err_code": 1,
                                                     "err_msg": "{} metric type is invalid for "
                                                                "float vector".format(metric)})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_not_support_metric(self, get_support_metric_field, get_not_support_metric):
        """
        target: test calculated distance with invalid metric
        method: input invalid metric
        expected: raise exception
        """
        self._connect()
        vectors_l = cf.gen_vectors(default_nb, default_dim)
        vectors_r = cf.gen_vectors(default_nb, default_dim)
        op_l = {"float_vectors": vectors_l}
        op_r = {"float_vectors": vectors_r}
        metric_field = get_support_metric_field
        metric = get_not_support_metric
        params = {metric_field: metric}
        self.utility_wrap.calc_distance(op_l, op_r, params,
                                        check_task=CheckTasks.err_res,
                                        check_items={"err_code": 1,
                                                     "err_msg": "{} metric type is invalid for "
                                                                "float vector".format(metric)})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_invalid_using(self, get_support_metric_field):
        """
        target: test calculated distance with invalid using
        method: input invalid using
        expected: raise exception
        """
        self._connect()
        vectors_l = cf.gen_vectors(default_nb, default_dim)
        vectors_r = cf.gen_vectors(default_nb, default_dim)
        op_l = {"float_vectors": vectors_l}
        op_r = {"float_vectors": vectors_r}
        metric_field = get_support_metric_field
        params = {metric_field: "L2", "sqrt": True}
        using = "empty"
        self.utility_wrap.calc_distance(op_l, op_r, params, using=using,
                                        check_task=CheckTasks.err_res,
                                        check_items={"err_code": 1,
                                                     "err_msg": "should create connect"})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_not_match_dim(self):
        """
        target: test calculated distance with invalid vectors
        method: input invalid vectors type and value
        expected: raise exception
        """
        self._connect()
        dim = 129
        vector_l = cf.gen_vectors(default_nb, default_dim)
        vector_r = cf.gen_vectors(default_nb, dim)
        op_l = {"float_vectors": vector_l}
        op_r = {"float_vectors": vector_r}
        self.utility_wrap.calc_distance(op_l, op_r,
                                        check_task=CheckTasks.err_res,
                                        check_items={"err_code": 1,
                                                     "err_msg": "Cannot calculate distance between "
                                                                "vectors with different dimension"})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_collection_before_load(self, get_support_metric_field):
        """
        target: test calculated distance when entities is not ready
        method: calculate distance before load
        expected: raise exception
        """
        self._connect()
        nb = 10
        collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix, True, nb,
                                                                               is_index=True)
        middle = len(insert_ids) // 2
        op_l = {"ids": insert_ids[:middle], "collection": collection_w.name,
                "field": default_field_name}
        op_r = {"ids": insert_ids[middle:], "collection": collection_w.name,
                "field": default_field_name}
        metric_field = get_support_metric_field
        params = {metric_field: "L2", "sqrt": True}
        self.utility_wrap.calc_distance(op_l, op_r, params,
                                        check_task=CheckTasks.err_res,
                                        check_items={"err_code": 1,
                                                     "err_msg": "collection {} was not "
                                                                "loaded into memory)".format(collection_w.name)})


class TestUtilityBase(TestcaseBase):
    """ Test case of index interface """

    @pytest.fixture(scope="function", params=["metric_type", "metric"])
    def metric_field(self, request):
        yield request.param

    @pytest.fixture(scope="function", params=[True, False])
    def sqrt(self, request):
        yield request.param

    @pytest.fixture(scope="function", params=["L2", "IP"])
    def metric(self, request):
        yield request.param

    @pytest.fixture(scope="function", params=["HAMMING", "TANIMOTO"])
    def metric_binary(self, request):
        yield request.param

    @pytest.mark.tags(CaseLabel.L1)
    def test_has_collection(self):
        """
        target: test has_collection with collection name
        method: input collection name created before
        expected: True
        """
        cw = self.init_collection_wrap()
        res, _ = self.utility_wrap.has_collection(cw.name)
        assert res is True

    @pytest.mark.tags(CaseLabel.L2)
    def test_has_collection_not_created(self):
        """
        target: test has_collection with collection name which is not created
        method: input random collection name
        expected: False
        """
        c_name = cf.gen_unique_str(prefix)
        _ = self.init_collection_wrap()
        res, _ = self.utility_wrap.has_collection(c_name)
        assert res is False

    @pytest.mark.tags(CaseLabel.L1)
    def test_has_collection_after_drop(self):
        """
        target: test has_collection with collection name droped before
        method: input random collection name
        expected: False
        """
        c_name = cf.gen_unique_str(prefix)
        cw = self.init_collection_wrap(name=c_name)
        res, _ = self.utility_wrap.has_collection(c_name)
        assert res is True
        cw.drop()
        res, _ = self.utility_wrap.has_collection(c_name)
        assert res is False

    @pytest.mark.tags(CaseLabel.L1)
    def test_has_partition(self):
        """
        target: test has_partition with partition name
        method: input collection name and partition name created before
        expected: True
        """
        c_name = cf.gen_unique_str(prefix)
        p_name = cf.gen_unique_str(prefix)
        cw = self.init_collection_wrap(name=c_name)
        self.init_partition_wrap(cw, p_name)
        res, _ = self.utility_wrap.has_partition(c_name, p_name)
        assert res is True

    @pytest.mark.tags(CaseLabel.L2)
    def test_has_partition_not_created(self):
        """
        target: test has_partition with partition name
        method: input collection name, and partition name not created before
        expected: True
        """
        c_name = cf.gen_unique_str(prefix)
        p_name = cf.gen_unique_str()
        self.init_collection_wrap(name=c_name)
        res, _ = self.utility_wrap.has_partition(c_name, p_name)
        assert res is False

    @pytest.mark.tags(CaseLabel.L1)
    def test_has_partition_after_drop(self):
        """
        target: test has_partition with partition name
        method: input collection name, and partition name dropped
        expected: True
        """
        c_name = cf.gen_unique_str(prefix)
        p_name = cf.gen_unique_str()
        cw = self.init_collection_wrap(name=c_name)
        pw = self.init_partition_wrap(cw, p_name)
        res, _ = self.utility_wrap.has_partition(c_name, p_name)
        assert res is True
        pw.drop()
        res, _ = self.utility_wrap.has_partition(c_name, p_name)
        assert res is False

    @pytest.mark.tags(CaseLabel.L2)
    def test_has_default_partition(self):
        """
        target: test has_partition with '_default' partition
        method: input collection name and partition name created before
        expected: True
        """
        c_name = cf.gen_unique_str(prefix)
        self.init_collection_wrap(name=c_name)
        res, _ = self.utility_wrap.has_partition(c_name, ct.default_partition_name)
        assert res is True

    @pytest.mark.tags(CaseLabel.L1)
    def test_list_collections(self):
        """
        target: test list_collections
        method: create collection, list_collections
        expected: in the result
        """
        c_name = cf.gen_unique_str(prefix)
        self.init_collection_wrap(name=c_name)
        res, _ = self.utility_wrap.list_collections()
        assert c_name in res

    # TODO: make sure all collections deleted
    @pytest.mark.tags(CaseLabel.L1)
    def _test_list_collections_no_collection(self):
        """
        target: test list_collections
        method: no collection created, list_collections
        expected: length of the result equals to 0
        """
        self._connect()
        res, _ = self.utility_wrap.list_collections()
        assert len(res) == 0

    @pytest.mark.tags(CaseLabel.L2)
    def test_index_process_collection_not_existed(self):
        """
        target: test building_process
        method: input collection not created before
        expected: raise exception
        """
        self._connect()
        c_name = cf.gen_unique_str(prefix)
        self.utility_wrap.index_building_progress(
            c_name,
            check_task=CheckTasks.err_res,
            check_items={ct.err_code: 1, ct.err_msg: "can't find collection"})

    @pytest.mark.tags(CaseLabel.L1)
    def test_index_process_collection_empty(self):
        """
        target: test building_process
        method: input empty collection
        expected: no exception raised
        """
        c_name = cf.gen_unique_str(prefix)
        cw = self.init_collection_wrap(name=c_name)
        self.index_wrap.init_index(cw.collection, default_field_name, default_index_params)
        res, _ = self.utility_wrap.index_building_progress(c_name)
        exp_res = {'total_rows': 0, 'indexed_rows': 0}
        assert res == exp_res

    @pytest.mark.tags(CaseLabel.L2)
    def test_index_process_collection_insert_no_index(self):
        """
        target: test building_process
        method: insert 1 entity, no index created
        expected: no exception raised
        """
        nb = 1
        c_name = cf.gen_unique_str(prefix)
        cw = self.init_collection_wrap(name=c_name)
        data = cf.gen_default_list_data(nb)
        cw.insert(data=data)
        error = {ct.err_code: 1, ct.err_msg: "no index is created"}
        self.utility_wrap.index_building_progress(c_name, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L1)
    def test_index_process_collection_index(self):
        """
        target: test building_process
        method: 1.insert 1024 (because minSegmentSizeToEnableIndex=1024)
                2.build(server does create index) and call building_process
        expected: indexed_rows=0
        """
        nb = 1024
        c_name = cf.gen_unique_str(prefix)
        cw = self.init_collection_wrap(name=c_name)
        data = cf.gen_default_list_data(nb)
        cw.insert(data=data)
        cw.create_index(default_field_name, default_index_params)
        res, _ = self.utility_wrap.index_building_progress(c_name)
        assert res['indexed_rows'] == 0
        assert res['total_rows'] == nb

    @pytest.mark.tags(CaseLabel.L1)
    def test_index_process_collection_indexing(self):
        """
        target: test building_process
        method: 1.insert 2048 entities to ensure that server will build
                2.call building_process during building
        expected: 2048 or less entities indexed
        """
        nb = 2048
        c_name = cf.gen_unique_str(prefix)
        cw = self.init_collection_wrap(name=c_name)
        data = cf.gen_default_list_data(nb)
        cw.insert(data=data)
        cw.create_index(default_field_name, default_index_params)
        start = time.time()
        while True:
            time.sleep(1)
            res, _ = self.utility_wrap.index_building_progress(c_name)
            if 0 < res['indexed_rows'] <= nb:
                break
            if time.time() - start > 5:
                raise MilvusException(1, f"Index build completed in more than 5s")

    @pytest.mark.tags(CaseLabel.L2)
    def test_wait_index_collection_not_existed(self):
        """
        target: test wait_index
        method: input collection not created before
        expected: raise exception
        """
        self._connect()
        c_name = cf.gen_unique_str(prefix)
        self.utility_wrap.wait_for_index_building_complete(
            c_name,
            check_task=CheckTasks.err_res,
            check_items={ct.err_code: 1, ct.err_msg: "can't find collection"})

    @pytest.mark.tags(CaseLabel.L1)
    def test_wait_index_collection_empty(self):
        """
        target: test wait_index
        method: input empty collection
        expected: no exception raised
        """
        self._connect()
        c_name = cf.gen_unique_str(prefix)
        cw = self.init_collection_wrap(name=c_name)
        cw.create_index(default_field_name, default_index_params)
        assert self.utility_wrap.wait_for_index_building_complete(c_name)[0]
        res, _ = self.utility_wrap.index_building_progress(c_name)
        exp_res = {'total_rows': 0, 'indexed_rows': 0}
        assert res == exp_res

    @pytest.mark.tags(CaseLabel.L1)
    def test_wait_index_collection_index(self):
        """
        target: test wait_index
        method: insert 5000 entities, build and call wait_index
        expected: 5000 entity indexed
        """
        nb = 5000
        c_name = cf.gen_unique_str(prefix)
        cw = self.init_collection_wrap(name=c_name)
        data = cf.gen_default_list_data(nb)
        cw.insert(data=data)
        cw.create_index(default_field_name, default_index_params)
        res, _ = self.utility_wrap.wait_for_index_building_complete(c_name)
        assert res is True
        res, _ = self.utility_wrap.index_building_progress(c_name)
        assert res["indexed_rows"] == nb

    @pytest.mark.tags(CaseLabel.L2)
    def test_loading_progress_without_loading(self):
        """
        target: test loading progress without loading
        method: insert and flush data, call loading_progress without loading
        expected: return successfully with 0%
        """
        collection_w = self.init_collection_wrap()
        df = cf.gen_default_dataframe_data()
        collection_w.insert(df)
        assert collection_w.num_entities == ct.default_nb
        res = self.utility_wrap.loading_progress(collection_w.name)[0]
        exp_res = {loading_progress: '0%', num_loaded_partitions: 0, not_loaded_partitions: ['_default']}

        assert exp_res == res

    @pytest.mark.tags(CaseLabel.L1)
    @pytest.mark.parametrize("nb", [ct.default_nb, 5000])
    def test_loading_progress_collection(self, nb):
        """
        target: test loading progress
        method: 1.insert flush and load 2.call loading_progress
        expected: all entities is loafed, because load is synchronous
        """
        # create, insert default_nb, flush and load
        collection_w = self.init_collection_general(prefix, insert_data=True, nb=nb)[0]
        res, _ = self.utility_wrap.loading_progress(collection_w.name)
        assert res[loading_progress] == '100%'

    @pytest.mark.tags(CaseLabel.L2)
    def test_loading_progress_with_async_load(self):
        """
        target: test loading progress with async collection load
        method: 1.load collection with async=True 2.loading_progress
        expected: loading part entities
        """
        collection_w = self.init_collection_wrap()
        df = cf.gen_default_dataframe_data()
        collection_w.insert(df)
        assert collection_w.num_entities == ct.default_nb
        collection_w.load(_async=True)
        res, _ = self.utility_wrap.loading_progress(collection_w.name)
        loading_int = cf.percent_to_int(res[loading_progress])
        if -1 != loading_int:
            assert (0 <= loading_int <= 100)
        else:
            log.info("The output of loading progress is not a string or a percentage")

    @pytest.mark.tags(CaseLabel.L2)
    def test_loading_progress_empty_collection(self):
        """
        target: test loading_progress on an empty collection
        method: 1.create collection and no insert 2.loading_progress
        expected: 0 entities is loaded
        """
        collection_w = self.init_collection_wrap()
        collection_w.load()
        res, _ = self.utility_wrap.loading_progress(collection_w.name)
        exp_res = {loading_progress: '100%', num_loaded_partitions: 1, not_loaded_partitions: []}

        assert exp_res == res

    @pytest.mark.tags(CaseLabel.L1)
    def test_loading_progress_after_release(self):
        """
        target: test loading progress after release
        method: insert and flush data, call loading_progress after release
        expected: return successfully with 0%
        """
        collection_w = self.init_collection_general(prefix, insert_data=True)[0]
        collection_w.release()
        res = self.utility_wrap.loading_progress(collection_w.name)[0]
        exp_res = {loading_progress: '0%', num_loaded_partitions: 0, not_loaded_partitions: ['_default']}

        assert exp_res == res

    @pytest.mark.tags(CaseLabel.L2)
    def test_loading_progress_with_release_partition(self):
        """
        target: test loading progress after release part partitions
        method: 1.insert data into two partitions and flush
                2.load one partition and release one partition
        expected: loaded one partition entities
        """
        half = ct.default_nb
        # insert entities into two partitions, collection flush and load
        collection_w, partition_w, _, _ = self.insert_entities_into_two_partitions_in_half(half)
        partition_w.release()
        res = self.utility_wrap.loading_progress(collection_w.name)[0]
        assert res[loading_progress] == '50%'

    @pytest.mark.tags(CaseLabel.L2)
    def test_loading_progress_with_load_partition(self):
        """
        target: test loading progress after load partition
        method: 1.insert data into two partitions and flush
                2.load one partition and loading progress
        expected: loaded one partition entities
        """
        half = ct.default_nb
        collection_w, partition_w, _, _ = self.insert_entities_into_two_partitions_in_half(half)
        collection_w.release()
        partition_w.load()
        res = self.utility_wrap.loading_progress(collection_w.name)[0]
        assert res[loading_progress] == '50%'

    @pytest.mark.tags(CaseLabel.L1)
    def test_loading_progress_with_partition(self):
        """
        target: test loading progress with partition
        method: 1.insert data into two partitions and flush, and load
                2.loading progress with one partition
        expected: loaded one partition entities
        """
        half = ct.default_nb
        collection_w, partition_w, _, _ = self.insert_entities_into_two_partitions_in_half(half)
        res = self.utility_wrap.loading_progress(collection_w.name, partition_names=[partition_w.name])[0]
        assert res[loading_progress] == '100%'

    @pytest.mark.tags(CaseLabel.ClusterOnly)
    def test_loading_progress_multi_replicas(self):
        """
        target: test loading progress with multi replicas
        method: 1.Create collection and insert data
                2.Load replicas and get loading progress
                3.Create partitions and insert data
                4.Get loading progress
                5.Release and re-load replicas, get loading progress
        expected: Verify loading progress result
        """
        collection_w = self.init_collection_wrap()
        collection_w.insert(cf.gen_default_dataframe_data())
        assert collection_w.num_entities == ct.default_nb
        collection_w.load(partition_names=[ct.default_partition_name], replica_number=2)
        res_collection, _ = self.utility_wrap.loading_progress(collection_w.name)
        assert res_collection == {loading_progress: '100%', num_loaded_partitions: 1, not_loaded_partitions: []}

        # create partition and insert
        partition_w = self.init_partition_wrap(collection_wrap=collection_w)
        partition_w.insert(cf.gen_default_dataframe_data(start=ct.default_nb))
        assert partition_w.num_entities == ct.default_nb
        res_part_partition, _ = self.utility_wrap.loading_progress(collection_w.name)
        assert res_part_partition == {'loading_progress': '50%', 'num_loaded_partitions': 1,
                                      'not_loaded_partitions': [partition_w.name]}

        res_part_partition, _ = self.utility_wrap.loading_progress(collection_w.name,
                                                                   partition_names=[partition_w.name])
        assert res_part_partition == {'loading_progress': '0%', 'num_loaded_partitions': 0,
                                      'not_loaded_partitions': [partition_w.name]}

        collection_w.release()
        collection_w.load(replica_number=2)
        res_all_partitions, _ = self.utility_wrap.loading_progress(collection_w.name)
        assert res_all_partitions == {'loading_progress': '100%', 'num_loaded_partitions': 2,
                                      'not_loaded_partitions': []}

    @pytest.mark.tags(CaseLabel.L1)
    def test_wait_loading_collection_empty(self):
        """
        target: test wait_for_loading
        method: input empty collection
        expected: no exception raised
        """
        self._connect()
        cw = self.init_collection_wrap(name=cf.gen_unique_str(prefix))
        cw.load()
        self.utility_wrap.wait_for_loading_complete(cw.name)
        res, _ = self.utility_wrap.loading_progress(cw.name)
        exp_res = {loading_progress: "100%", not_loaded_partitions: [], num_loaded_partitions: 1}
        assert res == exp_res

    @pytest.mark.tags(CaseLabel.L1)
    def test_wait_for_loading_complete(self):
        """
        target: test wait for loading collection
        method: insert 10000 entities and wait for loading complete
        expected: after loading complete, loaded entities is 10000
        """
        nb = 6000
        collection_w = self.init_collection_wrap()
        df = cf.gen_default_dataframe_data(nb)
        collection_w.insert(df, timeout=60)
        assert collection_w.num_entities == nb
        collection_w.load(_async=True)
        self.utility_wrap.wait_for_loading_complete(collection_w.name)
        res, _ = self.utility_wrap.loading_progress(collection_w.name)
        assert res[loading_progress] == '100%'

    @pytest.mark.tags(CaseLabel.L0)
    def test_drop_collection(self):
        """
        target: test utility drop collection by name
        method: input collection name and drop collection
        expected: collection is dropped
        """
        c_name = cf.gen_unique_str(prefix)
        self.init_collection_wrap(c_name)
        assert self.utility_wrap.has_collection(c_name)[0]
        self.utility_wrap.drop_collection(c_name)
        assert not self.utility_wrap.has_collection(c_name)[0]

    @pytest.mark.tags(CaseLabel.L0)
    def test_drop_collection_repeatedly(self):
        """
        target: test drop collection repeatedly
        method: 1.collection.drop 2.utility.drop_collection
        expected: raise exception
        """
        c_name = cf.gen_unique_str(prefix)
        collection_w = self.init_collection_wrap(c_name)
        assert self.utility_wrap.has_collection(c_name)[0]
        collection_w.drop()
        assert not self.utility_wrap.has_collection(c_name)[0]

        # error = {ct.err_code: 1, ct.err_msg: {"describe collection failed: can't find collection:"}}
        # self.utility_wrap.drop_collection(c_name, check_task=CheckTasks.err_res, check_items=error)
        # @longjiquan: dropping collection should be idempotent.
        self.utility_wrap.drop_collection(c_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_drop_collection_create_repeatedly(self):
        """
        target: test repeatedly create and drop same name collection
        method: repeatedly create and drop collection
        expected: no exception
        """
        from time import sleep
        loops = 3
        c_name = cf.gen_unique_str(prefix)
        for _ in range(loops):
            self.init_collection_wrap(c_name)
            assert self.utility_wrap.has_collection(c_name)[0]
            self.utility_wrap.drop_collection(c_name)
            assert not self.utility_wrap.has_collection(c_name)[0]
            sleep(1)

    @pytest.mark.tags(CaseLabel.L1)
    def test_calc_distance_default(self):
        """
        target: test calculated distance with default params
        method: calculated distance between two random vectors
        expected: distance calculated successfully
        """
        log.info("Creating connection")
        self._connect()
        log.info("Creating vectors for distance calculation")
        vectors_l = cf.gen_vectors(default_nb, default_dim)
        vectors_r = cf.gen_vectors(default_nb, default_dim)
        op_l = {"float_vectors": vectors_l}
        op_r = {"float_vectors": vectors_r}
        log.info("Calculating distance for generated vectors")
        self.utility_wrap.calc_distance(op_l, op_r,
                                        check_task=CheckTasks.check_distance,
                                        check_items={"vectors_l": vectors_l,
                                                     "vectors_r": vectors_r})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_default_sqrt(self, metric_field, metric):
        """
        target: test calculated distance with default param
        method: calculated distance with default sqrt
        expected: distance calculated successfully
        """
        log.info("Creating connection")
        self._connect()
        log.info("Creating vectors for distance calculation")
        vectors_l = cf.gen_vectors(default_nb, default_dim)
        vectors_r = cf.gen_vectors(default_nb, default_dim)
        op_l = {"float_vectors": vectors_l}
        op_r = {"float_vectors": vectors_r}
        log.info("Calculating distance for generated vectors within default sqrt")
        params = {metric_field: metric}
        self.utility_wrap.calc_distance(op_l, op_r, params,
                                        check_task=CheckTasks.check_distance,
                                        check_items={"vectors_l": vectors_l,
                                                     "vectors_r": vectors_r,
                                                     "metric": metric})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_default_metric(self, sqrt):
        """
        target: test calculated distance with default param
        method: calculated distance with default metric
        expected: distance calculated successfully
        """
        log.info("Creating connection")
        self._connect()
        log.info("Creating vectors for distance calculation")
        vectors_l = cf.gen_vectors(default_nb, default_dim)
        vectors_r = cf.gen_vectors(default_nb, default_dim)
        op_l = {"float_vectors": vectors_l}
        op_r = {"float_vectors": vectors_r}
        log.info("Calculating distance for generated vectors within default metric")
        params = {"sqrt": sqrt}
        self.utility_wrap.calc_distance(op_l, op_r, params,
                                        check_task=CheckTasks.check_distance,
                                        check_items={"vectors_l": vectors_l,
                                                     "vectors_r": vectors_r,
                                                     "sqrt": sqrt})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_binary_metric(self, metric_field, metric_binary):
        """
        target: test calculate distance with binary vectors
        method: calculate distance between binary vectors
        expected: distance calculated successfully
        """
        log.info("Creating connection")
        self._connect()
        log.info("Creating vectors for distance calculation")
        nb = 10
        raw_vectors_l, vectors_l = cf.gen_binary_vectors(nb, default_dim)
        raw_vectors_r, vectors_r = cf.gen_binary_vectors(nb, default_dim)
        op_l = {"bin_vectors": vectors_l}
        op_r = {"bin_vectors": vectors_r}
        log.info("Calculating distance for binary vectors")
        params = {metric_field: metric_binary}
        vectors_l = raw_vectors_l
        vectors_r = raw_vectors_r
        self.utility_wrap.calc_distance(op_l, op_r, params,
                                        check_task=CheckTasks.check_distance,
                                        check_items={"vectors_l": vectors_l,
                                                     "vectors_r": vectors_r,
                                                     "metric": metric_binary})

    @pytest.mark.tags(CaseLabel.L1)
    def test_calc_distance_from_collection_ids(self, metric_field, metric, sqrt):
        """
        target: test calculated distance from collection entities
        method: both left and right vectors are from collection
        expected: distance calculated successfully
        """
        log.info("Creating connection")
        self._connect()
        nb = 10
        collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix, True, nb)
        middle = len(insert_ids) // 2
        vectors = vectors[0].loc[:, default_field_name]
        vectors_l = vectors[:middle]
        vectors_r = []
        for i in range(middle):
            vectors_r.append(vectors[middle + i])
        log.info("Creating vectors from collections for distance calculation")
        op_l = {"ids": insert_ids[:middle], "collection": collection_w.name,
                "field": default_field_name}
        op_r = {"ids": insert_ids[middle:], "collection": collection_w.name,
                "field": default_field_name}
        log.info("Creating vectors for entities")
        params = {metric_field: metric, "sqrt": sqrt}
        self.utility_wrap.calc_distance(op_l, op_r, params,
                                        check_task=CheckTasks.check_distance,
                                        check_items={"vectors_l": vectors_l,
                                                     "vectors_r": vectors_r,
                                                     "metric": metric,
                                                     "sqrt": sqrt})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_from_collections(self, metric_field, metric, sqrt):
        """
        target: test calculated distance between entities from collections
        method: calculated distance between entities from two collections
        expected: distance calculated successfully
        """
        log.info("Creating connection")
        self._connect()
        nb = 10
        prefix_1 = "utility_distance"
        log.info("Creating two collections")
        collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix, True, nb)
        collection_w_1, vectors_1, _, insert_ids_1, _ = self.init_collection_general(prefix_1, True, nb)
        vectors_l = vectors[0].loc[:, default_field_name]
        vectors_r = vectors_1[0].loc[:, default_field_name]
        log.info("Extracting entities from collections for distance calculating")
        op_l = {"ids": insert_ids, "collection": collection_w.name,
                "field": default_field_name}
        op_r = {"ids": insert_ids_1, "collection": collection_w_1.name,
                "field": default_field_name}
        params = {metric_field: metric, "sqrt": sqrt}
        log.info("Calculating distance for entities from two collections")
        self.utility_wrap.calc_distance(op_l, op_r, params,
                                        check_task=CheckTasks.check_distance,
                                        check_items={"vectors_l": vectors_l,
                                                     "vectors_r": vectors_r,
                                                     "metric": metric,
                                                     "sqrt": sqrt})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_left_vector_and_collection_ids(self, metric_field, metric, sqrt):
        """
        target: test calculated distance from collection entities
        method: set left vectors as random vectors, right vectors from collection
        expected: distance calculated successfully
        """
        log.info("Creating connection")
        self._connect()
        nb = 10
        collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix, True, nb)
        middle = len(insert_ids) // 2
        vectors = vectors[0].loc[:, default_field_name]
        vectors_l = cf.gen_vectors(nb, default_dim)
        vectors_r = []
        for i in range(middle):
            vectors_r.append(vectors[middle + i])
        op_l = {"float_vectors": vectors_l}
        log.info("Extracting entities from collections for distance calculating")
        op_r = {"ids": insert_ids[middle:], "collection": collection_w.name,
                "field": default_field_name}
        params = {metric_field: metric, "sqrt": sqrt}
        log.info("Calculating distance between vectors and entities")
        self.utility_wrap.calc_distance(op_l, op_r, params,
                                        check_task=CheckTasks.check_distance,
                                        check_items={"vectors_l": vectors_l,
                                                     "vectors_r": vectors_r,
                                                     "metric": metric,
                                                     "sqrt": sqrt})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_right_vector_and_collection_ids(self, metric_field, metric, sqrt):
        """
        target: test calculated distance from collection entities
        method: set right vectors as random vectors, left vectors from collection
        expected: distance calculated successfully
        """
        log.info("Creating connection")
        self._connect()
        nb = 10
        collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix, True, nb)
        middle = len(insert_ids) // 2
        vectors = vectors[0].loc[:, default_field_name]
        vectors_l = vectors[:middle]
        vectors_r = cf.gen_vectors(nb, default_dim)
        log.info("Extracting entities from collections for distance calculating")
        op_l = {"ids": insert_ids[:middle], "collection": collection_w.name,
                "field": default_field_name}
        op_r = {"float_vectors": vectors_r}
        params = {metric_field: metric, "sqrt": sqrt}
        log.info("Calculating distance between right vector and entities")
        self.utility_wrap.calc_distance(op_l, op_r, params,
                                        check_task=CheckTasks.check_distance,
                                        check_items={"vectors_l": vectors_l,
                                                     "vectors_r": vectors_r,
                                                     "metric": metric,
                                                     "sqrt": sqrt})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_from_partition_ids(self, metric_field, metric, sqrt):
        """
        target: test calculated distance from one partition entities
        method: both left and right vectors are from partition
        expected: distance calculated successfully
        """
        log.info("Creating connection")
        self._connect()
        nb = 10
        collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix, True, nb, partition_num=1)
        partitions = collection_w.partitions
        middle = len(insert_ids) // 2
        params = {metric_field: metric, "sqrt": sqrt}
        start = 0
        end = middle
        for i in range(len(partitions)):
            log.info("Extracting entities from partitions for distance calculating")
            vectors_l = vectors[i].loc[:, default_field_name]
            vectors_r = vectors[i].loc[:, default_field_name]
            op_l = {"ids": insert_ids[start:end], "collection": collection_w.name,
                    "partition": partitions[i].name, "field": default_field_name}
            op_r = {"ids": insert_ids[start:end], "collection": collection_w.name,
                    "partition": partitions[i].name, "field": default_field_name}
            start += middle
            end += middle
            log.info("Calculating distance between entities from one partition")
            self.utility_wrap.calc_distance(op_l, op_r, params,
                                            check_task=CheckTasks.check_distance,
                                            check_items={"vectors_l": vectors_l,
                                                         "vectors_r": vectors_r,
                                                         "metric": metric,
                                                         "sqrt": sqrt})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_from_partitions(self, metric_field, metric, sqrt):
        """
        target: test calculated distance between entities from partitions
        method: calculate distance between entities from two partitions
        expected: distance calculated successfully
        """
        log.info("Create connection")
        self._connect()
        nb = 10
        collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix, True, nb, partition_num=1)
        partitions = collection_w.partitions
        middle = len(insert_ids) // 2
        params = {metric_field: metric, "sqrt": sqrt}
        vectors_l = vectors[0].loc[:, default_field_name]
        vectors_r = vectors[1].loc[:, default_field_name]
        log.info("Extract entities from two partitions for distance calculating")
        op_l = {"ids": insert_ids[:middle], "collection": collection_w.name,
                "partition": partitions[0].name, "field": default_field_name}
        op_r = {"ids": insert_ids[middle:], "collection": collection_w.name,
                "partition": partitions[1].name, "field": default_field_name}
        log.info("Calculate distance between entities from two partitions")
        self.utility_wrap.calc_distance(op_l, op_r, params,
                                        check_task=CheckTasks.check_distance,
                                        check_items={"vectors_l": vectors_l,
                                                     "vectors_r": vectors_r,
                                                     "metric": metric,
                                                     "sqrt": sqrt})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_left_vectors_and_partition_ids(self, metric_field, metric, sqrt):
        """
        target: test calculated distance between vectors and partition entities
        method: set left vectors as random vectors, right vectors are entities
        expected: distance calculated successfully
        """
        log.info("Creating connection")
        self._connect()
        nb = 10
        collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix, True, nb, partition_num=1)
        middle = len(insert_ids) // 2
        partitions = collection_w.partitions
        vectors_l = cf.gen_vectors(nb // 2, default_dim)
        log.info("Extract entities from collection as right vectors")
        op_l = {"float_vectors": vectors_l}
        params = {metric_field: metric, "sqrt": sqrt}
        start = 0
        end = middle
        log.info("Calculate distance between vector and entities")
        for i in range(len(partitions)):
            vectors_r = vectors[i].loc[:, default_field_name]
            op_r = {"ids": insert_ids[start:end], "collection": collection_w.name,
                    "partition": partitions[i].name, "field": default_field_name}
            start += middle
            end += middle
            self.utility_wrap.calc_distance(op_l, op_r, params,
                                            check_task=CheckTasks.check_distance,
                                            check_items={"vectors_l": vectors_l,
                                                         "vectors_r": vectors_r,
                                                         "metric": metric,
                                                         "sqrt": sqrt})

    @pytest.mark.tags(CaseLabel.L2)
    def test_calc_distance_right_vectors_and_partition_ids(self, metric_field, metric, sqrt):
        """
        target: test calculated distance between vectors and partition entities
        method: set right vectors as random vectors, left vectors are entities
        expected: distance calculated successfully
        """
        log.info("Create connection")
        self._connect()
        nb = 10
        collection_w, vectors, _, insert_ids, _ = self.init_collection_general(prefix, True, nb, partition_num=1)
        middle = len(insert_ids) // 2
        partitions = collection_w.partitions
        vectors_r = cf.gen_vectors(nb // 2, default_dim)
        op_r = {"float_vectors": vectors_r}
        params = {metric_field: metric, "sqrt": sqrt}
        start = 0
        end = middle
        for i in range(len(partitions)):
            vectors_l = vectors[i].loc[:, default_field_name]
            log.info("Extract entities from partition %d as left vector" % i)
            op_l = {"ids": insert_ids[start:end], "collection": collection_w.name,
                    "partition": partitions[i].name, "field": default_field_name}
            start += middle
            end += middle
            log.info("Calculate distance between vector and entities from partition %d" % i)
            self.utility_wrap.calc_distance(op_l, op_r, params,
                                            check_task=CheckTasks.check_distance,
                                            check_items={"vectors_l": vectors_l,
                                                         "vectors_r": vectors_r,
                                                         "metric": metric,
                                                         "sqrt": sqrt})


class TestUtilityAdvanced(TestcaseBase):
    """ Test case of index interface """

    @pytest.mark.tags(CaseLabel.L2)
    def test_has_collection_multi_collections(self):
        """
        target: test has_collection with collection name
        method: input collection name created before
        expected: True
        """
        c_name = cf.gen_unique_str(prefix)
        c_name_2 = cf.gen_unique_str(prefix)
        self.init_collection_wrap(name=c_name)
        self.init_collection_wrap(name=c_name_2)
        for name in [c_name, c_name_2]:
            res, _ = self.utility_wrap.has_collection(name)
            assert res is True

    @pytest.mark.tags(CaseLabel.L2)
    def test_list_collections_multi_collection(self):
        """
        target: test list_collections
        method: create collection, list_collections
        expected: in the result
        """
        c_name = cf.gen_unique_str(prefix)
        c_name_2 = cf.gen_unique_str(prefix)
        self.init_collection_wrap(name=c_name)
        self.init_collection_wrap(name=c_name_2)
        res, _ = self.utility_wrap.list_collections()
        for name in [c_name, c_name_2]:
            assert name in res

    @pytest.mark.tags(CaseLabel.L2)
    def test_drop_multi_collection_concurrent(self):
        """
        target: test concurrent drop collection
        method: multi thread drop one collection
        expected: drop successfully
        """
        thread_num = 3
        threads = []
        c_names = []
        num = 5

        for i in range(thread_num * num):
            c_name = cf.gen_unique_str(prefix)
            self.init_collection_wrap(c_name)
            c_names.append(c_name)

        def create_and_drop_collection(names):
            for name in names:
                assert self.utility_wrap.has_collection(name)[0]
                self.utility_wrap.drop_collection(name)
                assert not self.utility_wrap.has_collection(name)[0]

        for i in range(thread_num):
            x = threading.Thread(target=create_and_drop_collection, args=(c_names[i * num:(i + 1) * num],))
            threads.append(x)
            x.start()
        for t in threads:
            t.join()
        log.debug(self.utility_wrap.list_collections()[0])

    @pytest.mark.tags(CaseLabel.L2)
    def test_get_query_segment_info_empty_collection(self):
        """
        target: test getting query segment info of empty collection
        method: init a collection and get query segment info
        expected: length of segment is 0
        """
        c_name = cf.gen_unique_str(prefix)
        collection_w = self.init_collection_wrap(name=c_name)
        collection_w.load()
        res, _ = self.utility_wrap.get_query_segment_info(c_name)
        assert len(res) == 0

    @pytest.mark.tags(CaseLabel.L1)
    def test_get_sealed_query_segment_info(self):
        """
        target: test getting sealed query segment info of collection with data
        method: init a collection, insert data, flush, load, and get query segment info
        expected:
            1. length of segment is greater than 0
            2. the sum num_rows of each segment is equal to num of entities
        """
        c_name = cf.gen_unique_str(prefix)
        collection_w = self.init_collection_wrap(name=c_name)
        nb = 3000
        df = cf.gen_default_dataframe_data(nb)
        collection_w.insert(df)
        collection_w.num_entities
        collection_w.load()
        res, _ = self.utility_wrap.get_query_segment_info(c_name)
        assert len(res) > 0
        segment_ids = []
        cnt = 0
        for r in res:
            log.info(f"segmentID {r.segmentID}: state: {r.state}; num_rows: {r.num_rows} ")
            if r.segmentID not in segment_ids:
                segment_ids.append(r.segmentID)
                cnt += r.num_rows
        assert cnt == nb

    @pytest.mark.tags(CaseLabel.L1)
    def test_get_sealed_query_segment_info_after_create_index(self):
        """
        target: test getting sealed query segment info of collection with data
        method: init a collection, insert data, flush, create index, load, and get query segment info
        expected:
            1. length of segment is greater than 0
            2. the sum num_rows of each segment is equal to num of entities
        """
        c_name = cf.gen_unique_str(prefix)
        collection_w = self.init_collection_wrap(name=c_name)
        nb = 3000
        df = cf.gen_default_dataframe_data(nb)
        collection_w.insert(df)
        collection_w.num_entities
        collection_w.create_index(default_field_name, default_index_params)
        collection_w.load()
        res, _ = self.utility_wrap.get_query_segment_info(c_name)
        assert len(res) > 0
        segment_ids = []
        cnt = 0
        for r in res:
            log.info(f"segmentID {r.segmentID}: state: {r.state}; num_rows: {r.num_rows} ")
            if r.segmentID not in segment_ids:
                segment_ids.append(r.segmentID)
                cnt += r.num_rows
        assert cnt == nb

    @pytest.mark.tags(CaseLabel.L2)
    def test_load_balance_normal(self):
        """
        target: test load balance of collection
        method: init a collection and load balance
        expected: sealed_segment_ids is subset of des_sealed_segment_ids
        """
        # init a collection
        self._connect()
        querynode_num = len(MilvusSys().query_nodes)
        if querynode_num < 2:
            pytest.skip("skip load balance testcase when querynode number less than 2")
        c_name = cf.gen_unique_str(prefix)
        collection_w = self.init_collection_wrap(name=c_name)
        ms = MilvusSys()
        nb = 3000
        df = cf.gen_default_dataframe_data(nb)
        collection_w.insert(df)
        # get sealed segments
        collection_w.num_entities
        # get growing segments
        collection_w.insert(df)
        collection_w.load()
        # prepare load balance params
        res, _ = self.utility_wrap.get_query_segment_info(c_name)
        segment_distribution = cf.get_segment_distribution(res)
        all_querynodes = [node["identifier"] for node in ms.query_nodes]
        assert len(all_querynodes) > 1
        all_querynodes = sorted(all_querynodes,
                                key=lambda x: len(segment_distribution[x]["sealed"])
                                if x in segment_distribution else 0, reverse=True)
        src_node_id = all_querynodes[0]
        des_node_ids = all_querynodes[1:]
        sealed_segment_ids = segment_distribution[src_node_id]["sealed"]
        # load balance
        self.utility_wrap.load_balance(collection_w.name, src_node_id, des_node_ids, sealed_segment_ids)
        # get segments distribution after load balance
        res, _ = self.utility_wrap.get_query_segment_info(c_name)
        segment_distribution = cf.get_segment_distribution(res)
        sealed_segment_ids_after_load_banalce = segment_distribution[src_node_id]["sealed"]
        # assert src node has no sealed segments
        assert sealed_segment_ids_after_load_banalce == []
        des_sealed_segment_ids = []
        for des_node_id in des_node_ids:
            des_sealed_segment_ids += segment_distribution[des_node_id]["sealed"]
        # assert sealed_segment_ids is subset of des_sealed_segment_ids
        assert set(sealed_segment_ids).issubset(des_sealed_segment_ids)

    @pytest.mark.tags(CaseLabel.L1)
    def test_load_balance_with_src_node_not_exist(self):
        """
        target: test load balance of collection
        method: init a collection and load balance with src_node not exist
        expected: raise exception
        """
        # init a collection
        c_name = cf.gen_unique_str(prefix)
        collection_w = self.init_collection_wrap(name=c_name)
        ms = MilvusSys()
        nb = 3000
        df = cf.gen_default_dataframe_data(nb)
        collection_w.insert(df)
        # get sealed segments
        collection_w.num_entities
        # get growing segments
        collection_w.insert(df)
        collection_w.load()
        # prepare load balance params
        res, _ = self.utility_wrap.get_query_segment_info(c_name)
        segment_distribution = cf.get_segment_distribution(res)
        all_querynodes = [node["identifier"] for node in ms.query_nodes]
        all_querynodes = sorted(all_querynodes,
                                key=lambda x: len(segment_distribution[x]["sealed"])
                                if x in segment_distribution else 0, reverse=True)
        # set src_node_id as the id of indexnode's id, which is not exist for querynode
        invalid_src_node_id = [node["identifier"] for node in ms.index_nodes][0]
        src_node_id = all_querynodes[0]
        dst_node_ids = all_querynodes[1:]
        sealed_segment_ids = segment_distribution[src_node_id]["sealed"]
        # load balance
        self.utility_wrap.load_balance(collection_w.name, invalid_src_node_id, dst_node_ids, sealed_segment_ids,
                                       check_task=CheckTasks.err_res,
                                       check_items={ct.err_code: 1, ct.err_msg: "is not exist to balance"})

    @pytest.mark.tags(CaseLabel.L1)
    def test_load_balance_with_all_dst_node_not_exist(self):
        """
        target: test load balance of collection
        method: init a collection and load balance with all dst_node not exist
        expected: raise exception
        """
        # init a collection
        c_name = cf.gen_unique_str(prefix)
        collection_w = self.init_collection_wrap(name=c_name)
        ms = MilvusSys()
        nb = 3000
        df = cf.gen_default_dataframe_data(nb)
        collection_w.insert(df)
        # get sealed segments
        collection_w.num_entities
        # get growing segments
        collection_w.insert(df)
        collection_w.load()
        # prepare load balance params
        res, _ = self.utility_wrap.get_query_segment_info(c_name)
        segment_distribution = cf.get_segment_distribution(res)
        all_querynodes = [node["identifier"] for node in ms.query_nodes]
        all_querynodes = sorted(all_querynodes,
                                key=lambda x: len(segment_distribution[x]["sealed"])
                                if x in segment_distribution else 0, reverse=True)
        src_node_id = all_querynodes[0]
        # add indexnode's id, which is not exist for querynode, to dst_node_ids
        dst_node_ids = [node["identifier"] for node in ms.index_nodes]
        sealed_segment_ids = segment_distribution[src_node_id]["sealed"]
        # load balance
        self.utility_wrap.load_balance(collection_w.name, src_node_id, dst_node_ids, sealed_segment_ids,
                                       check_task=CheckTasks.err_res,
                                       check_items={ct.err_code: 1, ct.err_msg: "no available queryNode to allocate"})

    @pytest.mark.tags(CaseLabel.L1)
    def test_load_balance_with_one_sealed_segment_id_not_exist(self):
        """
        target: test load balance of collection
        method: init a collection and load balance with one of sealed segment ids not exist
        expected: raise exception
        """
        # init a collection
        c_name = cf.gen_unique_str(prefix)
        collection_w = self.init_collection_wrap(name=c_name)
        ms = MilvusSys()
        nb = 3000
        df = cf.gen_default_dataframe_data(nb)
        collection_w.insert(df)
        # get sealed segments
        collection_w.num_entities
        # get growing segments
        collection_w.insert(df)
        collection_w.load()
        # prepare load balance params
        res, _ = self.utility_wrap.get_query_segment_info(c_name)
        segment_distribution = cf.get_segment_distribution(res)
        all_querynodes = [node["identifier"] for node in ms.query_nodes]
        all_querynodes = sorted(all_querynodes,
                                key=lambda x: len(segment_distribution[x]["sealed"])
                                if x in segment_distribution else 0, reverse=True)
        src_node_id = all_querynodes[0]
        dst_node_ids = all_querynodes[1:]
        dst_node_ids.append([node["identifier"] for node in ms.index_nodes][0])
        sealed_segment_ids = segment_distribution[src_node_id]["sealed"]
        # add a segment id which is not exist
        sealed_segment_ids.append(max(segment_distribution[src_node_id]["sealed"]) + 1)
        # load balance
        self.utility_wrap.load_balance(collection_w.name, src_node_id, dst_node_ids, sealed_segment_ids,
                                       check_task=CheckTasks.err_res,
                                       check_items={ct.err_code: 1, ct.err_msg: "is not exist"})

    @pytest.mark.tags(CaseLabel.L2)
    def test_load_balance_in_one_group(self):
        """
        target: test load balance of collection in one group
        method: init a collection, load with multi replicas and load balance among the querynodes in one group
        expected: load balance successfully
        """
        self._connect()
        querynode_num = len(MilvusSys().query_nodes)
        if querynode_num < 3:
            pytest.skip("skip load balance for multi replicas testcase when querynode number less than 3")
        # init a collection
        c_name = cf.gen_unique_str(prefix)
        collection_w = self.init_collection_wrap(name=c_name)
        nb = 3000
        df = cf.gen_default_dataframe_data(nb)
        collection_w.insert(df)
        # get sealed segments
        collection_w.num_entities
        collection_w.load(replica_number=2)
        # get growing segments
        collection_w.insert(df)
        # get replicas information
        res, _ = collection_w.get_replicas()
        # prepare load balance params
        # find a group which has multi nodes
        group_nodes = []
        for g in res.groups:
            if len(g.group_nodes) >= 2:
                group_nodes = list(g.group_nodes)
                break
        res, _ = self.utility_wrap.get_query_segment_info(c_name)
        segment_distribution = cf.get_segment_distribution(res)
        group_nodes = sorted(group_nodes,
                             key=lambda x: len(
                                 segment_distribution[x]["sealed"])
                             if x in segment_distribution else 0, reverse=True)
        src_node_id = group_nodes[0]
        dst_node_ids = group_nodes[1:]
        sealed_segment_ids = segment_distribution[src_node_id]["sealed"]
        # load balance
        self.utility_wrap.load_balance(collection_w.name, src_node_id, dst_node_ids, sealed_segment_ids)
        # get segments distribution after load balance
        res, _ = self.utility_wrap.get_query_segment_info(c_name)
        segment_distribution = cf.get_segment_distribution(res)
        sealed_segment_ids_after_load_banalce = segment_distribution[src_node_id]["sealed"]
        # assert src node has no sealed segments
        assert sealed_segment_ids_after_load_banalce == []
        des_sealed_segment_ids = []
        for des_node_id in dst_node_ids:
            des_sealed_segment_ids += segment_distribution[des_node_id]["sealed"]
        # assert sealed_segment_ids is subset of des_sealed_segment_ids
        assert set(sealed_segment_ids).issubset(des_sealed_segment_ids)

    @pytest.mark.tags(CaseLabel.L3)
    def test_load_balance_not_in_one_group(self):
        """
        target: test load balance of collection in one group
        method: init a collection, load with multi replicas and load balance among the querynodes in different group
        expected: load balance failed
        """
        # init a collection
        c_name = cf.gen_unique_str(prefix)
        collection_w = self.init_collection_wrap(name=c_name)
        ms = MilvusSys()
        nb = 3000
        df = cf.gen_default_dataframe_data(nb)
        collection_w.insert(df)
        # get sealed segments
        collection_w.num_entities
        collection_w.load(replica_number=2)
        # get growing segments
        collection_w.insert(df)
        # get replicas information
        res, _ = collection_w.get_replicas()
        # prepare load balance params
        all_querynodes = [node["identifier"] for node in ms.query_nodes]
        # find a group which has multi nodes
        group_nodes = []
        for g in res.groups:
            if len(g.group_nodes) >= 2:
                group_nodes = list(g.group_nodes)
                break
        src_node_id = group_nodes[0]
        dst_node_ids = list(set(all_querynodes) - set(group_nodes))
        res, _ = self.utility_wrap.get_query_segment_info(c_name)
        segment_distribution = cf.get_segment_distribution(res)
        sealed_segment_ids = segment_distribution[src_node_id]["sealed"]
        # load balance
        self.utility_wrap.load_balance(collection_w.name, src_node_id, dst_node_ids, sealed_segment_ids,
                                       check_task=CheckTasks.err_res,
                                       check_items={ct.err_code: 1, ct.err_msg: "must be in the same replica group"})

    @pytest.mark.tags(CaseLabel.L1)
    def test_handoff_query_search(self):
        """
        target: test query search after handoff
        method: 1.load collection
                2.insert, query and search
                3.flush collection and triggere handoff
                4. search with handoff indexed segments
        expected: Search ids before and after handoff are different, because search from growing and search from index
        """
        collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix), shards_num=1)
        collection_w.create_index(default_field_name, default_index_params)
        collection_w.load()

        # handoff: insert and flush one segment
        df = cf.gen_default_dataframe_data()
        insert_res, _ = collection_w.insert(df)
        term_expr = f'{ct.default_int64_field_name} in {insert_res.primary_keys[:10]}'
        res = df.iloc[:10, :1].to_dict('records')
        collection_w.query(term_expr, check_task=CheckTasks.check_query_results,
                           check_items={'exp_res': res})
        search_res_before, _ = collection_w.search(df[ct.default_float_vec_field_name][:1].to_list(),
                                                   ct.default_float_vec_field_name,
                                                   ct.default_search_params, ct.default_limit)
        log.debug(collection_w.num_entities)

        start = time.time()
        while True:
            time.sleep(2)
            segment_infos, _ = self.utility_wrap.get_query_segment_info(collection_w.name)
            # handoff done
            if len(segment_infos) == 1 and segment_infos[0].state == SegmentState.Sealed:
                break
            if time.time() - start > 20:
                raise MilvusException(1, f"Get query segment info after handoff cost more than 20s")

        # query and search from handoff segments
        collection_w.query(term_expr, check_task=CheckTasks.check_query_results,
                           check_items={'exp_res': res})
        search_res_after, _ = collection_w.search(df[ct.default_float_vec_field_name][:1].to_list(),
                                                  ct.default_float_vec_field_name,
                                                  ct.default_search_params, ct.default_limit)
        # the ids between twice search is different because of index building
        # log.debug(search_res_before[0].ids)
        # log.debug(search_res_after[0].ids)
        assert search_res_before[0].ids != search_res_after[0].ids

        # assert search result includes the nq-vector before or after handoff
        assert search_res_after[0].ids[0] == 0
        assert search_res_before[0].ids[0] == search_res_after[0].ids[0]


class TestUtilityUserPassword(TestcaseBase):
    """ Test case of user interface """

    @pytest.mark.tags(ct.CaseLabel.L3)
    def test_create_user_with_user_password(self, host, port):
        """
        target: test the user creation with user and password
        method: create user with the default user and password parameter
        expected: connected is True
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = "nico"
        password = "wertyu567"
        self.utility_wrap.create_user(user=user, password=password)
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user, password=password,
                                     check_task=ct.CheckTasks.ccr)
        self.utility_wrap.list_collections()

    @pytest.mark.tags(ct.CaseLabel.L3)
    @pytest.mark.parametrize("old_password", ["abc1234"])
    @pytest.mark.parametrize("new_password", ["abc12345"])
    def test_reset_password_with_user_and_old_password(self, host, port, old_password, new_password):
        """
        target: test the password reset with old password
        method: get a connection with user and corresponding old password
        expected: connected is True
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = "robot2048"
        self.utility_wrap.create_user(user=user, password=old_password)
        self.utility_wrap.reset_password(user=user, old_password=old_password, new_password=new_password)
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=new_password, check_task=ct.CheckTasks.ccr)
        self.utility_wrap.list_collections()

    @pytest.mark.tags(ct.CaseLabel.L3)
    def test_list_usernames(self, host, port):
        """
        target: test the user list created successfully
        method: get a list of users
        expected: list all users
        """
        # 1. default user login
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)

        # 2. create 2 users
        self.utility_wrap.create_user(user="user1", password="abc123")
        self.utility_wrap.create_user(user="user2", password="abc123")

        # 3. list all users
        res = self.utility_wrap.list_usernames()[0]
        assert "user1" and "user2" in res

    @pytest.mark.tags(ct.CaseLabel.L3)
    @pytest.mark.parametrize("connect_name", [DefaultConfig.DEFAULT_USING])
    def test_delete_user_with_username(self, host, port, connect_name):
        """
        target: test deleting user with username
        method: delete user with username and connect with the wrong user then list collections
        expected: deleted successfully
        """
        user = "xiaoai"
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        self.utility_wrap.create_user(user=user, password="abc123")
        self.utility_wrap.delete_user(user=user)
        self.connection_wrap.disconnect(alias=connect_name)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password="abc123", check_task=ct.CheckTasks.ccr)
        self.utility_wrap.list_collections(check_task=ct.CheckTasks.err_res,
                                           check_items={ct.err_code: 1})

    @pytest.mark.tags(ct.CaseLabel.L3)
    def test_delete_user_with_invalid_username(self, host, port):
        """
        target: test the nonexistant user when deleting credential
        method: delete a credential with user wrong
        excepted: delete is true
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        self.utility_wrap.delete_user(user="asdfghj")

    @pytest.mark.tags(ct.CaseLabel.L3)
    def test_delete_all_users(self, host, port):
        """
        target: delete the users that created for test
        method: delete the users in list_usernames except root
        excepted: delete is true
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        res = self.utility_wrap.list_usernames()[0]
        for user in res:
            if user != "root":
                self.utility_wrap.delete_user(user=user)
        res = self.utility_wrap.list_usernames()[0]
        assert len(res) == 1


class TestUtilityInvalidUserPassword(TestcaseBase):
    """ Test invalid case of user interface """

    @pytest.mark.tags(ct.CaseLabel.L3)
    @pytest.mark.parametrize("user", ["qwertyuiopasdfghjklzxcvbnmqwertyui", "@*-.-*", "alisd/"])
    def test_create_user_with_invalid_username(self, host, port, user):
        """
        target: test the user when create user
        method: make the length of user beyond standard
        excepted: the creation is false
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        self.utility_wrap.create_user(user=user, password=ct.default_password,
                                      check_task=ct.CheckTasks.err_res,
                                      check_items={ct.err_code: 5})

    @pytest.mark.tags(ct.CaseLabel.L3)
    @pytest.mark.parametrize("user", ["alice123w"])
    def test_create_user_with_existed_username(self, host, port, user):
        """
        target: test the user when create user
        method: create a user, and then create a user with the same username
        excepted: the creation is false
        """
        # 1.default user login
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)

        # 2.create the first user successfully
        self.utility_wrap.create_user(user=user, password=ct.default_password)

        # 3.create the second user with the same username
        self.utility_wrap.create_user(user=user, password=ct.default_password,
                                      check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 29})

    @pytest.mark.tags(ct.CaseLabel.L3)
    @pytest.mark.parametrize("password", ["12345"])
    def test_create_user_with_invalid_password(self, host, port, password):
        """
        target: test the password when create user
        method: make the length of user exceed the limitation [6, 256]
        excepted: the creation is false
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = "alice"
        self.utility_wrap.create_user(user=user, password=password,
                                      check_task=ct.CheckTasks.err_res, check_items={ct.err_code: 5})

    @pytest.mark.tags(ct.CaseLabel.L3)
    @pytest.mark.parametrize("user", ["hobo89"])
    @pytest.mark.parametrize("old_password", ["qwaszx0"])
    def test_reset_password_with_invalid_username(self, host, port, user, old_password):
        """
        target: test the wrong user when resetting password
        method: create a user, and then reset the password with wrong username
        excepted: reset is false
        """
        # 1.default user login
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)

        # 2.create a user
        self.utility_wrap.create_user(user=user, password=old_password)

        # 3.reset password with the wrong username
        self.utility_wrap.reset_password(user="hobo", old_password=old_password, new_password="qwaszx1",
                                         check_task=ct.CheckTasks.err_res,
                                         check_items={ct.err_code: 30})

    @pytest.mark.tags(ct.CaseLabel.L3)
    @pytest.mark.parametrize("user", ["demo"])
    @pytest.mark.parametrize("old_password", ["qwaszx0"])
    @pytest.mark.parametrize("new_password", ["12345"])
    def test_reset_password_with_invalid_new_password(self, host, port, user, old_password, new_password):
        """
        target: test the new password when resetting password
        method: create a user, and then set a wrong new password
        excepted: reset is false
        """
        # 1.default user login
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)

        # 2.create a user
        self.utility_wrap.create_user(user=user, password=old_password)

        # 3.reset password with the wrong new password
        self.utility_wrap.reset_password(user=user, old_password=old_password, new_password=new_password,
                                         check_task=ct.CheckTasks.err_res,
                                         check_items={ct.err_code: 5})

    @pytest.mark.tags(ct.CaseLabel.L3)
    @pytest.mark.parametrize("user", ["genny"])
    def test_reset_password_with_invalid_old_password(self, host, port, user):
        """
        target: test the old password when resetting password
        method: create a credential, and then reset with a wrong old password
        excepted: reset is false
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        self.utility_wrap.create_user(user=user, password="qwaszx0")
        self.utility_wrap.reset_password(user=user, old_password="waszx0", new_password="123456",
                                         check_task=ct.CheckTasks.err_res,
                                         check_items={ct.err_code: 30})

    @pytest.mark.tags(ct.CaseLabel.L3)
    def test_delete_user_root(self, host, port):
        """
        target: test deleting user root when deleting credential
        method: connect and then delete the user root
        excepted: delete is false
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        self.utility_wrap.delete_user(user=ct.default_user, check_task=ct.CheckTasks.err_res,
                                      check_items={ct.err_code: 31})


class TestUtilityRBAC(TestcaseBase):
    @pytest.mark.tags(CaseLabel.L3)
    def test_clear_roles(self, host, port):
        """
        target: check get roles list and clear them
        method: remove all roles except admin and public
        expected: assert clear success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)

        # add user and bind to role
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)

        usernames, _ = self.utility_wrap.list_usernames()
        for username in usernames:
            if username != "root":
                self.utility_wrap.delete_user(username)

        u, _ = self.utility_wrap.create_user(user=user, password=password)

        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        self.utility_wrap.role_add_user(user)

        # get roles
        role_groups, _ = self.utility_wrap.list_roles(False)

        # drop roles
        for role_group in role_groups.groups:
            if role_group.role_name not in ['admin', 'public']:
                self.utility_wrap.init_role(role_group.role_name)
                g_list, _ = self.utility_wrap.role_list_grants()
                for g in g_list.groups:
                    self.utility_wrap.role_revoke(g.object, g.object_name, g.privilege)
                self.utility_wrap.role_drop()
        role_groups, _ = self.utility_wrap.list_roles(False)
        assert len(role_groups.groups) == 2

    @pytest.mark.tags(CaseLabel.L3)
    def test_role_list_user_with_root_user(self, host, port):
        """
        target: check list user
        method: check list user with root
        expected: assert list user success, and root has no roles
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)

        user_info, _ = self.utility_wrap.list_user("root", True)
        user_item = user_info.groups[0]
        assert user_item.roles == ()
        assert user_item.username == "root"

    @pytest.mark.tags(CaseLabel.L3)
    def test_role_list_users(self, host, port):
        """
        target: check list users
        method: check list users con
        expected: assert list users success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        # add user and bind to role
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)

        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        self.utility_wrap.role_add_user(user)

        # get users
        user_info, _ = self.utility_wrap.list_users(True)

        # check root user and new user
        root_exist = False
        new_user_exist = False
        for user_item in user_info.groups:
            if user_item.username == "root" and len(user_item.roles) == 0:
                root_exist = True
            if user_item.username == user and user_item.roles[0] == r_name:
                new_user_exist = True
        assert root_exist
        assert new_user_exist

    @pytest.mark.tags(CaseLabel.L3)
    def test_create_role(self, host, port):
        """
        target: test create role
        method: test create role with random name
        expected: assert role create success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name, check_task=CheckTasks.check_role_property,
                                    check_items={exp_name: r_name})
        assert not self.utility_wrap.role_is_exist()[0]
        self.utility_wrap.create_role()
        assert self.utility_wrap.role_is_exist()[0]

    @pytest.mark.tags(CaseLabel.L3)
    def test_drop_role(self, host, port):
        """
        target: test drop role
        method: create a role, drop this role
        expected: assert role drop success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        self.utility_wrap.role_drop()
        assert not self.utility_wrap.role_is_exist()[0]

    @pytest.mark.tags(CaseLabel.L3)
    def test_add_user_to_role(self, host, port):
        """
        target: test add user to role
        method: create a new user,add user to role
        expected: assert add user success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)

        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        self.utility_wrap.role_add_user(user)
        users, _ = self.utility_wrap.role_get_users()
        user_info, _ = self.utility_wrap.list_user(user, True)
        user_item = user_info.groups[0]
        assert r_name in user_item.roles
        assert user in users

    @pytest.mark.tags(CaseLabel.L3)
    def test_remove_user_from_role(self, host, port):
        """
        target: test remove user from role
        method: create a new user,add user to role, remove user from role
        expected: assert remove user from role success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)

        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_remove_user(user)
        users, _ = self.utility_wrap.role_get_users()
        assert len(users) == 0

    @pytest.mark.tags(CaseLabel.L3)
    def test_role_is_exist(self, host, port):
        """
        target: test role is existed
        method: check not exist role and exist role
        expected: assert is_exist interface is correct
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)
        r_not_exist = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        assert self.utility_wrap.role_is_exist()[0]
        self.utility_wrap.init_role(r_not_exist)
        assert not self.utility_wrap.role_is_exist()[0]

    @pytest.mark.tags(CaseLabel.L3)
    def test_role_grant_collection_insert(self, host, port):
        """
        target: test grant role collection insert privilege
        method: create one role and tow collections, grant one collection insert privilege
        expected: assert grant privilege success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        c_name_2 = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)

        self.utility_wrap.init_role(r_name, check_task=CheckTasks.check_role_property,
                                    check_items={exp_name: r_name})
        self.utility_wrap.create_role()
        self.utility_wrap.role_add_user(user)

        self.init_collection_wrap(name=c_name)
        self.init_collection_wrap(name=c_name_2)

        # verify user default privilege
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        collection_w = self.init_collection_wrap(name=c_name)
        data = cf.gen_default_list_data(ct.default_nb)
        collection_w.insert(data=data, check_task=CheckTasks.check_permission_deny)
        collection_w2 = self.init_collection_wrap(name=c_name_2)
        collection_w2.insert(data=data, check_task=CheckTasks.check_permission_deny)

        # grant user collection insert privilege
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.role_grant("Collection", c_name, "Insert")

        # verify user specific collection insert privilege
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        collection_w = self.init_collection_wrap(name=c_name)
        collection_w.insert(data=data)

        # verify grant scope
        index_params = {"index_type": "IVF_SQ8", "metric_type": "L2", "params": {"nlist": 64}}
        collection_w.create_index(ct.default_float_vec_field_name, index_params,
                                  check_task=CheckTasks.check_permission_deny)
        collection_w2 = self.init_collection_wrap(name=c_name_2)
        collection_w2.insert(data=data, check_task=CheckTasks.check_permission_deny)

    @pytest.mark.tags(CaseLabel.L3)
    def test_revoke_public_role_privilege(self, host, port):
        """
        target: revoke public role privilege
        method: revoke public role privilege
        expected: success to revoke
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        self.init_collection_wrap(name=c_name)
        u, _ = self.utility_wrap.create_user(user=user, password=password)

        self.utility_wrap.init_role("public")
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_revoke("Collection", c_name, "Insert")
        data = cf.gen_default_list_data(ct.default_nb)
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        collection_w = self.init_collection_wrap(name=c_name)
        collection_w.insert(data=data, check_task=CheckTasks.check_permission_deny)

        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        self.utility_wrap.init_role("public")
        self.utility_wrap.role_grant("Collection", c_name, "Insert")


    @pytest.mark.tags(CaseLabel.L3)
    def test_role_revoke_collection_privilege(self, host, port):
        """
        target: test revoke role collection privilege,
        method: create role and collection, grant role insert privilege, revoke privilege
        expected: assert revoke privilege success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)

        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        self.utility_wrap.role_add_user(user)

        self.init_collection_wrap(name=c_name)

        # grant user collection insert privilege
        self.utility_wrap.role_grant("Collection", c_name, "Insert")

        # verify user specific collection insert privilege
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        collection_w = self.init_collection_wrap(name=c_name)
        data = cf.gen_default_list_data(ct.default_nb)
        collection_w.insert(data=data)

        # revoke privilege
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.role_revoke("Collection", c_name, "Insert")

        # verify revoke is success
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        collection_w = self.init_collection_wrap(name=c_name)
        collection_w.insert(data=data, check_task=CheckTasks.check_permission_deny)

    @pytest.mark.tags(CaseLabel.L3)
    def test_role_revoke_global_privilege(self, host, port):
        """
        target: test revoke role global privilege,
        method: create role, grant role global createcollection privilege, revoke privilege
        expected: assert revoke privilege success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        c_name_2 = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)

        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        self.utility_wrap.role_add_user(user)

        # grant user Global CreateCollection privilege
        self.utility_wrap.role_grant("Global", "*", "CreateCollection")

        # verify user specific Global CreateCollection privilege
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        collection_w = self.init_collection_wrap(name=c_name)

        # revoke privilege
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.role_revoke("Global",  "*", "CreateCollection")

        # verify revoke is success
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        collection_w = self.init_collection_wrap(name=c_name_2,
                                                 check_task=CheckTasks.check_permission_deny)

    @pytest.mark.tags(CaseLabel.L3)
    def test_role_revoke_user_privilege(self, host, port):
        """
        target: test revoke role user privilege,
        method: create role, grant role user updateuser privilege, revoke privilege
        expected: assert revoke privilege success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)

        user_test = cf.gen_unique_str(prefix)
        password_test = cf.gen_unique_str(prefix)
        self.utility_wrap.create_user(user=user_test, password=password_test)

        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        self.utility_wrap.role_add_user(user)

        # grant user User UpdateUser privilege
        self.utility_wrap.role_grant("User", "*", "UpdateUser")
        self.utility_wrap.role_revoke("User", "*", "UpdateUser")

        # verify revoke is success
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        self.utility_wrap.reset_password(user=user_test, old_password=password_test, new_password=password,
                                         check_task=CheckTasks.check_permission_deny)

    @pytest.mark.tags(CaseLabel.L3)
    def test_role_list_grants(self, host, port):
        """
        target: test grant role privileges and list them
        method: grant role privileges and list them
        expected: assert list granted privileges success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)

        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        self.utility_wrap.role_add_user(user)

        self.init_collection_wrap(name=c_name)

        # grant user privilege
        self.utility_wrap.init_role(r_name)
        grant_list = cf.gen_grant_list(c_name)
        for grant_item in grant_list:
            self.utility_wrap.role_grant(grant_item["object"], grant_item["object_name"], grant_item["privilege"])

        # list grants
        g_list, _ = self.utility_wrap.role_list_grants()
        assert len(g_list.groups) == len(grant_list)

    @pytest.mark.tags(CaseLabel.L3)
    def test_drop_role_which_bind_user(self, host, port):
        """
        target: drop role which bind user
        method: create a role, bind user to the role, drop the role
        expected: drop success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)

        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        self.utility_wrap.role_add_user(user)

        self.utility_wrap.role_drop()
        assert not self.utility_wrap.role_is_exist()[0]

    @pytest.mark.tags(CaseLabel.L3)
    @pytest.mark.parametrize("name", ["admin", "public"])
    def test_add_user_to_default_role(self, name, host, port):
        """
        target: add user to admin role or public role
        method: create a user,add user to admin role or public role
        expected: add success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)

        self.utility_wrap.init_role(name)
        self.utility_wrap.role_add_user(user)
        users, _ = self.utility_wrap.role_get_users()
        user_info, _ = self.utility_wrap.list_user(user, True)
        user_item = user_info.groups[0]
        assert name in user_item.roles
        assert user in users

    @pytest.mark.tags(CaseLabel.L3)
    def test_add_root_to_new_role(self, host, port):
        """
        target: add root to new role
        method: add root to new role
        expected: add success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)

        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        self.utility_wrap.role_add_user("root")
        users, _ = self.utility_wrap.role_get_users()
        user_info, _ = self.utility_wrap.list_user("root", True)
        user_item = user_info.groups[0]
        assert r_name in user_item.roles
        assert "root" in users
        self.utility_wrap.role_drop()

    @pytest.mark.tags(CaseLabel.L3)
    def test_list_collection_grands_by_role_and_object(self, host, port):
        """
        target: list grants by role and object
        method: create a new role,grant role collection privilege,list grants by role and object
        expected: list success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        collection_w = self.init_collection_wrap(name=c_name)

        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        self.utility_wrap.role_grant("Collection", c_name, "Search")
        self.utility_wrap.role_grant("Collection", c_name, "Insert")

        g_list, _ = self.utility_wrap.role_list_grant("Collection", c_name)
        assert len(g_list.groups) == 2
        for g in g_list.groups:
            assert g.object == "Collection"
            assert g.object_name == c_name
            assert g.privilege in ["Search", "Insert"]
            self.utility_wrap.role_revoke(g.object, g.object_name, g.privilege)
        self.utility_wrap.role_drop()

    @pytest.mark.tags(CaseLabel.L3)
    def test_list_global_grands_by_role_and_object(self, host, port):
        """
        target: list grants by role and object
        method: create a new role,grant role global privilege,list grants by role and object
        expected: list success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)

        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        self.utility_wrap.role_grant("Global", "*", "CreateCollection")
        self.utility_wrap.role_grant("Global", "*", "All")

        g_list, _ = self.utility_wrap.role_list_grant("Global", "*")
        assert len(g_list.groups) == 2
        for g in g_list.groups:
            assert g.object == "Global"
            assert g.object_name == "*"
            assert g.privilege in ["CreateCollection", "All"]
            self.utility_wrap.role_revoke(g.object, g.object_name, g.privilege)
        self.utility_wrap.role_drop()

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_admin_role_privilege(self, host, port):
        """
        target: verify admin role privilege
        method: create a new user, bind to admin role, crud collection
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        self.utility_wrap.init_role("admin")
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)

        self.utility_wrap.role_add_user(user)

        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        collection_w = self.init_collection_wrap(name=c_name)
        data = cf.gen_default_list_data(ct.default_nb)
        collection_w.insert(data=data)
        collection_w.load()
        assert collection_w.num_entities == ct.default_nb
        collection_w.release()
        collection_w.drop()

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_grant_collection_load_privilege(self, host, port):
        """
        target: verify grant collection load privilege
        method: verify grant collection load privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_grant("Collection", c_name, "Load")
        collection_w = self.init_collection_wrap(name=c_name)
        data = cf.gen_default_list_data(ct.default_nb)
        mutation_res, _ = collection_w.insert(data=data)
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)

        collection_w.load()

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_grant_collection_release_privilege(self, host, port):
        """
        target: verify grant collection release privilege
        method: verify grant collection release privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_grant("Collection", c_name, "Release")
        collection_w = self.init_collection_wrap(name=c_name)
        data = cf.gen_default_list_data(ct.default_nb)
        mutation_res, _ = collection_w.insert(data=data)
        collection_w.load()
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)

        collection_w.release()

    @pytest.mark.tags(CaseLabel.L3)
    @pytest.mark.xfail(reason="https://github.com/milvus-io/milvus/issues/19012")
    def test_verify_grant_collection_compaction_privilege(self, host, port):
        """
        target: verify grant collection compaction privilege
        method: verify grant collection compaction privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        collection_w = self.init_collection_wrap(name=c_name)
        self.utility_wrap.role_grant("Collection", c_name, "Compaction")

        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        collection_w.compact()

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_grant_collection_insert_privilege(self, host, port):
        """
        target: verify grant collection insert privilege
        method: verify grant collection insert privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        collection_w = self.init_collection_wrap(name=c_name)
        self.utility_wrap.role_grant("Collection", c_name, "Insert")

        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        data = cf.gen_default_list_data(ct.default_nb)
        mutation_res, _ = collection_w.insert(data=data)

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_grant_collection_delete_privilege(self, host, port):
        """
        target: verify grant collection delete privilege
        method: verify grant collection delete privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        collection_w = self.init_collection_wrap(name=c_name)
        self.utility_wrap.role_grant("Collection", c_name, "Delete")
        data = cf.gen_default_list_data(ct.default_nb)
        mutation_res, _ = collection_w.insert(data=data)
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        tmp_expr = f'{ct.default_int64_field_name} in {[0]}'
        collection_w.delete(tmp_expr)

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_create_index_privilege(self, host, port):
        """
        target: verify grant create index privilege
        method: verify grant create index privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        collection_w = self.init_collection_wrap(name=c_name)
        self.utility_wrap.role_grant("Collection", c_name, "CreateIndex")
        self.utility_wrap.role_grant("Collection", c_name, "Flush")
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        self.index_wrap.init_index(collection_w.collection, ct.default_int64_field_name,
                                   default_index_params)

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_drop_index_privilege(self, host, port):
        """
        target: verify grant drop index privilege
        method: verify grant drop index privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        collection_w = self.init_collection_wrap(name=c_name)
        self.index_wrap.init_index(collection_w.collection, ct.default_int64_field_name,
                                   default_index_params)
        self.utility_wrap.role_grant("Collection", c_name, "DropIndex")
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        self.index_wrap.drop()

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_collection_search_privilege(self, host, port):
        """
        target: verify grant collection search privilege
        method: verify grant collection search privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        collection_w = self.init_collection_wrap(name=c_name)
        data = cf.gen_default_list_data(ct.default_nb)
        mutation_res, _ = collection_w.insert(data=data)
        collection_w.load()
        self.utility_wrap.role_grant("Collection", c_name, "Search")
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        vectors = [[random.random() for _ in range(ct.default_dim)] for _ in range(ct.default_nq)]
        collection_w.search(vectors[:ct.default_nq], ct.default_float_vec_field_name,
                            ct.default_search_params, ct.default_limit,
                            "int64 >= 0", check_task=CheckTasks.check_search_results,
                            check_items={"nq": ct.default_nq,
                                         "limit": ct.default_limit})

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_collection_flush_privilege(self, host, port):
        """
        target: verify grant collection flush privilege
        method: verify grant collection flush privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        collection_w = self.init_collection_wrap(name=c_name)
        self.utility_wrap.role_grant("Collection", c_name, "Flush")
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        collection_w.flush()

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_collection_query_privilege(self, host, port):
        """
        target: verify grant collection query privilege
        method: verify grant collection query privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        collection_w = self.init_collection_wrap(name=c_name)
        data = cf.gen_default_list_data(ct.default_nb)
        mutation_res, _ = collection_w.insert(data=data)
        collection_w.load()
        self.utility_wrap.role_grant("Collection", c_name, "Query")
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        default_term_expr = f'{ct.default_int64_field_name} in [0, 1]'
        res, _ = collection_w.query(default_term_expr)
        assert len(res) == 2

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_global_all_privilege(self, host, port):
        """
        target: verify grant global all privilege
        method: verify grant global all privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_grant("Global", "*", "All")
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)

        collection_w = self.init_collection_wrap(name=c_name)
        collection_w.drop()
        user_test = cf.gen_unique_str(prefix)
        password_test = cf.gen_unique_str(prefix)
        self.utility_wrap.create_user(user=user_test, password=password_test)
        r_test = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_test)
        self.utility_wrap.create_role()
        self.utility_wrap.role_add_user(user_test)
        self.utility_wrap.role_grant("Collection", c_name, "Insert")
        self.utility_wrap.role_revoke("Collection", c_name, "Insert")
        self.utility_wrap.role_remove_user(user_test)

        self.utility_wrap.delete_user(user=user_test)
        self.utility_wrap.role_drop()

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_global_create_collection_privilege(self, host, port):
        """
        target: verify grant global create collection privilege
        method: verify grant global create collection privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_grant("Global", "*", "CreateCollection")
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)

        self.init_collection_wrap(name=c_name)

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_global_drop_collection_privilege(self, host, port):
        """
        target: verify grant global drop collection privilege
        method: verify grant global drop collection privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_grant("Global", "*", "DropCollection")
        collection_w = self.init_collection_wrap(name=c_name)
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        collection_w.drop()

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_global_create_ownership_privilege(self, host, port):
        """
        target: verify grant global create ownership privilege
        method: verify grant global create ownership privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_grant("Global", "*", "CreateOwnership")
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        user_test = cf.gen_unique_str(prefix)
        password_test = cf.gen_unique_str(prefix)
        self.utility_wrap.create_user(user=user_test, password=password_test)
        r_test = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_test)
        self.utility_wrap.create_role()

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_global_drop_ownership_privilege(self, host, port):
        """
        target: verify grant global drop ownership privilege
        method: verify grant global drop ownership privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_grant("Global", "*", "DropOwnership")

        user_test = cf.gen_unique_str(prefix)
        password_test = cf.gen_unique_str(prefix)
        self.utility_wrap.create_user(user=user_test, password=password_test)
        r_test = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_test)
        self.utility_wrap.create_role()

        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)

        self.utility_wrap.role_drop()
        self.utility_wrap.delete_user(user=user_test)

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_global_select_ownership_privilege(self, host, port):
        """
        target: verify grant global select ownership privilege
        method: verify grant global select ownership privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_grant("Global", "*", "SelectOwnership")

        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)

        self.utility_wrap.list_usernames()
        self.utility_wrap.role_list_grants()
        self.utility_wrap.list_roles(False)

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_global_manage_ownership_privilege(self, host, port):
        """
        target: verify grant global manage ownership privilege
        method: verify grant global manage ownership privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        c_name = cf.gen_unique_str(prefix)
        collection_w = self.init_collection_wrap(name=c_name)
        user_test = cf.gen_unique_str(prefix)
        password_test = cf.gen_unique_str(prefix)
        self.utility_wrap.create_user(user=user_test, password=password_test)
        r_test = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_test)
        self.utility_wrap.create_role()

        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_grant("Global", "*", "ManageOwnership")

        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)

        self.utility_wrap.role_add_user(user_test)
        self.utility_wrap.role_remove_user(user_test)
        self.utility_wrap.role_grant("Collection", c_name, "Search")
        self.utility_wrap.role_revoke("Collection", c_name, "Search")

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_user_update_privilege(self, host, port):
        """
        target: verify grant user update privilege
        method: verify grant user update privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user_test = cf.gen_unique_str(prefix)
        password_test = cf.gen_unique_str(prefix)
        self.utility_wrap.create_user(user=user_test, password=password_test)
        r_test = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_test)
        self.utility_wrap.create_role()

        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_grant("User", "*", "UpdateUser")

        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)
        self.utility_wrap.reset_password(user=user_test, old_password=password_test, new_password=password)
        self.utility_wrap.update_password(user=user_test, old_password=password, new_password=password_test)

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_select_user_privilege(self, host, port):
        """
        target: verify grant select user privilege
        method: verify grant select user privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user_test = cf.gen_unique_str(prefix)
        password_test = cf.gen_unique_str(prefix)
        self.utility_wrap.create_user(user=user_test, password=password_test)
        r_test = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_test)
        self.utility_wrap.create_role()

        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_grant("User", "*", "SelectUser")

        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)

        self.utility_wrap.list_user(username=user_test, include_role_info=False)
        self.utility_wrap.list_users(include_role_info=False)

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_grant_privilege_with_wildcard_object_name(self, host, port):
        """
        target: verify grant privilege with wildcard instead of object name
        method: verify grant privilege with wildcard instead of object name
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        c_name_2 = cf.gen_unique_str(prefix)
        collection_w = self.init_collection_wrap(name=c_name)
        collection_w2 = self.init_collection_wrap(name=c_name_2)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_grant("Collection", "*", "Load")

        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)

        collection_w.load()
        collection_w2.load()

    @pytest.mark.tags(CaseLabel.L3)
    def test_verify_grant_privilege_with_wildcard_privilege(self, host, port):
        """
        target: verify grant privilege with wildcard instead of privilege
        method: verify grant privilege with wildcard instead of privilege
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        collection_w = self.init_collection_wrap(name=c_name)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        self.utility_wrap.role_add_user(user)
        self.utility_wrap.role_grant("Collection", "*", "*")

        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)

        collection_w.load()
        collection_w.release()
        collection_w.compact()
        data = cf.gen_default_list_data(ct.default_nb)
        mutation_res, _ = collection_w.insert(data=data)
        tmp_expr = f'{ct.default_int64_field_name} in {[0]}'
        collection_w.delete(tmp_expr)
        self.index_wrap.init_index(collection_w.collection, ct.default_int64_field_name,
                                   default_index_params)
        self.index_wrap.drop(ct.default_int64_field_name)
        vectors = [[random.random() for _ in range(ct.default_dim)] for _ in range(ct.default_nq)]
        collection_w.load()
        collection_w.search(vectors[:ct.default_nq], ct.default_float_vec_field_name,
                            ct.default_search_params, ct.default_limit,
                            "int64 >= 0")
        collection_w.flush()
        default_term_expr = f'{ct.default_int64_field_name} in [0, 1]'
        collection_w.query(default_term_expr)

    @pytest.mark.tags(CaseLabel.L3)
    def test_new_user_default_owns_public_role_permission(self, host, port):
        """
        target: new user owns public role privilege
        method: create a role,verify its permission
        expected: verify success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user_test = cf.gen_unique_str(prefix)
        password_test = cf.gen_unique_str(prefix)
        self.utility_wrap.create_user(user=user_test, password=password_test)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        c_name = cf.gen_unique_str(prefix)
        c_name_2 = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        collection_w = self.init_collection_wrap(name=c_name)
        _, _ = self.index_wrap.init_index(collection_w.collection, default_field_name, default_index_params)
        self.connection_wrap.disconnect(alias=DefaultConfig.DEFAULT_USING)
        self.connection_wrap.connect(host=host, port=port, user=user,
                                     password=password, check_task=ct.CheckTasks.ccr)

        # Collection permission deny
        collection_w.load(check_task=CheckTasks.check_permission_deny)
        collection_w.release(check_task=CheckTasks.check_permission_deny)
        collection_w.compact(check_task=CheckTasks.check_permission_deny)
        data = cf.gen_default_list_data(ct.default_nb)
        mutation_res, _ = collection_w.insert(data=data, check_task=CheckTasks.check_permission_deny)
        tmp_expr = f'{ct.default_int64_field_name} in {[0]}'
        collection_w.delete(tmp_expr, check_task=CheckTasks.check_permission_deny)
        self.index_wrap.drop(ct.default_int64_field_name, check_task=CheckTasks.check_permission_deny)
        self.index_wrap.init_index(collection_w.collection, ct.default_int64_field_name,
                                   default_index_params, check_task=CheckTasks.check_permission_deny)
        vectors = [[random.random() for _ in range(ct.default_dim)] for _ in range(ct.default_nq)]
        collection_w.search(vectors[:ct.default_nq], ct.default_float_vec_field_name,
                            ct.default_search_params, ct.default_limit,
                            "int64 >= 0", check_task=CheckTasks.check_permission_deny)
        collection_w.flush(check_task=CheckTasks.check_permission_deny)
        default_term_expr = f'{ct.default_int64_field_name} in [0, 1]'
        collection_w.query(default_term_expr, check_task=CheckTasks.check_permission_deny)
        # self.utility_wrap.bulk_load(c_name, check_task=CheckTasks.check_permission_deny)

        # Global permission deny
        self.init_collection_wrap(name=c_name_2, check_task=CheckTasks.check_permission_deny)
        collection_w.drop(check_task=CheckTasks.check_permission_deny)
        self.utility_wrap.create_user(user=c_name, password=password, check_task=CheckTasks.check_permission_deny)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role(check_task=CheckTasks.check_permission_deny)
        self.utility_wrap.delete_user(user=user, check_task=CheckTasks.check_permission_deny)
        self.utility_wrap.role_drop(check_task=CheckTasks.check_permission_deny)
        self.utility_wrap.list_usernames(check_task=CheckTasks.check_permission_deny)
        self.utility_wrap.role_list_grants(check_task=CheckTasks.check_permission_deny)
        self.utility_wrap.list_roles(False, check_task=CheckTasks.check_permission_deny)
        self.utility_wrap.role_add_user(user, check_task=CheckTasks.check_permission_deny)
        self.utility_wrap.role_remove_user(user, check_task=CheckTasks.check_permission_deny)
        self.utility_wrap.role_grant("Collection", c_name, "Insert", check_task=CheckTasks.check_permission_deny)
        self.utility_wrap.role_revoke("Collection", c_name, "Insert", check_task=CheckTasks.check_permission_deny)

        # User permission deny
        self.utility_wrap.reset_password(user=user_test, old_password=password_test, new_password=password,
                                         check_task=CheckTasks.check_permission_deny)
        self.utility_wrap.update_password(user=user_test, old_password=password, new_password=password_test,
                                         check_task=CheckTasks.check_permission_deny)
        self.utility_wrap.list_user(user_test, False, check_task=CheckTasks.check_permission_deny)

        # public role access
        collection_w.index()
        self.utility_wrap.list_collections()
        self.utility_wrap.has_collection(c_name)

    @pytest.mark.tags(CaseLabel.L3)
    @pytest.mark.parametrize("name", ["admin", "public"])
    def test_remove_user_from_default_role(self, name, host, port):
        """
        target: remove user from admin role or public role
        method: create a user,add user to admin role or public role,remove user from role
        expected: remove success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)

        self.utility_wrap.init_role(name)
        self.utility_wrap.role_add_user(user)
        users, _ = self.utility_wrap.role_get_users()
        user_info, _ = self.utility_wrap.list_user(user, True)
        user_item = user_info.groups[0]
        assert name in user_item.roles
        assert user in users

        self.utility_wrap.role_remove_user(user)
        users, _ = self.utility_wrap.role_get_users()
        assert user not in users

    @pytest.mark.tags(CaseLabel.L3)
    def test_remove_root_from_new_role(self, host, port):
        """
        target: remove root from new role
        method: create a new role, bind root to role,remove root from role
        expected: remove success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)

        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        assert self.utility_wrap.role_is_exist()[0]
        self.utility_wrap.role_add_user("root")
        users, _ = self.utility_wrap.role_get_users()
        user_info, _ = self.utility_wrap.list_user("root", True)
        user_item = user_info.groups[0]
        assert r_name in user_item.roles
        assert "root" in users

        self.utility_wrap.role_remove_user("root")
        users, _ = self.utility_wrap.role_get_users()
        assert "root" not in users
        self.utility_wrap.role_drop()


class TestUtilityNegativeRbac(TestcaseBase):
    @pytest.mark.tags(CaseLabel.L3)
    @pytest.mark.parametrize("name", ["longlonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglong"
                                      "longlonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglong"
                                      "longlonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglong"
                                      "longlonglonglong",
                                      "n%$#@!", "123n", " ", "''", "test-role", "ff ff", "中文"])
    def test_create_role_with_invalid_name(self, name, host, port):
        """
        target: create role with invalid name
        method: create role with invalid name
        expected: create fail
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        self.utility_wrap.init_role(name)

        error = {"err_code": 5}
        self.utility_wrap.create_role(check_task=CheckTasks.err_res, check_items=error)
        # get roles
        role_groups, _ = self.utility_wrap.list_roles(False)

        # drop roles
        for role_group in role_groups.groups:
            if role_group.role_name not in ['admin', 'public']:
                self.utility_wrap.init_role(role_group.role_name)
                g_list, _ = self.utility_wrap.role_list_grants()
                for g in g_list.groups:
                    self.utility_wrap.role_revoke(g.object, g.object_name, g.privilege)
                self.utility_wrap.role_drop()
        role_groups, _ = self.utility_wrap.list_roles(False)
        assert len(role_groups.groups) == 2

    @pytest.mark.tags(CaseLabel.L3)
    def test_create_exist_role(self, host, port):
        """
        target: check create an exist role fail
        method: double create a role with same name
        expected: fail to create
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        assert self.utility_wrap.role_is_exist()[0]
        error = {"err_code": 35,
                 "err_msg": "fail to create role"}
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role(check_task=CheckTasks.err_res, check_items=error)
        self.utility_wrap.role_drop()
        assert not self.utility_wrap.role_is_exist()[0]

    @pytest.mark.tags(CaseLabel.L3)
    @pytest.mark.parametrize("name", ["admin", "public"])
    def test_drop_admin_and_public_role(self, name, host, port):
        """
        target: drop admin and public role fail
        method: drop admin and public role fail
        expected: fail to drop
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(name)
        assert self.utility_wrap.role_is_exist()[0]
        error = {"err_code": 5,
                 "err_msg": "the role[%s] is a default role, which can\'t be dropped" % name}
        self.utility_wrap.role_drop(check_task=CheckTasks.err_res, check_items=error)
        assert self.utility_wrap.role_is_exist()[0]

    @pytest.mark.tags(CaseLabel.L3)
    def test_drop_role_which_not_exist(self, host, port):
        """
        target: drop role which not exist fail
        method: drop role which not exist
        expected: fail to drop
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        assert not self.utility_wrap.role_is_exist()[0]
        error = {"err_code": 36,
                 "err_msg": "the role isn\'t existed"}
        self.utility_wrap.role_drop(check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L3)
    def test_add_user_not_exist_role(self, host, port):
        """
        target: add user to not exist role
        method: create a user,add user to not exist role
        expected: fail to add
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)

        self.utility_wrap.init_role(r_name)
        assert not self.utility_wrap.role_is_exist()[0]

        error = {"err_code": 37,
                 "err_msg": "fail to check the role name"}
        self.utility_wrap.role_add_user(user, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L3)
    def test_add_not_exist_user_to_role(self, host, port):
        """
        target: add not exist user to role
        method: create a role,add not exist user to role
        expected: fail to add
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        user = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        assert self.utility_wrap.role_is_exist()[0]

        error = {"err_code": 37,
                 "err_msg": "fail to check the username"}
        self.utility_wrap.role_remove_user(user, check_task=CheckTasks.err_res, check_items=error)
        self.utility_wrap.role_add_user(user, check_task=CheckTasks.err_res, check_items=error)
        self.utility_wrap.role_drop()

    @pytest.mark.tags(CaseLabel.L3)
    @pytest.mark.parametrize("name", ["admin", "public"])
    def test_remove_root_from_default_role(self, name, host, port):
        """
        target: remove root from admin role or public role
        method: remove root from admin role or public role
        expected: remove success
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)

        self.utility_wrap.init_role(name)
        error = {"err_code": 37,
                 "err_msg": "fail to operate user to role"}
        self.utility_wrap.role_remove_user("root", check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L3)
    def test_remove_user_from_unbind_role(self, host, port):
        """
        target: remove user from unbind role
        method: create new role and new user, remove user from unbind role
        expected: fail to  remove
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)

        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        assert self.utility_wrap.role_is_exist()[0]

        error = {"err_code": 37,
                 "err_msg": "fail to operate user to role"}
        self.utility_wrap.role_remove_user(user, check_task=CheckTasks.err_res, check_items=error)
        self.utility_wrap.role_drop()

    @pytest.mark.tags(CaseLabel.L3)
    def test_remove_user_from_empty_role(self, host, port):
        """
        target: remove not exist user from role
        method: create new role, remove not exist user from unbind role
        expected: fail to remove
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)

        user = cf.gen_unique_str(prefix)
        password = cf.gen_unique_str(prefix)
        u, _ = self.utility_wrap.create_user(user=user, password=password)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        assert not self.utility_wrap.role_is_exist()[0]

        error = {"err_code": 37,
                 "err_msg": "fail to check the role name"}
        self.utility_wrap.role_remove_user(user, check_task=CheckTasks.err_res, check_items=error)
        users, _ = self.utility_wrap.role_get_users()
        assert user not in users

    @pytest.mark.tags(CaseLabel.L3)
    def test_remove_not_exist_user_from_role(self, host, port):
        """
        target: remove not exist user from role
        method: create new role, remove not exist user from unbind role
        expected: fail to remove
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)

        user = cf.gen_unique_str(prefix)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        assert self.utility_wrap.role_is_exist()[0]

        error = {"err_code": 37,
                 "err_msg": "fail to check the username"}
        self.utility_wrap.role_remove_user(user, check_task=CheckTasks.err_res, check_items=error)
        users, _ = self.utility_wrap.role_get_users()
        assert user not in users
        self.utility_wrap.role_drop()

    @pytest.mark.tags(CaseLabel.L3)
    def test_drop_role_with_bind_privilege(self, host, port):
        """
        target: drop role with bind privilege
        method: create a new role,grant role privilege,drop it
        expected: fail to drop
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)

        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        self.utility_wrap.role_grant("Collection", "*", "*")

        error = {"err_code": 36,
                 "err_msg": "fail to drop the role that it has privileges. Use REVOKE API to revoke privileges"}
        self.utility_wrap.role_drop(check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L3)
    def test_list_grant_by_not_exist_role(self, host, port):
        """
        target: list grants by not exist role
        method: list grants by not exist role
        expected: fail to list
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        error = {"err_code": 42,
                 "err_msg": "there is no value on key = by-dev/meta/root-coord/credential/roles/%s" % r_name}
        self.utility_wrap.role_list_grants(check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L3)
    def test_list_grant_by_role_and_not_exist_object(self, host, port):
        """
        target: list grants by role and not exist object
        method: list grants by role and not exist object
        expected: fail to list
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)
        o_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        error = {"err_code": 41,
                 "err_msg": "the object type in the object entity[name: %s] is invalid" % o_name}
        self.utility_wrap.role_list_grant(o_name, "*", check_task=CheckTasks.err_res, check_items=error)
        self.utility_wrap.role_drop()

    @pytest.mark.tags(CaseLabel.L3)
    def test_grant_privilege_with_object_not_exist(self, host, port):
        """
        target: grant privilege with not exist object
        method: grant privilege with not exist object
        expected: fail to grant
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)
        o_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        error = {"err_code": 41,
                 "err_msg": "the object type in the object entity[name: %s] is invalid" % o_name}
        self.utility_wrap.role_grant(o_name, "*", "*", check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L3)
    def test_grant_privilege_with_privilege_not_exist(self, host, port):
        """
        target: grant privilege with not exist privilege
        method: grant privilege with not exist privilege
        expected: fail to grant
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)
        p_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        error = {"err_code": 41, "err_msg": "the privilege name[%s] in the privilege entity is invalid" % p_name}
        self.utility_wrap.role_grant("Global", "*", p_name, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L3)
    def test_revoke_privilege_with_object_not_exist(self, host, port):
        """
        target: revoke privilege with not exist object
        method: revoke privilege with not exist object
        expected: fail to revoke
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)
        o_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        error = {"err_code": 41,
                 "err_msg": "the object type in the object entity[name: %s] is invalid" % o_name}
        self.utility_wrap.role_revoke(o_name, "*", "*", check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L3)
    def test_revoke_privilege_with_privilege_not_exist(self, host, port):
        """
        target: revoke privilege with not exist privilege
        method: revoke privilege with not exist privilege
        expected: fail to revoke
        """
        self.connection_wrap.connect(host=host, port=port, user=ct.default_user,
                                     password=ct.default_password, check_task=ct.CheckTasks.ccr)
        r_name = cf.gen_unique_str(prefix)
        p_name = cf.gen_unique_str(prefix)
        self.utility_wrap.init_role(r_name)
        self.utility_wrap.create_role()
        error = {"err_code": 41, "err_msg": "the privilege name[%s] in the privilege entity is invalid" % p_name}
        self.utility_wrap.role_revoke("Global", "*", p_name, check_task=CheckTasks.err_res, check_items=error)