From 3c4541244c4444cfb1bc8f118d8f575ee990d77c Mon Sep 17 00:00:00 2001
From: meteorgan <meteorite.gan@gmail.com>
Date: Sat, 21 Sep 2024 19:02:04 +0800
Subject: [PATCH 1/6] test(core): add tests for versioning

---
 core/src/services/s3/backend.rs             |  3 +-
 core/src/types/capability.rs                |  5 ++-
 core/src/types/operator/operator_futures.rs |  2 +-
 core/tests/behavior/async_delete.rs         | 40 ++++++++++++++++-
 core/tests/behavior/async_list.rs           |  6 +--
 core/tests/behavior/async_read.rs           | 37 +++++++++++++++-
 core/tests/behavior/async_stat.rs           | 48 ++++++++++++++++++---
 7 files changed, 127 insertions(+), 14 deletions(-)

diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index 745b292f2b5..9b6c8f95898 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -938,7 +938,6 @@ impl Access for S3Backend {
                 list_with_limit: true,
                 list_with_start_after: true,
                 list_with_recursive: true,
-                list_with_version: self.core.enable_versioning,
 
                 presign: true,
                 presign_stat: true,
@@ -948,6 +947,8 @@ impl Access for S3Backend {
                 batch: true,
                 batch_max_operations: Some(self.core.batch_max_operations),
 
+                versioning: self.core.enable_versioning,
+
                 ..Default::default()
             });
 
diff --git a/core/src/types/capability.rs b/core/src/types/capability.rs
index cae69f04ab6..b3b654feaeb 100644
--- a/core/src/types/capability.rs
+++ b/core/src/types/capability.rs
@@ -134,8 +134,6 @@ pub struct Capability {
     pub list_with_start_after: bool,
     /// If backend supports list with recursive.
     pub list_with_recursive: bool,
-    /// If backend supports list with object version.
-    pub list_with_version: bool,
 
     /// If operator supports presign.
     pub presign: bool,
@@ -155,6 +153,9 @@ pub struct Capability {
 
     /// If operator supports blocking.
     pub blocking: bool,
+
+    /// If operator supports versioning
+    pub versioning: bool,
 }
 
 impl Debug for Capability {
diff --git a/core/src/types/operator/operator_futures.rs b/core/src/types/operator/operator_futures.rs
index 777c817e0f8..b5ddd1d4c32 100644
--- a/core/src/types/operator/operator_futures.rs
+++ b/core/src/types/operator/operator_futures.rs
@@ -406,7 +406,7 @@ impl<F: Future<Output = Result<Writer>>> FutureWriter<F> {
     ///
     /// ## Notes
     ///
-    /// we don't need to include the user defined metadata prefix in the key
+    /// we don't need to include the user defined metadata prefix in the key.
     /// every service will handle it internally
     pub fn user_metadata(self, data: impl IntoIterator<Item = (String, String)>) -> Self {
         self.map(|(args, options)| (args.with_user_metadata(HashMap::from_iter(data)), options))
diff --git a/core/tests/behavior/async_delete.rs b/core/tests/behavior/async_delete.rs
index ab51712d696..f1d2f35dc3b 100644
--- a/core/tests/behavior/async_delete.rs
+++ b/core/tests/behavior/async_delete.rs
@@ -33,7 +33,8 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
             test_delete_with_special_chars,
             test_delete_not_existing,
             test_delete_stream,
-            test_remove_one_file
+            test_remove_one_file,
+            test_delete_with_version
         ));
         if cap.list_with_recursive {
             tests.extend(async_trials!(op, test_remove_all_basic));
@@ -212,3 +213,40 @@ pub async fn test_remove_all_with_prefix_exists(op: Operator) -> Result<()> {
         .expect("write must succeed");
     test_blocking_remove_all_with_objects(op, parent, ["a", "a/b", "a/c", "a/b/e"]).await
 }
+
+pub async fn test_delete_with_version(op: Operator) -> Result<()> {
+    if !op.info().full_capability().versioning {
+        return Ok(());
+    }
+
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+
+    //TODO: refactor these code after `write` can return metadata
+    op.write(path.as_str(), content)
+        .await
+        .expect("write must success");
+    let meta = op.stat(path.as_str()).await.expect("stat must success");
+    let version = meta.version().expect("must have version");
+
+    op.delete(path.as_str()).await.expect("delete must success");
+    assert!(!op.is_exist(path.as_str()).await?);
+
+    // after simple delete, we can still get the data using version
+    let meta = op
+        .stat_with(path.as_str())
+        .version(version)
+        .await
+        .expect("stat must success");
+    assert_eq!(version, meta.version().expect("must have version"));
+
+    // after delete with version, we can get the object with special version
+    op.delete_with(path.as_str())
+        .version(version)
+        .await
+        .expect("delete must success");
+    let ret = op.stat_with(path.as_str()).version(version).await;
+    assert!(ret.is_err());
+    assert_eq!(ret.unwrap_err().kind(), ErrorKind::NotFound);
+
+    Ok(())
+}
diff --git a/core/tests/behavior/async_list.rs b/core/tests/behavior/async_list.rs
index ba862f70252..f48c689ab7c 100644
--- a/core/tests/behavior/async_list.rs
+++ b/core/tests/behavior/async_list.rs
@@ -684,7 +684,7 @@ pub async fn test_list_only(op: Operator) -> Result<()> {
 }
 
 pub async fn test_list_files_with_version(op: Operator) -> Result<()> {
-    if !op.info().full_capability().list_with_version {
+    if !op.info().full_capability().versioning {
         return Ok(());
     }
 
@@ -722,7 +722,7 @@ pub async fn test_list_with_version_and_limit(op: Operator) -> Result<()> {
     if op.info().scheme() == Scheme::Gdrive {
         return Ok(());
     }
-    if !op.info().full_capability().list_with_version {
+    if !op.info().full_capability().versioning {
         return Ok(());
     }
 
@@ -775,7 +775,7 @@ pub async fn test_list_with_version_and_limit(op: Operator) -> Result<()> {
 }
 
 pub async fn test_list_with_version_and_start_after(op: Operator) -> Result<()> {
-    if !op.info().full_capability().list_with_version {
+    if !op.info().full_capability().versioning {
         return Ok(());
     }
 
diff --git a/core/tests/behavior/async_read.rs b/core/tests/behavior/async_read.rs
index 53e6f29a610..82f869c44cc 100644
--- a/core/tests/behavior/async_read.rs
+++ b/core/tests/behavior/async_read.rs
@@ -44,7 +44,8 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
             test_read_with_special_chars,
             test_read_with_override_cache_control,
             test_read_with_override_content_disposition,
-            test_read_with_override_content_type
+            test_read_with_override_content_type,
+            test_read_with_version
         ))
     }
 
@@ -553,3 +554,37 @@ pub async fn test_read_only_read_with_if_none_match(op: Operator) -> anyhow::Res
 
     Ok(())
 }
+
+pub async fn test_read_with_version(op: Operator) -> anyhow::Result<()> {
+    if !op.info().full_capability().versioning {
+        return Ok(());
+    }
+
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+    op.write(path.as_str(), content.clone())
+        .await
+        .expect("write must success");
+    let meta = op.stat(path.as_str()).await.expect("stat must success");
+    let version = meta.version().expect("must have version");
+
+    let data = op
+        .read_with(path.as_str())
+        .version(version)
+        .await
+        .expect("read must success");
+    assert_eq!(content, data.to_vec());
+
+    op.write(path.as_str(), "1")
+        .await
+        .expect("write must success");
+
+    // we can use read with version to get the first version data
+    let second_data = op
+        .read_with(path.as_str())
+        .version(version)
+        .await
+        .expect("read must success");
+    assert_eq!(content, second_data.to_vec());
+
+    Ok(())
+}
diff --git a/core/tests/behavior/async_stat.rs b/core/tests/behavior/async_stat.rs
index 2dcbde9b2df..146031c87d9 100644
--- a/core/tests/behavior/async_stat.rs
+++ b/core/tests/behavior/async_stat.rs
@@ -18,13 +18,12 @@
 use std::str::FromStr;
 use std::time::Duration;
 
+use crate::*;
 use anyhow::Result;
 use http::StatusCode;
 use log::warn;
 use reqwest::Url;
 
-use crate::*;
-
 pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
     let cap = op.info().full_capability();
 
@@ -42,7 +41,8 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
             test_stat_with_override_cache_control,
             test_stat_with_override_content_disposition,
             test_stat_with_override_content_type,
-            test_stat_root
+            test_stat_root,
+            test_stat_with_version
         ))
     }
 
@@ -166,12 +166,12 @@ pub async fn test_stat_not_cleaned_path(op: Operator) -> Result<()> {
 pub async fn test_stat_not_exist(op: Operator) -> Result<()> {
     let path = uuid::Uuid::new_v4().to_string();
 
-    // Stat not exist file should returns NotFound.
+    // Stat not exist file should return NotFound.
     let meta = op.stat(&path).await;
     assert!(meta.is_err());
     assert_eq!(meta.unwrap_err().kind(), ErrorKind::NotFound);
 
-    // Stat not exist dir should also returns NotFound.
+    // Stat not exist dir should also return NotFound.
     if op.info().full_capability().create_dir {
         let meta = op.stat(&format!("{path}/")).await;
         assert!(meta.is_err());
@@ -499,3 +499,41 @@ pub async fn test_read_only_stat_root(op: Operator) -> Result<()> {
 
     Ok(())
 }
+
+pub async fn test_stat_with_version(op: Operator) -> Result<()> {
+    if !op.info().full_capability().versioning {
+        return Ok(());
+    }
+
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+
+    op.write(path.as_str(), content.clone())
+        .await
+        .expect("write must success");
+    let first_meta = op.stat(path.as_str()).await.expect("stat must success");
+    let first_version = first_meta.version().expect("must have version");
+
+    let first_versioning_meta = op
+        .stat_with(path.as_str())
+        .version(first_version)
+        .await
+        .expect("stat must success");
+    assert_eq!(first_meta, first_versioning_meta);
+
+    op.write(path.as_str(), content)
+        .await
+        .expect("write must success");
+    let second_meta = op.stat(path.as_str()).await.expect("stat must success");
+    let second_version = second_meta.version().expect("must have version");
+    assert_ne!(first_version, second_version);
+
+    // we can still `stat` with first_version after writing new data
+    let meta = op
+        .stat_with(path.as_str())
+        .version(first_version)
+        .await
+        .expect("stat must success");
+    assert_eq!(first_meta, meta);
+
+    Ok(())
+}

From 27c5dd860303a1b280dae592434a2c0e9c49b142 Mon Sep 17 00:00:00 2001
From: meteorgan <meteorite.gan@gmail.com>
Date: Sat, 21 Sep 2024 23:48:05 +0800
Subject: [PATCH 2/6] add tests for version doesn't match

---
 core/src/services/s3/error.rs       |  2 +-
 core/tests/behavior/async_delete.rs | 25 +++++++++++++++++++++++--
 core/tests/behavior/async_read.rs   | 23 ++++++++++++++++++++++-
 core/tests/behavior/async_stat.rs   | 24 +++++++++++++++++++++++-
 4 files changed, 69 insertions(+), 5 deletions(-)

diff --git a/core/src/services/s3/error.rs b/core/src/services/s3/error.rs
index 503ca4ecbd7..2dd68e6686d 100644
--- a/core/src/services/s3/error.rs
+++ b/core/src/services/s3/error.rs
@@ -41,7 +41,7 @@ pub(super) fn parse_error(resp: Response<Buffer>) -> Error {
     let (mut kind, mut retryable) = match parts.status.as_u16() {
         403 => (ErrorKind::PermissionDenied, false),
         404 => (ErrorKind::NotFound, false),
-        304 | 412 => (ErrorKind::ConditionNotMatch, false),
+        304 | 400 | 412 => (ErrorKind::ConditionNotMatch, false),
         // Service like R2 could return 499 error with a message like:
         // Client Disconnect, we should retry it.
         499 => (ErrorKind::Unexpected, true),
diff --git a/core/tests/behavior/async_delete.rs b/core/tests/behavior/async_delete.rs
index f1d2f35dc3b..98e582d5fcb 100644
--- a/core/tests/behavior/async_delete.rs
+++ b/core/tests/behavior/async_delete.rs
@@ -34,7 +34,8 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
             test_delete_not_existing,
             test_delete_stream,
             test_remove_one_file,
-            test_delete_with_version
+            test_delete_with_version,
+            test_delete_with_not_existing_version
         ));
         if cap.list_with_recursive {
             tests.extend(async_trials!(op, test_remove_all_basic));
@@ -221,7 +222,7 @@ pub async fn test_delete_with_version(op: Operator) -> Result<()> {
 
     let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
 
-    //TODO: refactor these code after `write` can return metadata
+    //TODO: refactor these code after `write` operation can return metadata
     op.write(path.as_str(), content)
         .await
         .expect("write must success");
@@ -250,3 +251,23 @@ pub async fn test_delete_with_version(op: Operator) -> Result<()> {
 
     Ok(())
 }
+
+pub async fn test_delete_with_not_existing_version(op: Operator) -> Result<()> {
+    if !op.info().full_capability().versioning {
+        return Ok(());
+    }
+
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+    op.write(path.as_str(), content)
+        .await
+        .expect("write must success");
+
+    let ret = op
+        .delete_with(path.as_str())
+        .version("not-existing-version")
+        .await;
+    assert!(ret.is_err());
+    assert_eq!(ret.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+
+    Ok(())
+}
diff --git a/core/tests/behavior/async_read.rs b/core/tests/behavior/async_read.rs
index 82f869c44cc..4fd9a0ae880 100644
--- a/core/tests/behavior/async_read.rs
+++ b/core/tests/behavior/async_read.rs
@@ -45,7 +45,8 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
             test_read_with_override_cache_control,
             test_read_with_override_content_disposition,
             test_read_with_override_content_type,
-            test_read_with_version
+            test_read_with_version,
+            test_read_with_not_existing_version
         ))
     }
 
@@ -588,3 +589,23 @@ pub async fn test_read_with_version(op: Operator) -> anyhow::Result<()> {
 
     Ok(())
 }
+
+pub async fn test_read_with_not_existing_version(op: Operator) -> anyhow::Result<()> {
+    if !op.info().full_capability().versioning {
+        return Ok(());
+    }
+
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+    op.write(path.as_str(), content.clone())
+        .await
+        .expect("write must success");
+
+    let ret = op
+        .read_with(path.as_str())
+        .version("not-existing-version")
+        .await;
+    assert!(ret.is_err());
+    assert_eq!(ret.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+
+    Ok(())
+}
diff --git a/core/tests/behavior/async_stat.rs b/core/tests/behavior/async_stat.rs
index 146031c87d9..0c3aa63d67b 100644
--- a/core/tests/behavior/async_stat.rs
+++ b/core/tests/behavior/async_stat.rs
@@ -42,7 +42,8 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
             test_stat_with_override_content_disposition,
             test_stat_with_override_content_type,
             test_stat_root,
-            test_stat_with_version
+            test_stat_with_version,
+            stat_with_not_existing_version
         ))
     }
 
@@ -507,6 +508,7 @@ pub async fn test_stat_with_version(op: Operator) -> Result<()> {
 
     let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
 
+    //TODO: refactor these code after `write` operation can return metadata
     op.write(path.as_str(), content.clone())
         .await
         .expect("write must success");
@@ -537,3 +539,23 @@ pub async fn test_stat_with_version(op: Operator) -> Result<()> {
 
     Ok(())
 }
+
+pub async fn stat_with_not_existing_version(op: Operator) -> Result<()> {
+    if !op.info().full_capability().versioning {
+        return Ok(());
+    }
+
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+    op.write(path.as_str(), content.clone())
+        .await
+        .expect("write must success");
+
+    let ret = op
+        .stat_with(path.as_str())
+        .version("not-existing-version")
+        .await;
+    assert!(ret.is_err());
+    assert_eq!(ret.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+
+    Ok(())
+}

From 1f281aadec1e60b11e62d352648296d6e5c24bb2 Mon Sep 17 00:00:00 2001
From: meteorgan <meteorite.gan@gmail.com>
Date: Sun, 22 Sep 2024 16:59:06 +0800
Subject: [PATCH 3/6] update comments

---
 core/tests/behavior/async_delete.rs | 4 ++--
 core/tests/behavior/async_read.rs   | 2 +-
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/core/tests/behavior/async_delete.rs b/core/tests/behavior/async_delete.rs
index 98e582d5fcb..f8b2426791c 100644
--- a/core/tests/behavior/async_delete.rs
+++ b/core/tests/behavior/async_delete.rs
@@ -232,7 +232,7 @@ pub async fn test_delete_with_version(op: Operator) -> Result<()> {
     op.delete(path.as_str()).await.expect("delete must success");
     assert!(!op.is_exist(path.as_str()).await?);
 
-    // after simple delete, we can still get the data using version
+    // After a simple delete, the data can still be accessed using its version.
     let meta = op
         .stat_with(path.as_str())
         .version(version)
@@ -240,7 +240,7 @@ pub async fn test_delete_with_version(op: Operator) -> Result<()> {
         .expect("stat must success");
     assert_eq!(version, meta.version().expect("must have version"));
 
-    // after delete with version, we can get the object with special version
+    // After deleting with the version, the data is removed permanently
     op.delete_with(path.as_str())
         .version(version)
         .await
diff --git a/core/tests/behavior/async_read.rs b/core/tests/behavior/async_read.rs
index 4fd9a0ae880..e3fbbb26060 100644
--- a/core/tests/behavior/async_read.rs
+++ b/core/tests/behavior/async_read.rs
@@ -579,7 +579,7 @@ pub async fn test_read_with_version(op: Operator) -> anyhow::Result<()> {
         .await
         .expect("write must success");
 
-    // we can use read with version to get the first version data
+    // After writing new data, we can still read the first version data
     let second_data = op
         .read_with(path.as_str())
         .version(version)

From d76f7763c13a92edb2f0b71e5eb24ab9dc1f23a1 Mon Sep 17 00:00:00 2001
From: meteorgan <meteorite.gan@gmail.com>
Date: Tue, 24 Sep 2024 16:41:28 +0800
Subject: [PATCH 4/6] add stat_with_versioning, read_with_versioning,
 delete_with_versioning, list_with_versioning_capabilities

---
 core/src/services/s3/backend.rs     |  8 +++++---
 core/src/types/capability.rs        | 17 +++++++++++------
 core/tests/behavior/async_delete.rs |  5 ++---
 core/tests/behavior/async_list.rs   |  6 +++---
 core/tests/behavior/async_read.rs   |  4 ++--
 core/tests/behavior/async_stat.rs   |  5 ++---
 6 files changed, 25 insertions(+), 20 deletions(-)

diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index 9b6c8f95898..a054770bd68 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -902,14 +902,15 @@ impl Access for S3Backend {
                 stat_with_override_cache_control: !self.core.disable_stat_with_override,
                 stat_with_override_content_disposition: !self.core.disable_stat_with_override,
                 stat_with_override_content_type: !self.core.disable_stat_with_override,
+                stat_with_versioning: self.core.enable_versioning,
 
                 read: true,
-
                 read_with_if_match: true,
                 read_with_if_none_match: true,
                 read_with_override_cache_control: true,
                 read_with_override_content_disposition: true,
                 read_with_override_content_type: true,
+                read_with_versioning: self.core.enable_versioning,
 
                 write: true,
                 write_can_empty: true,
@@ -932,12 +933,15 @@ impl Access for S3Backend {
                 },
 
                 delete: true,
+                delete_with_versioning: self.core.enable_versioning,
+
                 copy: true,
 
                 list: true,
                 list_with_limit: true,
                 list_with_start_after: true,
                 list_with_recursive: true,
+                list_with_versioning: self.core.enable_versioning,
 
                 presign: true,
                 presign_stat: true,
@@ -947,8 +951,6 @@ impl Access for S3Backend {
                 batch: true,
                 batch_max_operations: Some(self.core.batch_max_operations),
 
-                versioning: self.core.enable_versioning,
-
                 ..Default::default()
             });
 
diff --git a/core/src/types/capability.rs b/core/src/types/capability.rs
index b3b654feaeb..78ecb00205f 100644
--- a/core/src/types/capability.rs
+++ b/core/src/types/capability.rs
@@ -61,12 +61,14 @@ pub struct Capability {
     pub stat_with_if_match: bool,
     /// If operator supports stat with if none match.
     pub stat_with_if_none_match: bool,
-    /// if operator supports read with override cache control.
+    /// if operator supports stat with override cache control.
     pub stat_with_override_cache_control: bool,
-    /// if operator supports read with override content disposition.
+    /// if operator supports stat with override content disposition.
     pub stat_with_override_content_disposition: bool,
-    /// if operator supports read with override content type.
+    /// if operator supports stat with override content type.
     pub stat_with_override_content_type: bool,
+    /// if operator supports stat with versioning.
+    pub stat_with_versioning: bool,
 
     /// If operator supports read.
     pub read: bool,
@@ -80,6 +82,8 @@ pub struct Capability {
     pub read_with_override_content_disposition: bool,
     /// if operator supports read with override content type.
     pub read_with_override_content_type: bool,
+    /// if operator supports read with versioning.
+    pub read_with_versioning: bool,
 
     /// If operator supports write.
     pub write: bool,
@@ -119,6 +123,8 @@ pub struct Capability {
 
     /// If operator supports delete.
     pub delete: bool,
+    /// if operator supports delete with versioning.
+    pub delete_with_versioning: bool,
 
     /// If operator supports copy.
     pub copy: bool,
@@ -134,6 +140,8 @@ pub struct Capability {
     pub list_with_start_after: bool,
     /// If backend supports list with recursive.
     pub list_with_recursive: bool,
+    /// if operator supports list with versioning.
+    pub list_with_versioning: bool,
 
     /// If operator supports presign.
     pub presign: bool,
@@ -153,9 +161,6 @@ pub struct Capability {
 
     /// If operator supports blocking.
     pub blocking: bool,
-
-    /// If operator supports versioning
-    pub versioning: bool,
 }
 
 impl Debug for Capability {
diff --git a/core/tests/behavior/async_delete.rs b/core/tests/behavior/async_delete.rs
index f8b2426791c..44ef3afc8ce 100644
--- a/core/tests/behavior/async_delete.rs
+++ b/core/tests/behavior/async_delete.rs
@@ -216,13 +216,12 @@ pub async fn test_remove_all_with_prefix_exists(op: Operator) -> Result<()> {
 }
 
 pub async fn test_delete_with_version(op: Operator) -> Result<()> {
-    if !op.info().full_capability().versioning {
+    if !op.info().full_capability().delete_with_versioning {
         return Ok(());
     }
 
     let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
 
-    //TODO: refactor these code after `write` operation can return metadata
     op.write(path.as_str(), content)
         .await
         .expect("write must success");
@@ -253,7 +252,7 @@ pub async fn test_delete_with_version(op: Operator) -> Result<()> {
 }
 
 pub async fn test_delete_with_not_existing_version(op: Operator) -> Result<()> {
-    if !op.info().full_capability().versioning {
+    if !op.info().full_capability().delete_with_versioning {
         return Ok(());
     }
 
diff --git a/core/tests/behavior/async_list.rs b/core/tests/behavior/async_list.rs
index f48c689ab7c..86d2aa33caf 100644
--- a/core/tests/behavior/async_list.rs
+++ b/core/tests/behavior/async_list.rs
@@ -684,7 +684,7 @@ pub async fn test_list_only(op: Operator) -> Result<()> {
 }
 
 pub async fn test_list_files_with_version(op: Operator) -> Result<()> {
-    if !op.info().full_capability().versioning {
+    if !op.info().full_capability().list_with_versioning {
         return Ok(());
     }
 
@@ -722,7 +722,7 @@ pub async fn test_list_with_version_and_limit(op: Operator) -> Result<()> {
     if op.info().scheme() == Scheme::Gdrive {
         return Ok(());
     }
-    if !op.info().full_capability().versioning {
+    if !op.info().full_capability().list_with_versioning {
         return Ok(());
     }
 
@@ -775,7 +775,7 @@ pub async fn test_list_with_version_and_limit(op: Operator) -> Result<()> {
 }
 
 pub async fn test_list_with_version_and_start_after(op: Operator) -> Result<()> {
-    if !op.info().full_capability().versioning {
+    if !op.info().full_capability().list_with_versioning {
         return Ok(());
     }
 
diff --git a/core/tests/behavior/async_read.rs b/core/tests/behavior/async_read.rs
index e3fbbb26060..19e48aad3f7 100644
--- a/core/tests/behavior/async_read.rs
+++ b/core/tests/behavior/async_read.rs
@@ -557,7 +557,7 @@ pub async fn test_read_only_read_with_if_none_match(op: Operator) -> anyhow::Res
 }
 
 pub async fn test_read_with_version(op: Operator) -> anyhow::Result<()> {
-    if !op.info().full_capability().versioning {
+    if !op.info().full_capability().read_with_versioning {
         return Ok(());
     }
 
@@ -591,7 +591,7 @@ pub async fn test_read_with_version(op: Operator) -> anyhow::Result<()> {
 }
 
 pub async fn test_read_with_not_existing_version(op: Operator) -> anyhow::Result<()> {
-    if !op.info().full_capability().versioning {
+    if !op.info().full_capability().read_with_versioning {
         return Ok(());
     }
 
diff --git a/core/tests/behavior/async_stat.rs b/core/tests/behavior/async_stat.rs
index 0c3aa63d67b..967985995d2 100644
--- a/core/tests/behavior/async_stat.rs
+++ b/core/tests/behavior/async_stat.rs
@@ -502,13 +502,12 @@ pub async fn test_read_only_stat_root(op: Operator) -> Result<()> {
 }
 
 pub async fn test_stat_with_version(op: Operator) -> Result<()> {
-    if !op.info().full_capability().versioning {
+    if !op.info().full_capability().stat_with_versioning {
         return Ok(());
     }
 
     let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
 
-    //TODO: refactor these code after `write` operation can return metadata
     op.write(path.as_str(), content.clone())
         .await
         .expect("write must success");
@@ -541,7 +540,7 @@ pub async fn test_stat_with_version(op: Operator) -> Result<()> {
 }
 
 pub async fn stat_with_not_existing_version(op: Operator) -> Result<()> {
-    if !op.info().full_capability().versioning {
+    if !op.info().full_capability().stat_with_versioning {
         return Ok(());
     }
 

From fc794d997ebbbbcb2d7c72bdf38360ca5967037a Mon Sep 17 00:00:00 2001
From: meteorgan <meteorite.gan@gmail.com>
Date: Wed, 25 Sep 2024 16:40:47 +0800
Subject: [PATCH 5/6] read/stat non-existent version returns NotFound, delete
 non-existent version returns Ok

---
 core/src/services/s3/backend.rs     |  8 ++++----
 core/src/services/s3/error.rs       |  2 +-
 core/src/types/capability.rs        |  8 ++++----
 core/tests/behavior/async_delete.rs | 21 ++++++++++++++++-----
 core/tests/behavior/async_list.rs   |  6 +++---
 core/tests/behavior/async_read.rs   | 23 ++++++++++++++++-------
 core/tests/behavior/async_stat.rs   | 23 ++++++++++++++++-------
 7 files changed, 60 insertions(+), 31 deletions(-)

diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index a054770bd68..8073cbfbb92 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -902,7 +902,7 @@ impl Access for S3Backend {
                 stat_with_override_cache_control: !self.core.disable_stat_with_override,
                 stat_with_override_content_disposition: !self.core.disable_stat_with_override,
                 stat_with_override_content_type: !self.core.disable_stat_with_override,
-                stat_with_versioning: self.core.enable_versioning,
+                stat_with_version: self.core.enable_versioning,
 
                 read: true,
                 read_with_if_match: true,
@@ -910,7 +910,7 @@ impl Access for S3Backend {
                 read_with_override_cache_control: true,
                 read_with_override_content_disposition: true,
                 read_with_override_content_type: true,
-                read_with_versioning: self.core.enable_versioning,
+                read_with_version: self.core.enable_versioning,
 
                 write: true,
                 write_can_empty: true,
@@ -933,7 +933,7 @@ impl Access for S3Backend {
                 },
 
                 delete: true,
-                delete_with_versioning: self.core.enable_versioning,
+                delete_with_version: self.core.enable_versioning,
 
                 copy: true,
 
@@ -941,7 +941,7 @@ impl Access for S3Backend {
                 list_with_limit: true,
                 list_with_start_after: true,
                 list_with_recursive: true,
-                list_with_versioning: self.core.enable_versioning,
+                list_with_version: self.core.enable_versioning,
 
                 presign: true,
                 presign_stat: true,
diff --git a/core/src/services/s3/error.rs b/core/src/services/s3/error.rs
index 2dd68e6686d..503ca4ecbd7 100644
--- a/core/src/services/s3/error.rs
+++ b/core/src/services/s3/error.rs
@@ -41,7 +41,7 @@ pub(super) fn parse_error(resp: Response<Buffer>) -> Error {
     let (mut kind, mut retryable) = match parts.status.as_u16() {
         403 => (ErrorKind::PermissionDenied, false),
         404 => (ErrorKind::NotFound, false),
-        304 | 400 | 412 => (ErrorKind::ConditionNotMatch, false),
+        304 | 412 => (ErrorKind::ConditionNotMatch, false),
         // Service like R2 could return 499 error with a message like:
         // Client Disconnect, we should retry it.
         499 => (ErrorKind::Unexpected, true),
diff --git a/core/src/types/capability.rs b/core/src/types/capability.rs
index 78ecb00205f..b3d276c1331 100644
--- a/core/src/types/capability.rs
+++ b/core/src/types/capability.rs
@@ -68,7 +68,7 @@ pub struct Capability {
     /// if operator supports stat with override content type.
     pub stat_with_override_content_type: bool,
     /// if operator supports stat with versioning.
-    pub stat_with_versioning: bool,
+    pub stat_with_version: bool,
 
     /// If operator supports read.
     pub read: bool,
@@ -83,7 +83,7 @@ pub struct Capability {
     /// if operator supports read with override content type.
     pub read_with_override_content_type: bool,
     /// if operator supports read with versioning.
-    pub read_with_versioning: bool,
+    pub read_with_version: bool,
 
     /// If operator supports write.
     pub write: bool,
@@ -124,7 +124,7 @@ pub struct Capability {
     /// If operator supports delete.
     pub delete: bool,
     /// if operator supports delete with versioning.
-    pub delete_with_versioning: bool,
+    pub delete_with_version: bool,
 
     /// If operator supports copy.
     pub copy: bool,
@@ -141,7 +141,7 @@ pub struct Capability {
     /// If backend supports list with recursive.
     pub list_with_recursive: bool,
     /// if operator supports list with versioning.
-    pub list_with_versioning: bool,
+    pub list_with_version: bool,
 
     /// If operator supports presign.
     pub presign: bool,
diff --git a/core/tests/behavior/async_delete.rs b/core/tests/behavior/async_delete.rs
index 44ef3afc8ce..0225f50ab66 100644
--- a/core/tests/behavior/async_delete.rs
+++ b/core/tests/behavior/async_delete.rs
@@ -216,7 +216,7 @@ pub async fn test_remove_all_with_prefix_exists(op: Operator) -> Result<()> {
 }
 
 pub async fn test_delete_with_version(op: Operator) -> Result<()> {
-    if !op.info().full_capability().delete_with_versioning {
+    if !op.info().full_capability().delete_with_version {
         return Ok(());
     }
 
@@ -252,21 +252,32 @@ pub async fn test_delete_with_version(op: Operator) -> Result<()> {
 }
 
 pub async fn test_delete_with_not_existing_version(op: Operator) -> Result<()> {
-    if !op.info().full_capability().delete_with_versioning {
+    if !op.info().full_capability().delete_with_version {
         return Ok(());
     }
 
+    // retrieve a valid version
     let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
     op.write(path.as_str(), content)
         .await
         .expect("write must success");
+    let version = op
+        .stat(path.as_str())
+        .await
+        .expect("stat must success")
+        .version()
+        .expect("must have stat")
+        .to_string();
 
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+    op.write(path.as_str(), content)
+        .await
+        .expect("write must success");
     let ret = op
         .delete_with(path.as_str())
-        .version("not-existing-version")
+        .version(version.as_str())
         .await;
-    assert!(ret.is_err());
-    assert_eq!(ret.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+    assert!(ret.is_ok());
 
     Ok(())
 }
diff --git a/core/tests/behavior/async_list.rs b/core/tests/behavior/async_list.rs
index 86d2aa33caf..ba862f70252 100644
--- a/core/tests/behavior/async_list.rs
+++ b/core/tests/behavior/async_list.rs
@@ -684,7 +684,7 @@ pub async fn test_list_only(op: Operator) -> Result<()> {
 }
 
 pub async fn test_list_files_with_version(op: Operator) -> Result<()> {
-    if !op.info().full_capability().list_with_versioning {
+    if !op.info().full_capability().list_with_version {
         return Ok(());
     }
 
@@ -722,7 +722,7 @@ pub async fn test_list_with_version_and_limit(op: Operator) -> Result<()> {
     if op.info().scheme() == Scheme::Gdrive {
         return Ok(());
     }
-    if !op.info().full_capability().list_with_versioning {
+    if !op.info().full_capability().list_with_version {
         return Ok(());
     }
 
@@ -775,7 +775,7 @@ pub async fn test_list_with_version_and_limit(op: Operator) -> Result<()> {
 }
 
 pub async fn test_list_with_version_and_start_after(op: Operator) -> Result<()> {
-    if !op.info().full_capability().list_with_versioning {
+    if !op.info().full_capability().list_with_version {
         return Ok(());
     }
 
diff --git a/core/tests/behavior/async_read.rs b/core/tests/behavior/async_read.rs
index 19e48aad3f7..d691165e5fb 100644
--- a/core/tests/behavior/async_read.rs
+++ b/core/tests/behavior/async_read.rs
@@ -557,7 +557,7 @@ pub async fn test_read_only_read_with_if_none_match(op: Operator) -> anyhow::Res
 }
 
 pub async fn test_read_with_version(op: Operator) -> anyhow::Result<()> {
-    if !op.info().full_capability().read_with_versioning {
+    if !op.info().full_capability().read_with_version {
         return Ok(());
     }
 
@@ -591,21 +591,30 @@ pub async fn test_read_with_version(op: Operator) -> anyhow::Result<()> {
 }
 
 pub async fn test_read_with_not_existing_version(op: Operator) -> anyhow::Result<()> {
-    if !op.info().full_capability().read_with_versioning {
+    if !op.info().full_capability().read_with_version {
         return Ok(());
     }
 
+    // retrieve a valid version
     let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
     op.write(path.as_str(), content.clone())
         .await
         .expect("write must success");
+    let version = op
+        .stat(path.as_str())
+        .await
+        .expect("stat must success")
+        .version()
+        .expect("must have version")
+        .to_string();
 
-    let ret = op
-        .read_with(path.as_str())
-        .version("not-existing-version")
-        .await;
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+    op.write(path.as_str(), content)
+        .await
+        .expect("write must success");
+    let ret = op.read_with(path.as_str()).version(&version).await;
     assert!(ret.is_err());
-    assert_eq!(ret.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+    assert_eq!(ret.unwrap_err().kind(), ErrorKind::NotFound);
 
     Ok(())
 }
diff --git a/core/tests/behavior/async_stat.rs b/core/tests/behavior/async_stat.rs
index 967985995d2..38f50704acd 100644
--- a/core/tests/behavior/async_stat.rs
+++ b/core/tests/behavior/async_stat.rs
@@ -502,7 +502,7 @@ pub async fn test_read_only_stat_root(op: Operator) -> Result<()> {
 }
 
 pub async fn test_stat_with_version(op: Operator) -> Result<()> {
-    if !op.info().full_capability().stat_with_versioning {
+    if !op.info().full_capability().stat_with_version {
         return Ok(());
     }
 
@@ -540,21 +540,30 @@ pub async fn test_stat_with_version(op: Operator) -> Result<()> {
 }
 
 pub async fn stat_with_not_existing_version(op: Operator) -> Result<()> {
-    if !op.info().full_capability().stat_with_versioning {
+    if !op.info().full_capability().stat_with_version {
         return Ok(());
     }
 
+    // retrieve a valid version
     let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
     op.write(path.as_str(), content.clone())
         .await
         .expect("write must success");
+    let version = op
+        .stat(path.as_str())
+        .await
+        .expect("stat must success")
+        .version()
+        .expect("must have version")
+        .to_string();
 
-    let ret = op
-        .stat_with(path.as_str())
-        .version("not-existing-version")
-        .await;
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+    op.write(path.as_str(), content)
+        .await
+        .expect("write must success");
+    let ret = op.stat_with(path.as_str()).version(version.as_str()).await;
     assert!(ret.is_err());
-    assert_eq!(ret.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+    assert_eq!(ret.unwrap_err().kind(), ErrorKind::NotFound);
 
     Ok(())
 }

From c2532a6ff071ac98765a9645f66cbe80f949ed2f Mon Sep 17 00:00:00 2001
From: meteorgan <meteorite.gan@gmail.com>
Date: Wed, 25 Sep 2024 16:55:36 +0800
Subject: [PATCH 6/6] fix docs

---
 core/src/types/capability.rs | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/core/src/types/capability.rs b/core/src/types/capability.rs
index b3d276c1331..4a874b23877 100644
--- a/core/src/types/capability.rs
+++ b/core/src/types/capability.rs
@@ -67,7 +67,7 @@ pub struct Capability {
     pub stat_with_override_content_disposition: bool,
     /// if operator supports stat with override content type.
     pub stat_with_override_content_type: bool,
-    /// if operator supports stat with versioning.
+    /// if operator supports stat with version.
     pub stat_with_version: bool,
 
     /// If operator supports read.
@@ -82,7 +82,7 @@ pub struct Capability {
     pub read_with_override_content_disposition: bool,
     /// if operator supports read with override content type.
     pub read_with_override_content_type: bool,
-    /// if operator supports read with versioning.
+    /// if operator supports read with version.
     pub read_with_version: bool,
 
     /// If operator supports write.
@@ -123,7 +123,7 @@ pub struct Capability {
 
     /// If operator supports delete.
     pub delete: bool,
-    /// if operator supports delete with versioning.
+    /// if operator supports delete with version.
     pub delete_with_version: bool,
 
     /// If operator supports copy.
@@ -140,7 +140,7 @@ pub struct Capability {
     pub list_with_start_after: bool,
     /// If backend supports list with recursive.
     pub list_with_recursive: bool,
-    /// if operator supports list with versioning.
+    /// if operator supports list with version.
     pub list_with_version: bool,
 
     /// If operator supports presign.