test(e2e): activate conditional write regressions

crates/e2e_test/src/reliant/conditional_writes.rs

@@ -1,52 +1,26 @@
 #![cfg(test)]
 
-use aws_config::meta::region::RegionProviderChain;
+use crate::common::{RustFSTestEnvironment, TEST_BUCKET, init_logging};
 use aws_sdk_s3::Client;
-use aws_sdk_s3::config::{Credentials, Region};
-use aws_sdk_s3::error::SdkError;
+use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
 use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
 use bytes::Bytes;
 use std::error::Error;
+use std::fmt::Debug;
 
-const ENDPOINT: &str = "http://localhost:9000";
-const ACCESS_KEY: &str = "rustfsadmin";
-const SECRET_KEY: &str = "rustfsadmin";
-const BUCKET: &str = "api-test";
+type TestResult = Result<(), Box<dyn Error + Send + Sync>>;
 
-async fn create_aws_s3_client() -> Result<Client, Box<dyn Error>> {
-    let region_provider = RegionProviderChain::default_provider().or_else(Region::new("us-east-1"));
-    let shared_config = aws_config::defaults(aws_config::BehaviorVersion::latest())
-        .region(region_provider)
-        .credentials_provider(Credentials::new(ACCESS_KEY, SECRET_KEY, None, None, "static"))
-        .endpoint_url(ENDPOINT)
-        .load()
-        .await;
-
-    let client = Client::from_conf(
-        aws_sdk_s3::Config::from(&shared_config)
-            .to_builder()
-            .force_path_style(true)
-            .build(),
+fn assert_s3_error_code<T, E>(result: Result<T, SdkError<E>>, expected: &str)
+where
+    T: Debug,
+    E: ProvideErrorMetadata + Debug,
+{
+    let error = result.expect_err("conditional request must fail");
+    assert_eq!(
+        error.as_service_error().and_then(ProvideErrorMetadata::code),
+        Some(expected),
+        "unexpected conditional request error: {error:?}"
     );
-    Ok(client)
-}
-
-/// Setup test bucket, creating it if it doesn't exist
-async fn setup_test_bucket(client: &Client) -> Result<(), Box<dyn Error>> {
-    match client.create_bucket().bucket(BUCKET).send().await {
-        Ok(_) => {}
-        Err(SdkError::ServiceError(e)) => {
-            let e = e.into_err();
-            let error_code = e.meta().code().unwrap_or("");
-            if !error_code.eq("BucketAlreadyExists") {
-                return Err(e.into());
-            }
-        }
-        Err(e) => {
-            return Err(e.into());
-        }
-    }
-    Ok(())
 }
 
 /// Generate test data of specified size
@@ -60,7 +34,12 @@ fn generate_test_data(size: usize) -> Vec<u8> {
 }
 
 /// Upload an object and return its ETag
-async fn upload_object_with_metadata(client: &Client, bucket: &str, key: &str, data: &[u8]) -> Result<String, Box<dyn Error>> {
+async fn upload_object_with_metadata(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    data: &[u8],
+) -> Result<String, Box<dyn Error + Send + Sync>> {
     let response = client
         .put_object()
         .bucket(bucket)
@@ -69,188 +48,164 @@ async fn upload_object_with_metadata(client: &Client, bucket: &str, key: &str, d
         .send()
         .await?;
 
-    let etag = response.e_tag().unwrap_or("").to_string();
-    Ok(etag)
+    response
+        .e_tag()
+        .map(str::to_owned)
+        .ok_or_else(|| std::io::Error::other("put object response did not include an ETag").into())
 }
 
-/// Cleanup test objects from bucket
-async fn cleanup_objects(client: &Client, bucket: &str, keys: &[&str]) {
-    for key in keys {
-        let _ = client.delete_object().bucket(bucket).key(*key).send().await;
-    }
-}
-
-/// Generate unique test object key
-fn generate_test_key(prefix: &str) -> String {
-    use std::time::{SystemTime, UNIX_EPOCH};
-    let timestamp = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
-    format!("{prefix}-{timestamp}")
+async fn object_body(client: &Client, key: &str) -> Result<Bytes, Box<dyn Error + Send + Sync>> {
+    let response = client.get_object().bucket(TEST_BUCKET).key(key).send().await?;
+    Ok(response.body.collect().await?.into_bytes())
 }
 
 #[tokio::test]
-#[ignore = "requires running RustFS server at localhost:9000"]
-async fn test_conditional_put_okay() -> Result<(), Box<dyn std::error::Error>> {
-    let client = create_aws_s3_client().await?;
-    setup_test_bucket(&client).await?;
+async fn test_conditional_put_okay() -> TestResult {
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    env.create_test_bucket(TEST_BUCKET).await?;
+    let client = env.create_s3_client();
 
-    let test_key = generate_test_key("conditional-put-ok");
+    let test_key = "conditional-put-ok";
     let initial_data = generate_test_data(1024); // 1KB test data
-    let updated_data = generate_test_data(2048); // 2KB updated data
+    let matching_data = generate_test_data(2048); // 2KB updated data
+    let non_matching_data = generate_test_data(3072); // 3KB updated data
 
     // Upload initial object and get its ETag
-    let initial_etag = upload_object_with_metadata(&client, BUCKET, &test_key, &initial_data).await?;
+    let initial_etag = upload_object_with_metadata(&client, TEST_BUCKET, test_key, &initial_data).await?;
 
     // Test 1: PUT with matching If-Match condition (should succeed)
-    let response1 = client
+    client
         .put_object()
-        .bucket(BUCKET)
-        .key(&test_key)
-        .body(Bytes::from(updated_data.clone()).into())
+        .bucket(TEST_BUCKET)
+        .key(test_key)
+        .body(Bytes::from(matching_data.clone()).into())
         .if_match(&initial_etag)
         .send()
-        .await;
-    assert!(response1.is_ok(), "PUT with matching If-Match should succeed");
+        .await?;
+    assert_eq!(object_body(&client, test_key).await?.as_ref(), matching_data);
 
     // Test 2: PUT with non-matching If-None-Match condition (should succeed)
     let fake_etag = "\"fake-etag-12345\"";
-    let response2 = client
+    client
         .put_object()
-        .bucket(BUCKET)
-        .key(&test_key)
-        .body(Bytes::from(updated_data.clone()).into())
+        .bucket(TEST_BUCKET)
+        .key(test_key)
+        .body(Bytes::from(non_matching_data.clone()).into())
         .if_none_match(fake_etag)
         .send()
-        .await;
-    assert!(response2.is_ok(), "PUT with non-matching If-None-Match should succeed");
-
-    // Cleanup
-    cleanup_objects(&client, BUCKET, &[&test_key]).await;
+        .await?;
+    assert_eq!(object_body(&client, test_key).await?.as_ref(), non_matching_data);
 
     Ok(())
 }
 
 #[tokio::test]
-#[ignore = "requires running RustFS server at localhost:9000"]
-async fn test_conditional_put_failed() -> Result<(), Box<dyn std::error::Error>> {
-    let client = create_aws_s3_client().await?;
-    setup_test_bucket(&client).await?;
+async fn test_conditional_put_failed() -> TestResult {
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    env.create_test_bucket(TEST_BUCKET).await?;
+    let client = env.create_s3_client();
 
-    let test_key = generate_test_key("conditional-put-failed");
+    let test_key = "conditional-put-failed";
     let initial_data = generate_test_data(1024);
     let updated_data = generate_test_data(2048);
 
     // Upload initial object and get its ETag
-    let initial_etag = upload_object_with_metadata(&client, BUCKET, &test_key, &initial_data).await?;
+    let initial_etag = upload_object_with_metadata(&client, TEST_BUCKET, test_key, &initial_data).await?;
 
     // Test 1: PUT with non-matching If-Match condition (should fail with 412)
     let fake_etag = "\"fake-etag-should-not-match\"";
     let response1 = client
         .put_object()
-        .bucket(BUCKET)
-        .key(&test_key)
+        .bucket(TEST_BUCKET)
+        .key(test_key)
         .body(Bytes::from(updated_data.clone()).into())
         .if_match(fake_etag)
         .send()
         .await;
 
-    assert!(response1.is_err(), "PUT with non-matching If-Match should fail");
-    if let Err(e) = response1 {
-        if let SdkError::ServiceError(e) = e {
-            let e = e.into_err();
-            let error_code = e.meta().code().unwrap_or("");
-            assert_eq!("PreconditionFailed", error_code);
-        } else {
-            panic!("Unexpected error: {e:?}");
-        }
-    }
+    assert_s3_error_code(response1, "PreconditionFailed");
+    assert_eq!(object_body(&client, test_key).await?.as_ref(), initial_data);
 
     // Test 2: PUT with matching If-None-Match condition (should fail with 412)
     let response2 = client
         .put_object()
-        .bucket(BUCKET)
-        .key(&test_key)
+        .bucket(TEST_BUCKET)
+        .key(test_key)
         .body(Bytes::from(updated_data.clone()).into())
         .if_none_match(&initial_etag)
         .send()
         .await;
 
-    assert!(response2.is_err(), "PUT with matching If-None-Match should fail");
-    if let Err(e) = response2 {
-        if let SdkError::ServiceError(e) = e {
-            let e = e.into_err();
-            let error_code = e.meta().code().unwrap_or("");
-            assert_eq!("PreconditionFailed", error_code);
-        } else {
-            panic!("Unexpected error: {e:?}");
-        }
-    }
-
-    // Cleanup - only need to clean up the initial object since failed PUTs shouldn't create objects
-    cleanup_objects(&client, BUCKET, &[&test_key]).await;
+    assert_s3_error_code(response2, "PreconditionFailed");
+    assert_eq!(object_body(&client, test_key).await?.as_ref(), initial_data);
 
     Ok(())
 }
 
 #[tokio::test]
-#[ignore = "requires running RustFS server at localhost:9000"]
-async fn test_conditional_put_when_object_does_not_exist() -> Result<(), Box<dyn std::error::Error>> {
-    let client = create_aws_s3_client().await?;
-    setup_test_bucket(&client).await?;
+async fn test_conditional_put_when_object_does_not_exist() -> TestResult {
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    env.create_test_bucket(TEST_BUCKET).await?;
+    let client = env.create_s3_client();
 
-    let key = "some_key";
-    cleanup_objects(&client, BUCKET, &[key]).await;
+    let key = "conditional-put-missing";
 
     // When the object does not exist, the If-Match condition should always fail
     let response1 = client
         .put_object()
-        .bucket(BUCKET)
+        .bucket(TEST_BUCKET)
         .key(key)
         .body(Bytes::from(generate_test_data(1024)).into())
         .if_match("*")
         .send()
         .await;
-    assert!(response1.is_err());
-    if let Err(e) = response1 {
-        if let SdkError::ServiceError(e) = e {
-            let e = e.into_err();
-            let error_code = e.meta().code().unwrap_or("");
-            assert_eq!("NoSuchKey", error_code);
-        } else {
-            panic!("Unexpected error: {e:?}");
-        }
-    }
+    assert_s3_error_code(response1, "NoSuchKey");
 
     // When the object does not exist, the If-None-Match condition should be able to succeed
-    let response2 = client
+    let created_data = generate_test_data(1024);
+    client
         .put_object()
-        .bucket(BUCKET)
+        .bucket(TEST_BUCKET)
         .key(key)
-        .body(Bytes::from(generate_test_data(1024)).into())
+        .body(Bytes::from(created_data.clone()).into())
         .if_none_match("*")
         .send()
-        .await;
-    assert!(response2.is_ok());
+        .await?;
+    assert_eq!(object_body(&client, key).await?.as_ref(), created_data);
 
-    cleanup_objects(&client, BUCKET, &[key]).await;
     Ok(())
 }
 
 #[tokio::test]
-#[ignore = "requires running RustFS server at localhost:9000"]
-async fn test_conditional_multi_part_upload() -> Result<(), Box<dyn std::error::Error>> {
-    let client = create_aws_s3_client().await?;
-    setup_test_bucket(&client).await?;
+async fn test_conditional_multi_part_upload() -> TestResult {
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    env.create_test_bucket(TEST_BUCKET).await?;
+    let client = env.create_s3_client();
 
-    let test_key = generate_test_key("multipart-upload-ok");
+    let test_key = "conditional-multipart-upload";
     let test_data = generate_test_data(1024);
-    let initial_etag = upload_object_with_metadata(&client, BUCKET, &test_key, &test_data).await?;
+    let initial_etag = upload_object_with_metadata(&client, TEST_BUCKET, test_key, &test_data).await?;
 
     let part_size = 5 * 1024 * 1024; // 5MB per part (minimum for multipart)
     let num_parts = 3;
     let mut parts = Vec::new();
+    let mut expected_data = Vec::with_capacity(part_size * usize::try_from(num_parts)?);
 
     // Initiate multipart upload
-    let initiate_response = client.create_multipart_upload().bucket(BUCKET).key(&test_key).send().await?;
+    let initiate_response = client
+        .create_multipart_upload()
+        .bucket(TEST_BUCKET)
+        .key(test_key)
+        .send()
+        .await?;
 
     let upload_id = initiate_response
         .upload_id()
@@ -258,12 +213,13 @@ async fn test_conditional_multi_part_upload() -> Result<(), Box<dyn std::error::
 
     // Upload parts
     for part_number in 1..=num_parts {
-        let part_data = generate_test_data(part_size);
+        let part_data = vec![u8::try_from(part_number)?; part_size];
+        expected_data.extend_from_slice(&part_data);
 
         let upload_part_response = client
             .upload_part()
-            .bucket(BUCKET)
-            .key(&test_key)
+            .bucket(TEST_BUCKET)
+            .key(test_key)
             .upload_id(upload_id)
             .part_number(part_number)
             .body(Bytes::from(part_data).into())
@@ -286,57 +242,62 @@ async fn test_conditional_multi_part_upload() -> Result<(), Box<dyn std::error::
     // Test 1: Multipart upload with wildcard If-None-Match, should fail
     let complete_response = client
         .complete_multipart_upload()
-        .bucket(BUCKET)
-        .key(&test_key)
+        .bucket(TEST_BUCKET)
+        .key(test_key)
         .upload_id(upload_id)
         .multipart_upload(completed_upload.clone())
         .if_none_match("*")
         .send()
         .await;
 
-    assert!(complete_response.is_err());
+    assert_s3_error_code(complete_response, "PreconditionFailed");
 
     // Test 2: Multipart upload with matching If-None-Match, should fail
     let complete_response = client
         .complete_multipart_upload()
-        .bucket(BUCKET)
-        .key(&test_key)
+        .bucket(TEST_BUCKET)
+        .key(test_key)
         .upload_id(upload_id)
         .multipart_upload(completed_upload.clone())
         .if_none_match(initial_etag.clone())
         .send()
         .await;
 
-    assert!(complete_response.is_err());
+    assert_s3_error_code(complete_response, "PreconditionFailed");
 
     // Test 3: Multipart upload with unmatching If-Match, should fail
     let complete_response = client
         .complete_multipart_upload()
-        .bucket(BUCKET)
-        .key(&test_key)
+        .bucket(TEST_BUCKET)
+        .key(test_key)
         .upload_id(upload_id)
         .multipart_upload(completed_upload.clone())
         .if_match("\"abcdef\"")
         .send()
         .await;
 
-    assert!(complete_response.is_err());
+    assert_s3_error_code(complete_response, "PreconditionFailed");
+
+    let staged_parts = client
+        .list_parts()
+        .bucket(TEST_BUCKET)
+        .key(test_key)
+        .upload_id(upload_id)
+        .send()
+        .await?;
+    assert_eq!(staged_parts.parts().len(), usize::try_from(num_parts)?);
 
     // Test 4: Multipart upload with matching If-Match, should succeed
-    let complete_response = client
+    client
         .complete_multipart_upload()
-        .bucket(BUCKET)
-        .key(&test_key)
+        .bucket(TEST_BUCKET)
+        .key(test_key)
         .upload_id(upload_id)
-        .multipart_upload(completed_upload.clone())
+        .multipart_upload(completed_upload)
         .if_match(initial_etag)
         .send()
-        .await;
-
-    assert!(complete_response.is_ok());
-
-    // Cleanup
-    cleanup_objects(&client, BUCKET, &[&test_key]).await;
+        .await?;
+    assert_eq!(object_body(&client, test_key).await?.as_ref(), expected_data);
 
     Ok(())
 }