111 lines
2.6 KiB
Python
111 lines
2.6 KiB
Python
"""Integration tests for functions exposed by the datastore module."""
|
|
|
|
import os
|
|
from io import BytesIO
|
|
from random import randbytes
|
|
|
|
import pytest
|
|
from minio import Minio
|
|
from testcontainers.minio import MinioContainer
|
|
|
|
from shared.datastore import connect_minio, get, put
|
|
from shared.utils import setup_logging
|
|
|
|
MINIO_BUCKET_NAME = 'test-bucket'
|
|
minio = MinioContainer()
|
|
|
|
|
|
@pytest.fixture(scope='module', autouse=True)
|
|
def setup(request: pytest.FixtureRequest):
|
|
"""Setup function for datastore module integration tests."""
|
|
# start minio
|
|
minio.start()
|
|
|
|
# ensure minio is stopped after testing
|
|
def remove_container():
|
|
minio.stop()
|
|
|
|
request.addfinalizer(remove_container)
|
|
# extract information
|
|
minio_ip = minio.get_container_host_ip()
|
|
minio_port = minio.get_exposed_port(port=9000)
|
|
os.environ['MINIO_ENDPOINT'] = f'{minio_ip}:{minio_port}'
|
|
os.environ['MINIO_ACCESS_KEY'] = minio.access_key
|
|
os.environ['MINIO_SECRET_KEY'] = minio.secret_key
|
|
os.environ['MINIO_BUCKET_NAME'] = MINIO_BUCKET_NAME
|
|
|
|
|
|
def test_connect_minio():
|
|
"""Test function connect_minio."""
|
|
minio_client = connect_minio()
|
|
assert isinstance(minio_client, Minio)
|
|
|
|
|
|
def test_put():
|
|
"""Test function put."""
|
|
# connect to minio
|
|
minio_client = connect_minio()
|
|
# generate random bytes
|
|
buffer = BytesIO(randbytes(2**20))
|
|
# put data in Minio
|
|
put(
|
|
client=minio_client,
|
|
buffer=buffer,
|
|
bucket_name=MINIO_BUCKET_NAME,
|
|
object_name='test_put_data',
|
|
)
|
|
|
|
|
|
def test_get():
|
|
"""Test function get."""
|
|
# connect to minio
|
|
minio_client = connect_minio()
|
|
# generate random bytes
|
|
buffer = BytesIO(randbytes(2**20))
|
|
# put data in minio
|
|
put(
|
|
client=minio_client,
|
|
buffer=buffer,
|
|
bucket_name=MINIO_BUCKET_NAME,
|
|
object_name='test_get_data',
|
|
)
|
|
# get data back from minio
|
|
buffer_rec = get(
|
|
client=minio_client,
|
|
bucket_name=MINIO_BUCKET_NAME,
|
|
object_name='test_get_data',
|
|
)
|
|
# check that data are the same
|
|
assert buffer.read() == buffer_rec.read()
|
|
|
|
|
|
def test_delete():
|
|
"""Test function delete."""
|
|
raise NotImplementedError()
|
|
|
|
|
|
def test_get_image():
|
|
"""Test function get_image."""
|
|
raise NotImplementedError()
|
|
|
|
|
|
def test_put_image():
|
|
"""Test function put_image."""
|
|
raise NotImplementedError()
|
|
|
|
|
|
def test_get_model():
|
|
"""Test function get_model."""
|
|
raise NotImplementedError()
|
|
|
|
|
|
def test_put_model():
|
|
"""Test function put_model."""
|
|
raise NotImplementedError()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
os.environ['LOG_LEVEL'] = 'debug'
|
|
setup_logging()
|
|
pytest.main()
|