MinIO Testing Guide: Self-Hosted S3-Compatible Storage Testing with TestContainers
MinIO is an S3-compatible object storage server you can run locally or in production. Testing it properly means running a real MinIO instance — TestContainers makes this easy in both Java and Python by spinning up a MinIO Docker container per test suite and tearing it down automatically.
Key Takeaways
TestContainers gives you a real MinIO, not a mock. Unlike moto or localstack, TestContainers starts an actual MinIO Docker container — your tests hit real MinIO behavior including its policy engine and presigned URL signing.
MinIO's default credentials in tests are just "minioadmin/minioadmin". Change them per-container using environment variables; never use production credentials in test fixtures.
Bucket policies in MinIO follow AWS IAM JSON syntax. Test them by setting a policy and verifying anonymous vs authenticated access behaves correctly.
Java and Python SDKs differ in API style but MinIO compatibility is identical. The same container works for both; you can share a MinIO TestContainer across mixed-language integration tests.
Always configure endpoint override in your SDK client. Both boto3 and the MinIO Java SDK need the container's host+port as the endpoint — don't let them default to real AWS/MinIO endpoints.
MinIO is the go-to choice for teams who need S3-compatible storage on-premises — Kubernetes clusters, air-gapped environments, or just keeping cloud costs low. But "S3-compatible" doesn't mean "identical," and mocking MinIO with moto misses MinIO-specific behavior like its bucket policy evaluation and quota enforcement.
TestContainers is the answer. It starts a real MinIO Docker container in your test process, runs your tests against it, and shuts it down when you're done. No mocking. No approximations.
Java Setup with TestContainers
Add the dependencies to your pom.xml:
<dependencies>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
<version>1.19.7</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.minio</groupId>
<artifactId>minio</artifactId>
<version>8.5.9</version>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
<version>5.10.2</version>
<scope>test</scope>
</dependency>
</dependencies>MinIO doesn't have an official TestContainers module, but GenericContainer works perfectly:
import io.minio.*;
import io.minio.messages.*;
import org.junit.jupiter.api.*;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import java.io.ByteArrayInputStream;
import java.util.List;
@Testcontainers
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class MinioIntegrationTest {
@Container
static GenericContainer<?> minio = new GenericContainer<>("minio/minio:latest")
.withCommand("server /data --console-address :9001")
.withEnv("MINIO_ROOT_USER", "testuser")
.withEnv("MINIO_ROOT_PASSWORD", "testpassword")
.withExposedPorts(9000, 9001);
private MinioClient minioClient;
private static final String BUCKET = "test-bucket";
@BeforeAll
void setupClient() throws Exception {
String endpoint = "http://localhost:" + minio.getMappedPort(9000);
minioClient = MinioClient.builder()
.endpoint(endpoint)
.credentials("testuser", "testpassword")
.build();
// Create test bucket
minioClient.makeBucket(MakeBucketArgs.builder().bucket(BUCKET).build());
}
@Test
void uploadAndDownloadObject() throws Exception {
byte[] content = "Hello from MinIO test".getBytes();
minioClient.putObject(
PutObjectArgs.builder()
.bucket(BUCKET)
.object("hello.txt")
.stream(new ByteArrayInputStream(content), content.length, -1)
.contentType("text/plain")
.build()
);
var response = minioClient.getObject(
GetObjectArgs.builder()
.bucket(BUCKET)
.object("hello.txt")
.build()
);
byte[] downloaded = response.readAllBytes();
Assertions.assertArrayEquals(content, downloaded);
}
@Test
void listObjectsInBucket() throws Exception {
// Upload a few objects
for (int i = 1; i <= 3; i++) {
byte[] data = ("object " + i).getBytes();
minioClient.putObject(
PutObjectArgs.builder()
.bucket(BUCKET)
.object("prefix/file" + i + ".txt")
.stream(new ByteArrayInputStream(data), data.length, -1)
.contentType("text/plain")
.build()
);
}
Iterable<Result<Item>> results = minioClient.listObjects(
ListObjectsArgs.builder()
.bucket(BUCKET)
.prefix("prefix/")
.build()
);
int count = 0;
for (Result<Item> result : results) {
Item item = result.get();
Assertions.assertTrue(item.objectName().startsWith("prefix/"));
count++;
}
Assertions.assertEquals(3, count);
}
}Java: Testing Presigned URLs
Presigned URLs from MinIO are fully functional within the container's network. In tests, you generate the URL and then use it with a plain HTTP client:
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.ZonedDateTime;
@Test
void presignedUrlAllowsDownloadWithoutCredentials() throws Exception {
// Upload a file
byte[] secret = "secret document".getBytes();
minioClient.putObject(
PutObjectArgs.builder()
.bucket(BUCKET)
.object("docs/secret.txt")
.stream(new ByteArrayInputStream(secret), secret.length, -1)
.contentType("text/plain")
.build()
);
// Generate presigned URL (valid 10 minutes)
String presignedUrl = minioClient.getPresignedObjectUrl(
GetPresignedObjectUrlArgs.builder()
.method(Method.GET)
.bucket(BUCKET)
.object("docs/secret.txt")
.expiry(10, TimeUnit.MINUTES)
.build()
);
// Verify the URL contains expected query params
Assertions.assertTrue(presignedUrl.contains("X-Amz-Expires=600"));
Assertions.assertTrue(presignedUrl.contains("X-Amz-Signature"));
// Use the presigned URL with a plain HTTP client (no credentials needed)
HttpClient httpClient = HttpClient.newHttpClient();
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(presignedUrl))
.GET()
.build();
HttpResponse<byte[]> response = httpClient.send(request, HttpResponse.BodyHandlers.ofByteArray());
Assertions.assertEquals(200, response.statusCode());
Assertions.assertArrayEquals(secret, response.body());
}Python Setup with TestContainers
pip install testcontainers boto3 pytest minioimport pytest
import boto3
from testcontainers.core.container import DockerContainer
from testcontainers.core.waiting_utils import wait_for_logs
import time
@pytest.fixture(scope="session")
def minio_container():
container = DockerContainer("minio/minio:latest")
container.with_command("server /data --console-address :9001")
container.with_env("MINIO_ROOT_USER", "testuser")
container.with_env("MINIO_ROOT_PASSWORD", "testpassword")
container.with_exposed_ports(9000)
container.start()
wait_for_logs(container, "API:", timeout=30)
yield container
container.stop()
@pytest.fixture(scope="session")
def s3_client(minio_container):
port = minio_container.get_exposed_port(9000)
endpoint = f"http://localhost:{port}"
client = boto3.client(
"s3",
endpoint_url=endpoint,
aws_access_key_id="testuser",
aws_secret_access_key="testpassword",
region_name="us-east-1"
)
return client
@pytest.fixture(scope="session")
def test_bucket(s3_client):
bucket = "integration-test-bucket"
s3_client.create_bucket(Bucket=bucket)
yield bucket
def test_upload_and_retrieve_file(s3_client, test_bucket):
s3_client.put_object(
Bucket=test_bucket,
Key="data/records.csv",
Body=b"id,name\n1,Alice\n2,Bob",
ContentType="text/csv"
)
response = s3_client.get_object(Bucket=test_bucket, Key="data/records.csv")
content = response["Body"].read()
assert b"Alice" in content
assert b"Bob" in content
def test_object_metadata(s3_client, test_bucket):
s3_client.put_object(
Bucket=test_bucket,
Key="meta/file.txt",
Body=b"content",
Metadata={"uploaded-by": "test-suite", "environment": "ci"}
)
head = s3_client.head_object(Bucket=test_bucket, Key="meta/file.txt")
assert head["Metadata"]["uploaded-by"] == "test-suite"
assert head["Metadata"]["environment"] == "ci"Testing MinIO Bucket Policies
MinIO's bucket policies use IAM JSON syntax. Testing them means verifying anonymous access behaves exactly as the policy dictates:
import json
import requests
def test_public_read_bucket_policy(s3_client, test_bucket, minio_container):
port = minio_container.get_exposed_port(9000)
endpoint = f"http://localhost:{port}"
# Upload a public file
s3_client.put_object(
Bucket=test_bucket,
Key="public/readme.txt",
Body=b"This is public content"
)
# Set a public-read policy for the public/ prefix
policy = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": ["*"]},
"Action": ["s3:GetObject"],
"Resource": [f"arn:aws:s3:::{test_bucket}/public/*"]
}
]
}
s3_client.put_bucket_policy(
Bucket=test_bucket,
Policy=json.dumps(policy)
)
# Verify: anonymous GET on public/ prefix should succeed
public_url = f"{endpoint}/{test_bucket}/public/readme.txt"
response = requests.get(public_url)
assert response.status_code == 200
assert response.text == "This is public content"
# Verify: anonymous GET on private/ prefix should fail
s3_client.put_object(
Bucket=test_bucket,
Key="private/secret.txt",
Body=b"secret"
)
private_url = f"{endpoint}/{test_bucket}/private/secret.txt"
private_response = requests.get(private_url)
assert private_response.status_code == 403
def test_read_back_bucket_policy(s3_client, test_bucket):
policy = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": ["*"]},
"Action": ["s3:GetObject"],
"Resource": [f"arn:aws:s3:::{test_bucket}/*"]
}
]
}
s3_client.put_bucket_policy(
Bucket=test_bucket,
Policy=json.dumps(policy)
)
# Read it back and verify
response = s3_client.get_bucket_policy(Bucket=test_bucket)
returned_policy = json.loads(response["Policy"])
assert returned_policy["Version"] == "2012-10-17"
assert len(returned_policy["Statement"]) == 1
assert returned_policy["Statement"][0]["Effect"] == "Allow"Testing Large File Uploads with MinIO SDK
The MinIO Python SDK (not boto3) handles multipart uploads automatically for files over 5MB:
from minio import Minio
from minio.error import S3Error
import io
@pytest.fixture(scope="session")
def minio_client(minio_container):
port = minio_container.get_exposed_port(9000)
return Minio(
f"localhost:{port}",
access_key="testuser",
secret_key="testpassword",
secure=False
)
def test_large_file_upload(minio_client, test_bucket):
# Generate a 10MB file (above MinIO's 5MB multipart threshold)
large_content = b"x" * (10 * 1024 * 1024)
data_stream = io.BytesIO(large_content)
minio_client.put_object(
test_bucket,
"large/10mb-file.bin",
data_stream,
length=len(large_content),
content_type="application/octet-stream"
)
# Verify the object exists and has the right size
stat = minio_client.stat_object(test_bucket, "large/10mb-file.bin")
assert stat.size == len(large_content)
assert stat.content_type == "application/octet-stream"
def test_object_not_found_raises_s3error(minio_client, test_bucket):
with pytest.raises(S3Error) as exc_info:
minio_client.stat_object(test_bucket, "nonexistent/file.txt")
assert exc_info.value.code == "NoSuchKey"CI Configuration
In GitHub Actions, TestContainers works out of the box because Docker is available:
# .github/workflows/integration-tests.yml
name: MinIO Integration Tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install dependencies
run: pip install -r requirements-test.txt
- name: Run integration tests
run: pytest tests/integration/ -v --timeout=120
env:
TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE: /var/run/docker.sockNo additional MinIO service configuration needed — TestContainers handles the Docker setup automatically. The container image is pulled on first run and cached by Docker's layer cache on subsequent runs, keeping CI fast.
