146 lines
3.8 KiB
Python
146 lines
3.8 KiB
Python
"""Integration test configurations."""
|
|
|
|
import os
|
|
import random
|
|
from collections.abc import Iterator
|
|
from io import BytesIO
|
|
|
|
import pytest
|
|
from minio import Minio
|
|
from PIL import Image
|
|
from testcontainers.minio import MinioContainer
|
|
|
|
env_var_map = {
|
|
'MINIO_ENDPOINT': 'localhost:9000',
|
|
'MINIO_ACCESS_KEY': 'test-access-key',
|
|
'MINIO_SECRET_KEY': 'test-secret-key',
|
|
'MINIO_BUCKET_NAME': 'test-bucket',
|
|
'MINIO_OBJECT_NAME': '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX',
|
|
}
|
|
container_map = {
|
|
'minio': MinioContainer(
|
|
port=env_var_map['MINIO_ENDPOINT'].rsplit(':', maxsplit=1)[-1],
|
|
access_key=env_var_map['MINIO_ACCESS_KEY'],
|
|
secret_key=env_var_map['MINIO_SECRET_KEY'],
|
|
),
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope='session', autouse=True)
|
|
def setup_infrastructure(request: pytest.FixtureRequest) -> None:
|
|
"""Prepare infrastructure for integration test."""
|
|
# prepare infrastructure
|
|
for container in container_map.values():
|
|
container.start()
|
|
# update env var map
|
|
minio_host = container_map['minio'].get_container_host_ip()
|
|
minio_port = container_map['minio'].get_exposed_port(9000)
|
|
env_var_map['MINIO_ENDPOINT'] = f'{minio_host}:{minio_port}'
|
|
|
|
# ensure cleanup
|
|
def cleanup_infrastructure():
|
|
for container in container_map.values():
|
|
container.stop()
|
|
|
|
request.addfinalizer(cleanup_infrastructure)
|
|
|
|
|
|
@pytest.fixture(scope='session', autouse=True)
|
|
def populate_env(
|
|
request: pytest.FixtureRequest,
|
|
setup_infrastructure,
|
|
) -> None:
|
|
"""Populate environment with variables used for testing."""
|
|
# update env
|
|
for key, val in env_var_map.items():
|
|
os.environ[key] = val
|
|
|
|
# ensure cleanup
|
|
def cleanup_env():
|
|
for key in env_var_map:
|
|
_ = os.environ.pop(key, default=None)
|
|
|
|
request.addfinalizer(cleanup_env)
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def minio_client(
|
|
setup_infrastructure,
|
|
populate_env,
|
|
) -> Iterator[Minio]:
|
|
# prepare arguments
|
|
minio_endpoint = str(os.getenv('MINIO_ENDPOINT'))
|
|
minio_access_key = str(os.getenv('MINIO_ACCESS_KEY'))
|
|
minio_secret_key = str(os.getenv('MINIO_SECRET_KEY'))
|
|
minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
|
|
# connect to minio
|
|
client = Minio(
|
|
endpoint=minio_endpoint,
|
|
access_key=minio_access_key,
|
|
secret_key=minio_secret_key,
|
|
secure=False,
|
|
)
|
|
# ensure bucket exists
|
|
if not client.bucket_exists(bucket_name=minio_bucket_name):
|
|
client.make_bucket(bucket_name=minio_bucket_name)
|
|
# expose client
|
|
yield client
|
|
|
|
|
|
@pytest.fixture
|
|
def data() -> Iterator[bytes]:
|
|
# generate random data
|
|
num_bytes = 2**21 # 2 MB
|
|
data = random.randbytes(n=num_bytes)
|
|
# expose data
|
|
yield data
|
|
|
|
|
|
@pytest.fixture
|
|
def data_in_minio(
|
|
minio_client,
|
|
data,
|
|
) -> Iterator[tuple[BytesIO, str, str]]:
|
|
# prepare arguments
|
|
minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
|
|
minio_object_name = str(os.getenv('MINIO_OBJECT_NAME'))
|
|
# convert data
|
|
buffer = BytesIO(data)
|
|
# prepare for saving
|
|
num_bytes = buffer.tell()
|
|
buffer.seek(0)
|
|
# send data to bucket
|
|
minio_client.put_object(
|
|
bucket_name=minio_bucket_name,
|
|
object_name=minio_object_name,
|
|
length=num_bytes,
|
|
data=buffer,
|
|
)
|
|
# expose data
|
|
yield buffer, minio_bucket_name, minio_object_name
|
|
# clean up
|
|
minio_client.remove_object(
|
|
bucket_name=minio_bucket_name,
|
|
object_name=minio_object_name,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def image() -> Iterator[Image.Image]:
|
|
# generate image
|
|
image = Image.new(mode='RGB', size=(480, 480))
|
|
# expose image
|
|
yield image
|
|
|
|
|
|
# @pytest.fixture
|
|
# def image_in_minio(
|
|
# image: Image.Image,
|
|
# ) -> tuple(Image.Image, str):
|
|
# #
|
|
|
|
|
|
# # expose image and object name
|
|
# yield image, object_name
|
|
# # cleanup
|