129 lines
3.3 KiB
Python
129 lines
3.3 KiB
Python
"""Integration test configurations."""
|
|
|
|
import os
|
|
import random
|
|
from collections.abc import Iterator
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from dotenv import load_dotenv
|
|
from minio import Minio
|
|
from PIL import Image
|
|
|
|
env_var_map = {
|
|
'MINIO_BUCKET_NAME': 'test-bucket',
|
|
'MINIO_OBJECT_NAME': '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX',
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope='session', autouse=True)
|
|
def populate_env(
|
|
request: pytest.FixtureRequest,
|
|
) -> None:
|
|
"""Populate environment with variables used for testing."""
|
|
# read env-file for local testing
|
|
load_dotenv(
|
|
dotenv_path=Path(__file__).parent.parent.parent.parent.parent / 'server.env',
|
|
)
|
|
# 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(
|
|
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
|
|
# remove objects left behind by tests
|
|
for obj_name in client.list_objects(bucket_name=minio_bucket_name, recursive=True):
|
|
client.remove_object(
|
|
bucket_name=minio_bucket_name,
|
|
object_name=obj_name,
|
|
)
|
|
# remove bucket
|
|
client.remove_bucket(bucket_name=minio_bucket_name)
|
|
assert not client.bucket_exists(bucket_name=minio_bucket_name)
|
|
|
|
|
|
@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 = len(buffer.getvalue())
|
|
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
|