Brian Bjarke JensenandCursor
b28ac6e803
PR Title Check / check-title (pull_request) Successful in 7s
Test Python Package / unit-tests (pull_request) Successful in 13s
Test Python Package / integration-tests (pull_request) Successful in 23s
Test Python Package / coverage-report (pull_request) Successful in 10s
Code Quality Pipeline / code-quality (pull_request) Successful in 49s
Split Redis and MinIO container setup into backend-specific conftest modules so only the containers needed for collected tests are imported and started. Co-authored-by: Cursor <[email protected]>
87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
"""MinIO container session helpers for integration tests."""
|
|
|
|
from collections.abc import Generator
|
|
|
|
from minio import Minio
|
|
from testcontainers.minio import MinioContainer
|
|
|
|
from python_repositories.config import MinioConfig
|
|
|
|
MINIO_ACCESS_KEY = "minioadmin"
|
|
MINIO_SECRET_KEY = "minioadmin"
|
|
MINIO_BUCKET = "test-bucket"
|
|
|
|
_minio_env: dict[str, str] | None = None
|
|
_minio_container: MinioContainer | None = None
|
|
_raw_minio_client: Minio | None = None
|
|
_minio_env_refs = 0
|
|
_raw_minio_client_refs = 0
|
|
|
|
|
|
def minio_env() -> Generator[dict[str, str], None, None]:
|
|
"""Yield session-scoped MinIO connection settings, starting the container once."""
|
|
global _minio_env, _minio_container, _minio_env_refs
|
|
if _minio_env is None:
|
|
_minio_container = MinioContainer(
|
|
image="minio/minio:latest",
|
|
access_key=MINIO_ACCESS_KEY,
|
|
secret_key=MINIO_SECRET_KEY,
|
|
)
|
|
_minio_container.start()
|
|
minio_host = _minio_container.get_container_host_ip()
|
|
minio_port = _minio_container.get_exposed_port(9000)
|
|
minio_endpoint = f"{minio_host}:{minio_port}"
|
|
_minio_env = {
|
|
"MINIO_ENDPOINT": minio_endpoint,
|
|
"MINIO_ACCESS_KEY": MINIO_ACCESS_KEY,
|
|
"MINIO_SECRET_KEY": MINIO_SECRET_KEY,
|
|
"MINIO_BUCKET": MINIO_BUCKET,
|
|
"MINIO_SECURE": "false",
|
|
}
|
|
|
|
_minio_env_refs += 1
|
|
yield _minio_env
|
|
_minio_env_refs -= 1
|
|
|
|
if _minio_env_refs == 0 and _minio_container is not None:
|
|
_minio_container.stop()
|
|
_minio_container = None
|
|
_minio_env = None
|
|
|
|
|
|
def minio_config_from_env(env: dict[str, str]) -> MinioConfig:
|
|
"""Build MinioConfig from container environment settings."""
|
|
return MinioConfig(
|
|
endpoint=env["MINIO_ENDPOINT"],
|
|
access_key=env["MINIO_ACCESS_KEY"],
|
|
secret_key=env["MINIO_SECRET_KEY"],
|
|
bucket=env["MINIO_BUCKET"],
|
|
secure=False,
|
|
)
|
|
|
|
|
|
def raw_minio_client_from_env(env: dict[str, str]) -> Generator[Minio, None, None]:
|
|
"""Yield a session-scoped raw MinIO client, reusing one client per session."""
|
|
global _raw_minio_client, _raw_minio_client_refs
|
|
if _raw_minio_client is None:
|
|
_raw_minio_client = Minio(
|
|
endpoint=env["MINIO_ENDPOINT"],
|
|
access_key=env["MINIO_ACCESS_KEY"],
|
|
secret_key=env["MINIO_SECRET_KEY"],
|
|
secure=False,
|
|
)
|
|
bucket_name = env["MINIO_BUCKET"]
|
|
if not _raw_minio_client.bucket_exists(bucket_name):
|
|
_raw_minio_client.make_bucket(bucket_name)
|
|
|
|
_raw_minio_client_refs += 1
|
|
yield _raw_minio_client
|
|
_raw_minio_client_refs -= 1
|
|
|
|
if _raw_minio_client_refs == 0 and _raw_minio_client is not None:
|
|
bucket_name = env["MINIO_BUCKET"]
|
|
objects = _raw_minio_client.list_objects(bucket_name, recursive=True)
|
|
for obj in objects:
|
|
_raw_minio_client.remove_object(bucket_name, obj.object_name)
|
|
_raw_minio_client = None
|