406 lines
11 KiB
Python
406 lines
11 KiB
Python
"""Integration tests configuration."""
|
|
|
|
import os
|
|
import random
|
|
from collections.abc import Iterator
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
|
|
import minio
|
|
import pytest
|
|
from dotenv import load_dotenv
|
|
from PIL import Image
|
|
from pymongo import MongoClient
|
|
|
|
from model.src.models import VisualCommunicationModel
|
|
from shared.repositories import (
|
|
HexadecimalString,
|
|
ImageData,
|
|
ImageRepository,
|
|
ModelData,
|
|
ModelRepository,
|
|
VisualCommunicationData,
|
|
VisualCommunicationRepository,
|
|
VisualCommunicationValues,
|
|
)
|
|
from shared.repositories.src.implementations import (
|
|
MinioImplementation,
|
|
MongoImplementation,
|
|
)
|
|
|
|
# set random seed for reproducibility
|
|
random.seed(13)
|
|
|
|
# load local test environment variables
|
|
test_env_path = Path(__file__).parent.parent.parent.parent.parent / 'test.env'
|
|
if test_env_path.exists():
|
|
load_dotenv(test_env_path.as_posix())
|
|
|
|
# define test environment variables
|
|
necessary_env_vars = {
|
|
'MINIO_ENDPOINT',
|
|
'MINIO_ACCESS_KEY',
|
|
'MINIO_SECRET_KEY',
|
|
'MONGO_ENDPOINT',
|
|
}
|
|
env_var_map = {
|
|
'MINIO_BUCKET_NAME': 'test-bucket',
|
|
'MINIO_OBJECT_NAME': 'test-object',
|
|
'MINIO_IMAGE_NAME': 'test-image',
|
|
'MONGO_DB': 'test-db',
|
|
'MONGO_COLLECTION': 'test-collection',
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope='session', autouse=True)
|
|
def setup_env(
|
|
request: pytest.FixtureRequest,
|
|
) -> None:
|
|
"""Populate environment with variables used for testing."""
|
|
# check if necessary env vars are set
|
|
for key in necessary_env_vars:
|
|
assert key in os.environ, f'{key} not set'
|
|
# set testing env vars
|
|
for key, val in env_var_map.items():
|
|
# set env var
|
|
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 raw_minio_client(
|
|
setup_env,
|
|
) -> Iterator[minio.Minio]:
|
|
"""Raw Minio client fixture."""
|
|
# 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 client
|
|
client = minio.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
|
|
# cleanup
|
|
object_list = client.list_objects(minio_bucket_name, recursive=True)
|
|
for obj in object_list:
|
|
client.remove_object(
|
|
bucket_name=obj.bucket_name,
|
|
object_name=obj.object_name,
|
|
)
|
|
client.remove_bucket(minio_bucket_name)
|
|
|
|
|
|
@pytest.fixture
|
|
def minio_client(
|
|
setup_env,
|
|
) -> Iterator[MinioImplementation]:
|
|
"""MinioImplementation fixture."""
|
|
# instantiate and connect client
|
|
minio_client = MinioImplementation()
|
|
minio_client.connect()
|
|
# expose client
|
|
yield minio_client
|
|
# cleanup
|
|
object_name_list = minio_client._list_objects(Path('*'))
|
|
for name in object_name_list:
|
|
minio_client._delete(Path(name))
|
|
minio_client.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def buffer() -> Iterator[BytesIO]:
|
|
"""Bytes buffer fixture."""
|
|
# generate reproducible random data
|
|
data = random.randbytes(n=2**21) # 2 MB
|
|
# convert data
|
|
buffer = BytesIO(data)
|
|
# expose buffer
|
|
yield buffer
|
|
|
|
|
|
@pytest.fixture
|
|
def buffer_in_minio(
|
|
raw_minio_client: minio.Minio,
|
|
buffer: BytesIO,
|
|
) -> Iterator[tuple[Path, BytesIO]]:
|
|
"""Buffer in Minio fixture."""
|
|
# prepare arguments
|
|
minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
|
|
minio_object_name = str(os.getenv('MINIO_OBJECT_NAME'))
|
|
# prepare for saving
|
|
num_bytes = len(buffer.getvalue())
|
|
buffer.seek(0)
|
|
# put data in bucket
|
|
raw_minio_client.put_object(
|
|
bucket_name=minio_bucket_name,
|
|
object_name=minio_object_name,
|
|
length=num_bytes,
|
|
data=buffer,
|
|
)
|
|
# expose data
|
|
yield Path(minio_object_name), buffer
|
|
# cleanup
|
|
raw_minio_client.remove_object(
|
|
bucket_name=minio_bucket_name,
|
|
object_name=minio_object_name,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def image_data() -> Iterator[ImageData]:
|
|
"""Image data fixture."""
|
|
# prepare arguments
|
|
name = str(os.getenv('MINIO_IMAGE_NAME'))
|
|
image = Image.new(mode='RGB', size=(480, 480))
|
|
# instantiate data
|
|
image_data = ImageData(image=image, name=name)
|
|
# expose data
|
|
yield image_data
|
|
|
|
|
|
@pytest.fixture
|
|
def image_data_in_minio(
|
|
raw_minio_client: minio.Minio,
|
|
image_data: ImageData,
|
|
) -> Iterator[ImageData]:
|
|
"""Image data in Minio fixture."""
|
|
# prepare arguments
|
|
minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
|
|
image_name = image_data.name
|
|
# build object path
|
|
object_path = ImageRepository._build_path(image_name)
|
|
# save image to buffer
|
|
buffer = BytesIO()
|
|
image_data.image.save(buffer, 'png')
|
|
# prepare for saving
|
|
num_bytes = len(buffer.getvalue())
|
|
buffer.seek(0)
|
|
# put data in bucket
|
|
raw_minio_client.put_object(
|
|
bucket_name=minio_bucket_name,
|
|
object_name=object_path.as_posix(),
|
|
length=num_bytes,
|
|
data=buffer,
|
|
)
|
|
# expose data
|
|
yield image_data
|
|
# cleanup
|
|
raw_minio_client.remove_object(
|
|
bucket_name=minio_bucket_name,
|
|
object_name=object_path.as_posix(),
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def image_repo(
|
|
setup_env,
|
|
) -> Iterator[ImageRepository]:
|
|
"""Image repository fixture."""
|
|
# define test environment variables
|
|
assert 'MINIO_ENDPOINT' in os.environ, 'MINIO_ENDPOINT not set'
|
|
assert 'MINIO_ACCESS_KEY' in os.environ, 'MINIO_ACCESS_KEY not set'
|
|
assert 'MINIO_SECRET_KEY' in os.environ, 'MINIO_SECRET_KEY not set'
|
|
assert 'MONGO_ENDPOINT' in os.environ, 'MONGO_ENDPOINT not set'
|
|
repo = ImageRepository()
|
|
repo.connect()
|
|
yield repo
|
|
repo.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def model_data() -> Iterator[ModelData]:
|
|
"""Model data fixture."""
|
|
# prepare arguments
|
|
vis_com_model = VisualCommunicationModel().to('cpu')
|
|
class_name = type(vis_com_model).__name__
|
|
buffer = ModelData.model_to_buffer(vis_com_model)
|
|
buffer_checksum = HexadecimalString('77dcab1769563654a6e24f92d40f29bd')
|
|
# instantiate data
|
|
model_data = ModelData(
|
|
buffer=buffer,
|
|
buffer_checksum=buffer_checksum,
|
|
class_name=class_name,
|
|
)
|
|
# expose model
|
|
yield model_data
|
|
|
|
|
|
@pytest.fixture
|
|
def model_data_in_minio(
|
|
raw_minio_client: minio.Minio,
|
|
model_data: ModelData,
|
|
) -> Iterator[ModelData]:
|
|
"""Model data in Minio fixture."""
|
|
# prepare arguments
|
|
minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
|
|
object_name = ModelRepository._build_object_name(model_data)
|
|
buffer = model_data.buffer
|
|
# build object path
|
|
object_path = ModelRepository._prefix() / object_name
|
|
# prepare for saving
|
|
num_bytes = len(buffer.getvalue())
|
|
buffer.seek(0)
|
|
# put data in bucket
|
|
raw_minio_client.put_object(
|
|
bucket_name=minio_bucket_name,
|
|
object_name=object_path.as_posix(),
|
|
length=num_bytes,
|
|
data=buffer,
|
|
)
|
|
# expose data
|
|
yield model_data
|
|
# cleanup
|
|
raw_minio_client.remove_object(
|
|
bucket_name=minio_bucket_name,
|
|
object_name=object_path.as_posix(),
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def model_repo(
|
|
setup_env,
|
|
) -> Iterator[ModelRepository]:
|
|
"""Model repository fixture."""
|
|
repo = ModelRepository()
|
|
repo.connect()
|
|
yield repo
|
|
repo.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def raw_mongo_client(
|
|
setup_env,
|
|
) -> Iterator[MongoClient]:
|
|
"""Raw mongo client fixture."""
|
|
# prepare arguments
|
|
mongo_endpoint = str(os.getenv('MONGO_ENDPOINT'))
|
|
mongo_database = str(os.getenv('MONGO_DB'))
|
|
# connect client
|
|
client: MongoClient = MongoClient(mongo_endpoint)
|
|
_ = client[mongo_database]
|
|
# expose client
|
|
yield client
|
|
# cleanup
|
|
client.drop_database(mongo_database)
|
|
|
|
|
|
@pytest.fixture
|
|
def mongo_client(
|
|
setup_env,
|
|
) -> Iterator[MongoImplementation]:
|
|
"""MongoImplementation fixture."""
|
|
# instantiate and connect client
|
|
mongo_client = MongoImplementation()
|
|
mongo_client.connect()
|
|
# expose client
|
|
yield mongo_client
|
|
# cleanup
|
|
mongo_client._collection.drop()
|
|
mongo_client.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def dictionary() -> Iterator[dict]:
|
|
"""Dictionary fixture."""
|
|
# prepare data
|
|
data = {
|
|
'name': 'test-dictionary',
|
|
'str_key': 'value',
|
|
'int_key': 100,
|
|
'float_key': 3.14,
|
|
'list_key': [1, 2, 3],
|
|
'dict_key': {
|
|
'nested_key': 'nested_value',
|
|
},
|
|
}
|
|
# expose data
|
|
yield data
|
|
|
|
|
|
@pytest.fixture
|
|
def dictionary_in_mongo(
|
|
raw_mongo_client: MongoClient,
|
|
dictionary: dict,
|
|
) -> Iterator[dict]:
|
|
"""Dictionary in Mongo fixture."""
|
|
# prepare arguments
|
|
database = str(os.getenv('MONGO_DB'))
|
|
collection = str(os.getenv('MONGO_COLLECTION'))
|
|
# save data
|
|
_ = raw_mongo_client[database][collection].insert_one(dictionary.copy())
|
|
# expose data
|
|
yield dictionary
|
|
# cleanup
|
|
raw_mongo_client[database][collection].delete_one(dictionary)
|
|
|
|
|
|
@pytest.fixture
|
|
def visual_communication_values() -> Iterator[VisualCommunicationValues]:
|
|
"""Visual communication values fixture."""
|
|
# instantiate with random values
|
|
visual_communication_values = VisualCommunicationValues.from_random()
|
|
# expose values
|
|
yield visual_communication_values
|
|
|
|
|
|
@pytest.fixture
|
|
def visual_communication_data(
|
|
visual_communication_values: VisualCommunicationValues,
|
|
) -> Iterator[VisualCommunicationData]:
|
|
"""Visual communication data fixture."""
|
|
# prepare arguments
|
|
name = 'test-visual-communication'
|
|
annotation = visual_communication_values
|
|
# instantiate data
|
|
visual_communication_data = VisualCommunicationData(
|
|
name=name,
|
|
annotation=annotation,
|
|
)
|
|
# expose data
|
|
yield visual_communication_data
|
|
|
|
|
|
@pytest.fixture
|
|
def visual_communication_data_in_mongo(
|
|
raw_mongo_client: MongoClient,
|
|
visual_communication_data: VisualCommunicationData,
|
|
) -> Iterator[VisualCommunicationData]:
|
|
"""Visual communication data in Mongo fixture."""
|
|
# prepare arguments
|
|
database = str(os.getenv('MONGO_DB'))
|
|
collection = str(os.getenv('MONGO_COLLECTION'))
|
|
# convert data
|
|
dictionary = visual_communication_data.model_dump(mode='dict')
|
|
# save data
|
|
_ = raw_mongo_client[database][collection].insert_one(dictionary.copy())
|
|
# expose data
|
|
yield visual_communication_data
|
|
# cleanup
|
|
raw_mongo_client[database][collection].delete_one(dictionary)
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def visual_communication_repo(
|
|
setup_env,
|
|
) -> Iterator[VisualCommunicationRepository]:
|
|
"""Visual communication repository fixture."""
|
|
repo = VisualCommunicationRepository()
|
|
repo.connect()
|
|
yield repo
|
|
repo.close()
|