test(cdc): add collection properties, resource group, multi-db, search verification, and switchover tests

- test_collection_properties: TTL, mmap, autocompaction, multi-property, drop property
- test_resource_group: create, drop, update, transfer replica
- test_multi_database: cross-DB operations, drop DB
- test_search_verification: 7 vector types × search/query/iterator/hybrid/filtered
- test_switchover: basic, during-writes, all-types, loaded, indexed, rapid stress, failover

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>

Author: bigsheeper

tests/python_client/cdc/testcases/test_collection_properties.py

@@ -0,0 +1,281 @@
+"""
+CDC sync tests for collection property operations.
+"""
+
+import time
+from .base import TestCDCSyncBase, logger
+
+
+class TestCDCSyncCollectionProperties(TestCDCSyncBase):
+    """Test CDC sync for collection property (alter/drop) operations."""
+
+    def setup_method(self):
+        """Setup for each test method."""
+        self.resources_to_cleanup = []
+
+    def teardown_method(self):
+        """Cleanup after each test method - only cleanup upstream, downstream will sync."""
+        upstream_client = getattr(self, "_upstream_client", None)
+
+        if upstream_client:
+            for resource_type, resource_name in self.resources_to_cleanup:
+                if resource_type == "collection":
+                    self.cleanup_collection(upstream_client, resource_name)
+
+            time.sleep(1)  # Allow cleanup to sync to downstream
+
+    def _create_basic_collection(self, client, c_name):
+        """Create a default schema collection, insert 100 rows, and create HNSW index."""
+        schema = self.create_default_schema(client)
+        client.create_collection(
+            collection_name=c_name,
+            schema=schema,
+            consistency_level="Strong",
+        )
+
+        # Insert 100 rows
+        test_data = self.generate_test_data(100)
+        client.insert(c_name, test_data)
+
+        # Create HNSW index on vector field
+        index_params = client.prepare_index_params()
+        index_params.add_index(
+            field_name="vector",
+            index_type="HNSW",
+            metric_type="L2",
+            params={"M": 8, "efConstruction": 64},
+        )
+        client.create_index(c_name, index_params)
+
+    def test_ttl_sync(self, upstream_client, downstream_client, sync_timeout):
+        """Test ALTER_COLLECTION_PROPERTIES (TTL) sync."""
+        # Store upstream client for teardown
+        self._upstream_client = upstream_client
+
+        c_name = self.gen_unique_name("test_col_ttl")
+        self.resources_to_cleanup.append(("collection", c_name))
+
+        # Initial cleanup
+        self.cleanup_collection(upstream_client, c_name)
+
+        # Create collection
+        self._create_basic_collection(upstream_client, c_name)
+
+        # Wait for collection to sync downstream
+        def check_create():
+            return downstream_client.has_collection(c_name)
+
+        assert self.wait_for_sync(check_create, sync_timeout, f"create collection {c_name}")
+
+        # Alter collection TTL property
+        upstream_client.alter_collection_properties(
+            collection_name=c_name,
+            properties={"collection.ttl.seconds": "3600"},
+        )
+
+        # Wait for property to sync downstream
+        def check_ttl():
+            try:
+                desc = downstream_client.describe_collection(c_name)
+                props = desc.get("properties", {})
+                logger.info(f"Downstream collection properties: {props}")
+                return str(props.get("collection.ttl.seconds", "")) == "3600"
+            except Exception as e:
+                logger.warning(f"Check TTL sync failed: {e}")
+                return False
+
+        assert self.wait_for_sync(check_ttl, sync_timeout, f"TTL property sync for {c_name}")
+
+    def test_mmap_sync(self, upstream_client, downstream_client, sync_timeout):
+        """Test ALTER_COLLECTION_PROPERTIES (mmap.enabled) sync."""
+        # Store upstream client for teardown
+        self._upstream_client = upstream_client
+
+        c_name = self.gen_unique_name("test_col_mmap")
+        self.resources_to_cleanup.append(("collection", c_name))
+
+        # Initial cleanup
+        self.cleanup_collection(upstream_client, c_name)
+
+        # Create collection
+        self._create_basic_collection(upstream_client, c_name)
+
+        # Wait for collection to sync downstream
+        def check_create():
+            return downstream_client.has_collection(c_name)
+
+        assert self.wait_for_sync(check_create, sync_timeout, f"create collection {c_name}")
+
+        # Alter mmap.enabled property
+        upstream_client.alter_collection_properties(
+            collection_name=c_name,
+            properties={"mmap.enabled": "true"},
+        )
+
+        # Wait for property to sync downstream
+        def check_mmap():
+            try:
+                desc = downstream_client.describe_collection(c_name)
+                props = desc.get("properties", {})
+                logger.info(f"Downstream collection properties: {props}")
+                return str(props.get("mmap.enabled", "")).lower() == "true"
+            except Exception as e:
+                logger.warning(f"Check mmap sync failed: {e}")
+                return False
+
+        assert self.wait_for_sync(check_mmap, sync_timeout, f"mmap property sync for {c_name}")
+
+    def test_autocompaction_sync(self, upstream_client, downstream_client, sync_timeout):
+        """Test ALTER_COLLECTION_PROPERTIES (autocompaction.enabled) sync."""
+        # Store upstream client for teardown
+        self._upstream_client = upstream_client
+
+        c_name = self.gen_unique_name("test_col_autocomp")
+        self.resources_to_cleanup.append(("collection", c_name))
+
+        # Initial cleanup
+        self.cleanup_collection(upstream_client, c_name)
+
+        # Create collection
+        self._create_basic_collection(upstream_client, c_name)
+
+        # Wait for collection to sync downstream
+        def check_create():
+            return downstream_client.has_collection(c_name)
+
+        assert self.wait_for_sync(check_create, sync_timeout, f"create collection {c_name}")
+
+        # Alter autocompaction property
+        upstream_client.alter_collection_properties(
+            collection_name=c_name,
+            properties={"collection.autocompaction.enabled": "true"},
+        )
+
+        # Wait for property to sync downstream
+        def check_autocomp():
+            try:
+                desc = downstream_client.describe_collection(c_name)
+                props = desc.get("properties", {})
+                logger.info(f"Downstream collection properties: {props}")
+                return str(props.get("collection.autocompaction.enabled", "")).lower() == "true"
+            except Exception as e:
+                logger.warning(f"Check autocompaction sync failed: {e}")
+                return False
+
+        assert self.wait_for_sync(
+            check_autocomp, sync_timeout, f"autocompaction property sync for {c_name}"
+        )
+
+    def test_alter_multiple_properties(self, upstream_client, downstream_client, sync_timeout):
+        """Test ALTER_COLLECTION_PROPERTIES with multiple properties at once sync."""
+        # Store upstream client for teardown
+        self._upstream_client = upstream_client
+
+        c_name = self.gen_unique_name("test_col_multi_props")
+        self.resources_to_cleanup.append(("collection", c_name))
+
+        # Initial cleanup
+        self.cleanup_collection(upstream_client, c_name)
+
+        # Create collection
+        self._create_basic_collection(upstream_client, c_name)
+
+        # Wait for collection to sync downstream
+        def check_create():
+            return downstream_client.has_collection(c_name)
+
+        assert self.wait_for_sync(check_create, sync_timeout, f"create collection {c_name}")
+
+        # Alter all 3 properties at once
+        upstream_client.alter_collection_properties(
+            collection_name=c_name,
+            properties={
+                "collection.ttl.seconds": "3600",
+                "mmap.enabled": "true",
+                "collection.autocompaction.enabled": "true",
+            },
+        )
+
+        # Wait for all 3 properties to sync downstream
+        def check_all_props():
+            try:
+                desc = downstream_client.describe_collection(c_name)
+                props = desc.get("properties", {})
+                logger.info(f"Downstream collection properties: {props}")
+                ttl_ok = str(props.get("collection.ttl.seconds", "")) == "3600"
+                mmap_ok = str(props.get("mmap.enabled", "")).lower() == "true"
+                autocomp_ok = str(props.get("collection.autocompaction.enabled", "")).lower() == "true"
+                return ttl_ok and mmap_ok and autocomp_ok
+            except Exception as e:
+                logger.warning(f"Check all properties sync failed: {e}")
+                return False
+
+        assert self.wait_for_sync(
+            check_all_props, sync_timeout, f"all properties sync for {c_name}"
+        )
+
+    def test_drop_properties_sync(self, upstream_client, downstream_client, sync_timeout):
+        """Test DROP_COLLECTION_PROPERTIES — set TTL + mmap, drop TTL, verify TTL gone and mmap remains."""
+        # Store upstream client for teardown
+        self._upstream_client = upstream_client
+
+        c_name = self.gen_unique_name("test_col_drop_props")
+        self.resources_to_cleanup.append(("collection", c_name))
+
+        # Initial cleanup
+        self.cleanup_collection(upstream_client, c_name)
+
+        # Create collection
+        self._create_basic_collection(upstream_client, c_name)
+
+        # Wait for collection to sync downstream
+        def check_create():
+            return downstream_client.has_collection(c_name)
+
+        assert self.wait_for_sync(check_create, sync_timeout, f"create collection {c_name}")
+
+        # Set TTL and mmap properties
+        upstream_client.alter_collection_properties(
+            collection_name=c_name,
+            properties={
+                "collection.ttl.seconds": "3600",
+                "mmap.enabled": "true",
+            },
+        )
+
+        # Wait for both properties to sync downstream
+        def check_props_set():
+            try:
+                desc = downstream_client.describe_collection(c_name)
+                props = desc.get("properties", {})
+                ttl_set = str(props.get("collection.ttl.seconds", "")) == "3600"
+                mmap_set = str(props.get("mmap.enabled", "")).lower() == "true"
+                return ttl_set and mmap_set
+            except Exception as e:
+                logger.warning(f"Check properties set failed: {e}")
+                return False
+
+        assert self.wait_for_sync(check_props_set, sync_timeout, f"set properties for {c_name}")
+
+        # Drop TTL property
+        upstream_client.drop_collection_properties(
+            collection_name=c_name,
+            property_keys=["collection.ttl.seconds"],
+        )
+
+        # Wait for TTL to be gone but mmap to remain on downstream
+        def check_drop_ttl():
+            try:
+                desc = downstream_client.describe_collection(c_name)
+                props = desc.get("properties", {})
+                logger.info(f"Downstream collection properties after drop: {props}")
+                ttl_gone = "collection.ttl.seconds" not in props
+                mmap_remains = str(props.get("mmap.enabled", "")).lower() == "true"
+                return ttl_gone and mmap_remains
+            except Exception as e:
+                logger.warning(f"Check drop TTL sync failed: {e}")
+                return False
+
+        assert self.wait_for_sync(
+            check_drop_ttl, sync_timeout, f"drop TTL property sync for {c_name}"
+        )