test: add restful testcases for database api (#38281)

Signed-off-by: zhuwenxing <wenxing.zhu@zilliz.com>

tests/restful_client_v2/api/milvus.py

@@ -10,6 +10,7 @@ from tenacity import retry, retry_if_exception_type, stop_after_attempt
 from requests.exceptions import ConnectionError
 import urllib.parse
 
+
 ENABLE_LOG_SAVE = False
 
 
@@ -902,6 +903,56 @@ class ImportJobClient(Requests):
         return rsp, finished
 
 
+class DatabaseClient(Requests):
+    def __init__(self, endpoint, token):
+        super().__init__(url=endpoint, api_key=token)
+        self.endpoint = endpoint
+        self.api_key = token
+        self.headers = self.update_headers()
+        self.db_name = None
+        self.db_names = []  # Track created databases
+
+    @classmethod
+    def update_headers(cls):
+        headers = {
+            'Content-Type': 'application/json',
+            'Authorization': f'Bearer {cls.api_key}'
+        }
+        return headers
+
+    def database_create(self, payload):
+        """Create a database"""
+        url = f"{self.endpoint}/v2/vectordb/databases/create"
+        rsp = self.post(url, data=payload).json()
+        if rsp['code'] == 0:
+            self.db_name = payload['dbName']
+            self.db_names.append(payload['dbName'])
+        return rsp
+
+    def database_list(self, payload):
+        """List all databases"""
+        url = f"{self.endpoint}/v2/vectordb/databases/list"
+        return self.post(url, data=payload).json()
+
+    def database_describe(self, payload):
+        """Describe a database"""
+        url = f"{self.endpoint}/v2/vectordb/databases/describe"
+        return self.post(url, data=payload).json()
+
+    def database_alter(self, payload):
+        """Alter database properties"""
+        url = f"{self.endpoint}/v2/vectordb/databases/alter"
+        return self.post(url, data=payload).json()
+
+    def database_drop(self, payload):
+        """Drop a database"""
+        url = f"{self.endpoint}/v2/vectordb/databases/drop"
+        rsp = self.post(url, data=payload).json()
+        if rsp['code'] == 0 and payload['dbName'] in self.db_names:
+            self.db_names.remove(payload['dbName'])
+        return rsp
+
+
 class StorageClient():
 
     def __init__(self, endpoint, access_key, secret_key, bucket_name, root_path="file"):

tests/restful_client_v2/base/testbase.py

@@ -6,7 +6,7 @@ import uuid
 from pymilvus import connections, db, MilvusClient
 from utils.util_log import test_log as logger
 from api.milvus import (VectorClient, CollectionClient, PartitionClient, IndexClient, AliasClient,
-                        UserClient, RoleClient, ImportJobClient, StorageClient, Requests)
+                        UserClient, RoleClient, ImportJobClient, StorageClient, Requests, DatabaseClient)
 from utils.utils import get_data_by_payload
 
 
@@ -34,12 +34,14 @@ class Base:
     import_job_client = None
     storage_client = None
     milvus_client = None
+    database_client = None
 
 
 class TestBase(Base):
     req = None
 
     def teardown_method(self):
+        # Clean up collections
         self.collection_client.api_key = self.api_key
         all_collections = self.collection_client.collection_list()['data']
         if self.name in all_collections:
@@ -50,7 +52,8 @@ class TestBase(Base):
             try:
                 rsp = self.collection_client.collection_drop(payload)
             except Exception as e:
-                logger.error(e)
+                logger.error(f"drop collection error: {e}")
+
         for item in self.collection_client.name_list:
             db_name = item[0]
             c_name = item[1]
@@ -61,7 +64,17 @@ class TestBase(Base):
             try:
                 self.collection_client.collection_drop(payload)
             except Exception as e:
-                logger.error(e)
+                logger.error(f"drop collection error: {e}")
+
+
+        # Clean up databases created by this client
+        self.database_client.api_key = self.api_key
+        for db_name in self.database_client.db_names[:]:  # Create a copy of the list to iterate
+            logger.info(f"database {db_name} exist, drop it")
+            try:
+                rsp = self.database_client.database_drop({"dbName": db_name})
+            except Exception as e:
+                logger.error(f"drop database error: {e}")
 
     @pytest.fixture(scope="function", autouse=True)
     def init_client(self, endpoint, token, minio_host, bucket_name, root_path):
@@ -88,6 +101,8 @@ class TestBase(Base):
         self.import_job_client = ImportJobClient(self.endpoint, self.api_key)
         self.import_job_client.update_uuid(_uuid)
         self.storage_client = StorageClient(f"{minio_host}:9000", "minioadmin", "minioadmin", bucket_name, root_path)
+        self.database_client = DatabaseClient(self.endpoint, self.api_key)
+        self.database_client.update_uuid(_uuid)
         if token is None:
             self.vector_client.api_key = None
             self.collection_client.api_key = None

tests/restful_client_v2/testcases/test_database_operation.py

@@ -0,0 +1,164 @@
+import pytest
+from base.testbase import TestBase
+from utils.utils import gen_unique_str
+
+
+@pytest.mark.L0
+class TestDatabaseOperation(TestBase):
+    """
+    Test cases for database operations
+    """
+
+    def test_create_database_with_default_properties(self):
+        """
+        Test creating a database with default properties
+        """
+        db_name = f"test_db_{gen_unique_str()}"
+        payload = {"dbName": db_name}
+        rsp = self.database_client.database_create(payload)
+        assert rsp["code"] == 0
+
+        # Verify database exists
+        list_rsp = self.database_client.database_list({})
+        assert rsp["code"] == 0
+        assert db_name in list_rsp["data"]
+
+    def test_create_database_with_custom_properties(self):
+        """
+        Test creating a database with custom properties
+        """
+        db_name = f"test_db_{gen_unique_str()}"
+        payload = {"dbName": db_name, "properties": {"mmap.enabled": True}}
+        rsp = self.database_client.database_create(payload)
+        assert rsp["code"] == 0
+
+        # Verify properties
+        describe_rsp = self.database_client.database_describe({"dbName": db_name})
+        assert describe_rsp["code"] == 0
+        assert any(
+            prop["key"] == "mmap.enabled" and prop["value"] == "true"
+            for prop in describe_rsp["data"]["properties"]
+        )
+
+    def test_alter_database_properties(self):
+        """
+        Test altering database properties
+        """
+        db_name = f"test_db_{gen_unique_str()}"
+
+        # Create database with initial properties
+        create_payload = {"dbName": db_name, "properties": {"mmap.enabled": True}}
+        rsp = self.database_client.database_create(create_payload)
+        assert rsp["code"] == 0
+        # Verify properties
+        describe_rsp = self.database_client.database_describe({"dbName": db_name})
+        assert describe_rsp["code"] == 0
+        assert any(
+            prop["key"] == "mmap.enabled" and prop["value"] == "true"
+            for prop in describe_rsp["data"]["properties"]
+        )
+
+        # Alter properties
+        alter_payload = {"dbName": db_name, "properties": {"mmap.enabled": False}}
+        alter_rsp = self.database_client.database_alter(alter_payload)
+        assert alter_rsp["code"] == 0
+
+        # Verify altered properties
+        describe_rsp = self.database_client.database_describe({"dbName": db_name})
+        assert describe_rsp["code"] == 0
+        assert any(
+            prop["key"] == "mmap.enabled" and prop["value"] == "false"
+            for prop in describe_rsp["data"]["properties"]
+        )
+
+    def test_list_databases(self):
+        """
+        Test listing databases
+        """
+        # Create test database
+        db_name = f"test_db_{gen_unique_str()}"
+        self.database_client.database_create({"dbName": db_name})
+
+        # List databases
+        rsp = self.database_client.database_list({})
+        assert rsp["code"] == 0
+        assert "default" in rsp["data"]  # Default database should always exist
+        assert db_name in rsp["data"]
+
+    def test_describe_database(self):
+        """
+        Test describing database
+        """
+        db_name = f"test_db_{gen_unique_str()}"
+        properties = {"mmap.enabled": True}
+
+        # Create database
+        self.database_client.database_create(
+            {"dbName": db_name, "properties": properties}
+        )
+
+        # Describe database
+        rsp = self.database_client.database_describe({"dbName": db_name})
+        assert rsp["code"] == 0
+        assert rsp["data"]["dbName"] == db_name
+        assert "dbID" in rsp["data"]
+        assert len(rsp["data"]["properties"]) > 0
+
+
+@pytest.mark.L0
+class TestDatabaseOperationNegative(TestBase):
+    """
+    Negative test cases for database operations
+    """
+
+    def test_create_database_with_invalid_name(self):
+        """
+        Test creating database with invalid name
+        """
+        invalid_names = ["", " ", "test db", "test/db", "test\\db"]
+        for name in invalid_names:
+            rsp = self.database_client.database_create({"dbName": name})
+            assert rsp["code"] != 0
+
+    def test_create_duplicate_database(self):
+        """
+        Test creating database with duplicate name
+        """
+        db_name = f"test_db_{gen_unique_str()}"
+
+        # Create first database
+        rsp1 = self.database_client.database_create({"dbName": db_name})
+        assert rsp1["code"] == 0
+
+        # Try to create duplicate
+        rsp2 = self.database_client.database_create({"dbName": db_name})
+        assert rsp2["code"] != 0
+
+    def test_describe_non_existent_database(self):
+        """
+        Test describing non-existent database
+        """
+        rsp = self.database_client.database_describe({"dbName": "non_existent_db"})
+        assert rsp["code"] != 0
+
+    def test_alter_non_existent_database(self):
+        """
+        Test altering non-existent database
+        """
+        payload = {"dbName": "non_existent_db", "properties": {"mmap.enabled": False}}
+        rsp = self.database_client.database_alter(payload)
+        assert rsp["code"] != 0
+
+    def test_drop_non_existent_database(self):
+        """
+        Test dropping non-existent database
+        """
+        rsp = self.database_client.database_drop({"dbName": "non_existent_db"})
+        assert rsp["code"] == 0
+
+    def test_drop_default_database(self):
+        """
+        Test dropping default database (should not be allowed)
+        """
+        rsp = self.database_client.database_drop({"dbName": "default"})
+        assert rsp["code"] != 0