MinIOを使ってS3にアクセスするpytestの単体テストを書く

import boto3


def get_object(bucket, key, s3cli_args=None):
    if not s3cli_args:
        s3cli_args = {}
    s3cli = boto3.client("s3", **s3cli_args)
    resp = s3cli.get_object(
        Bucket=bucket,
        Key=key
    )
    return resp["Body"].read().decode('utf-8')

import contextlib
import hashlib
import os
import socket
import subprocess
import tempfile

import boto3
import pytest

from target import get_object


def find_minio_bin():
    # minioのパス特定
    which_minio = subprocess.run(["which", "minio"], capture_output=True)
    minio_bin = which_minio.stdout.decode('utf-8').strip()
    return minio_bin


@pytest.fixture(scope="session")
def minio():
    # minioのパス特定
    minio_bin = find_minio_bin()

    # 空きポートの特定
    with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
        sock.bind(('', 0))
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        port = sock.getsockname()[1]

    # minioの起動
    with tempfile.TemporaryDirectory() as tmpdir:
        print(minio_bin)
        proc = subprocess.Popen([minio_bin, 'server', tmpdir, "--address", f":{port}"])
        minio_args = {
            "endpoint_url": f"http://127.0.0.1:{port}",
            "aws_access_key_id":
                os.environ["MINIO_ROOT_USER"] if "MINIO_ROOT_USER" in os.environ else "minioadmin",
            "aws_secret_access_key":
                os.environ["MINIO_ROOT_PASSWORD"] if "MINIO_ROOT_PASSWORD" in os.environ else "minioadmin"
        }
        yield minio_args
        proc.kill()


@pytest.mark.skipif(not find_minio_bin(), reason="minio not found")
def test_get_object(tmpdir, minio):
    # 準備
    # - テスト毎にbucketを作る
    # ※ tmpdirはテスト毎にbucket名をユニークにするためだけに使っています
    test_bucket = hashlib.sha224(str(tmpdir).encode()).hexdigest()
    s3cli = boto3.client("s3", **minio)
    s3cli.create_bucket(Bucket=test_bucket, CreateBucketConfiguration={'LocationConstraint': test_bucket})
    # - オブジェクトをupload
    s3cli.put_object(
        Bucket=test_bucket,
        Key="foo",
        Body="body string"
    )

    # 呼び出し
    resp_body = get_object(test_bucket, "foo", minio)

    # 検証
    assert resp_body == "body string"