feat: add a test macro (#170)

.github/workflows/rust.yml

@@ -11,26 +11,53 @@ env:
   CARGO_TERM_COLOR: always
 
 jobs:
-  build:
+  check-format:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v4
+    - name: Check format
+      run: |
+        cargo fmt --all -- --check
 
+  clippy:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: clippy
+        run: cargo clippy --all-targets --all-features --workspace -- -D warnings
+  test-multi-thread:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Run tests
+        run: |
+          ./tests/start-server.sh
+          export SERVER_ENDPOINT=localhost:9000
+          export ACCESS_KEY=minioadmin
+          export SECRET_KEY=minioadmin
+          export ENABLE_HTTPS=1
+          export MINIO_SSL_CERT_FILE=./tests/public.crt
+          MINIO_TEST_TOKIO_RUNTIME_FLAVOR="multi_thread" cargo test -- --nocapture
+  test-current-thread:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Run tests
+        run: |
+          ./tests/start-server.sh
+          export SERVER_ENDPOINT=localhost:9000
+          export ACCESS_KEY=minioadmin
+          export SECRET_KEY=minioadmin
+          export ENABLE_HTTPS=1
+          export MINIO_SSL_CERT_FILE=./tests/public.crt
+          MINIO_TEST_TOKIO_RUNTIME_FLAVOR="current_thread" cargo test -- --nocapture
+
+  build:
     runs-on: ubuntu-latest
     timeout-minutes: 5
-
     steps:
     - uses: actions/checkout@v4
     - name: Build
       run: |
         cargo --version
-        cargo fmt --all -- --check
-        cargo clippy --all-targets --all-features
         cargo build --bins --examples --tests --benches --verbose
-
-    - name: Run tests
-      run: |
-        ./tests/start-server.sh
-        export SERVER_ENDPOINT=localhost:9000
-        export ACCESS_KEY=minioadmin
-        export SECRET_KEY=minioadmin
-        export ENABLE_HTTPS=1
-        export SSL_CERT_FILE=./tests/public.crt
-        cargo test --verbose -- --nocapture

Cargo.toml

@@ -62,6 +62,7 @@ async-std = { version = "1.13.1", features = ["attributes", "tokio1"] }
 clap = { version = "4.5.40", features = ["derive"] }
 quickcheck = "1.0.3"
 criterion = "0.6.0"
+minio-macros = { path = "./macros" }
 
 [lib]
 name = "minio"

benches/s3/api_benchmarks.rs

@@ -12,6 +12,7 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
+#![allow(unused_must_use)]
 
 mod bench_bucket_exists;
 mod bench_bucket_lifecycle;

benches/s3/bench_bucket_replication.rs

@@ -42,7 +42,7 @@ pub(crate) fn bench_put_bucket_replication(criterion: &mut Criterion) {
 
             let _resp: PutBucketVersioningResponse = ctx
                 .client
-                .put_bucket_versioning(&ctx.aux_bucket.clone().unwrap())
+                .put_bucket_versioning(ctx.aux_bucket.clone().unwrap())
                 .versioning_status(VersioningStatus::Enabled)
                 .send()
                 .await
@@ -77,7 +77,7 @@ pub(crate) fn bench_get_bucket_replication(criterion: &mut Criterion) {
 
             let _resp: PutBucketVersioningResponse = ctx
                 .client
-                .put_bucket_versioning(&ctx.aux_bucket.clone().unwrap())
+                .put_bucket_versioning(ctx.aux_bucket.clone().unwrap())
                 .versioning_status(VersioningStatus::Enabled)
                 .send()
                 .await
@@ -107,7 +107,7 @@ pub(crate) fn bench_delete_bucket_replication(criterion: &mut Criterion) {
 
             let _resp: PutBucketVersioningResponse = ctx
                 .client
-                .put_bucket_versioning(&ctx.aux_bucket.clone().unwrap())
+                .put_bucket_versioning(ctx.aux_bucket.clone().unwrap())
                 .versioning_status(VersioningStatus::Enabled)
                 .send()
                 .await

benches/s3/common_benches.rs

@@ -40,7 +40,7 @@ impl Ctx2 {
     /// Create a new context with a bucket
     pub async fn new() -> Self {
         unsafe {
-            env::set_var("SSL_CERT_FILE", "./tests/public.crt");
+            env::set_var("MINIO_SSL_CERT_FILE", "./tests/public.crt");
         }
         let ctx = TestContext::new_from_env();
         let (bucket_name, cleanup) = ctx.create_bucket_helper().await;
@@ -57,7 +57,7 @@ impl Ctx2 {
     /// Create a new context with a bucket and an object
     pub async fn new_with_object(object_lock: bool) -> Self {
         unsafe {
-            env::set_var("SSL_CERT_FILE", "./tests/public.crt");
+            env::set_var("MINIO_SSL_CERT_FILE", "./tests/public.crt");
         }
         let ctx = TestContext::new_from_env();
         let bucket_name: String = rand_bucket_name();

common/src/cleanup_guard.rs

@@ -13,11 +13,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use async_std::future::timeout;
 use minio::s3::Client;
 
-use std::thread;
-
 /// Cleanup guard that removes the bucket when it is dropped
 pub struct CleanupGuard {
     client: Client,
@@ -32,41 +29,26 @@ impl CleanupGuard {
             bucket_name: bucket_name.into(),
         }
     }
-}
 
-impl Drop for CleanupGuard {
+    pub async fn cleanup(&self) {
-    fn drop(&mut self) {
+        cleanup(self.client.clone(), &self.bucket_name).await;
-        let client = self.client.clone();
-        let bucket_name = self.bucket_name.clone();
-        //println!("Going to remove bucket {}", bucket_name);
-
-        // Spawn the cleanup task in a way that detaches it from the current runtime
-        thread::spawn(move || {
-            // Create a new runtime for this thread
-            let rt = tokio::runtime::Runtime::new().unwrap();
-
-            // Execute the async cleanup in this new runtime
-            rt.block_on(async {
-                // do the actual removal of the bucket
-                match timeout(
-                    std::time::Duration::from_secs(60),
-                    client.delete_and_purge_bucket(&bucket_name),
-                )
-                .await
-                {
-                    Ok(result) => match result {
-                        Ok(_) => {
-                            //println!("Bucket {} removed successfully", bucket_name),
-                        }
-                        Err(_e) => {
-                            //println!("Error removing bucket {}: {:?}", bucket_name, e)
-                        }
-                    },
-                    Err(_) => println!("Timeout after 60s while removing bucket {}", bucket_name),
-                }
-            });
-        })
-        .join()
-        .unwrap(); // This blocks the current thread until cleanup is done
     }
 }
+
+pub async fn cleanup(client: Client, bucket_name: &str) {
+    tokio::select!(
+                _ = tokio::time::sleep(std::time::Duration::from_secs(60)) => {
+                    eprintln!("Cleanup timeout after 60s while removing bucket {}", bucket_name);
+                },
+                outcome = client.delete_and_purge_bucket(bucket_name) => {
+                    match outcome {
+                        Ok(_) => {
+                            eprintln!("Bucket {} removed successfully", bucket_name);
+                        }
+                        Err(e) => {
+                            eprintln!("Error removing bucket {}: {:?}", bucket_name, e);
+                        }
+                    }
+                }
+    );
+}

common/src/rand_reader.rs

@@ -13,7 +13,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use futures::AsyncRead;
 use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
 
 pub struct RandReader {
     size: u64,
@@ -28,10 +31,7 @@ impl RandReader {
 
 impl io::Read for RandReader {
     fn read(&mut self, buf: &mut [u8]) -> Result<usize, io::Error> {
-        let bytes_read: usize = match (self.size as usize) > buf.len() {
+        let bytes_read = buf.len().min(self.size as usize);
-            true => buf.len(),
-            false => self.size as usize,
-        };
 
         if bytes_read > 0 {
             let random: &mut dyn rand::RngCore = &mut rand::thread_rng();
@@ -43,3 +43,22 @@ impl io::Read for RandReader {
         Ok(bytes_read)
     }
 }
+
+impl AsyncRead for RandReader {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        let bytes_read = buf.len().min(self.size as usize);
+
+        if bytes_read > 0 {
+            let random: &mut dyn rand::RngCore = &mut rand::thread_rng();
+            random.fill_bytes(&mut buf[0..bytes_read]);
+        }
+
+        self.get_mut().size -= bytes_read as u64;
+
+        Poll::Ready(Ok(bytes_read))
+    }
+}

common/src/rand_src.rs

@@ -14,7 +14,6 @@
 // limitations under the License.
 
 use async_std::stream::Stream;
-use async_std::task;
 use bytes::Bytes;
 use futures::io::AsyncRead;
 use rand::prelude::SmallRng;
@@ -41,26 +40,21 @@ impl Stream for RandSrc {
 
     fn poll_next(
         self: std::pin::Pin<&mut Self>,
-        _cx: &mut task::Context<'_>,
+        _cx: &mut Context<'_>,
-    ) -> task::Poll<Option<Self::Item>> {
+    ) -> Poll<Option<Self::Item>> {
         if self.size == 0 {
-            return task::Poll::Ready(None);
+            return Poll::Ready(None);
         }
-
+        // Limit to 8 KiB per read
-        let bytes_read = match self.size > 64 * 1024 {
+        let bytes_read = self.size.min(8 * 1024) as usize;
-            true => 64 * 1024,
-            false => self.size as usize,
-        };
 
         let this = self.get_mut();
 
         let mut buf = vec![0; bytes_read];
         let random: &mut dyn rand::RngCore = &mut this.rng;
         random.fill_bytes(&mut buf);
-
         this.size -= bytes_read as u64;
-
+        Poll::Ready(Some(Ok(Bytes::from(buf))))
-        task::Poll::Ready(Some(Ok(Bytes::from(buf))))
     }
 }
 

common/src/test_context.rs

@@ -42,7 +42,7 @@ impl TestContext {
             let access_key = std::env::var("ACCESS_KEY").unwrap();
             let secret_key = std::env::var("SECRET_KEY").unwrap();
             let secure = std::env::var("ENABLE_HTTPS").is_ok();
-            let value = std::env::var("SSL_CERT_FILE").unwrap();
+            let value = std::env::var("MINIO_SSL_CERT_FILE").unwrap();
             let mut ssl_cert_file = None;
             if !value.is_empty() {
                 ssl_cert_file = Some(Path::new(&value));
@@ -97,8 +97,8 @@ impl TestContext {
                 .unwrap_or(false);
             log::debug!("ENABLE_HTTPS={}", secure);
             let ssl_cert: String =
-                std::env::var("SSL_CERT_FILE").unwrap_or(DEFAULT_SSL_CERT_FILE.to_string());
+                std::env::var("MINIO_SSL_CERT_FILE").unwrap_or(DEFAULT_SSL_CERT_FILE.to_string());
-            log::debug!("SSL_CERT_FILE={}", ssl_cert);
+            log::debug!("MINIO_SSL_CERT_FILE={}", ssl_cert);
             let ssl_cert_file: PathBuf = ssl_cert.into();
             let ignore_cert_check: bool = std::env::var("IGNORE_CERT_CHECK")
                 .unwrap_or(DEFAULT_IGNORE_CERT_CHECK.to_string())

macros/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "minio-macros"
+version = "0.1.0"
+edition = "2024"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+[lib]
+proc-macro = true
+
+
+[dependencies]
+syn = "2.0.53"
+proc-macro2 = "1.0.37"
+quote = "1.0.18"
+darling = "0.20.8"
+darling_core = "0.20.8"
+uuid = { version = "1.17.0", features = ["v4"] }
+
+[dev-dependencies]
+minio_common = { path = "../common" }

macros/src/lib.rs

@@ -0,0 +1,109 @@
+// MinIO Rust Library for Amazon S3 Compatible Cloud Storage
+// Copyright 2025 MinIO, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+mod test_attr;
+
+use darling::ast::NestedMeta;
+use darling::{Error, FromMeta};
+use syn::ItemFn;
+extern crate proc_macro;
+
+/// A proc macro attribute for writing MinIO tests.
+///
+/// This macro extends the `#[tokio::test]` attribute to provide additional functionality for
+/// testing MinIO operations. The macro takes care of setting up and tearing down the test
+/// environment, it automatically creates a bucket for the test if needed and cleans it up after
+/// the test is done.
+///
+/// By default, it requires the test function to have two parameters:
+///
+/// - `ctx: TestContext` - The test context which will give you access to a minio-client.
+/// - `bucket_name: String` - The name of the bucket to be used in the test.
+///
+/// ```no_run
+/// use minio_common::test_context::TestContext;
+/// #[minio_macros::test]
+/// async fn my_test(ctx: TestContext, bucket_name: String) {
+///    // Your test code here
+/// }
+/// ```
+///
+/// If the `no_bucket` argument is provided, the test function must have only one parameter:
+///
+/// - `ctx: TestContext` - The test context which will give you access to a minio-client.
+///
+/// ```no_run
+/// use minio_common::test_context::TestContext;
+/// #[minio_macros::test(no_bucket)]
+/// async fn my_test(ctx: TestContext) {
+///    // Your test code here
+/// }
+///```
+/// The macro also supports additional arguments:
+///
+/// - `flavor`: Specifies the flavor of the Tokio test (e.g., "multi_thread").
+/// - `worker_threads`: Specifies the number of worker threads for the Tokio test.
+/// - `bucket_name`: Specifies the name of the bucket to be used in the test. If not provided, a random bucket name will be generated.
+/// - `skip_if_express`: If set, the test will be skipped if the MinIO server is running in Express mode.
+/// - `object_lock`: If set, the test bucket is created with `.object_lock(true)`
+/// - `no_cleanup`: If set, the test bucket is not deleted after the test is run.
+///
+/// ```no_run
+/// use minio_common::test_context::TestContext;
+/// #[minio_macros::test(skip_if_express)]
+/// async fn my_test(ctx: TestContext) {
+///    // this test will not run if the MinIO server is running in Express mode
+/// }
+/// ```
+/// - `skip_if_not_express`: If set, the test will be skipped if the MinIO server is NOT running in Express mode.
+/// ```no_run
+/// use minio_common::test_context::TestContext;
+/// #[minio_macros::test(skip_if_not_express)]
+/// async fn my_test(ctx: TestContext) {
+///    // this test will not run if the MinIO server is NOT running in Express mode
+/// }
+/// ```
+#[proc_macro_attribute]
+pub fn test(
+    args: proc_macro::TokenStream,
+    input: proc_macro::TokenStream,
+) -> proc_macro::TokenStream {
+    // Parse the function
+    let input_fn = match syn::parse::<ItemFn>(input.clone()) {
+        Ok(input_fn) => input_fn,
+        Err(err) => return err.to_compile_error().into(),
+    };
+
+    // Parse the macro arguments
+    let attr_args = match NestedMeta::parse_meta_list(args.into()) {
+        Ok(v) => v,
+        Err(e) => return Error::from(e).write_errors().into(),
+    };
+
+    let args = match test_attr::MacroArgs::from_list(&attr_args) {
+        Ok(v) => v,
+        Err(e) => return e.write_errors().into(),
+    };
+
+    // Validate the function arguments
+    if let Err(err) = args.validate(&input_fn) {
+        return err;
+    }
+
+    // Expand the macro
+    match test_attr::expand_test_macro(args, input_fn) {
+        Ok(expanded) => expanded.into(),
+        Err(err) => err.into(),
+    }
+}

macros/src/test_attr.rs

@@ -0,0 +1,278 @@
+// MinIO Rust Library for Amazon S3 Compatible Cloud Storage
+// Copyright 2025 MinIO, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use darling::FromMeta;
+use darling_core::Error;
+use proc_macro2::TokenStream;
+use quote::{ToTokens, quote, quote_spanned};
+use syn::punctuated::Punctuated;
+use syn::spanned::Spanned;
+use syn::{FnArg, ItemFn, ReturnType};
+use uuid::Uuid;
+
+#[derive(Debug, FromMeta)]
+pub(crate) struct MacroArgs {
+    flavor: Option<String>,
+    worker_threads: Option<usize>,
+    bucket_name: Option<String>,
+    skip_if_express: darling::util::Flag,
+    skip_if_not_express: darling::util::Flag,
+    no_bucket: darling::util::Flag,
+    object_lock: darling::util::Flag,
+    no_cleanup: darling::util::Flag,
+}
+
+impl MacroArgs {
+    pub(crate) fn validate(&self, func: &ItemFn) -> Result<(), proc_macro::TokenStream> {
+        if self.no_bucket.is_present() && self.bucket_name.is_some() {
+            let error_msg = "The `no_bucket` argument cannot be used with `bucket_name`";
+            return Err(proc_macro::TokenStream::from(
+                Error::custom(error_msg)
+                    .with_span(&func.sig.span())
+                    .write_errors(),
+            ));
+        }
+
+        if self.no_bucket.is_present() && func.sig.inputs.len() != 1 {
+            let error_msg = "When using `no_bucket`, the test function must have exactly one argument: (ctx: TestContext)";
+            return Err(proc_macro::TokenStream::from(
+                Error::custom(error_msg)
+                    .with_span(&func.sig.inputs.span())
+                    .write_errors(),
+            ));
+        }
+
+        // Validate that the function has exactly two arguments: ctx and bucket_name
+        if func.sig.inputs.len() != 2 && !self.no_bucket.is_present() {
+            let error_msg = "Minio test function must have exactly two arguments: (ctx: TestContext, bucket_name: String)";
+            return Err(proc_macro::TokenStream::from(
+                Error::custom(error_msg)
+                    .with_span(&func.sig.inputs.span())
+                    .write_errors(),
+            ));
+        }
+
+        // Check the argument types
+        let mut iter = func.sig.inputs.iter();
+
+        // Check first argument (ctx: &mut TestContext)
+        if let Some(FnArg::Typed(pat_type)) = iter.next() {
+            let type_str = pat_type.ty.to_token_stream().to_string();
+            if !type_str.contains("TestContext") {
+                let error_msg = "First argument must be of type TestContext";
+                return Err(proc_macro::TokenStream::from(
+                    Error::custom(error_msg)
+                        .with_span(&pat_type.span())
+                        .write_errors(),
+                ));
+            }
+        }
+
+        // Check second argument (bucket_name: String)
+        if !self.no_bucket.is_present() {
+            if let Some(FnArg::Typed(pat_type)) = iter.next() {
+                let type_str = pat_type.ty.to_token_stream().to_string();
+                if !type_str.contains("String") {
+                    let error_msg = "Second argument must be of type String";
+                    return Err(proc_macro::TokenStream::from(
+                        Error::custom(error_msg)
+                            .with_span(&pat_type.span())
+                            .write_errors(),
+                    ));
+                }
+            }
+        }
+
+        Ok(())
+    }
+}
+
+/// Expands the test macro into the final TokenStream
+pub(crate) fn expand_test_macro(
+    args: MacroArgs,
+    mut func: ItemFn,
+) -> Result<TokenStream, proc_macro::TokenStream> {
+    let input_span = func.sig.paren_token.span.span();
+    func.sig.output = ReturnType::Default;
+    let old_inps = func.sig.inputs.clone();
+    func.sig.inputs = Punctuated::default();
+    let sig = func.sig.clone().into_token_stream();
+
+    // Generate the tokio test attribute based on the provided arguments
+    let header = generate_tokio_test_header(&args, sig);
+
+    let test_function_block = func.block.clone().into_token_stream();
+
+    let inner_inputs = quote_spanned!(input_span=> #old_inps);
+    let inner_fn_name = create_inner_func_name(&func);
+    let inner_header = quote_spanned!(func.sig.span()=> async fn #inner_fn_name(#inner_inputs));
+
+    // Generate the skip logic for express mode if required
+    let maybe_skip_if_express = generate_express_skip_logic(&args, func.sig.span());
+
+    // Setup common prelude
+    let prelude = quote!(
+            use ::futures::FutureExt;
+            use ::std::panic::AssertUnwindSafe;
+            use ::minio::s3::types::S3Api;
+            use ::minio::s3::response::a_response_traits::HasBucket;
+
+            let ctx = ::minio_common::test_context::TestContext::new_from_env();
+    );
+
+    // Generate the outer function body based on whether a bucket is needed
+    let outer_body = if args.no_bucket.is_present() {
+        generate_no_bucket_body(
+            prelude,
+            maybe_skip_if_express,
+            inner_fn_name,
+            func.block.span(),
+        )
+    } else {
+        generate_with_bucket_body(
+            prelude,
+            maybe_skip_if_express,
+            inner_fn_name,
+            &args,
+            func.block.span(),
+        )
+    };
+
+    // Generate the inner function implementation
+    let inner_impl = quote_spanned!(func.span()=>
+        #inner_header
+        #test_function_block
+    );
+
+    // Combine all parts into the final output
+    let mut out = TokenStream::new();
+    out.extend(header);
+    out.extend(outer_body);
+    out.extend(inner_impl);
+
+    Ok(out)
+}
+
+fn generate_tokio_test_header(args: &MacroArgs, sig: TokenStream) -> TokenStream {
+    let flavor = args
+        .flavor
+        .as_ref()
+        .map(ToString::to_string)
+        .or(std::env::var("MINIO_TEST_TOKIO_RUNTIME_FLAVOR").ok());
+    match (flavor, args.worker_threads) {
+        (Some(flavor), None) => {
+            quote!(#[::tokio::test(flavor = #flavor)]
+            #sig
+                )
+        }
+        (None, Some(worker_threads)) => {
+            quote!(#[::tokio::test(worker_threads = #worker_threads)]
+            #sig
+                )
+        }
+        (None, None) => {
+            quote!(#[::tokio::test]
+            #sig
+                )
+        }
+        (Some(flavor), Some(worker_threads)) => {
+            quote!(#[::tokio::test(flavor = #flavor, worker_threads = #worker_threads)]
+            #sig
+                )
+        }
+    }
+}
+
+fn generate_express_skip_logic(args: &MacroArgs, span: proc_macro2::Span) -> TokenStream {
+    if args.skip_if_express.is_present() {
+        quote_spanned!(span=>
+        if ctx.client.is_minio_express().await {
+            println!("Skipping test because it is running in MinIO Express mode");
+            return;
+        })
+    } else if args.skip_if_not_express.is_present() {
+        quote_spanned!(span=>
+        if !ctx.client.is_minio_express().await {
+            println!("Skipping test because it is NOT running in MinIO Express mode");
+            return;
+        })
+    } else {
+        TokenStream::new()
+    }
+}
+
+fn generate_no_bucket_body(
+    prelude: TokenStream,
+    maybe_skip_if_express: TokenStream,
+    inner_fn_name: TokenStream,
+    span: proc_macro2::Span,
+) -> TokenStream {
+    quote_spanned!(span=> {
+        #prelude
+        #maybe_skip_if_express
+        #inner_fn_name(ctx).await;
+    })
+}
+
+fn generate_with_bucket_body(
+    prelude: TokenStream,
+    maybe_skip_if_express: TokenStream,
+    inner_fn_name: TokenStream,
+    args: &MacroArgs,
+    span: proc_macro2::Span,
+) -> TokenStream {
+    let bucket_name = args
+        .bucket_name
+        .as_ref()
+        .map(|b| b.to_token_stream())
+        .unwrap_or_else(|| {
+            let random_name = format!("test-bucket-{}", Uuid::new_v4());
+            proc_macro2::Literal::string(&random_name).into_token_stream()
+        });
+    let maybe_lock = if args.object_lock.is_present() {
+        quote! {
+            .object_lock(true)
+        }
+    } else {
+        TokenStream::new()
+    };
+    let maybe_cleanup = if args.no_cleanup.is_present() {
+        quote! {}
+    } else {
+        quote! {
+            ::minio_common::cleanup_guard::cleanup(client_clone, resp.bucket()).await;
+        }
+    };
+    quote_spanned!(span=> {
+        #prelude
+        #maybe_skip_if_express
+
+        let client_clone = ctx.client.clone();
+        let bucket_name = #bucket_name;
+        let resp = client_clone.create_bucket(bucket_name)#maybe_lock.send().await.expect("Failed to create bucket");
+        assert_eq!(resp.bucket(), bucket_name);
+        let res = AssertUnwindSafe(#inner_fn_name(ctx, resp.bucket().to_string())).catch_unwind().await;
+        #maybe_cleanup
+        if let Err(e) = res {
+            ::std::panic::resume_unwind(e);
+        }
+    })
+}
+
+fn create_inner_func_name(func: &ItemFn) -> TokenStream {
+    let inner_name = format!("{}_test_impl", func.sig.ident);
+    let ident = proc_macro2::Ident::new(&inner_name, func.sig.span());
+    quote! { #ident }
+}

src/s3/client.rs

@@ -424,7 +424,7 @@ impl Client {
         query_params: &Multimap,
         bucket_name: Option<&str>,
         object_name: Option<&str>,
-        body: Option<&SegmentedBytes>,
+        body: Option<Arc<SegmentedBytes>>,
         retry: bool,
     ) -> Result<reqwest::Response, Error> {
         let url = self.shared.base_url.build_url(
@@ -437,7 +437,6 @@ impl Client {
 
         {
             headers.add("Host", url.host_header_value());
-
             let sha256: String = match *method {
                 Method::PUT | Method::POST => {
                     if !headers.contains_key("Content-Type") {
@@ -445,10 +444,12 @@ impl Client {
                     }
                     let len: usize = body.as_ref().map_or(0, |b| b.len());
                     headers.add("Content-Length", len.to_string());
-
                     match body {
                         None => EMPTY_SHA256.into(),
-                        Some(v) => sha256_hash_sb(v),
+                        Some(ref v) => {
+                            let clone = v.clone();
+                            async_std::task::spawn_blocking(move || sha256_hash_sb(clone)).await
+                        }
                     }
                 }
                 _ => EMPTY_SHA256.into(),
@@ -457,7 +458,6 @@ impl Client {
 
             let date = utc_now();
             headers.add("x-amz-date", to_amz_date(date));
-
             if let Some(p) = &self.shared.provider {
                 let creds = p.fetch();
                 if creds.session_token.is_some() {
@@ -498,14 +498,14 @@ impl Client {
                 method,
                 url.path,
                 header_strings.join("; "),
-                body.unwrap()
+                body.as_ref().unwrap()
             );
         }
 
         if (*method == Method::PUT) || (*method == Method::POST) {
             //TODO: why-oh-why first collect into a vector and then iterate to a stream?
             let bytes_vec: Vec<Bytes> = match body {
-                Some(v) => v.into_iter().collect(),
+                Some(v) => v.iter().collect(),
                 None => Vec::new(),
             };
             let stream = futures_util::stream::iter(
@@ -557,7 +557,7 @@ impl Client {
         query_params: &Multimap,
         bucket_name: &Option<&str>,
         object_name: &Option<&str>,
-        data: Option<&SegmentedBytes>,
+        data: Option<Arc<SegmentedBytes>>,
     ) -> Result<reqwest::Response, Error> {
         let resp: Result<reqwest::Response, Error> = self
             .execute_internal(
@@ -567,7 +567,7 @@ impl Client {
                 query_params,
                 bucket_name.as_deref(),
                 object_name.as_deref(),
-                data,
+                data.as_ref().map(Arc::clone),
                 true,
             )
             .await;

src/s3/types.rs

@@ -27,7 +27,7 @@ use http::Method;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::fmt;
-
+use std::sync::Arc;
 use xmltree::Element;
 
 #[derive(Clone, Default, Debug)]
@@ -41,7 +41,7 @@ pub struct S3Request {
     pub(crate) object: Option<String>,
     pub(crate) query_params: Multimap,
     headers: Multimap,
-    body: Option<SegmentedBytes>,
+    body: Option<Arc<SegmentedBytes>>,
 
     /// region computed by [`S3Request::execute`]
     pub(crate) inner_region: String,
@@ -83,7 +83,7 @@ impl S3Request {
     }
 
     pub fn body(mut self, body: Option<SegmentedBytes>) -> Self {
-        self.body = body;
+        self.body = body.map(Arc::new);
         self
     }
 
@@ -105,7 +105,7 @@ impl S3Request {
                 &self.query_params,
                 &self.bucket.as_deref(),
                 &self.object.as_deref(),
-                self.body.as_ref(),
+                self.body.as_ref().map(Arc::clone),
             )
             .await
     }

src/s3/utils.rs

@@ -33,6 +33,7 @@ use ring::digest::{Context, SHA256};
 #[cfg(not(feature = "ring"))]
 use sha2::{Digest, Sha256};
 use std::collections::HashMap;
+use std::sync::Arc;
 pub use urlencoding::decode as urldecode;
 pub use urlencoding::encode as urlencode;
 use xmltree::Element;
@@ -71,7 +72,7 @@ pub fn sha256_hash(data: &[u8]) -> String {
     }
 }
 
-pub fn sha256_hash_sb(sb: &SegmentedBytes) -> String {
+pub fn sha256_hash_sb(sb: Arc<SegmentedBytes>) -> String {
     #[cfg(feature = "ring")]
     {
         let mut context = Context::new(&SHA256);
@@ -96,10 +97,14 @@ pub fn sha256_hash_sb(sb: &SegmentedBytes) -> String {
 mod tests {
     use crate::s3::utils::SegmentedBytes;
     use crate::s3::utils::sha256_hash_sb;
+    use std::sync::Arc;
 
     #[test]
     fn test_empty_sha256_segmented_bytes() {
-        assert_eq!(super::EMPTY_SHA256, sha256_hash_sb(&SegmentedBytes::new()));
+        assert_eq!(
+            super::EMPTY_SHA256,
+            sha256_hash_sb(Arc::new(SegmentedBytes::new()))
+        );
     }
 }
 

tests/run-tests-windows.ps1

@@ -3,7 +3,7 @@ $Env:SERVER_ENDPOINT = "http://localhost:9000/"
 $Env:ACCESS_KEY = "minioadmin"
 $Env:SECRET_KEY = "minioadmin"
 $Env:ENABLE_HTTPS = "false"
-$Env:SSL_CERT_FILE = "./tests/public.crt"
+$Env:MINIO_SSL_CERT_FILE = "./tests/public.crt"
 $Env:IGNORE_CERT_CHECK = "false"
 $Env:SERVER_REGION = ""
 

tests/test_append_object.rs

@@ -73,15 +73,8 @@ async fn create_object_helper(
 }
 
 /// Append to the end of an existing object (happy flow)
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_not_express)]
-async fn append_object_0() {
+async fn append_object_0(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if !ctx.client.is_minio_express().await {
-        println!("Skipping test because it is NOT running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
 
     let content1 = "aaaa";
@@ -129,15 +122,8 @@ async fn append_object_0() {
 }
 
 /// Append to the beginning of an existing object (happy flow)
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_not_express)]
-async fn append_object_1() {
+async fn append_object_1(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if !ctx.client.is_minio_express().await {
-        println!("Skipping test because it is NOT running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
 
     let content1 = "aaaa";
@@ -185,15 +171,8 @@ async fn append_object_1() {
 }
 
 /// Append to the middle of an existing object (error InvalidWriteOffset)
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_not_express)]
-async fn append_object_2() {
+async fn append_object_2(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if !ctx.client.is_minio_express().await {
-        println!("Skipping test because it is NOT running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
 
     let content1 = "aaaa";
@@ -221,15 +200,8 @@ async fn append_object_2() {
 }
 
 /// Append beyond the size of an existing object (error InvalidWriteOffset)
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_not_express)]
-async fn append_object_3() {
+async fn append_object_3(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if !ctx.client.is_minio_express().await {
-        println!("Skipping test because it is NOT running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
 
     let content1 = "aaaa";
@@ -257,15 +229,8 @@ async fn append_object_3() {
 }
 
 /// Append to the beginning/end of a non-existing object (happy flow)
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_not_express)]
-async fn append_object_4() {
+async fn append_object_4(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if !ctx.client.is_minio_express().await {
-        println!("Skipping test because it is NOT running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
 
     let content1 = "aaaa";
@@ -309,15 +274,8 @@ async fn append_object_4() {
 }
 
 /// Append beyond the size of a non-existing object (error NoSuchKey)
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_not_express)]
-async fn append_object_5() {
+async fn append_object_5(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if !ctx.client.is_minio_express().await {
-        println!("Skipping test because it is NOT running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
 
     let content1 = "aaaa";
@@ -339,15 +297,8 @@ async fn append_object_5() {
     }
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_not_express)]
-async fn append_object_content_0() {
+async fn append_object_content_0(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if !ctx.client.is_minio_express().await {
-        println!("Skipping test because it is NOT running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
 
     let content1 = "aaaaa";
@@ -389,15 +340,8 @@ async fn append_object_content_0() {
     assert_eq!(content, format!("{}{}", content1, content2));
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_not_express)]
-async fn append_object_content_1() {
+async fn append_object_content_1(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if !ctx.client.is_minio_express().await {
-        println!("Skipping test because it is NOT running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
 
     let n_parts = 3;
@@ -441,15 +385,8 @@ async fn append_object_content_1() {
     }
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_not_express)]
-async fn append_object_content_2() {
+async fn append_object_content_2(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if !ctx.client.is_minio_express().await {
-        println!("Skipping test because it is NOT running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
 
     let sizes = [16_u64, 5 * 1024 * 1024, 16 + 5 * 1024 * 1024];
@@ -492,15 +429,8 @@ async fn append_object_content_2() {
 }
 
 /// Test sending AppendObject across async tasks.
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_not_express)]
-async fn append_object_content_3() {
+async fn append_object_content_3(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if !ctx.client.is_minio_express().await {
-        println!("Skipping test because it is NOT running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
     let sizes = vec![16_u64, 5 * 1024 * 1024, 16 + 5 * 1024 * 1024];
 

tests/test_bucket_create_delete.rs

@@ -21,9 +21,8 @@ use minio::s3::types::S3Api;
 use minio_common::test_context::TestContext;
 use minio_common::utils::rand_bucket_name;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(no_bucket)]
-async fn bucket_create() {
+async fn bucket_create(ctx: TestContext) {
-    let ctx = TestContext::new_from_env();
     let bucket_name = rand_bucket_name();
 
     // try to create a bucket that does not exist
@@ -49,9 +48,8 @@ async fn bucket_create() {
     }
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(no_bucket)]
-async fn bucket_delete() {
+async fn bucket_delete(ctx: TestContext) {
-    let ctx = TestContext::new_from_env();
     let bucket_name = rand_bucket_name();
 
     // try to remove a bucket that does not exist

tests/test_bucket_encryption.rs

@@ -21,11 +21,8 @@ use minio::s3::response::{
 use minio::s3::types::{S3Api, SseConfig};
 use minio_common::test_context::TestContext;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn bucket_encryption() {
+async fn bucket_encryption(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-
     let config = SseConfig::default();
 
     if false {

tests/test_bucket_exists.rs

@@ -19,11 +19,8 @@ use minio::s3::response::{BucketExistsResponse, DeleteBucketResponse};
 use minio::s3::types::S3Api;
 use minio_common::test_context::TestContext;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(no_cleanup)]
-async fn bucket_exists() {
+async fn bucket_exists(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-
     let resp: BucketExistsResponse = ctx.client.bucket_exists(&bucket_name).send().await.unwrap();
     assert!(resp.exists());
     assert_eq!(resp.bucket(), bucket_name);

tests/test_bucket_lifecycle.rs

@@ -24,11 +24,8 @@ use minio::s3::types::S3Api;
 use minio_common::example::create_bucket_lifecycle_config_examples;
 use minio_common::test_context::TestContext;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn bucket_lifecycle() {
+async fn bucket_lifecycle(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-
     let config: LifecycleConfig = create_bucket_lifecycle_config_examples();
 
     let resp: PutBucketLifecycleResponse = ctx

tests/test_bucket_notification.rs

@@ -24,16 +24,8 @@ use minio_common::test_context::TestContext;
 
 const SQS_ARN: &str = "arn:minio:sqs::miniojavatest:webhook";
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_express)]
-async fn test_bucket_notification() {
+async fn test_bucket_notification(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if ctx.client.is_minio_express().await {
-        println!("Skipping test because it is running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-
     let config: NotificationConfig = create_bucket_notification_config_example();
 
     let resp: PutBucketNotificationResponse = ctx

tests/test_bucket_policy.rs

@@ -22,11 +22,8 @@ use minio::s3::types::S3Api;
 use minio_common::example::create_bucket_policy_config_example;
 use minio_common::test_context::TestContext;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn bucket_policy() {
+async fn bucket_policy(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-
     let config: String = create_bucket_policy_config_example(&bucket_name);
 
     let resp: PutBucketPolicyResponse = ctx

tests/test_bucket_replication.rs

@@ -27,17 +27,10 @@ use minio_common::example::{
 };
 use minio_common::test_context::TestContext;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_express)]
-async fn bucket_replication_s3() {
+async fn bucket_replication_s3(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if ctx.client.is_minio_express().await {
-        println!("Skipping test because it is running in MinIO Express mode");
-        return;
-    }
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-
     let ctx2 = TestContext::new_from_env();
-    let (bucket_name2, _cleanup2) = ctx2.create_bucket_helper().await;
+    let (bucket_name2, cleanup2) = ctx2.create_bucket_helper().await;
 
     {
         let resp: PutBucketVersioningResponse = ctx
@@ -132,19 +125,12 @@ async fn bucket_replication_s3() {
         .send()
         .await
         .unwrap();
+    cleanup2.cleanup().await;
     //println!("response of getting replication: resp={:?}", resp);
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_not_express)]
-async fn bucket_replication_s3express() {
+async fn bucket_replication_s3express(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-
-    if !ctx.client.is_minio_express().await {
-        println!("Skipping test because it is NOT running in MinIO Express mode");
-        return;
-    }
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-
     let config: ReplicationConfig = create_bucket_replication_config_example(&bucket_name);
 
     let resp: Result<PutBucketReplicationResponse, Error> = ctx

tests/test_bucket_tagging.rs

@@ -23,15 +23,8 @@ use minio::s3::types::S3Api;
 use minio_common::example::create_tags_example;
 use minio_common::test_context::TestContext;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_express)]
-async fn bucket_tags_s3() {
+async fn bucket_tags_s3(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if ctx.client.is_minio_express().await {
-        println!("Skipping test because it is running in MinIO Express mode");
-        return;
-    }
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-
     let tags = create_tags_example();
 
     let resp: PutBucketTaggingResponse = ctx
@@ -74,15 +67,8 @@ async fn bucket_tags_s3() {
     assert_eq!(resp.region(), DEFAULT_REGION);
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_not_express)]
-async fn bucket_tags_s3express() {
+async fn bucket_tags_s3express(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if !ctx.client.is_minio_express().await {
-        println!("Skipping test because it is NOT running in MinIO Express mode");
-        return;
-    }
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-
     let tags = create_tags_example();
 
     let resp: Result<PutBucketTaggingResponse, Error> = ctx

tests/test_bucket_versioning.rs

@@ -21,15 +21,8 @@ use minio::s3::response::{GetBucketVersioningResponse, PutBucketVersioningRespon
 use minio::s3::types::S3Api;
 use minio_common::test_context::TestContext;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_express)]
-async fn bucket_versioning_s3() {
+async fn bucket_versioning_s3(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if ctx.client.is_minio_express().await {
-        println!("Skipping test because it is running in MinIO Express mode");
-        return;
-    }
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-
     let resp: PutBucketVersioningResponse = ctx
         .client
         .put_bucket_versioning(&bucket_name)
@@ -71,15 +64,8 @@ async fn bucket_versioning_s3() {
     assert_eq!(resp.region(), DEFAULT_REGION);
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_not_express)]
-async fn bucket_versioning_s3express() {
+async fn bucket_versioning_s3express(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if !ctx.client.is_minio_express().await {
-        println!("Skipping test because it is NOT running in MinIO Express mode");
-        return;
-    }
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-
     let resp: Result<PutBucketVersioningResponse, Error> = ctx
         .client
         .put_bucket_versioning(&bucket_name)

tests/test_get_object.rs

@@ -20,10 +20,8 @@ use minio::s3::types::S3Api;
 use minio_common::test_context::TestContext;
 use minio_common::utils::rand_object_name;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn get_object() {
+async fn get_object(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
 
     let data: Bytes = Bytes::from("hello, world".to_string().into_bytes());

tests/test_get_presigned_object_url.rs

@@ -19,11 +19,8 @@ use minio::s3::response::GetPresignedObjectUrlResponse;
 use minio_common::test_context::TestContext;
 use minio_common::utils::rand_object_name;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn get_presigned_object_url() {
+async fn get_presigned_object_url(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-
     let object_name = rand_object_name();
     let resp: GetPresignedObjectUrlResponse = ctx
         .client

tests/test_get_presigned_post_form_data.rs

@@ -19,10 +19,8 @@ use minio_common::test_context::TestContext;
 use minio_common::utils::rand_object_name;
 use std::collections::HashMap;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn get_presigned_post_form_data() {
+async fn get_presigned_post_form_data(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
 
     let policy: PostPolicy = create_post_policy_example(&bucket_name, &object_name);

tests/test_list_buckets.rs

@@ -18,10 +18,9 @@ use minio::s3::types::S3Api;
 use minio_common::cleanup_guard::CleanupGuard;
 use minio_common::test_context::TestContext;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(no_bucket)]
-async fn list_buckets() {
+async fn list_buckets(ctx: TestContext) {
     const N_BUCKETS: usize = 3;
-    let ctx = TestContext::new_from_env();
 
     let mut names: Vec<String> = Vec::new();
     let mut guards: Vec<CleanupGuard> = Vec::new();
@@ -48,4 +47,7 @@ async fn list_buckets() {
     }
     assert_eq!(guards.len(), N_BUCKETS);
     assert_eq!(count, N_BUCKETS);
+    for guard in guards {
+        guard.cleanup().await;
+    }
 }

tests/test_list_objects.rs

@@ -27,6 +27,8 @@ async fn list_objects(
     express: bool,
     n_prefixes: usize,
     n_objects: usize,
+    ctx: TestContext,
+    bucket_name: String,
 ) {
     if express {
         if use_api_v1 {
@@ -36,7 +38,6 @@ async fn list_objects(
             panic!("S3-Express does not support versioning");
         }
     }
-    let ctx = TestContext::new_from_env();
 
     let is_express = ctx.client.is_minio_express().await;
     if is_express && !express {
@@ -47,8 +48,6 @@ async fn list_objects(
         return;
     }
 
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-
     let mut names_set_before: HashSet<String> = HashSet::new();
     let mut names_vec_after: Vec<String> = Vec::with_capacity(n_prefixes * n_objects);
 
@@ -98,29 +97,29 @@ async fn list_objects(
     assert_eq!(names_set_after, names_set_before);
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn list_objects_v1_no_versions() {
+async fn list_objects_v1_no_versions(ctx: TestContext, bucket_name: String) {
-    list_objects(true, false, false, 5, 5).await;
+    list_objects(true, false, false, 5, 5, ctx, bucket_name).await;
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn list_objects_v1_with_versions() {
+async fn list_objects_v1_with_versions(ctx: TestContext, bucket_name: String) {
-    list_objects(true, true, false, 5, 5).await;
+    list_objects(true, true, false, 5, 5, ctx, bucket_name).await;
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn list_objects_v2_no_versions() {
+async fn list_objects_v2_no_versions(ctx: TestContext, bucket_name: String) {
-    list_objects(false, false, false, 5, 5).await;
+    list_objects(false, false, false, 5, 5, ctx, bucket_name).await;
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn list_objects_v2_with_versions() {
+async fn list_objects_v2_with_versions(ctx: TestContext, bucket_name: String) {
-    list_objects(false, true, false, 5, 5).await;
+    list_objects(false, true, false, 5, 5, ctx, bucket_name).await;
 }
 
 /// Test for S3-Express: List objects with S3-Express are only supported with V2 API, without
 /// versions, and yield unsorted results.
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn list_objects_express() {
+async fn list_objects_express(ctx: TestContext, bucket_name: String) {
-    list_objects(false, false, true, 5, 5).await;
+    list_objects(false, false, true, 5, 5, ctx, bucket_name).await;
 }

tests/test_listen_bucket_notification.rs

@@ -24,10 +24,8 @@ use minio_common::test_context::TestContext;
 use minio_common::utils::rand_object_name;
 use tokio::sync::mpsc;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(flavor = "multi_thread", worker_threads = 10)]
-async fn listen_bucket_notification() {
+async fn listen_bucket_notification(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
 
     type MessageType = u32;
@@ -57,7 +55,7 @@ async fn listen_bucket_notification() {
 
             if let Some(record) = record {
                 let key: &str = &record.s3.object.key;
-                if key == &object_name2 {
+                if key == object_name2 {
                     // Do something with the record, check if you received an event triggered
                     // by the put_object that will happen in a few ms.
                     assert_eq!(record.event_name, "s3:ObjectCreated:Put");

tests/test_object_compose.rs

@@ -21,10 +21,8 @@ use minio_common::rand_src::RandSrc;
 use minio_common::test_context::TestContext;
 use minio_common::utils::rand_object_name;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn compose_object() {
+async fn compose_object(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name_src: String = rand_object_name();
     let object_name_dst: String = rand_object_name();
 

tests/test_object_copy.rs

@@ -21,15 +21,8 @@ use minio_common::rand_src::RandSrc;
 use minio_common::test_context::TestContext;
 use minio_common::utils::rand_object_name;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_express)]
-async fn copy_object() {
+async fn copy_object(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if ctx.client.is_minio_express().await {
-        println!("Skipping test because it is running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name_src: String = rand_object_name();
     let object_name_dst: String = rand_object_name();
 

tests/test_object_legal_hold.rs

@@ -18,30 +18,14 @@ use bytes::Bytes;
 use minio::s3::client::DEFAULT_REGION;
 use minio::s3::response::a_response_traits::{HasBucket, HasObject, HasRegion, HasVersion};
 use minio::s3::response::{
-    CreateBucketResponse, GetObjectLegalHoldResponse, PutObjectContentResponse,
+    GetObjectLegalHoldResponse, PutObjectContentResponse, PutObjectLegalHoldResponse,
-    PutObjectLegalHoldResponse,
 };
 use minio::s3::types::S3Api;
-use minio_common::cleanup_guard::CleanupGuard;
 use minio_common::test_context::TestContext;
-use minio_common::utils::{rand_bucket_name, rand_object_name};
+use minio_common::utils::rand_object_name;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_express, object_lock)]
-async fn object_legal_hold_s3() {
+async fn object_legal_hold_s3(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if ctx.client.is_minio_express().await {
-        println!("Skipping test because it is running in MinIO Express mode");
-        return;
-    }
-    let bucket_name: String = rand_bucket_name();
-    let _resp: CreateBucketResponse = ctx
-        .client
-        .create_bucket(&bucket_name)
-        .object_lock(true)
-        .send()
-        .await
-        .unwrap();
-    let _cleanup = CleanupGuard::new(ctx.client.clone(), &bucket_name);
     let object_name = rand_object_name();
 
     let data = Bytes::from("hello, world".to_string().into_bytes());

tests/test_object_lock_config.rs

@@ -19,27 +19,10 @@ use minio::s3::response::{
     DeleteObjectLockConfigResponse, GetObjectLockConfigResponse, PutObjectLockConfigResponse,
 };
 use minio::s3::types::{ObjectLockConfig, RetentionMode, S3Api};
-use minio_common::cleanup_guard::CleanupGuard;
 use minio_common::test_context::TestContext;
-use minio_common::utils::rand_bucket_name;
-
-#[tokio::test(flavor = "multi_thread")]
-async fn object_lock_config() {
-    let ctx = TestContext::new_from_env();
-    if ctx.client.is_minio_express().await {
-        println!("Skipping test because it is running in MinIO Express mode");
-        return;
-    }
-
-    let bucket_name: String = rand_bucket_name();
-    ctx.client
-        .create_bucket(&bucket_name)
-        .object_lock(true)
-        .send()
-        .await
-        .unwrap();
-    let _cleanup = CleanupGuard::new(ctx.client.clone(), &bucket_name);
 
+#[minio_macros::test(skip_if_express, object_lock)]
+async fn object_lock_config(ctx: TestContext, bucket_name: String) {
     const DURATION_DAYS: i32 = 7;
     let config =
         ObjectLockConfig::new(RetentionMode::GOVERNANCE, Some(DURATION_DAYS), None).unwrap();

tests/test_object_put.rs

@@ -26,10 +26,8 @@ use minio_common::test_context::TestContext;
 use minio_common::utils::rand_object_name;
 use tokio::sync::mpsc;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn put_object() {
+async fn put_object(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name: String = rand_object_name();
 
     let size = 16_u64;
@@ -64,7 +62,7 @@ async fn put_object() {
         .send()
         .await
         .unwrap();
-    assert!(!resp.version_id().is_some());
+    assert!(resp.version_id().is_none());
 
     // Validate delete succeeded.
     let resp: Result<StatObjectResponse, Error> = ctx
@@ -81,11 +79,9 @@ async fn put_object() {
     }
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn put_object_multipart() {
+async fn put_object_multipart(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
+    let object_name: String = rand_object_name();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-    let object_name = rand_object_name();
 
     let size: u64 = 16 + MIN_PART_SIZE;
 
@@ -122,10 +118,8 @@ async fn put_object_multipart() {
     assert_eq!(resp.version_id(), None);
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn put_object_content_1() {
+async fn put_object_content_1(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
     let sizes = [16_u64, MIN_PART_SIZE, 16 + MIN_PART_SIZE];
 
@@ -168,10 +162,8 @@ async fn put_object_content_1() {
     }
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 10)]
+#[minio_macros::test]
-async fn put_object_content_2() {
+async fn put_object_content_2(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
     let sizes = [16_u64, MIN_PART_SIZE, 16 + MIN_PART_SIZE];
 
@@ -212,10 +204,8 @@ async fn put_object_content_2() {
 }
 
 /// Test sending PutObject across async tasks.
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn put_object_content_3() {
+async fn put_object_content_3(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
     let sizes = vec![16_u64, MIN_PART_SIZE, 16 + MIN_PART_SIZE];
 

tests/test_object_remove.rs

@@ -21,11 +21,8 @@ use minio::s3::types::ToStream;
 use minio_common::test_context::TestContext;
 use minio_common::utils::rand_object_name;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn remove_objects() {
+async fn remove_objects(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
-
     let mut names: Vec<String> = Vec::new();
     for _ in 1..=3 {
         let object_name = rand_object_name();

tests/test_object_retention.rs

@@ -17,34 +17,16 @@ use minio::s3::builders::ObjectContent;
 use minio::s3::client::DEFAULT_REGION;
 use minio::s3::response::a_response_traits::{HasBucket, HasObject, HasRegion, HasVersion};
 use minio::s3::response::{
-    CreateBucketResponse, GetObjectRetentionResponse, PutObjectContentResponse,
+    GetObjectRetentionResponse, PutObjectContentResponse, PutObjectRetentionResponse,
-    PutObjectRetentionResponse,
 };
 use minio::s3::types::{RetentionMode, S3Api};
 use minio::s3::utils::{to_iso8601utc, utc_now};
-use minio_common::cleanup_guard::CleanupGuard;
 use minio_common::rand_src::RandSrc;
 use minio_common::test_context::TestContext;
-use minio_common::utils::{rand_bucket_name, rand_object_name};
+use minio_common::utils::rand_object_name;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_express, object_lock)]
-async fn object_retention() {
+async fn object_retention(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if ctx.client.is_minio_express().await {
-        println!("Skipping test because it is running in MinIO Express mode");
-        return;
-    }
-
-    let bucket_name: String = rand_bucket_name();
-    let resp: CreateBucketResponse = ctx
-        .client
-        .create_bucket(&bucket_name)
-        .object_lock(true)
-        .send()
-        .await
-        .unwrap();
-    let _cleanup = CleanupGuard::new(ctx.client.clone(), &bucket_name);
-    assert_eq!(resp.bucket(), bucket_name);
     let object_name = rand_object_name();
 
     let size = 16_u64;

tests/test_object_tagging.rs

@@ -28,15 +28,8 @@ use minio_common::test_context::TestContext;
 use minio_common::utils::rand_object_name;
 use std::collections::HashMap;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_express)]
-async fn object_tags() {
+async fn object_tags(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if ctx.client.is_minio_express().await {
-        println!("Skipping test because it is running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
 
     let size = 16_u64;

tests/test_select_object_content.rs

@@ -21,15 +21,8 @@ use minio_common::example::{create_select_content_data, create_select_content_re
 use minio_common::test_context::TestContext;
 use minio_common::utils::rand_object_name;
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_express)]
-async fn select_object_content_s3() {
+async fn select_object_content_s3(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if ctx.client.is_minio_express().await {
-        println!("Skipping test because it is running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name: String = rand_object_name();
     let (select_body, select_data) = create_select_content_data();
 
@@ -62,15 +55,8 @@ async fn select_object_content_s3() {
     assert_eq!(got, select_data);
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test(skip_if_not_express)]
-async fn select_object_content_express() {
+async fn select_object_content_express(ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    if !ctx.client.is_minio_express().await {
-        println!("Skipping test because it is NOT running in MinIO Express mode");
-        return;
-    }
-
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name = rand_object_name();
     let (select_body, _) = create_select_content_data();
 

tests/test_upload_download_object.rs

@@ -13,6 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use async_std::io::ReadExt;
 use hex::ToHex;
 use minio::s3::builders::ObjectContent;
 use minio::s3::response::a_response_traits::{HasBucket, HasObject};
@@ -25,38 +26,38 @@ use minio_common::utils::rand_object_name;
 use ring::digest::{Context, SHA256};
 #[cfg(not(feature = "ring"))]
 use sha2::{Digest, Sha256};
-#[cfg(feature = "ring")]
-use std::io::Read;
 use std::path::PathBuf;
-use std::{fs, io};
 
-fn get_hash(filename: &String) -> String {
+async fn get_hash(filename: &String) -> String {
     #[cfg(feature = "ring")]
     {
         let mut context = Context::new(&SHA256);
-        let mut file = fs::File::open(filename).unwrap();
+        let mut file = async_std::fs::File::open(filename).await.unwrap();
         let mut buf = Vec::new();
-        file.read_to_end(&mut buf).unwrap();
+        file.read_to_end(&mut buf).await.unwrap();
         context.update(&buf);
         context.finish().encode_hex()
     }
     #[cfg(not(feature = "ring"))]
     {
         let mut hasher = Sha256::new();
-        let mut file = fs::File::open(filename).unwrap();
+        let mut file = async_std::fs::File::open(filename).await.unwrap();
-        io::copy(&mut file, &mut hasher).unwrap();
+        let mut buf = Vec::new();
+        file.read_to_end(&mut buf).await.unwrap();
+        hasher.update(&buf);
         hasher.finalize().encode_hex()
     }
 }
 
-async fn upload_download_object(size: u64) {
+async fn upload_download_object(size: u64, ctx: TestContext, bucket_name: String) {
-    let ctx = TestContext::new_from_env();
-    let (bucket_name, _cleanup) = ctx.create_bucket_helper().await;
     let object_name: String = rand_object_name();
+    let mut file = async_std::fs::File::create(&object_name).await.unwrap();
 
-    let mut file = fs::File::create(&object_name).unwrap();
+    async_std::io::copy(&mut RandReader::new(size), &mut file)
-    io::copy(&mut RandReader::new(size), &mut file).unwrap();
+        .await
-    file.sync_all().unwrap();
+        .unwrap();
+
+    file.sync_all().await.unwrap();
 
     let obj: ObjectContent = PathBuf::from(&object_name).as_path().into();
 
@@ -86,18 +87,18 @@ async fn upload_download_object(size: u64) {
         .to_file(PathBuf::from(&filename).as_path())
         .await
         .unwrap();
-    assert_eq!(get_hash(&object_name), get_hash(&filename));
+    assert_eq!(get_hash(&object_name).await, get_hash(&filename).await);
 
-    fs::remove_file(&object_name).unwrap();
+    async_std::fs::remove_file(&object_name).await.unwrap();
-    fs::remove_file(&filename).unwrap();
+    async_std::fs::remove_file(&filename).await.unwrap();
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn upload_download_object_1() {
+async fn upload_download_object_1(ctx: TestContext, bucket_name: String) {
-    upload_download_object(16).await;
+    upload_download_object(16, ctx, bucket_name).await;
 }
 
-#[tokio::test(flavor = "multi_thread")]
+#[minio_macros::test]
-async fn upload_download_object_2() {
+async fn upload_download_object_2(ctx: TestContext, bucket_name: String) {
-    upload_download_object(16 + 5 * 1024 * 1024).await;
+    upload_download_object(16 + 5 * 1024 * 1024, ctx, bucket_name).await;
 }