import datetime
import logging
import random
import threading
import time

import numpy as np
import pytest
from api.milvus import CollectionClient
from base.testbase import TestBase
from pymilvus import Collection, CollectionSchema, DataType, FieldSchema
from utils.constant import CaseLabel, default_nb
from utils.util_log import test_log as logger
from utils.utils import gen_collection_name, gen_vector, get_data_by_payload


@pytest.mark.tags(CaseLabel.L0)
class TestCreateCollection(TestBase):
    @pytest.mark.parametrize("dim", [128])
    def test_create_collections_quick_setup(self, dim):
        """
        target: test create collection
        method: create a collection with a simple schema
        expected: create collection success
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "dimension": dim,
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()

        all_collections = rsp["data"]
        assert name in all_collections
        # describe collection
        rsp = client.collection_describe(name)
        assert rsp["code"] == 0
        assert rsp["data"]["collectionName"] == name
        assert rsp["data"]["autoId"] is False
        assert rsp["data"]["enableDynamicField"] is True
        assert "COSINE" in str(rsp["data"]["indexes"])

    @pytest.mark.parametrize("dim", [128])
    @pytest.mark.parametrize("metric_type", ["L2", "COSINE", "IP"])
    @pytest.mark.parametrize("id_type", ["Int64", "VarChar"])
    @pytest.mark.parametrize("primary_field", ["id", "url"])
    @pytest.mark.parametrize("vector_field", ["vector", "embedding"])
    def test_create_collection_quick_setup_with_custom(self, vector_field, primary_field, dim, id_type, metric_type):
        """
        Insert a vector with a simple payload
        """
        # create a collection
        name = gen_collection_name()
        collection_payload = {
            "collectionName": name,
            "dimension": dim,
            "metricType": metric_type,
            "primaryFieldName": primary_field,
            "vectorFieldName": vector_field,
            "idType": id_type,
        }
        if id_type == "VarChar":
            collection_payload["params"] = {"max_length": "256"}
        rsp = self.collection_client.collection_create(collection_payload)
        assert rsp["code"] == 0
        rsp = self.collection_client.collection_describe(name)
        logger.info(f"rsp: {rsp}")
        assert rsp["code"] == 0
        assert rsp["data"]["collectionName"] == name
        fields = [f["name"] for f in rsp["data"]["fields"]]
        assert primary_field in fields
        assert vector_field in fields
        for f in rsp["data"]["fields"]:
            if f["name"] == primary_field:
                assert f["type"] == id_type
                assert f["primaryKey"] is True
        for index in rsp["data"]["indexes"]:
            assert index["metricType"] == metric_type

    @pytest.mark.parametrize(
        "vector_field_type,dimension,expected_metric",
        [
            ("Float16Vector", 4, "COSINE"),
            ("BFloat16Vector", 4, "COSINE"),
            ("BinaryVector", 8, "HAMMING"),
            ("SparseFloatVector", None, "IP"),
        ],
    )
    def test_create_collection_quick_setup_vector_field_type_default_metric(
        self, vector_field_type, dimension, expected_metric
    ):
        """
        target: test quick create collection with vectorFieldType and default metricType
        method: create collection with BinaryVector/SparseFloatVector without metricType
        expected: create collection succeeds with compatible default index metric
        """
        name = gen_collection_name()
        client = self.collection_client
        payload = {
            "collectionName": name,
            "vectorFieldType": vector_field_type,
        }
        if dimension is not None:
            payload["dimension"] = dimension

        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0

        rsp = client.collection_describe(name)
        assert rsp["code"] == 0
        assert rsp["data"]["collectionName"] == name
        vector_fields = [field for field in rsp["data"]["fields"] if field["name"] == "vector"]
        assert len(vector_fields) == 1
        assert vector_fields[0]["type"] == vector_field_type
        assert len(rsp["data"]["indexes"]) == 1
        assert rsp["data"]["indexes"][0]["metricType"] == expected_metric

    @pytest.mark.parametrize("enable_dynamic_field", [False, "False", "0"])
    @pytest.mark.parametrize("request_shards_num", [2, "2"])
    @pytest.mark.parametrize("request_ttl_seconds", [360, "360"])
    def test_create_collections_without_params(self, enable_dynamic_field, request_shards_num, request_ttl_seconds):
        """
        target: test create collection
        method: create a collection with a simple schema
        expected: create collection success
        """
        name = gen_collection_name()
        dim = 128
        metric_type = "COSINE"
        client = self.collection_client
        num_shards = 2
        consistency_level = "Strong"
        ttl_seconds = 360
        payload = {
            "collectionName": name,
            "dimension": dim,
            "metricType": metric_type,
            "params": {
                "enableDynamicField": enable_dynamic_field,
                "shardsNum": request_shards_num,
                "consistencyLevel": f"{consistency_level}",
                "ttlSeconds": request_ttl_seconds,
            },
        }

        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()

        all_collections = rsp["data"]
        assert name in all_collections
        # describe collection by pymilvus
        c = Collection(name)
        res = c.describe()
        logger.info(f"describe collection: {res}")
        # describe collection
        time.sleep(10)
        rsp = client.collection_describe(name)
        logger.info(f"describe collection: {rsp}")

        ttl_seconds_actual = None
        for d in rsp["data"]["properties"]:
            if d["key"] == "collection.ttl.seconds":
                ttl_seconds_actual = int(d["value"])
        assert rsp["code"] == 0
        assert rsp["data"]["enableDynamicField"] is False
        assert rsp["data"]["collectionName"] == name
        assert rsp["data"]["shardsNum"] == num_shards
        assert rsp["data"]["consistencyLevel"] == consistency_level
        assert ttl_seconds_actual == ttl_seconds

    @pytest.mark.parametrize("primary_key_field", ["book_id"])
    @pytest.mark.parametrize("partition_key_field", ["word_count"])
    @pytest.mark.parametrize("clustering_key_field", ["book_category"])
    @pytest.mark.parametrize("shardsNum", [4])
    @pytest.mark.parametrize("partitionsNum", [16])
    @pytest.mark.parametrize("ttl_seconds", [60])
    @pytest.mark.parametrize("metric_type", ["L2", "COSINE", "IP"])
    @pytest.mark.parametrize("consistency_level", ["Strong", "Bounded"])
    @pytest.mark.parametrize("enable_dynamic_field", [True, False])
    @pytest.mark.parametrize("index_type", ["AUTOINDEX", "IVF_SQ8", "HNSW"])
    @pytest.mark.parametrize("dim", [128])
    def test_create_collections_with_all_params(
        self,
        dim,
        index_type,
        enable_dynamic_field,
        consistency_level,
        metric_type,
        ttl_seconds,
        partitionsNum,
        shardsNum,
        clustering_key_field,
        partition_key_field,
        primary_key_field,
    ):
        """
        target: test create collection
        method: create a collection with a simple schema
        expected: create collection success
        """
        name = gen_collection_name()
        dim = dim
        metric_type = metric_type
        client = self.collection_client
        num_shards = shardsNum
        num_partitions = partitionsNum
        consistency_level = consistency_level
        ttl_seconds = ttl_seconds
        index_param_map = {
            "FLAT": {},
            "IVF_SQ8": {"nlist": 16384},
            "HNSW": {"M": 16, "efConstruction": 500},
            "BM25_SPARSE_INVERTED_INDEX": {"bm25_k1": 0.5, "bm25_b": 0.5},
            "AUTOINDEX": {},
        }

        payload = {
            "collectionName": name,
            "params": {
                "shardsNum": f"{num_shards}",
                "partitionsNum": f"{num_partitions}",
                "consistencyLevel": f"{consistency_level}",
                "ttlSeconds": f"{ttl_seconds}",
            },
            "schema": {
                "enableDynamicField": enable_dynamic_field,
                "fields": [
                    {"fieldName": "user_id", "dataType": "Int64", "elementTypeParams": {}},
                    {
                        "fieldName": "book_id",
                        "dataType": "Int64",
                        "isPrimary": primary_key_field == "book_id",
                        "elementTypeParams": {},
                    },
                    {
                        "fieldName": "word_count",
                        "dataType": "Int64",
                        "isPartitionKey": partition_key_field == "word_count",
                        "isClusteringKey": clustering_key_field == "word_count",
                        "elementTypeParams": {},
                    },
                    {
                        "fieldName": "book_category",
                        "dataType": "Int64",
                        "isPartitionKey": partition_key_field == "book_category",
                        "isClusteringKey": clustering_key_field == "book_category",
                        "elementTypeParams": {},
                    },
                    {"fieldName": "book_describe", "dataType": "VarChar", "elementTypeParams": {"max_length": "256"}},
                    {
                        "fieldName": "document_content",
                        "dataType": "VarChar",
                        "elementTypeParams": {
                            "max_length": "1000",
                            "enable_analyzer": True,
                            "analyzer_params": {"tokenizer": "standard"},
                            "enable_match": True,
                        },
                    },
                    {"fieldName": "json", "dataType": "JSON", "elementTypeParams": {}},
                    {
                        "fieldName": "int_array",
                        "dataType": "Array",
                        "elementDataType": "Int64",
                        "elementTypeParams": {"max_capacity": "1024"},
                    },
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                    {"fieldName": "sparse_vector", "dataType": "SparseFloatVector"},
                ],
                "functions": [
                    {
                        "name": "bm25_fn",
                        "type": "BM25",
                        "inputFieldNames": ["document_content"],
                        "outputFieldNames": ["sparse_vector"],
                        "params": {},
                    }
                ],
            },
            "indexParams": [
                {
                    "fieldName": "book_intro",
                    "indexName": "book_intro_vector",
                    "metricType": f"{metric_type}",
                    "indexType": index_type,
                    "params": index_param_map[index_type],
                },
                {
                    "fieldName": "sparse_vector",
                    "indexName": "sparse_vector_index",
                    "metricType": "BM25",
                    "indexType": "SPARSE_INVERTED_INDEX",
                    "params": index_param_map["BM25_SPARSE_INVERTED_INDEX"],
                },
            ],
        }

        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()

        all_collections = rsp["data"]
        assert name in all_collections
        # describe collection by pymilvus
        c = Collection(name)
        res = c.describe()
        logger.info(f"pymilvus describe collection: {res}")
        # describe collection
        time.sleep(10)
        rsp = client.collection_describe(name)
        logger.info(f"restful describe collection: {rsp}")

        ttl_seconds_actual = None
        for d in rsp["data"]["properties"]:
            if d["key"] == "collection.ttl.seconds":
                ttl_seconds_actual = int(d["value"])
        assert rsp["code"] == 0
        assert rsp["data"]["collectionName"] == name
        assert rsp["data"]["enableDynamicField"] == enable_dynamic_field
        assert rsp["data"]["shardsNum"] == num_shards
        assert rsp["data"]["partitionsNum"] == num_partitions
        assert rsp["data"]["consistencyLevel"] == consistency_level
        assert ttl_seconds_actual == ttl_seconds
        assert len(rsp["data"]["functions"]) == len(payload["schema"]["functions"])
        #
        # # check fields properties
        fields = rsp["data"]["fields"]
        assert len(fields) == len(payload["schema"]["fields"])
        for field in fields:
            if field["name"] == primary_key_field:
                assert field["primaryKey"] is True
            if field["name"] == partition_key_field:
                assert field["partitionKey"] is True
            if field["name"] == clustering_key_field:
                assert field["clusteringKey"] is True

        # check index
        index_info = [index.to_dict() for index in c.indexes]
        logger.info(f"index_info: {index_info}")
        assert len(index_info) == 2
        for index in index_info:
            index_param = index["index_param"]
            if index_param["index_type"] == "SPARSE_INVERTED_INDEX":
                assert index_param["metric_type"] == "BM25"
                assert index_param.get("params", {}) == index_param_map["BM25_SPARSE_INVERTED_INDEX"]
            else:
                assert index_param["metric_type"] == metric_type
                assert index_param["index_type"] == index_type
                assert index_param.get("params", {}) == index_param_map[index_type]

    @pytest.mark.parametrize("auto_id", [True, False])
    @pytest.mark.parametrize("enable_dynamic_field", [True, False])
    @pytest.mark.parametrize("enable_partition_key", [True, False])
    @pytest.mark.parametrize("dim", [128])
    def test_create_collections_custom_without_index(self, dim, auto_id, enable_dynamic_field, enable_partition_key):
        """
        target: test create collection
        method: create a collection with a simple schema
        expected: create collection success
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "autoId": auto_id,
                "enableDynamicField": enable_dynamic_field,
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {
                        "fieldName": "user_id",
                        "dataType": "Int64",
                        "isPartitionKey": enable_partition_key,
                        "elementTypeParams": {},
                    },
                    {"fieldName": "word_count", "dataType": "Int64", "elementTypeParams": {}},
                    {"fieldName": "book_describe", "dataType": "VarChar", "elementTypeParams": {"max_length": "256"}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                    {"fieldName": "image_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ],
            },
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()

        all_collections = rsp["data"]
        assert name in all_collections
        c = Collection(name)
        logger.info(f"schema: {c.schema}")
        # describe collection
        rsp = client.collection_describe(name)
        assert rsp["code"] == 0
        assert rsp["data"]["collectionName"] == name
        assert rsp["data"]["autoId"] == auto_id
        assert c.schema.auto_id == auto_id
        assert rsp["data"]["enableDynamicField"] == enable_dynamic_field
        assert c.schema.enable_dynamic_field == enable_dynamic_field
        # assert no index created
        indexes = rsp["data"]["indexes"]
        assert len(indexes) == 0
        # assert not loaded
        assert rsp["data"]["load"] == "LoadStateNotLoad"
        for field in rsp["data"]["fields"]:
            if field["name"] == "user_id":
                assert field["partitionKey"] == enable_partition_key
        for field in c.schema.fields:
            if field.name == "user_id":
                assert field.is_partition_key == enable_partition_key

    @pytest.mark.parametrize("metric_type", ["L2", "IP", "COSINE"])
    @pytest.mark.parametrize("dim", [128])
    def test_create_collections_one_float_vector_with_index(self, dim, metric_type):
        """
        target: test create collection
        method: create a collection with a simple schema
        expected: create collection success
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "word_count", "dataType": "Int64", "elementTypeParams": {}},
                    {"fieldName": "book_describe", "dataType": "VarChar", "elementTypeParams": {"max_length": "256"}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
            "indexParams": [
                {"fieldName": "book_intro", "indexName": "book_intro_vector", "metricType": f"{metric_type}"}
            ],
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()

        all_collections = rsp["data"]
        assert name in all_collections
        # describe collection
        time.sleep(10)
        rsp = client.collection_describe(name)
        assert rsp["code"] == 0
        assert rsp["data"]["collectionName"] == name
        # assert index created
        indexes = rsp["data"]["indexes"]
        assert len(indexes) == len(payload["indexParams"])
        # assert load success
        assert rsp["data"]["load"] == "LoadStateLoaded"

    @pytest.mark.parametrize("metric_type", ["L2", "IP", "COSINE"])
    @pytest.mark.parametrize("dim", [128])
    def test_create_collections_multi_float_vector_with_one_index(self, dim, metric_type):
        """
        target: test create collection
        method: create a collection with multiple vector fields and only one index
        expected: create reports the missing vector index and the collection stays unloaded
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "word_count", "dataType": "Int64", "elementTypeParams": {}},
                    {"fieldName": "book_describe", "dataType": "VarChar", "elementTypeParams": {"max_length": "256"}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                    {"fieldName": "image_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
            "indexParams": [
                {"fieldName": "book_intro", "indexName": "book_intro_vector", "metricType": f"{metric_type}"}
            ],
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 1100
        assert "there is no vector index on field: [image_intro]" in rsp["message"]
        rsp = client.collection_list()

        all_collections = rsp["data"]
        assert name in all_collections
        # describe collection
        time.sleep(10)
        rsp = client.collection_describe(name)
        assert rsp["code"] == 0
        assert rsp["data"]["collectionName"] == name
        # assert index created
        indexes = rsp["data"]["indexes"]
        assert len(indexes) == len(payload["indexParams"])
        # assert load success
        assert rsp["data"]["load"] == "LoadStateNotLoad"

    @pytest.mark.parametrize("metric_type", ["L2", "IP", "COSINE"])
    @pytest.mark.parametrize("dim", [128])
    def test_create_collections_multi_float_vector_with_all_index(self, dim, metric_type):
        """
        target: test create collection
        method: create a collection with a simple schema
        expected: create collection success
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "word_count", "dataType": "Int64", "elementTypeParams": {}},
                    {"fieldName": "book_describe", "dataType": "VarChar", "elementTypeParams": {"max_length": "256"}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                    {"fieldName": "image_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
            "indexParams": [
                {"fieldName": "book_intro", "indexName": "book_intro_vector", "metricType": f"{metric_type}"},
                {"fieldName": "image_intro", "indexName": "image_intro_vector", "metricType": f"{metric_type}"},
            ],
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()

        all_collections = rsp["data"]
        assert name in all_collections
        # describe collection
        time.sleep(10)
        rsp = client.collection_describe(name)
        assert rsp["code"] == 0
        assert rsp["data"]["collectionName"] == name
        # assert index created
        indexes = rsp["data"]["indexes"]
        assert len(indexes) == len(payload["indexParams"])
        # assert load success
        assert rsp["data"]["load"] in ["LoadStateLoaded", "LoadStateLoading"]

    @pytest.mark.parametrize("auto_id", [True])
    @pytest.mark.parametrize("enable_dynamic_field", [True])
    @pytest.mark.parametrize("enable_partition_key", [True])
    @pytest.mark.parametrize("dim", [128])
    @pytest.mark.parametrize("metric_type", ["L2", "IP", "COSINE"])
    def test_create_collections_float16_vector_datatype(
        self, dim, auto_id, enable_dynamic_field, enable_partition_key, metric_type
    ):
        """
        target: test create collection
        method: create a collection with a simple schema
        expected: create collection success
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "autoId": auto_id,
                "enableDynamicField": enable_dynamic_field,
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {
                        "fieldName": "float16_vector",
                        "dataType": "Float16Vector",
                        "elementTypeParams": {"dim": f"{dim}"},
                    },
                    {
                        "fieldName": "bfloat16_vector",
                        "dataType": "BFloat16Vector",
                        "elementTypeParams": {"dim": f"{dim}"},
                    },
                ],
            },
            "indexParams": [
                {"fieldName": "float16_vector", "indexName": "float16_vector_index", "metricType": f"{metric_type}"},
                {"fieldName": "bfloat16_vector", "indexName": "bfloat16_vector_index", "metricType": f"{metric_type}"},
            ],
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()

        all_collections = rsp["data"]
        assert name in all_collections
        c = Collection(name)
        logger.info(f"schema: {c.schema}")
        # describe collection
        rsp = client.collection_describe(name)
        assert rsp["code"] == 0
        assert rsp["data"]["collectionName"] == name
        assert len(rsp["data"]["fields"]) == len(c.schema.fields)

    @pytest.mark.parametrize("auto_id", [True])
    @pytest.mark.parametrize("enable_dynamic_field", [True])
    @pytest.mark.parametrize("enable_partition_key", [True])
    @pytest.mark.parametrize("dim", [128])
    @pytest.mark.parametrize("metric_type", ["JACCARD", "HAMMING"])
    def test_create_collections_binary_vector_datatype(
        self, dim, auto_id, enable_dynamic_field, enable_partition_key, metric_type
    ):
        """
        target: test create collection
        method: create a collection with a simple schema
        expected: create collection success
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "autoId": auto_id,
                "enableDynamicField": enable_dynamic_field,
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "binary_vector", "dataType": "BinaryVector", "elementTypeParams": {"dim": f"{dim}"}},
                ],
            },
            "indexParams": [
                {"fieldName": "binary_vector", "indexName": "binary_vector_index", "metricType": f"{metric_type}"}
            ],
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()

        all_collections = rsp["data"]
        assert name in all_collections
        c = Collection(name)
        logger.info(f"schema: {c.schema}")
        # describe collection
        rsp = client.collection_describe(name)
        assert rsp["code"] == 0
        assert rsp["data"]["collectionName"] == name
        assert len(rsp["data"]["fields"]) == len(c.schema.fields)

    def test_create_collections_concurrent_with_same_param(self):
        """
        target: test create collection with same param
        method: concurrent create collections with same param with multi thread
        expected: create collections all success
        """
        concurrent_rsp = []

        def create_collection(c_name, vector_dim, c_metric_type):
            collection_payload = {
                "collectionName": c_name,
                "dimension": vector_dim,
                "metricType": c_metric_type,
            }
            rsp = client.collection_create(collection_payload)
            concurrent_rsp.append(rsp)
            logger.info(rsp)

        name = gen_collection_name()
        dim = 128
        metric_type = "L2"
        client = self.collection_client
        threads = []
        for i in range(10):
            t = threading.Thread(
                target=create_collection,
                args=(
                    name,
                    dim,
                    metric_type,
                ),
            )
            threads.append(t)
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        time.sleep(10)
        success_cnt = 0
        for rsp in concurrent_rsp:
            if rsp["code"] == 0:
                success_cnt += 1
        logger.info(concurrent_rsp)
        assert success_cnt == 10
        rsp = client.collection_list()
        all_collections = rsp["data"]
        assert name in all_collections
        # describe collection
        rsp = client.collection_describe(name)
        assert rsp["code"] == 0
        assert rsp["data"]["collectionName"] == name

    def test_create_collections_concurrent_with_different_param(self):
        """
        target: test create collection with different param
        method: concurrent create collections with different param with multi thread
        expected: only one collection can success
        """
        concurrent_rsp = []

        def create_collection(c_name, vector_dim, c_metric_type):
            collection_payload = {
                "collectionName": c_name,
                "dimension": vector_dim,
                "metricType": c_metric_type,
            }
            rsp = client.collection_create(collection_payload)
            concurrent_rsp.append(rsp)
            logger.info(rsp)

        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        threads = []
        for i in range(0, 5):
            t = threading.Thread(
                target=create_collection,
                args=(
                    name,
                    dim + i,
                    "L2",
                ),
            )
            threads.append(t)
        for i in range(5, 10):
            t = threading.Thread(
                target=create_collection,
                args=(
                    name,
                    dim + i,
                    "IP",
                ),
            )
            threads.append(t)
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        time.sleep(10)
        success_cnt = 0
        for rsp in concurrent_rsp:
            if rsp["code"] == 0:
                success_cnt += 1
        logger.info(concurrent_rsp)
        assert success_cnt == 1
        rsp = client.collection_list()
        all_collections = rsp["data"]
        assert name in all_collections
        # describe collection
        rsp = client.collection_describe(name)
        assert rsp["code"] == 0
        assert rsp["data"]["collectionName"] == name

    def test_create_collections_with_nullable_default(self):
        """
        target: test create collection
        method: create a collection with default none
        expected: create collection success
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "word_count", "dataType": "Int64", "elementTypeParams": {}, "defaultValue": 100},
                    {
                        "fieldName": "book_describe",
                        "dataType": "VarChar",
                        "elementTypeParams": {"max_length": "256"},
                        "nullable": True,
                        "defaultValue": "123",
                    },
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()

        all_collections = rsp["data"]
        assert name in all_collections
        # describe collection
        rsp = client.collection_describe(name)
        assert rsp["code"] == 0
        assert rsp["data"]["fields"][2]["defaultValue"] == {"Data": {"StringData": "123"}}
        assert rsp["data"]["fields"][2]["nullable"] is True


@pytest.mark.tags(CaseLabel.L1)
class TestCreateCollectionNegative(TestBase):
    def test_create_collections_custom_with_invalid_datatype(self):
        """
        VARCHAR is not a valid data type, it should be VarChar
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "word_count", "dataType": "Int64", "elementTypeParams": {}},
                    {"fieldName": "book_describe", "dataType": "VARCHAR", "elementTypeParams": {"max_length": "256"}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 1100

    def test_create_collections_custom_with_invalid_params(self):
        """ """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "enableDynamicField": 1,
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "word_count", "dataType": "Int64", "elementTypeParams": {}},
                    {"fieldName": "book_describe", "dataType": "VarChar", "elementTypeParams": {"max_length": "256"}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ],
            },
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 1801

    def test_create_collection_quick_setup_with_invalid_vector_field_type(self):
        """
        target: test quick create collection with invalid vectorFieldType
        method: create collection with invalid vectorFieldType
        expected: create collection failed with right error message
        """
        name = gen_collection_name()
        client = self.collection_client
        payload = {
            "collectionName": name,
            "dimension": 4,
            "metricType": "L2",
            "idType": "Int64",
            "autoID": True,
            "vectorFieldType": "InvalidVectorType",
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 1100
        assert "vectorFieldType can only be" in rsp["message"]

        rsp = client.collection_list()
        assert name not in rsp["data"]

    def test_create_collection_quick_setup_with_invalid_consistency_level(self):
        """
        target: test quick create collection with invalid top-level consistencyLevel
        method: create collection with invalid consistencyLevel
        expected: create collection failed with right error message
        """
        name = gen_collection_name()
        client = self.collection_client
        payload = {
            "collectionName": name,
            "dimension": 4,
            "consistencyLevel": "Invalid",
        }
        logging.info(f"create collection {name} with payload: {payload}")
        url = f"{client.endpoint}/v2/vectordb/collections/create"
        rsp = client.post(url, headers=client.update_headers(), data=payload).json()
        assert rsp["code"] == 1100
        assert "consistencyLevel can only be" in rsp["message"]

        rsp = client.collection_list()
        assert name not in rsp["data"]

    def test_create_collection_quick_setup_sparse_vector_with_dimension(self):
        """
        target: test quick create collection with SparseFloatVector and dimension
        method: create collection with SparseFloatVector and dimension
        expected: create collection failed with right error message
        """
        name = gen_collection_name()
        client = self.collection_client
        payload = {
            "collectionName": name,
            "dimension": 4,
            "vectorFieldType": "SparseFloatVector",
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 1100
        assert "dimension should not be specified for SparseFloatVector quick create" in rsp["message"]

        rsp = client.collection_list()
        assert name not in rsp["data"]

    @pytest.mark.parametrize(
        "vector_field_type,dimension,metric_type,expected_message",
        [
            ("BinaryVector", 8, "COSINE", "binary vector index does not support metric type: COSINE"),
            ("BinaryVector", 8, "SUBSTRUCTURE", "binary vector index does not support metric type: SUBSTRUCTURE"),
            ("BinaryVector", 8, "SUPERSTRUCTURE", "binary vector index does not support metric type: SUPERSTRUCTURE"),
            ("BinaryVector", 8, "MHJACCARD", "binary vector index does not support metric type: MHJACCARD"),
            ("SparseFloatVector", None, "COSINE", "only IP&BM25 is the supported metric type for sparse index"),
            ("SparseFloatVector", None, "BM25", "only BM25 Function output field support BM25 metric type"),
        ],
    )
    def test_create_collection_quick_setup_vector_field_type_invalid_metric(
        self, vector_field_type, dimension, metric_type, expected_message
    ):
        """
        target: test quick create collection with vectorFieldType and invalid metricType
        method: create collection with BinaryVector/SparseFloatVector and incompatible metricType
        expected: create collection failed before collection is created
        """
        name = gen_collection_name()
        client = self.collection_client
        payload = {
            "collectionName": name,
            "metricType": metric_type,
            "vectorFieldType": vector_field_type,
        }
        if dimension is not None:
            payload["dimension"] = dimension
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 1100
        assert expected_message in rsp["message"]

        rsp = client.collection_list()
        assert name not in rsp["data"]

    @pytest.mark.parametrize(
        "name", [" ", "test_collection_" * 100, "test collection", "test/collection", r"test\collection"]
    )
    def test_create_collections_with_invalid_collection_name(self, name):
        """
        target: test create collection with invalid collection name
        method: create collections with invalid collection name
        expected: create collection failed with right error message
        """
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "dimension": dim,
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 1100
        assert "Invalid collection name" in rsp["message"] or "invalid parameter" in rsp["message"]

    def test_create_collections_with_partition_key_nullable(self):
        """
        partition key field not support nullable
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {
                        "fieldName": "word_count",
                        "dataType": "Int64",
                        "elementTypeParams": {},
                        "isPartitionKey": True,
                        "nullable": True,
                    },
                    {"fieldName": "book_describe", "dataType": "VarChar", "elementTypeParams": {"max_length": "256"}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 1100
        assert "partition key field not support nullable" in rsp["message"]

    def test_create_collections_with_primary_default(self):
        """
        primary key field not support defaultValue
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {
                        "fieldName": "book_id",
                        "dataType": "Int64",
                        "isPrimary": True,
                        "elementTypeParams": {},
                        "defaultValue": 123,
                    },
                    {"fieldName": "book_describe", "dataType": "VarChar", "elementTypeParams": {"max_length": "256"}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 1100
        assert "primary field not support default_value" in rsp["message"]

    def test_create_collections_with_json_field_default(self):
        """
        json field not support default value
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "word_count", "dataType": "Int64", "elementTypeParams": {}},
                    {"fieldName": "book_describe", "dataType": "VarChar", "elementTypeParams": {"max_length": "256"}},
                    {"fieldName": "json", "dataType": "JSON", "elementTypeParams": {}, "defaultValue": {"key": 1}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 1100
        assert "convert defaultValue fail" in rsp["message"]

    def test_create_collections_with_array_field_default(self):
        """
        array field not support default value
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "word_count", "dataType": "Int64", "elementTypeParams": {}},
                    {
                        "fieldName": "int_array",
                        "dataType": "Array",
                        "elementDataType": "Int64",
                        "defaultValue": [1, 2],
                        "elementTypeParams": {"max_capacity": "1024"},
                    },
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
        }
        logging.info(f"create collection {name} with payload: {payload}")
        rsp = client.collection_create(payload)
        assert rsp["code"] == 1100
        assert "convert defaultValue fail" in rsp["message"]


@pytest.mark.tags(CaseLabel.L0)
class TestHasCollections(TestBase):
    def test_has_collections_default(self):
        """
        target: test list collection with a simple schema
        method: create collections and list them
        expected: created collections are in list
        """
        client = self.collection_client
        name_list = []
        for i in range(2):
            name = gen_collection_name()
            dim = 128
            payload = {
                "collectionName": name,
                "metricType": "L2",
                "dimension": dim,
            }
            time.sleep(1)
            rsp = client.collection_create(payload)
            assert rsp["code"] == 0
            name_list.append(name)
        rsp = client.collection_list()
        all_collections = rsp["data"]
        for name in name_list:
            assert name in all_collections
            rsp = client.collection_has(collection_name=name)
            assert rsp["data"]["has"] is True

    def test_has_collections_with_not_exist_name(self):
        """
        target: test list collection with a simple schema
        method: create collections and list them
        expected: created collections are in list
        """
        client = self.collection_client
        name_list = []
        for i in range(2):
            name = gen_collection_name()
            name_list.append(name)
        rsp = client.collection_list()
        all_collections = rsp["data"]
        for name in name_list:
            assert name not in all_collections
            rsp = client.collection_has(collection_name=name)
            assert rsp["data"]["has"] is False


@pytest.mark.tags(CaseLabel.L0)
class TestGetCollectionStats(TestBase):
    def test_get_collections_stats(self):
        """
        target: test list collection with a simple schema
        method: create collections and list them
        expected: created collections are in list
        """
        client = self.collection_client
        name = gen_collection_name()
        dim = 128
        payload = {
            "collectionName": name,
            "metricType": "L2",
            "dimension": dim,
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        # describe collection
        client.collection_describe(collection_name=name)
        rsp = client.collection_stats(collection_name=name)
        assert rsp["code"] == 0
        assert rsp["data"]["rowCount"] == 0
        # insert data
        nb = default_nb
        data = get_data_by_payload(payload, nb)
        payload = {"collectionName": name, "data": data}
        self.vector_client.vector_insert(payload=payload)
        c = Collection(name)
        count = c.query(expr="", output_fields=["count(*)"])
        logger.info(f"count: {count}")
        c.flush()
        rsp = client.collection_stats(collection_name=name)
        assert rsp["data"]["rowCount"] == nb


@pytest.mark.tags(CaseLabel.L0)
class TestLoadReleaseCollection(TestBase):
    def test_load_and_release_collection(self):
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "word_count", "dataType": "Int64", "elementTypeParams": {}},
                    {"fieldName": "book_describe", "dataType": "VarChar", "elementTypeParams": {"max_length": "256"}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        # create index before load
        index_params = [{"fieldName": "book_intro", "indexName": "book_intro_vector", "metricType": "L2"}]
        payload = {"collectionName": name, "indexParams": index_params}
        rsp = self.index_client.index_create(payload)

        # get load state before load
        rsp = client.collection_load_state(collection_name=name)
        assert rsp["data"]["loadState"] == "LoadStateNotLoad"

        # describe collection
        client.collection_describe(collection_name=name)
        rsp = client.collection_load(collection_name=name)
        assert rsp["code"] == 0
        rsp = client.collection_load_state(collection_name=name)
        assert rsp["data"]["loadState"] in ["LoadStateLoaded", "LoadStateLoading"]
        time.sleep(5)
        rsp = client.collection_load_state(collection_name=name)
        assert rsp["data"]["loadState"] == "LoadStateLoaded"

        # release collection
        rsp = client.collection_release(collection_name=name)
        time.sleep(5)
        rsp = client.collection_load_state(collection_name=name)
        assert rsp["data"]["loadState"] == "LoadStateNotLoad"


@pytest.mark.tags(CaseLabel.L0)
class TestGetCollectionLoadState(TestBase):
    def test_get_collection_load_state(self):
        """
        target: test list collection with a simple schema
        method: create collections and list them
        expected: created collections are in list
        """
        client = self.collection_client
        name = gen_collection_name()
        dim = 128
        payload = {
            "collectionName": name,
            "metricType": "L2",
            "dimension": dim,
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        # describe collection
        client.collection_describe(collection_name=name)
        rsp = client.collection_load_state(collection_name=name)
        assert rsp["code"] == 0
        t0 = time.time()
        while time.time() - t0 < 10:
            rsp = client.collection_load_state(collection_name=name)
            if rsp["data"]["loadState"] != "LoadStateNotLoad":
                break
            time.sleep(1)
        assert rsp["data"]["loadState"] in ["LoadStateLoading", "LoadStateLoaded"]
        # insert data
        nb = default_nb
        data = get_data_by_payload(payload, nb)
        payload = {"collectionName": name, "data": data}
        self.vector_client.vector_insert(payload=payload)
        rsp = client.collection_load_state(collection_name=name)
        assert rsp["data"]["loadState"] in ["LoadStateLoading", "LoadStateLoaded"]
        time.sleep(10)
        rsp = client.collection_load_state(collection_name=name)
        assert rsp["data"]["loadState"] == "LoadStateLoaded"


@pytest.mark.tags(CaseLabel.L0)
class TestListCollections(TestBase):
    def test_list_collections_default(self):
        """
        target: test list collection with a simple schema
        method: create collections and list them
        expected: created collections are in list
        """
        client = self.collection_client
        name_list = []
        for i in range(2):
            name = gen_collection_name()
            dim = 128
            payload = {
                "collectionName": name,
                "metricType": "L2",
                "dimension": dim,
            }
            time.sleep(1)
            rsp = client.collection_create(payload)
            assert rsp["code"] == 0
            name_list.append(name)
        rsp = client.collection_list()
        all_collections = rsp["data"]
        for name in name_list:
            assert name in all_collections


@pytest.mark.tags(CaseLabel.L0)
class TestDescribeCollection(TestBase):
    def test_describe_collections_default(self):
        """
        target: test describe collection with a simple schema
        method: describe collection
        expected: info of description is same with param passed to create collection
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {"collectionName": name, "dimension": dim, "metricType": "L2"}
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()
        all_collections = rsp["data"]
        assert name in all_collections
        # describe collection
        rsp = client.collection_describe(name)
        assert rsp["code"] == 0
        assert rsp["data"]["collectionName"] == name
        assert rsp["data"]["autoId"] is False
        assert rsp["data"]["enableDynamicField"] is True
        assert len(rsp["data"]["indexes"]) == 1

    def test_describe_collections_custom(self):
        """
        target: test describe collection with a simple schema
        method: describe collection
        expected: info of description is same with param passed to create collection
        """
        name = gen_collection_name()
        client = self.collection_client
        fields = [
            FieldSchema(name="reviewer_id", dtype=DataType.INT64, description="", is_primary=True),
            FieldSchema(
                name="store_address", dtype=DataType.VARCHAR, description="", max_length=512, is_partition_key=True
            ),
            FieldSchema(name="review", dtype=DataType.VARCHAR, description="", max_length=16384),
            FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, description="", dim=384, is_index=True),
        ]

        schema = CollectionSchema(
            fields=fields,
            description="",
            enable_dynamic_field=True,
            # The following is an alternative to setting `is_partition_key` in a field schema.
            partition_key_field="store_address",
        )

        Collection(
            name=name,
            schema=schema,
        )
        logger.info(f"schema: {schema}")
        rsp = client.collection_list()
        all_collections = rsp["data"]
        assert name in all_collections
        # describe collection
        rsp = client.collection_describe(name)
        assert rsp["code"] == 0
        assert rsp["data"]["collectionName"] == name

        for field in rsp["data"]["fields"]:
            if field["name"] == "store_address":
                assert field["partitionKey"] is True
            if field["name"] == "reviewer_id":
                assert field["primaryKey"] is True
        assert rsp["data"]["autoId"] is False
        assert rsp["data"]["enableDynamicField"] is True


@pytest.mark.tags(CaseLabel.L0)
class TestDescribeCollectionNegative(TestBase):
    def test_describe_collections_with_invalid_collection_name(self):
        """
        target: test describe collection with invalid collection name
        method: describe collection with invalid collection name
        expected: raise error with right error code and message
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "dimension": dim,
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()
        all_collections = rsp["data"]
        assert name in all_collections
        # describe collection
        invalid_name = "invalid_name"
        rsp = client.collection_describe(invalid_name)
        assert rsp["code"] == 100
        assert "can't find collection" in rsp["message"]


@pytest.mark.tags(CaseLabel.L0)
class TestDropCollection(TestBase):
    def test_drop_collections_default(self):
        """
        Drop a collection with a simple schema
        target: test drop collection with a simple schema
        method: drop collection
        expected: dropped collection was not in collection list
        """
        clo_list = []
        for i in range(5):
            time.sleep(1)
            name = "test_collection_" + datetime.datetime.now().strftime("%Y_%m_%d_%H_%M_%S_%f_%f")
            payload = {"collectionName": name, "dimension": 128, "metricType": "L2"}
            rsp = self.collection_client.collection_create(payload)
            assert rsp["code"] == 0
            clo_list.append(name)
        rsp = self.collection_client.collection_list()
        all_collections = rsp["data"]
        for name in clo_list:
            assert name in all_collections
        for name in clo_list:
            time.sleep(0.2)
            payload = {
                "collectionName": name,
            }
            rsp = self.collection_client.collection_drop(payload)
            assert rsp["code"] == 0
        rsp = self.collection_client.collection_list()
        all_collections = rsp["data"]
        for name in clo_list:
            assert name not in all_collections


@pytest.mark.tags(CaseLabel.L0)
class TestDropCollectionNegative(TestBase):
    def test_drop_collections_with_invalid_collection_name(self):
        """
        target: test drop collection with invalid collection name
        method: drop collection with invalid collection name
        expected: raise error with right error code and message
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "dimension": dim,
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()
        all_collections = rsp["data"]
        assert name in all_collections
        # drop collection
        invalid_name = "invalid_name"
        payload = {
            "collectionName": invalid_name,
        }
        rsp = client.collection_drop(payload)
        assert rsp["code"] == 0


@pytest.mark.tags(CaseLabel.L0)
class TestRenameCollection(TestBase):
    def test_rename_collection(self):
        """
        target: test rename collection
        method: rename collection
        expected: renamed collection is in collection list
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "metricType": "L2",
            "dimension": dim,
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()
        all_collections = rsp["data"]
        assert name in all_collections
        new_name = gen_collection_name()
        payload = {
            "collectionName": name,
            "newCollectionName": new_name,
        }
        rsp = client.collection_rename(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()
        all_collections = rsp["data"]
        assert new_name in all_collections
        assert name not in all_collections


@pytest.mark.tags(CaseLabel.RBAC)
class TestCollectionWithAuth(TestBase):
    def test_drop_collections_with_invalid_api_key(self):
        """
        target: test drop collection with invalid api key
        method: drop collection with invalid api key
        expected: raise error with right error code and message; collection still in collection list
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "dimension": dim,
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()
        all_collections = rsp["data"]
        assert name in all_collections
        # drop collection
        payload = {
            "collectionName": name,
        }
        illegal_client = CollectionClient(self.endpoint, "invalid_api_key")
        rsp = illegal_client.collection_drop(payload)
        assert rsp["code"] == 1800
        rsp = client.collection_list()
        all_collections = rsp["data"]
        assert name in all_collections

    def test_describe_collections_with_invalid_api_key(self):
        """
        target: test describe collection with invalid api key
        method: describe collection with invalid api key
        expected: raise error with right error code and message
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "dimension": dim,
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        rsp = client.collection_list()
        all_collections = rsp["data"]
        assert name in all_collections
        # describe collection
        illegal_client = CollectionClient(self.endpoint, "illegal_api_key")
        rsp = illegal_client.collection_describe(name)
        assert rsp["code"] == 1800

    def test_list_collections_with_invalid_api_key(self):
        """
        target: test list collection with an invalid api key
        method: list collection with invalid api key
        expected: raise error with right error code and message
        """
        client = self.collection_client
        name_list = []
        for i in range(2):
            name = gen_collection_name()
            dim = 128
            payload = {
                "collectionName": name,
                "metricType": "L2",
                "dimension": dim,
            }
            time.sleep(1)
            rsp = client.collection_create(payload)
            assert rsp["code"] == 0
            name_list.append(name)
        client = self.collection_client
        client.api_key = "illegal_api_key"
        rsp = client.collection_list()
        assert rsp["code"] == 1800

    def test_create_collections_with_invalid_api_key(self):
        """
        target: test create collection with invalid api key(wrong username and password)
        method: create collections with invalid api key
        expected: create collection failed
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        client.api_key = "illegal_api_key"
        payload = {
            "collectionName": name,
            "dimension": dim,
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 1800


@pytest.mark.tags(CaseLabel.L0)
class TestCollectionProperties(TestBase):
    """Test collection property operations"""

    def test_refresh_load_collection(self):
        """
        target: test refresh load collection
        method: create collection, refresh load
        expected: refresh load success
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "dimension": dim,
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0

        # release collection
        client.collection_release(collection_name=name)
        # load collection
        client.collection_load(collection_name=name)
        client.wait_load_completed(collection_name=name)
        # refresh load
        rsp = client.refresh_load(collection_name=name)

        assert rsp["code"] == 0

    def test_alter_collection_properties(self):
        """
        target: test alter collection properties
        method: create collection, alter properties
        expected: alter properties success
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "dimension": dim,
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        client.collection_release(collection_name=name)
        # alter properties
        properties = {"mmap.enabled": "true"}
        rsp = client.alter_collection_properties(name, properties)
        assert rsp["code"] == 0
        rsp = client.collection_describe(name)
        enabled_mmap = False
        for prop in rsp["data"]["properties"]:
            if prop["key"] == "mmap.enabled":
                assert prop["value"] == "true"
                enabled_mmap = True
        assert enabled_mmap

    def test_drop_collection_properties(self):
        """
        target: test drop collection properties
        method: create collection, alter properties, drop properties
        expected: drop properties success
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "dimension": dim,
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        client.collection_release(collection_name=name)

        # alter properties
        properties = {"mmap.enabled": "true"}
        rsp = client.alter_collection_properties(name, properties)
        assert rsp["code"] == 0
        rsp = client.collection_describe(name)
        enabled_mmap = False
        for prop in rsp["data"]["properties"]:
            if prop["key"] == "mmap.enabled":
                assert prop["value"] == "true"
                enabled_mmap = True
        assert enabled_mmap

        # drop properties
        delete_keys = ["mmap.enabled"]
        rsp = client.drop_collection_properties(name, delete_keys)
        assert rsp["code"] == 0
        rsp = client.collection_describe(name)
        enabled_mmap = False
        for prop in rsp["data"]["properties"]:
            if prop["key"] == "mmap.enabled":
                enabled_mmap = True
        assert not enabled_mmap

    def test_alter_field_properties(self):
        """
        target: test alter field properties
        method: create collection with varchar field, alter field properties
        expected: alter field properties success
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "autoId": True,
                "enableDynamicField": True,
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "user_id", "dataType": "Int64", "isPartitionKey": True, "elementTypeParams": {}},
                    {"fieldName": "word_count", "dataType": "Int64", "elementTypeParams": {}},
                    {"fieldName": "book_describe", "dataType": "VarChar", "elementTypeParams": {"max_length": "256"}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                    {"fieldName": "image_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ],
            },
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0
        # release collection
        client.collection_release(collection_name=name)

        # describe collection
        rsp = client.collection_describe(name)
        for field in rsp["data"]["fields"]:
            if field["name"] == "book_describe":
                for p in field["params"]:
                    if p["key"] == "max_length":
                        assert p["value"] == "256"

        # alter field properties
        field_params = {"max_length": "100"}
        rsp = client.alter_field_properties(name, "book_describe", field_params)
        assert rsp["code"] == 0

        # describe collection
        rsp = client.collection_describe(name)
        for field in rsp["data"]["fields"]:
            if field["name"] == "book_describe":
                for p in field["params"]:
                    if p["key"] == "max_length":
                        assert p["value"] == "100"


@pytest.mark.tags(CaseLabel.L1)
class TestCollectionAddField(TestBase):
    """Test collection add field operations"""

    @pytest.mark.parametrize(
        "field_params,test_data_generator,expected_validations",
        [
            # Test case 1: Int64 nullable field
            (
                {"fieldName": "new_int_field", "dataType": "Int64", "nullable": True, "elementTypeParams": {}},
                lambda i: i * 10,  # Generate int values
                {
                    "field_type": "Int64",
                    "nullable": True,
                    "has_default": False,
                    "data_validator": lambda item, i: item["new_int_field"] == i * 10,
                },
            ),
            # Test case 2: VarChar field with max_length
            (
                {
                    "fieldName": "new_varchar_field",
                    "dataType": "VarChar",
                    "nullable": True,
                    "elementTypeParams": {"max_length": "256"},
                },
                lambda i: f"description_{i}",  # Generate string values
                {
                    "field_type": "VarChar",
                    "nullable": True,
                    "has_default": False,
                    "max_length": "256",
                    "data_validator": lambda item, i: item["new_varchar_field"] == f"description_{i}",
                },
            ),
            # Test case 3: Int64 field with default value
            (
                {
                    "fieldName": "new_field_with_default",
                    "dataType": "Int64",
                    "nullable": True,
                    "defaultValue": 42,
                    "elementTypeParams": {},
                },
                lambda i: i * 100,  # Generate int values when explicitly provided
                {
                    "field_type": "Int64",
                    "nullable": True,
                    "has_default": True,
                    "default_value": 42,
                    "data_validator": lambda item, i: item["new_field_with_default"] == i * 100,
                },
            ),
            # Test case 4: Array field
            (
                {
                    "fieldName": "new_array_field",
                    "dataType": "Array",
                    "elementDataType": "Int64",
                    "nullable": True,
                    "elementTypeParams": {"max_capacity": "1024"},
                },
                lambda i: [i * 10, i * 20, i * 30],  # Generate array values
                {
                    "field_type": "Array",
                    "nullable": True,
                    "has_default": False,
                    "element_type": "Int64",
                    "data_validator": lambda item, i: item["new_array_field"] == [i * 10, i * 20, i * 30],
                },
            ),
        ],
    )
    def test_add_field_parametrized(self, field_params, test_data_generator, expected_validations):
        """
        target: test add various types of fields
        method: create collection, insert data, add field, insert and query again
        expected: add field success and data operations work before and after
        """
        name = gen_collection_name()
        dim = 128
        nb = default_nb  # Number of records to insert in each batch
        client = self.collection_client
        vector_client = self.vector_client
        field_name = field_params["fieldName"]

        # Create collection first
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
            "indexParams": [{"fieldName": "book_intro", "indexName": "book_intro_index", "metricType": "L2"}],
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0

        # Wait for collection to be loaded
        client.wait_load_completed(collection_name=name)

        # Insert data before adding field
        insert_data_before = []
        for i in range(nb):
            insert_data_before.append({"book_id": i, "book_intro": gen_vector(dim=dim)})

        insert_payload = {"collectionName": name, "data": insert_data_before}
        rsp = vector_client.vector_insert(insert_payload)
        assert rsp["code"] == 0

        # Query data before adding field
        query_payload = {"collectionName": name, "expr": "book_id >= 0", "outputFields": ["book_id"], "limit": nb}
        rsp = vector_client.vector_query(query_payload)
        assert rsp["code"] == 0
        assert len(rsp["data"]) == nb

        # Search data before adding field
        search_payload = {
            "collectionName": name,
            "data": [gen_vector(dim=dim)],
            "annsField": "book_intro",
            "limit": 100,
        }
        rsp = vector_client.vector_search(search_payload)
        assert rsp["code"] == 0
        assert len(rsp["data"]) > 0

        # Add field
        rsp = client.add_field(name, field_params)
        logger.info(f"add field response: {rsp}")
        assert rsp["code"] == 0

        # Verify field was added
        rsp = client.collection_describe(name)
        assert rsp["code"] == 0
        field_names = [field["name"] for field in rsp["data"]["fields"]]
        assert field_name in field_names

        # Check the field properties
        for field in rsp["data"]["fields"]:
            if field["name"] == field_name:
                assert field["type"] == expected_validations["field_type"]
                assert field["nullable"] == expected_validations["nullable"]

                # Check specific field type properties
                if expected_validations.get("max_length"):
                    for param in field.get("params", []):
                        if param["key"] == "max_length":
                            assert param["value"] == expected_validations["max_length"]

                if expected_validations.get("element_type"):
                    assert field.get("elementType") == expected_validations["element_type"]

                if expected_validations.get("has_default") and expected_validations["has_default"]:
                    assert field.get("defaultValue") is not None

        # Insert data after adding field
        insert_data_after = []
        for i in range(nb, nb * 2):
            data_item = {"book_id": i, "book_intro": gen_vector(dim=dim)}

            # For default value test, sometimes omit the field to test default behavior
            if expected_validations.get("has_default") and expected_validations["has_default"] and i < nb + nb // 2:
                # Don't add the field for first half of records to test default value
                pass
            else:
                # Add the field with generated test data
                data_item[field_name] = test_data_generator(i)

            insert_data_after.append(data_item)

        insert_payload = {"collectionName": name, "data": insert_data_after}
        rsp = vector_client.vector_insert(insert_payload)
        assert rsp["code"] == 0

        # Query data after adding field
        query_payload = {
            "collectionName": name,
            "expr": f"book_id >= {nb}",
            "outputFields": ["book_id", field_name],
            "limit": nb,
        }
        rsp = vector_client.vector_query(query_payload)
        assert rsp["code"] == 0
        assert len(rsp["data"]) == nb

        # Validate field data for records that have explicit values
        for item in rsp["data"]:
            assert field_name in item
            book_id = item["book_id"]
            # Only validate explicit values (not default values)
            if not (
                expected_validations.get("has_default")
                and expected_validations["has_default"]
                and book_id < nb + nb // 2
            ):
                if expected_validations.get("data_validator"):
                    expected_validations["data_validator"](item, book_id)

        # Search data after adding field
        search_payload = {
            "collectionName": name,
            "data": [gen_vector(dim=dim)],
            "annsField": "book_intro",
            "limit": 100,
            "outputFields": ["book_id", field_name],
        }
        rsp = vector_client.vector_search(search_payload)
        assert rsp["code"] == 0
        assert len(rsp["data"]) > 0

    def test_drop_field_by_name_and_id(self):
        name = gen_collection_name("drop_collection_field")
        client = self.collection_client
        payload = {
            "collectionName": name,
            "schema": {
                "enableDynamicField": False,
                "fields": [
                    {"fieldName": "id", "dataType": "Int64", "isPrimary": True},
                    {"fieldName": "dense", "dataType": "FloatVector", "elementTypeParams": {"dim": "4"}},
                ],
            },
            "indexParams": [
                {"fieldName": "dense", "indexName": "dense_idx", "indexType": "AUTOINDEX", "metricType": "L2"}
            ],
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0, rsp

        rsp = client.add_field(
            name,
            {
                "fieldName": "tag",
                "dataType": "VarChar",
                "nullable": True,
                "elementTypeParams": {"max_length": "64"},
            },
        )
        assert rsp["code"] == 0, rsp
        rsp = client.collection_describe(name)
        assert any(field["name"] == "tag" for field in rsp["data"]["fields"]), rsp

        rsp = client.drop_field(name, field_name="tag")
        assert rsp["code"] == 0, rsp
        rsp = client.collection_describe(name)
        assert not any(field["name"] == "tag" for field in rsp["data"]["fields"]), rsp

        rsp = client.add_field(name, {"fieldName": "score", "dataType": "Int64", "nullable": True})
        assert rsp["code"] == 0, rsp
        rsp = client.collection_describe(name)
        score_id = next(field["id"] for field in rsp["data"]["fields"] if field["name"] == "score")

        rsp = client.drop_field(name, field_id=score_id)
        assert rsp["code"] == 0, rsp
        rsp = client.collection_describe(name)
        assert not any(field["name"] == "score" for field in rsp["data"]["fields"]), rsp

    @pytest.mark.parametrize(
        "schema_variant,field_params",
        [
            (
                "array_with_struct_element",
                {
                    "fieldName": "dynamic_struct",
                    "dataType": "Array",
                    "elementDataType": "Struct",
                    "nullable": True,
                    "elementTypeParams": {"max_capacity": 16},
                    "fields": [
                        {"fieldName": "sub_int", "dataType": "Array", "elementDataType": "Int32"},
                        {
                            "fieldName": "sub_vec",
                            "dataType": "ArrayOfVector",
                            "elementDataType": "FloatVector",
                            "elementTypeParams": {"dim": 8},
                        },
                    ],
                },
            ),
            (
                "array_of_struct",
                {
                    "fieldName": "dynamic_struct",
                    "dataType": "ArrayOfStruct",
                    "nullable": True,
                    "typeParams": {"max_capacity": 16},
                    "fields": [
                        {"fieldName": "sub_int", "dataType": "Array", "elementDataType": "Int32"},
                        {
                            "fieldName": "sub_vec",
                            "dataType": "ArrayOfVector",
                            "elementDataType": "FloatVector",
                            "elementTypeParams": {"dim": 8},
                        },
                    ],
                },
            ),
        ],
    )
    def test_add_struct_array_field(self, schema_variant, field_params):
        """
        target: test REST v2 add StructArray field
        method: create collection, add StructArray field, insert and query rows with the new field
        expected: StructArray field is added and usable through REST v2
        """
        name = gen_collection_name()
        dim = DEFAULT_STRUCT_ARRAY_DIM
        field_name = field_params["fieldName"]
        client = self.collection_client
        vector_client = self.vector_client

        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
            "indexParams": [{"fieldName": "book_intro", "indexName": "book_intro_index", "metricType": "L2"}],
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0, rsp
        client.wait_load_completed(collection_name=name, timeout=60)

        rsp = vector_client.vector_insert(
            {
                "collectionName": name,
                "data": [{"book_id": 0, "book_intro": gen_vector(dim=dim)}],
            }
        )
        assert rsp["code"] == 0, rsp

        rsp = client.add_struct_field(name, field_params)
        logger.info(f"add struct array field response ({schema_variant}): {rsp}")
        assert rsp["code"] == 0, rsp

        rsp = client.collection_describe(name)
        assert rsp["code"] == 0, rsp
        struct_fields = {field["name"]: field for field in rsp["data"].get("structFields", [])}
        assert field_name in struct_fields, rsp
        struct_field = struct_fields[field_name]
        assert struct_field["type"] == "ArrayOfStruct"

        sub_fields = {field["name"]: field for field in struct_field.get("fields", [])}
        assert sorted(sub_fields) == ["sub_int", "sub_vec"]
        assert sub_fields["sub_int"]["type"] == "Array"
        assert sub_fields["sub_int"]["elementType"] == "Int32"
        assert sub_fields["sub_vec"]["type"] == "ArrayOfVector"
        assert sub_fields["sub_vec"]["elementType"] == "FloatVector"
        for sub_field in sub_fields.values():
            sub_params = {param["key"]: param["value"] for param in sub_field.get("params", [])}
            assert str(sub_params["max_capacity"]) == str(DEFAULT_STRUCT_ARRAY_SUB_CAPACITY)

        new_row = {
            "book_id": 1,
            "book_intro": gen_vector(dim=dim),
            field_name: [
                {"sub_int": 11, "sub_vec": _rand_struct_array_vector(dim)},
                {"sub_int": 12, "sub_vec": _rand_struct_array_vector(dim)},
            ],
        }
        rsp = vector_client.vector_insert({"collectionName": name, "data": [new_row]})
        assert rsp["code"] == 0, rsp

        rsp = vector_client.vector_query(
            {
                "collectionName": name,
                "filter": "book_id == 1",
                "outputFields": ["book_id", field_name],
                "limit": 1,
            }
        )
        assert rsp["code"] == 0, rsp
        assert len(rsp["data"]) == 1
        got = rsp["data"][0]
        assert got["book_id"] == 1
        assert len(got[field_name]) == len(new_row[field_name])
        for actual, expected in zip(got[field_name], new_row[field_name]):
            assert int(actual["sub_int"]) == expected["sub_int"]
            np.testing.assert_allclose(actual["sub_vec"], expected["sub_vec"], rtol=1e-5, atol=1e-5)


@pytest.mark.tags(CaseLabel.L1)
class TestCollectionAddFieldNegative(TestBase):
    """Test collection add field negative cases"""

    def test_add_field_missing_data_type(self):
        """
        target: test add field with missing dataType
        method: create collection, add field without dataType parameter
        expected: add field failed with proper error message
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client

        # Create collection first
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0

        # Try to add field without dataType
        field_params = {"fieldName": "new_field", "nullable": True, "elementTypeParams": {}}
        rsp = client.add_field(name, field_params)
        logger.info(f"add field response: {rsp}")
        assert rsp["code"] != 0
        assert "dataType" in rsp.get("message", "").lower() or "required" in rsp.get("message", "").lower()

    def test_add_field_invalid_default_value_type(self):
        """
        target: test add field with invalid defaultValue type
        method: create collection, add Int64 field with string defaultValue
        expected: add field failed with proper error message
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client

        # Create collection first
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0

        # Try to add Int64 field with string defaultValue
        field_params = {
            "fieldName": "new_field",
            "dataType": "Int64",
            "nullable": True,
            "defaultValue": "aaa",  # Invalid type for Int64
            "elementTypeParams": {},
        }
        rsp = client.add_field(name, field_params)
        logger.info(f"add field response: {rsp}")
        assert rsp["code"] != 0
        assert "defaultValue" in rsp.get("message", "") or "invalid" in rsp.get("message", "").lower()

    def test_add_field_invalid_data_type(self):
        """
        target: test add field with invalid dataType
        method: create collection, add field with invalid dataType
        expected: add field failed with proper error message
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client

        # Create collection first
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0

        # Try to add field with invalid dataType
        field_params = {
            "fieldName": "new_field",
            "dataType": "LONGLONGLONGLONGTEXT",  # Invalid dataType
            "nullable": True,
            "elementTypeParams": {},
        }
        rsp = client.add_field(name, field_params)
        logger.info(f"add field response: {rsp}")
        assert rsp["code"] != 0
        assert "invalid" in rsp.get("message", "").lower() or "data type" in rsp.get("message", "").lower()

    def test_add_field_array_missing_element_data_type(self):
        """
        target: test add Array field without elementDataType
        method: create collection, add Array field without elementDataType
        expected: add field failed with proper error message
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client

        # Create collection first
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0

        # Try to add Array field without elementDataType
        field_params = {"fieldName": "new_array_field", "dataType": "Array", "nullable": True, "elementTypeParams": {}}
        rsp = client.add_field(name, field_params)
        logger.info(f"add field response: {rsp}")
        assert rsp["code"] != 0
        assert "element" in rsp.get("message", "").lower() or "invalid" in rsp.get("message", "").lower()

    def test_add_field_array_invalid_element_data_type(self):
        """
        target: test add Array field with invalid elementDataType
        method: create collection, add Array field with invalid elementDataType
        expected: add field failed with proper error message
        """
        name = gen_collection_name()
        dim = 128
        client = self.collection_client

        # Create collection first
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "book_intro", "dataType": "FloatVector", "elementTypeParams": {"dim": f"{dim}"}},
                ]
            },
        }
        rsp = client.collection_create(payload)
        assert rsp["code"] == 0

        # Try to add Array field with invalid elementDataType
        field_params = {
            "fieldName": "new_array_field",
            "dataType": "Array",
            "elementDataType": "MYBLOB",  # Invalid elementDataType
            "nullable": True,
            "elementTypeParams": {},
        }
        rsp = client.add_field(name, field_params)
        logger.info(f"add field response: {rsp}")
        assert rsp["code"] != 0
        assert "element" in rsp.get("message", "").lower() or "invalid" in rsp.get("message", "").lower()

    def test_add_field_to_nonexistent_collection(self):
        """
        target: test add field to non-existent collection
        method: add field to a collection that doesn't exist
        expected: add field failed with proper error message
        """
        name = "nonexistent_collection"
        client = self.collection_client

        # Try to add field to non-existent collection
        field_params = {"fieldName": "new_field", "dataType": "Int64", "nullable": True, "elementTypeParams": {}}
        rsp = client.add_field(name, field_params)
        logger.info(f"add field response: {rsp}")
        assert rsp["code"] != 0
        assert "collection" in rsp.get("message", "").lower() or "not found" in rsp.get("message", "").lower()


@pytest.mark.tags(CaseLabel.L0)
class TestCollectionMaintenance(TestBase):
    """Test collection maintenance operations"""

    def test_collection_flush(self):
        """
        target: test collection flush
        method: create collection, insert data multiple times and flush
        expected: flush successfully
        """
        # Create collection
        name = gen_collection_name()
        client = self.collection_client
        vector_client = self.vector_client
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {"fieldName": "my_vector", "dataType": "FloatVector", "elementTypeParams": {"dim": 128}},
                ]
            },
        }
        client.collection_create(payload)

        # Insert small batches of data multiple times
        for i in range(3):
            vectors = [gen_vector(dim=128) for _ in range(10)]
            insert_data = {
                "collectionName": name,
                "data": [
                    {"book_id": i * 10 + j, "my_vector": vector} for i, vector in enumerate(vectors) for j in range(10)
                ],
            }
            response = vector_client.vector_insert(insert_data)
            assert response["code"] == 0
        c = Collection(name)
        num_entities_before_flush = c.num_entities
        # Flush collection
        response = client.flush(name)
        assert response["code"] == 0
        # check segments
        num_entities_after_flush = c.num_entities
        logger.info(
            f"num_entities_before_flush: {num_entities_before_flush}, num_entities_after_flush: {num_entities_after_flush}"
        )
        assert num_entities_after_flush > num_entities_before_flush

    def test_collection_compact(self):
        """
        target: test collection compact
        method: create collection with a clustering key, insert data, flush multiple times, then compact
        expected: compact successfully
        """
        # Create collection
        name = gen_collection_name()
        client = self.collection_client
        vector_client = self.vector_client
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "book_id", "dataType": "Int64", "isPrimary": True, "elementTypeParams": {}},
                    {
                        "fieldName": "word_count",
                        "dataType": "Int64",
                        "isClusteringKey": True,
                        "elementTypeParams": {},
                    },
                    {"fieldName": "my_vector", "dataType": "FloatVector", "elementTypeParams": {"dim": 128}},
                ]
            },
        }
        client.collection_create(payload)

        # Insert and flush multiple times
        batch_count = 4
        rows_per_batch = 100
        for batch in range(batch_count):
            # Insert data
            vectors = [gen_vector(dim=128) for _ in range(rows_per_batch)]
            insert_data = {
                "collectionName": name,
                "data": [
                    {
                        "book_id": batch * rows_per_batch + i,
                        "word_count": batch,
                        "my_vector": vector,
                    }
                    for i, vector in enumerate(vectors)
                ],
            }
            response = vector_client.vector_insert(insert_data)
            assert response["code"] == 0

            # Flush after each insert
            c = Collection(name)
            c.flush()
        # Compact collection
        response = client.compact(name, is_clustering=True)
        assert response["code"] == 0
        compaction_id = response.get("data", {}).get("compactionID")
        assert isinstance(compaction_id, int) and compaction_id > 0, response

        # Get compaction state
        response = client.get_compaction_state(compaction_id)
        assert response["code"] == 0
        assert "state" in response["data"]
        assert "compactionID" in response["data"]
        # TODO need verification by pymilvus


DEFAULT_STRUCT_ARRAY_DIM = 8
DEFAULT_STRUCT_ARRAY_SUB_CAPACITY = 16


def _rand_struct_array_vector(dim=DEFAULT_STRUCT_ARRAY_DIM):
    return [random.random() for _ in range(dim)]


def _build_struct_array_schema_payload(
    name,
    dim=DEFAULT_STRUCT_ARRAY_DIM,
    max_capacity=DEFAULT_STRUCT_ARRAY_SUB_CAPACITY,
    include_index_params=False,
    metric_type="COSINE",
    sub_metric_type="COSINE",
    enable_dynamic_field=False,
):
    payload = {
        "collectionName": name,
        "schema": {
            "autoId": False,
            "enableDynamicField": enable_dynamic_field,
            "fields": [
                {"fieldName": "id", "dataType": "Int64", "isPrimary": True},
                {
                    "fieldName": "vec",
                    "dataType": "FloatVector",
                    "elementTypeParams": {"dim": f"{dim}"},
                },
            ],
            "structFields": [
                {
                    "fieldName": "my_struct",
                    "fields": [
                        {
                            "fieldName": "sub_int",
                            "dataType": "Array",
                            "elementDataType": "Int32",
                            "elementTypeParams": {"max_capacity": max_capacity},
                        },
                        {
                            "fieldName": "sub_vec",
                            "dataType": "ArrayOfVector",
                            "elementDataType": "FloatVector",
                            "elementTypeParams": {
                                "dim": dim,
                                "max_capacity": max_capacity,
                            },
                        },
                    ],
                }
            ],
        },
    }
    if include_index_params:
        payload["indexParams"] = [
            {"fieldName": "vec", "indexName": "vec_idx", "metricType": metric_type},
            {
                "fieldName": "my_struct[sub_vec]",
                "indexName": "sub_vec_idx",
                "metricType": sub_metric_type,
            },
        ]
    return payload


def _gen_struct_array_row(row_id, num_elems, dim=DEFAULT_STRUCT_ARRAY_DIM):
    struct_elems = []
    for j in range(num_elems):
        struct_elems.append(
            {
                "sub_int": row_id * 100 + j,
                "sub_vec": _rand_struct_array_vector(dim),
            }
        )
    return {
        "id": row_id,
        "vec": _rand_struct_array_vector(dim),
        "my_struct": struct_elems,
    }


@pytest.mark.tags(CaseLabel.L0)
class TestStructArrayCollection(TestBase):
    def test_create_struct_array_collection(self):
        name = gen_collection_name()
        payload = _build_struct_array_schema_payload(name, include_index_params=True)
        rsp = self.collection_client.collection_create(payload)
        assert rsp["code"] == 0

        rsp = self.collection_client.collection_describe(name)
        assert rsp["code"] == 0
        struct_fields = rsp["data"].get("structFields", [])
        assert len(struct_fields) == 1, f"expected 1 struct field, got {struct_fields}"

        struct_field = struct_fields[0]
        assert struct_field["name"] == "my_struct"
        assert struct_field["type"] == "ArrayOfStruct"

        sub_fields = struct_field.get("fields", [])
        sub_names = sorted(s["name"] for s in sub_fields)
        assert sub_names == ["sub_int", "sub_vec"], sub_names

        by_name = {s["name"]: s for s in sub_fields}
        assert by_name["sub_int"]["type"] == "Array"
        assert by_name["sub_int"]["elementType"] == "Int32"
        assert by_name["sub_vec"]["elementType"] == "FloatVector"


@pytest.mark.tags(CaseLabel.L1)
class TestStructArraySchemaValidation(TestBase):
    def _create_with_bad_sub_field(self, name, bad_sub_field):
        payload = {
            "collectionName": name,
            "schema": {
                "autoId": False,
                "enableDynamicField": False,
                "fields": [
                    {"fieldName": "id", "dataType": "Int64", "isPrimary": True},
                    {
                        "fieldName": "vec",
                        "dataType": "FloatVector",
                        "elementTypeParams": {"dim": f"{DEFAULT_STRUCT_ARRAY_DIM}"},
                    },
                ],
                "structFields": [
                    {
                        "fieldName": "bad_struct",
                        "fields": [bad_sub_field],
                    }
                ],
            },
            "indexParams": [
                {"fieldName": "vec", "indexName": "vec_idx", "metricType": "L2"},
            ],
        }
        return self.collection_client.collection_create(payload)

    def test_reject_nullable_sub_field(self):
        name = gen_collection_name()
        rsp = self._create_with_bad_sub_field(
            name,
            {
                "fieldName": "sub",
                "dataType": "Array",
                "elementDataType": "Int32",
                "elementTypeParams": {"max_capacity": DEFAULT_STRUCT_ARRAY_SUB_CAPACITY},
                "nullable": True,
            },
        )
        assert rsp["code"] != 0, rsp
        assert "nullable" in rsp.get("message", "").lower()

    def test_reject_default_value_sub_field(self):
        name = gen_collection_name()
        rsp = self._create_with_bad_sub_field(
            name,
            {
                "fieldName": "sub",
                "dataType": "Array",
                "elementDataType": "Int32",
                "elementTypeParams": {"max_capacity": DEFAULT_STRUCT_ARRAY_SUB_CAPACITY},
                "defaultValue": 1,
            },
        )
        assert rsp["code"] != 0, rsp
        assert "default" in rsp.get("message", "").lower()

    def test_reject_primary_key_sub_field(self):
        name = gen_collection_name()
        rsp = self._create_with_bad_sub_field(
            name,
            {
                "fieldName": "sub",
                "dataType": "Array",
                "elementDataType": "Int64",
                "elementTypeParams": {"max_capacity": DEFAULT_STRUCT_ARRAY_SUB_CAPACITY},
                "isPrimary": True,
            },
        )
        assert rsp["code"] != 0, rsp

    def test_reject_partition_key_sub_field(self):
        name = gen_collection_name()
        rsp = self._create_with_bad_sub_field(
            name,
            {
                "fieldName": "sub",
                "dataType": "Array",
                "elementDataType": "Int64",
                "elementTypeParams": {"max_capacity": DEFAULT_STRUCT_ARRAY_SUB_CAPACITY},
                "isPartitionKey": True,
            },
        )
        assert rsp["code"] != 0, rsp

    def test_reject_clustering_key_sub_field(self):
        name = gen_collection_name()
        rsp = self._create_with_bad_sub_field(
            name,
            {
                "fieldName": "sub",
                "dataType": "Array",
                "elementDataType": "Int64",
                "elementTypeParams": {"max_capacity": DEFAULT_STRUCT_ARRAY_SUB_CAPACITY},
                "isClusteringKey": True,
            },
        )
        assert rsp["code"] != 0, rsp

    def test_reject_non_array_sub_field(self):
        name = gen_collection_name()
        rsp = self._create_with_bad_sub_field(
            name,
            {
                "fieldName": "sub",
                "dataType": "Int32",
            },
        )
        assert rsp["code"] != 0, rsp

    def test_reject_empty_sub_fields(self):
        name = gen_collection_name()
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "id", "dataType": "Int64", "isPrimary": True},
                    {
                        "fieldName": "vec",
                        "dataType": "FloatVector",
                        "elementTypeParams": {"dim": f"{DEFAULT_STRUCT_ARRAY_DIM}"},
                    },
                ],
                "structFields": [{"fieldName": "empty_struct", "fields": []}],
            },
            "indexParams": [
                {"fieldName": "vec", "indexName": "vec_idx", "metricType": "L2"},
            ],
        }
        rsp = self.collection_client.collection_create(payload)
        assert rsp["code"] != 0, rsp

    def test_reject_duplicated_sub_field_name(self):
        name = gen_collection_name()
        payload = {
            "collectionName": name,
            "schema": {
                "fields": [
                    {"fieldName": "id", "dataType": "Int64", "isPrimary": True},
                    {
                        "fieldName": "vec",
                        "dataType": "FloatVector",
                        "elementTypeParams": {"dim": f"{DEFAULT_STRUCT_ARRAY_DIM}"},
                    },
                ],
                "structFields": [
                    {
                        "fieldName": "dup_struct",
                        "fields": [
                            {
                                "fieldName": "s",
                                "dataType": "Array",
                                "elementDataType": "Int32",
                                "elementTypeParams": {"max_capacity": DEFAULT_STRUCT_ARRAY_SUB_CAPACITY},
                            },
                            {
                                "fieldName": "s",
                                "dataType": "Array",
                                "elementDataType": "Int32",
                                "elementTypeParams": {"max_capacity": DEFAULT_STRUCT_ARRAY_SUB_CAPACITY},
                            },
                        ],
                    }
                ],
            },
            "indexParams": [
                {"fieldName": "vec", "indexName": "vec_idx", "metricType": "L2"},
            ],
        }
        rsp = self.collection_client.collection_create(payload)
        assert rsp["code"] != 0, rsp


@pytest.mark.tags(CaseLabel.L0)
class TestStructArrayInsertQuery(TestBase):
    def _create_and_load(self, name, dim=DEFAULT_STRUCT_ARRAY_DIM):
        payload = _build_struct_array_schema_payload(name, dim=dim, include_index_params=True)
        rsp = self.collection_client.collection_create(payload)
        assert rsp["code"] == 0, rsp
        self.collection_client.wait_load_completed(name, timeout=60)

    def test_insert_and_query_struct_rows(self):
        name = gen_collection_name()
        self._create_and_load(name)

        nb = 10
        rows = [_gen_struct_array_row(i, num_elems=random.randint(1, 4)) for i in range(nb)]
        rsp = self.vector_client.vector_insert({"collectionName": name, "data": rows})
        assert rsp["code"] == 0, rsp
        assert rsp["data"]["insertCount"] == nb

        rsp = self.vector_client.vector_query(
            {
                "collectionName": name,
                "filter": "id >= 0",
                "outputFields": ["id", "my_struct"],
                "limit": nb,
            }
        )
        assert rsp["code"] == 0, rsp
        got = sorted(rsp["data"], key=lambda r: r["id"])
        assert len(got) == nb

        for orig, back in zip(rows, got):
            assert back["id"] == orig["id"]
            elems_back = back["my_struct"]
            assert len(elems_back) == len(orig["my_struct"]), (
                f"row {orig['id']} elem count mismatch: {len(elems_back)} vs {len(orig['my_struct'])}"
            )
            for eb, eo in zip(elems_back, orig["my_struct"]):
                assert int(eb["sub_int"]) == eo["sub_int"]
                np.testing.assert_allclose(eb["sub_vec"], eo["sub_vec"], rtol=1e-5, atol=1e-5)

    def test_insert_reject_missing_sub_field(self):
        name = gen_collection_name()
        self._create_and_load(name)

        bad_row = {
            "id": 1,
            "vec": _rand_struct_array_vector(),
            "my_struct": [
                {"sub_int": 1},
            ],
        }
        rsp = self.vector_client.vector_insert({"collectionName": name, "data": [bad_row]})
        assert rsp["code"] != 0, rsp
        assert "sub_vec" in rsp.get("message", "") or "missing" in rsp.get("message", "").lower()

    def test_insert_reject_struct_value_not_array(self):
        name = gen_collection_name()
        self._create_and_load(name)

        bad_row = {
            "id": 1,
            "vec": _rand_struct_array_vector(),
            "my_struct": {"sub_int": 1, "sub_vec": _rand_struct_array_vector()},
        }
        rsp = self.vector_client.vector_insert({"collectionName": name, "data": [bad_row]})
        assert rsp["code"] != 0, rsp


@pytest.mark.tags(CaseLabel.L0)
class TestStructSubVectorSearch(TestBase):
    def _setup_collection_with_sub_index(self, name, dim=DEFAULT_STRUCT_ARRAY_DIM, nb=50, sub_metric="COSINE"):
        payload = _build_struct_array_schema_payload(name, dim=dim, include_index_params=False)
        rsp = self.collection_client.collection_create(payload)
        assert rsp["code"] == 0, rsp

        rsp = self.index_client.index_create(
            {
                "collectionName": name,
                "indexParams": [
                    {"fieldName": "vec", "indexName": "vec_idx", "metricType": "COSINE"},
                ],
            }
        )
        assert rsp["code"] == 0, rsp

        rsp = self.index_client.index_create(
            {
                "collectionName": name,
                "indexParams": [
                    {
                        "fieldName": "my_struct[sub_vec]",
                        "indexName": "sub_vec_idx",
                        "metricType": sub_metric,
                    },
                ],
            }
        )
        assert rsp["code"] == 0, rsp

        rsp = self.collection_client.collection_load(collection_name=name)
        assert rsp["code"] == 0, rsp
        self.collection_client.wait_load_completed(name, timeout=60)

        rows = [_gen_struct_array_row(i, num_elems=random.randint(2, 5), dim=dim) for i in range(nb)]
        rsp = self.vector_client.vector_insert({"collectionName": name, "data": rows})
        assert rsp["code"] == 0, rsp
        self.collection_client.flush(name)
        self.collection_client.wait_load_completed(name, timeout=60)
        return rows

    def test_element_level_sub_vector_search(self):
        name = gen_collection_name()
        nq = 2
        self._setup_collection_with_sub_index(name)

        search_payload = {
            "collectionName": name,
            "annsField": "my_struct[sub_vec]",
            "data": [_rand_struct_array_vector() for _ in range(nq)],
            "limit": 5,
            "outputFields": ["id"],
        }
        rsp = self.vector_client.vector_search(search_payload)
        assert rsp["code"] == 0, rsp
        assert len(rsp["data"]) > 0, rsp

    def test_embedding_list_sub_vector_search(self):
        name = gen_collection_name()
        nq = 2
        per_query_vecs = 3
        self._setup_collection_with_sub_index(name, sub_metric="MAX_SIM_COSINE")

        search_payload = {
            "collectionName": name,
            "annsField": "my_struct[sub_vec]",
            "data": [[_rand_struct_array_vector() for _ in range(per_query_vecs)] for _ in range(nq)],
            "limit": 5,
            "outputFields": ["id"],
        }
        rsp = self.vector_client.vector_search(search_payload)
        assert rsp["code"] == 0, rsp
        assert len(rsp["data"]) > 0, rsp

    def test_sub_vector_search_dim_mismatch(self):
        name = gen_collection_name()
        self._setup_collection_with_sub_index(name)

        search_payload = {
            "collectionName": name,
            "annsField": "my_struct[sub_vec]",
            "data": [_rand_struct_array_vector(DEFAULT_STRUCT_ARRAY_DIM + 1)],
            "limit": 5,
        }
        rsp = self.vector_client.vector_search(search_payload)
        assert rsp["code"] != 0, rsp

    def test_embedding_list_search_dim_mismatch(self):
        name = gen_collection_name()
        self._setup_collection_with_sub_index(name, sub_metric="MAX_SIM_COSINE")

        search_payload = {
            "collectionName": name,
            "annsField": "my_struct[sub_vec]",
            "data": [
                [
                    _rand_struct_array_vector(DEFAULT_STRUCT_ARRAY_DIM + 1),
                    _rand_struct_array_vector(DEFAULT_STRUCT_ARRAY_DIM + 1),
                ]
            ],
            "limit": 5,
        }
        rsp = self.vector_client.vector_search(search_payload)
        assert rsp["code"] != 0, rsp


@pytest.mark.tags(CaseLabel.L0)
class TestStructSubVectorSearchOneStep(TestBase):
    def _create_load_insert(self, name, sub_metric, nb=50):
        payload = _build_struct_array_schema_payload(
            name,
            include_index_params=True,
            sub_metric_type=sub_metric,
        )
        rsp = self.collection_client.collection_create(payload)
        assert rsp["code"] == 0, rsp
        self.collection_client.wait_load_completed(name, timeout=60)

        rows = [_gen_struct_array_row(i, num_elems=random.randint(2, 4)) for i in range(nb)]
        rsp = self.vector_client.vector_insert({"collectionName": name, "data": rows})
        assert rsp["code"] == 0, rsp
        self.collection_client.flush(name)
        self.collection_client.wait_load_completed(name, timeout=60)

    def test_create_with_sub_vector_index_element_level(self):
        name = gen_collection_name()
        self._create_load_insert(name, sub_metric="COSINE")
        rsp = self.vector_client.vector_search(
            {
                "collectionName": name,
                "annsField": "my_struct[sub_vec]",
                "data": [_rand_struct_array_vector()],
                "limit": 5,
            }
        )
        assert rsp["code"] == 0, rsp
        assert len(rsp["data"]) > 0, rsp

    def test_create_with_sub_vector_index_embedding_list(self):
        name = gen_collection_name()
        self._create_load_insert(name, sub_metric="MAX_SIM_COSINE")
        rsp = self.vector_client.vector_search(
            {
                "collectionName": name,
                "annsField": "my_struct[sub_vec]",
                "data": [
                    [
                        _rand_struct_array_vector(),
                        _rand_struct_array_vector(),
                        _rand_struct_array_vector(),
                    ]
                ],
                "limit": 5,
            }
        )
        assert rsp["code"] == 0, rsp
        assert len(rsp["data"]) > 0, rsp

    def test_create_with_unknown_sub_field_rejected(self):
        name = gen_collection_name()
        payload = _build_struct_array_schema_payload(name, include_index_params=False)
        payload["indexParams"] = [
            {"fieldName": "vec", "indexName": "vec_idx", "metricType": "L2"},
            {
                "fieldName": "my_struct[no_such_sub]",
                "indexName": "bad",
                "metricType": "L2",
            },
        ]
        rsp = self.collection_client.collection_create(payload)
        assert rsp["code"] != 0, rsp
        assert "hasn't defined in schema" in rsp.get("message", ""), rsp
