chore: Add MinIO support for S3 snapshot tests

Author: vyavdoshenko

.github/actions/regression-tests/action.yml

@@ -37,6 +37,11 @@ inputs:
     default: ""
     required: false
     type: string
+  with-s3:
+    required: false
+    type: string
+    default: 'false'
+    description: "Whether the build has S3 (AWS) support"
 
 runs:
   using: "composite"
@@ -74,6 +79,32 @@ runs:
         echo "===================After freeing up space ============================================"
         df -h
 
+    - name: Run S3 snapshot tests with MinIO
+      if: inputs.with-s3 == 'true'
+      shell: bash
+      run: |
+        cd ${GITHUB_WORKSPACE}/tests
+        pip3 install -r dragonfly/requirements.txt
+
+        export DRAGONFLY_PATH="${GITHUB_WORKSPACE}/${{inputs.build-folder-name}}/${{inputs.dfly-executable}}"
+
+        # Download MinIO binary (atomic: download to .tmp, then rename)
+        ARCH=$(uname -m)
+        case "$ARCH" in
+          x86_64)  ARCH="amd64" ;;
+          aarch64) ARCH="arm64" ;;
+          *) echo "Unsupported MinIO architecture: $ARCH"; exit 1 ;;
+        esac
+        MINIO_DIR="$HOME/.cache/dragonfly-tests"
+        mkdir -p "$MINIO_DIR"
+        if [ ! -f "$MINIO_DIR/minio" ]; then
+          curl -fsSL "https://dl.min.io/server/minio/release/linux-${ARCH}/minio" -o "$MINIO_DIR/minio.tmp"
+          chmod +x "$MINIO_DIR/minio.tmp"
+          mv "$MINIO_DIR/minio.tmp" "$MINIO_DIR/minio"
+        fi
+
+        S3_ENDPOINT=http://localhost:9000 timeout 10m pytest -k "s3" --timeout=300 --color=yes dragonfly/snapshot_test.py --log-cli-level=INFO -v
+
     - name: Run PyTests
       id: main
       shell: bash

.github/workflows/epoll-regression-tests.yml

@@ -64,6 +64,7 @@ jobs:
           build-folder-name: build
           filter: ${{ matrix.build-type == 'Release' && 'not empty' || 'not opt_only' }}
           s3-bucket: ${{ secrets.S3_REGTEST_BUCKET }}
+          with-s3: 'true'
           # Chain ternary oprator of the form (which can be nested)
           # (expression == condition && <true expression> || <false expression>)
           epoll: ${{ matrix.proactor == 'Epoll' && 'epoll' || 'iouring' }}

.github/workflows/ioloop-v2-regtests.yml

@@ -56,6 +56,7 @@ jobs:
           aws-access-key-id: ${{ secrets.AWS_S3_ACCESS_KEY }}
           aws-secret-access-key: ${{ secrets.AWS_S3_ACCESS_SECRET }}
           s3-bucket: ${{ secrets.S3_REGTEST_BUCKET }}
+          with-s3: 'true'
           df-arg: "experimental_io_loop_v2"
 
       - name: Upload logs on failure

.github/workflows/regression-tests.yml

@@ -64,6 +64,7 @@ jobs:
           build-folder-name: build
           filter: ${{ matrix.build-type == 'Release' && 'not debug_only' || 'not opt_only' }}
           s3-bucket: ${{ secrets.S3_REGTEST_BUCKET }}
+          with-s3: 'true'
 
       - name: Upload logs on failure
         if: failure()

.github/workflows/release.yml

@@ -111,6 +111,7 @@ jobs:
         gspace-secret: ${{ secrets.GSPACES_BOT_DF_BUILD }}
         build-folder-name: ${{ env.RELEASE_DIR }}
         filter: 'not debug_only'
+        with-s3: 'true'
     - name: Save artifacts
       run: |
           # place all artifacts at the same location

tests/dragonfly/conftest.py

@@ -35,14 +35,131 @@
 LAST_LOGS = "/tmp/last_test_log_dir.txt"
 
 
+def _download_minio_binary(dest: Path):
+    """Download MinIO binary to dest if not already cached.
+
+    Downloads to a temporary file first, then renames atomically to avoid
+    leaving a corrupt binary on interrupted downloads.
+    """
+    import platform
+    import urllib.request
+
+    system = platform.system().lower()
+    arch = platform.machine()
+    arch_map = {"x86_64": "amd64", "aarch64": "arm64", "arm64": "arm64"}
+    arch = arch_map.get(arch, arch)
+    url = f"https://dl.min.io/server/minio/release/{system}-{arch}/minio"
+    logging.info(f"Downloading MinIO binary from {url}")
+    tmp_dest = dest.with_suffix(".tmp")
+    try:
+        urllib.request.urlretrieve(url, tmp_dest)
+        tmp_dest.chmod(0o755)
+        tmp_dest.rename(dest)
+    except Exception:
+        tmp_dest.unlink(missing_ok=True)
+        raise
+
+
+def _start_minio_server(endpoint):
+    """Start MinIO subprocess and configure env vars for S3 tests."""
+    import boto3
+    from urllib.parse import urlparse
+
+    cache_dir = Path.home() / ".cache" / "dragonfly-tests"
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    minio_bin = cache_dir / "minio"
+
+    if not minio_bin.exists():
+        _download_minio_binary(minio_bin)
+
+    parsed = urlparse(endpoint)
+    address = f":{parsed.port or 9000}"
+
+    data_dir = Path(mkdtemp(prefix="minio_data_"))
+    minio_log = data_dir / "minio.log"
+    log_file = open(minio_log, "w")
+    proc = subprocess.Popen(
+        [str(minio_bin), "server", str(data_dir), "--address", address],
+        env={**os.environ, "MINIO_ROOT_USER": "minioadmin", "MINIO_ROOT_PASSWORD": "minioadmin"},
+        stdout=log_file,
+        stderr=subprocess.STDOUT,
+    )
+
+    bucket = "dragonfly-test"
+    s3 = boto3.client(
+        "s3",
+        endpoint_url=endpoint,
+        aws_access_key_id="minioadmin",
+        aws_secret_access_key="minioadmin",
+        region_name="us-east-1",
+    )
+
+    for attempt in range(30):
+        try:
+            s3.create_bucket(Bucket=bucket)
+            break
+        except Exception:
+            if proc.poll() is not None:
+                log_file.close()
+                logs = minio_log.read_text()
+                shutil.rmtree(data_dir, ignore_errors=True)
+                raise RuntimeError(
+                    f"MinIO process exited with code {proc.returncode}.\nLogs:\n{logs}"
+                )
+            time.sleep(1)
+    else:
+        proc.terminate()
+        log_file.close()
+        logs = minio_log.read_text()
+        shutil.rmtree(data_dir, ignore_errors=True)
+        raise RuntimeError(f"MinIO did not become ready in time.\nLogs:\n{logs}")
+
+    log_file.close()
+    os.environ["DRAGONFLY_S3_BUCKET"] = bucket
+    os.environ["AWS_ACCESS_KEY_ID"] = "minioadmin"
+    os.environ["AWS_SECRET_ACCESS_KEY"] = "minioadmin"
+    os.environ["AWS_ENDPOINT_URL"] = endpoint
+
+    return proc, data_dir
+
+
+_minio_proc = None
+_minio_data_dir = None
+
+
 # runs on pytest start
 def pytest_configure(config):
+    global _minio_proc, _minio_data_dir
+
     # clean everything
     if os.path.exists(FAILED_PATH):
         shutil.rmtree(FAILED_PATH)
     if os.path.exists(BASE_LOG_DIR):
         shutil.rmtree(BASE_LOG_DIR)
 
+    # Start MinIO if S3_ENDPOINT is set (must happen before test collection
+    # so that @pytest.mark.skipif checking DRAGONFLY_S3_BUCKET sees it)
+    endpoint = os.environ.get("S3_ENDPOINT")
+    if endpoint:
+        _minio_proc, _minio_data_dir = _start_minio_server(endpoint)
+
+
+def pytest_unconfigure(config):
+    global _minio_proc, _minio_data_dir
+
+    if _minio_proc is not None:
+        _minio_proc.terminate()
+        try:
+            _minio_proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            _minio_proc.kill()
+            _minio_proc.wait()
+        _minio_proc = None
+
+    if _minio_data_dir is not None:
+        shutil.rmtree(_minio_data_dir, ignore_errors=True)
+        _minio_data_dir = None
+
 
 @pytest.fixture(scope="class")
 def df_log_dir(request):

tests/dragonfly/instance.py

@@ -439,6 +439,22 @@ def create(self, existing_port=None, path=None, version=100, **kwargs) -> DflyIn
         if version >= 1.26:
             args.setdefault("fiber_safety_margin=4096")
 
+        # When a custom S3 endpoint is configured (e.g. MinIO), pass it to Dragonfly
+        s3_endpoint = os.environ.get("S3_ENDPOINT")
+        if s3_endpoint:
+            from urllib.parse import urlparse
+
+            # Normalize scheme-less values (e.g. "localhost:9000") so urlparse
+            # correctly populates hostname/port instead of treating it as a path.
+            to_parse = s3_endpoint if "://" in s3_endpoint else "http://" + s3_endpoint
+            parsed = urlparse(to_parse)
+            endpoint_host = parsed.hostname or ""
+            if parsed.port:
+                endpoint_host = f"{endpoint_host}:{parsed.port}"
+            if endpoint_host:
+                args.setdefault("s3_endpoint", endpoint_host)
+            args.setdefault("s3_use_https", "false" if parsed.scheme == "http" else "true")
+
         for k, v in args.items():
             args[k] = v.format(**self.params.env) if isinstance(v, str) else v