test: Add tests for KvBackend trait implement

src/catalog/Cargo.toml

@@ -12,8 +12,8 @@ workspace = true
 
 [dependencies]
 api.workspace = true
-arrow.workspace = true
 arrow-schema.workspace = true
+arrow.workspace = true
 async-stream.workspace = true
 async-trait = "0.1"
 common-catalog.workspace = true

src/common/meta/src/kv_backend/etcd.rs

@@ -626,4 +626,117 @@ mod tests {
         assert_eq!(b"test_key".to_vec(), delete.key);
         let _ = delete.options.unwrap();
     }
+
+    use crate::kv_backend::test::{
+        prepare_kv_with_prefix, test_kv_batch_delete_with_prefix, test_kv_batch_get_with_prefix,
+        test_kv_compare_and_put_with_prefix, test_kv_delete_range_with_prefix,
+        test_kv_put_with_prefix, test_kv_range_2_with_prefix, test_kv_range_with_prefix,
+        unprepare_kv,
+    };
+
+    async fn build_kv_backend() -> Option<EtcdStore> {
+        let endpoints = std::env::var("GT_ETCD_ENDPOINTS").unwrap_or_default();
+        if endpoints.is_empty() {
+            return None;
+        }
+        let endpoints = endpoints
+            .split(',')
+            .map(|s| s.to_string())
+            .collect::<Vec<String>>();
+
+        std::panic::set_hook(Box::new(|_| {
+            println!("panic: malformed endpoints");
+        }));
+        let client = Client::connect(endpoints, None)
+            .await
+            .expect("malformed endpoints");
+        Some(EtcdStore {
+            client,
+            max_txn_ops: 128,
+        })
+    }
+
+    #[tokio::test]
+    async fn test_put() {
+        match build_kv_backend().await {
+            Some(kv_backend) => {
+                let prefix = b"put/";
+                prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
+                test_kv_put_with_prefix(&kv_backend, prefix.to_vec()).await;
+                unprepare_kv(&kv_backend, prefix).await;
+            }
+            None => {}
+        }
+    }
+
+    #[tokio::test]
+    async fn test_range() {
+        match build_kv_backend().await {
+            Some(kv_backend) => {
+                let prefix = b"range/";
+                prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
+                test_kv_range_with_prefix(&kv_backend, prefix.to_vec()).await;
+                unprepare_kv(&kv_backend, prefix).await;
+            }
+            None => {}
+        }
+    }
+
+    #[tokio::test]
+    async fn test_range_2() {
+        match build_kv_backend().await {
+            Some(kv_backend) => {
+                test_kv_range_2_with_prefix(kv_backend, b"range2/".to_vec()).await;
+            }
+            None => {}
+        }
+    }
+
+    #[tokio::test]
+    async fn test_batch_get() {
+        match build_kv_backend().await {
+            Some(kv_backend) => {
+                let prefix = b"batchGet/";
+                prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
+                test_kv_batch_get_with_prefix(&kv_backend, prefix.to_vec()).await;
+                unprepare_kv(&kv_backend, prefix).await;
+            }
+            None => {}
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_compare_and_put() {
+        match build_kv_backend().await {
+            Some(kv_backend) => {
+                let kv_backend = Arc::new(kv_backend);
+                test_kv_compare_and_put_with_prefix(kv_backend, b"compareAndPut/".to_vec()).await;
+            }
+            None => {}
+        }
+    }
+
+    #[tokio::test]
+    async fn test_delete_range() {
+        match build_kv_backend().await {
+            Some(kv_backend) => {
+                let prefix = b"deleteRange/";
+                prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
+                test_kv_delete_range_with_prefix(kv_backend, prefix.to_vec()).await;
+            }
+            None => {}
+        }
+    }
+
+    #[tokio::test]
+    async fn test_batch_delete() {
+        match build_kv_backend().await {
+            Some(kv_backend) => {
+                let prefix = b"batchDelete/";
+                prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
+                test_kv_batch_delete_with_prefix(kv_backend, prefix.to_vec()).await;
+            }
+            None => {}
+        }
+    }
 }

src/common/meta/src/kv_backend/memory.rs

@@ -357,14 +357,14 @@ mod tests {
     async fn test_put() {
         let kv_backend = mock_mem_store_with_data().await;
 
-        test_kv_put(kv_backend).await;
+        test_kv_put(&kv_backend).await;
     }
 
     #[tokio::test]
     async fn test_range() {
         let kv_backend = mock_mem_store_with_data().await;
 
-        test_kv_range(kv_backend).await;
+        test_kv_range(&kv_backend).await;
     }
 
     #[tokio::test]
@@ -378,7 +378,7 @@ mod tests {
     async fn test_batch_get() {
         let kv_backend = mock_mem_store_with_data().await;
 
-        test_kv_batch_get(kv_backend).await;
+        test_kv_batch_get(&kv_backend).await;
     }
 
     #[tokio::test(flavor = "multi_thread")]