feat: add a test macro (#170)

Author: twuebi

.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
-    timeout-minutes: 5
+    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 {
-    fn drop(&mut self) {
-        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();
+    pub async fn cleanup(&self) {
+        cleanup(self.client.clone(), &self.bucket_name).await;
+    }
+}
 
-            // 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 {
+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(_) => {
-                            //println!("Bucket {} removed successfully", bucket_name),
+                            eprintln!("Bucket {} removed successfully", bucket_name);
                         }
-                        Err(_e) => {
-                            //println!("Error removing bucket {}: {:?}", bucket_name, e)
+                        Err(e) => {
+                            eprintln!("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
-    }
+    );
 }

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() {
-            true => buf.len(),
-            false => self.size as 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();
@@ -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<'_>,
-    ) -> task::Poll<Option<Self::Item>> {
+        _cx: &mut Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
         if self.size == 0 {
-            return task::Poll::Ready(None);
+            return Poll::Ready(None);
         }
-
-        let bytes_read = match self.size > 64 * 1024 {
-            true => 64 * 1024,
-            false => self.size as usize,
-        };
+        // Limit to 8 KiB per read
+        let bytes_read = self.size.min(8 * 1024) 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;
-
-        task::Poll::Ready(Some(Ok(Bytes::from(buf))))
+        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());
-            log::debug!("SSL_CERT_FILE={}", ssl_cert);
+                std::env::var("MINIO_SSL_CERT_FILE").unwrap_or(DEFAULT_SSL_CERT_FILE.to_string());
+            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" }