updated tests to match new interface
Code Quality Pipeline / Check Code (pull_request) Failing after 3m5s

This commit is contained in:
brian
2025-01-06 11:34:02 +00:00
parent 88698dd735
commit 16ad2b80ee
14 changed files with 495 additions and 596 deletions
@@ -1,33 +0,0 @@
"""Definition of unittests for connect_minio function."""
import os
import unittest
from shared.datastore import connect_minio
class TestConnectMinio(unittest.TestCase):
def setUp(self):
# define relevant env vars
self.env_var_map = {
'MINIO_ENDPOINT': '192.168.1.2',
'MINIO_ACCESS_KEY': 'randomAccess_key',
'MINIO_SECRET_KEY': 'randomSecret_key',
'MINIO_BUCKET_NAME': 'test-bucket-name',
}
# set env vars
for key, val in self.env_var_map.items():
os.environ[key] = val
def tearDown(self):
# clear env vars
for key in self.env_var_map:
_ = os.environ.pop(key, default=None)
def test_should_fail_when_env_not_set(self):
# ensure env not set
self.tearDown()
# run test
with self.assertRaises(AssertionError):
_ = connect_minio()
@@ -0,0 +1,230 @@
"""Definition of unittests for Datastore instantiation."""
import os
from hashlib import md5
from io import BytesIO
from unittest import TestCase
from unittest.mock import ANY, MagicMock, patch
from minio import Minio
from PIL import Image
from urllib3 import BaseHTTPResponse
from shared.datastore.src.datastore_minio import DatastoreMinio
class TestDatastoreMinioInstantiation(TestCase):
def setUp(self):
# define relevant env vars
self.env_var_map = {
'MINIO_ENDPOINT': '192.168.1.2',
'MINIO_ACCESS_KEY': 'randomAccess_key',
'MINIO_SECRET_KEY': 'randomSecret_key',
'MINIO_BUCKET_NAME': 'test-bucket-name',
}
# set env vars
for key, val in self.env_var_map.items():
os.environ[key] = val
# set other variables
self.object_name = '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX'
self.image = Image.new(mode='RGB', size=(480, 480))
self.buffer = BytesIO()
self.image.save(self.buffer, 'png')
self.num_bytes = len(self.buffer.getvalue())
self.checksum = md5(self.buffer.getbuffer()).hexdigest()
def tearDown(self):
# clear env vars
for key in self.env_var_map:
_ = os.environ.pop(key, default=None)
def test_instantiation_should_fail_when_env_not_set(self):
# ensure env not set
self.tearDown()
# run test
with self.assertRaises(OSError):
_ = DatastoreMinio()
@patch('shared.datastore.src.datastore_minio.Minio')
def test_connect_should_call_Minio_with_env_vars(self, minio_mock):
# ARRANGE
datastore = DatastoreMinio()
minio_mock().bucket_exists.return_value = False
# ACT
datastore.connect()
# ASSERT
minio_mock.assert_called_with(
endpoint=self.env_var_map['MINIO_ENDPOINT'],
access_key=self.env_var_map['MINIO_ACCESS_KEY'],
secret_key=self.env_var_map['MINIO_SECRET_KEY'],
secure=False,
)
minio_mock().bucket_exists.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
)
minio_mock().make_bucket.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
)
def test_close_should_overwrite_private_variables(self):
# ARRANGE
datastore = DatastoreMinio()
datastore._client = MagicMock()
datastore._bucket_name = MagicMock()
assert isinstance(datastore._client, MagicMock)
assert isinstance(datastore._bucket_name, MagicMock)
# ACT
datastore.close()
# ASSERT
assert datastore._client is None
assert datastore._bucket_name is None
@patch('shared.datastore.src.datastore_minio.DatastoreMinio.connect')
@patch('shared.datastore.src.datastore_minio.DatastoreMinio.close')
def test_context_management_implemented(
self,
mocked_close_method,
mocked_connect_method,
):
# ARRANGE, ACT and ASSERT
with DatastoreMinio() as ds:
mocked_connect_method.assert_called_once()
assert isinstance(ds, DatastoreMinio)
mocked_close_method.assert_called_once()
def test_should_call_put_object_with_arguments(self):
# ARRANGE
datastore = DatastoreMinio()
datastore._client = MagicMock(Minio)
datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
self.buffer.seek(0)
# ACT
datastore._put(
object_name=self.object_name,
buffer=self.buffer,
)
# ASSERT
datastore._client.put_object.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
object_name=self.object_name,
length=self.num_bytes,
data=self.buffer,
)
def test_should_call_get_object_with_arguments(self):
# ARRANGE
datastore = DatastoreMinio()
datastore._client = MagicMock(Minio)
datastore._client.get_object.return_value = MagicMock(
BaseHTTPResponse,
status=200,
)
# datastore._client.get_object.read.side_effect = [b'random ', b'test', b'text']
datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
# ACT
with self.assertRaises(TypeError): # dont care to mock even more...
datastore._get(
object_name=self.object_name,
)
# ASSERT
datastore._client.get_object.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
object_name=self.object_name,
)
def test_should_call_remove_object_with_arguments(self):
# ARRANGE
datastore = DatastoreMinio()
datastore._client = MagicMock(Minio)
datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
# ACT
datastore._delete(
object_name=self.object_name,
)
# ASSERT
datastore._client.remove_object.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
object_name=self.object_name,
)
def test_put_image_should_call_put_object_with_arguments(self):
# ARRANGE
datastore = DatastoreMinio()
datastore._client = MagicMock(Minio)
datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
object_path = f'images/{self.checksum}'
# ACT
datastore.put_image(
image=self.image,
)
# ASSERT
datastore._client.put_object.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
object_name=object_path,
length=self.num_bytes,
data=ANY, # saved to different buffer when converting image
)
def test_get_image_should_call_get_object_with_arguments(self):
# ARRANGE
datastore = DatastoreMinio()
datastore._client = MagicMock(Minio)
datastore._client.get_object.return_value = MagicMock(
BaseHTTPResponse,
status=200,
)
datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
object_path = f'images/{self.checksum}'
# ACT
with self.assertRaises(TypeError): # dont care to mock even more...
datastore.get_image(
object_name=self.checksum,
)
# ASSERT
datastore._client.get_object.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
object_name=object_path,
)
# @patch('shared.datastore.src.datastore_minio.torch.serialization.save')
# def test_put_model_should_call_put_object_with_arguments(self, mocked_torch_fn):
# # ARRANGE
# datastore = DatastoreMinio()
# datastore._client = MagicMock(Minio)
# datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
# object_path = f'images/{self.checksum}'
# mocked_torch_fn.return_value = None
# model = MagicMock(torch.nn.Module)
# # ACT
# datastore.put_model(
# model=model,
# )
# # ASSERT
# datastore._client.put_object.assert_called_with(
# bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
# object_name=object_path,
# length=self.num_bytes,
# data=ANY, # saved to different buffer when converting data
# )
def test_get_model_should_call_get_object_with_arguments(self):
# ARRANGE
datastore = DatastoreMinio()
datastore._client = MagicMock(Minio)
datastore._client.get_object.return_value = MagicMock(
BaseHTTPResponse,
status=200,
)
datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
object_path = f'models/{self.checksum}'
# ACT
with self.assertRaises(TypeError): # dont care to mock even more...
datastore.get_model(
object_name=self.checksum,
)
# ASSERT
datastore._client.get_object.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
object_name=object_path,
)
@@ -1,61 +0,0 @@
"""Definition of unittests for delete function."""
import unittest
from unittest.mock import Mock
from minio import Minio
from shared.datastore import delete
class TestDelete(unittest.TestCase):
def setUp(self):
# set relevant variables
self.client = Mock(spec=Minio)
self.bucket_name = 'test-bucket'
self.object_name = '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX'
# set bad arguments
self.bad_client = 'not-minio-type'
self.bad_string = float(0.0)
self.len_0_string = ''
def test_should_fail_on_wrong_input_type_client(self):
with self.assertRaises(AssertionError):
delete(
client=self.bad_client,
bucket_name=self.bucket_name,
object_name=self.object_name,
)
def test_should_fail_on_wrong_input_type_bucket_name(self):
with self.assertRaises(AssertionError):
delete(
client=self.client,
bucket_name=self.bad_string,
object_name=self.object_name,
)
def test_should_fail_on_wrong_input_length_bucket_name(self):
with self.assertRaises(AssertionError):
delete(
client=self.client,
bucket_name=self.len_0_string,
object_name=self.object_name,
)
def test_should_fail_on_wrong_input_type_object_name(self):
with self.assertRaises(AssertionError):
delete(
client=self.client,
bucket_name=self.bucket_name,
object_name=self.bad_string,
)
def test_should_fail_on_wrong_input_length_object_name(self):
with self.assertRaises(AssertionError):
delete(
client=self.client,
bucket_name=self.bucket_name,
object_name=self.len_0_string,
)
@@ -1,63 +0,0 @@
"""Definition of unittests for get_image function."""
import os
import unittest
from unittest.mock import MagicMock
from minio import Minio
from shared.datastore import get_image
class TestGetImage(unittest.TestCase):
def setUp(self):
# set relevant variables
self.client = MagicMock(spec=Minio)
self.object_name = '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX'
self.env_var_map = {
'MINIO_BUCKET_NAME': 'test-bucket',
}
# set bad arguments
self.bad_client = 'not-minio-type'
self.bad_string = float(0.0)
self.len_0_string = ''
# populate env
for key, val in self.env_var_map.items():
os.environ[key] = val
def tearDown(self):
# clean env
for key in self.env_var_map:
_ = os.environ.pop(key, default=None)
def test_should_fail_when_env_not_set(self):
# ensure env not set
self.tearDown()
# run test
with self.assertRaises(AssertionError):
get_image(
client=self.client,
object_name=self.object_name,
)
def test_should_fail_on_wrong_input_type_client(self):
with self.assertRaises(AssertionError):
get_image(
client=self.bad_client,
object_name=self.object_name,
)
def test_should_fail_on_wrong_input_type_object_name(self):
with self.assertRaises(AssertionError):
get_image(
client=self.client,
object_name=self.bad_string,
)
def test_should_fail_on_wrong_input_length_object_name(self):
with self.assertRaises(AssertionError):
get_image(
client=self.client,
object_name=self.len_0_string,
)
@@ -1,63 +0,0 @@
"""Definition of unittest for get_model function."""
import os
import unittest
from unittest.mock import MagicMock
from minio import Minio
from shared.datastore import get_model
class TestGetModel(unittest.TestCase):
def setUp(self):
# set relevant variables
self.client = MagicMock(spec=Minio)
self.object_name = '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX'
self.env_var_map = {
'MINIO_BUCKET_NAME': 'test-bucket',
}
# set bad arguments
self.bad_client = 'not-minio-type'
self.bad_string = float(0.0)
self.len_0_string = ''
# populate env
for key, val in self.env_var_map.items():
os.environ[key] = val
def tearDown(self):
# clean env
for key in self.env_var_map:
_ = os.environ.pop(key, default=None)
def test_should_fail_when_env_not_set(self):
# ensure env not set
self.tearDown()
# run test
with self.assertRaises(AssertionError):
get_model(
client=self.client,
object_name=self.object_name,
)
def test_should_fail_on_wrong_input_type_client(self):
with self.assertRaises(AssertionError):
get_model(
client=self.bad_client,
object_name=self.object_name,
)
def test_should_fail_on_wrong_input_type_object_name(self):
with self.assertRaises(AssertionError):
get_model(
client=self.client,
object_name=self.bad_string,
)
def test_should_fail_on_wrong_input_length_object_name(self):
with self.assertRaises(AssertionError):
get_model(
client=self.client,
object_name=self.len_0_string,
)
-60
View File
@@ -1,60 +0,0 @@
"""Definition of unittests for get function."""
import unittest
from unittest.mock import MagicMock
from minio import Minio
from shared.datastore import get
class TestGet(unittest.TestCase):
def setUp(self):
# set relevant variables
self.client = MagicMock(spec=Minio)
self.bucket_name = 'test-bucket'
self.object_name = '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX'
# set bad arguments
self.bad_client = 'not-minio-type'
self.bad_string = float(0.0)
self.len_0_string = ''
def test_should_fail_on_wrong_input_type_client(self):
with self.assertRaises(AssertionError):
get(
client=self.bad_client,
bucket_name=self.bucket_name,
object_name=self.object_name,
)
def test_should_fail_on_wrong_input_type_bucket_name(self):
with self.assertRaises(AssertionError):
get(
client=self.client,
bucket_name=self.bad_string,
object_name=self.object_name,
)
def test_should_fail_on_wrong_input_length_bucket_name(self):
with self.assertRaises(AssertionError):
get(
client=self.client,
bucket_name=self.len_0_string,
object_name=self.object_name,
)
def test_should_fail_on_wrong_input_type_object_name(self):
with self.assertRaises(AssertionError):
get(
client=self.client,
bucket_name=self.bucket_name,
object_name=self.bad_string,
)
def test_should_fail_on_wrong_input_length_object_name(self):
with self.assertRaises(AssertionError):
get(
client=self.client,
bucket_name=self.bucket_name,
object_name=self.len_0_string,
)
@@ -1,41 +0,0 @@
"""Definition of unittests for put_image function."""
import os
import unittest
from unittest.mock import MagicMock
from minio import Minio
from PIL import Image
from shared.datastore import put_image
class TestPutImage(unittest.TestCase):
def setUp(self):
# set relevant variables
self.client = MagicMock(spec=Minio)
self.image = Image.new(mode='RGB', size=(480, 480))
self.env_var_map = {
'MINIO_BUCKET_NAME': 'test-bucket',
}
# set bad arguments
self.bad_client = 'not-minio-type'
self.bad_image = 'not-image-type'
# populate env
for key, val in self.env_var_map.items():
os.environ[key] = val
def test_should_fail_on_wrong_input_type_client(self):
with self.assertRaises(AssertionError):
put_image(
client=self.bad_client,
image=self.image,
)
def test_should_fail_on_wrong_input_type_image(self):
with self.assertRaises(AssertionError):
put_image(
client=self.client,
image=self.bad_image,
)
@@ -1,40 +0,0 @@
"""Definition of unittests for put_model function."""
import os
import unittest
from unittest.mock import MagicMock
from minio import Minio
from torch.nn import Module
from shared.datastore import put_model
class TestPutModel(unittest.TestCase):
def setUp(self):
# set relevant variables
self.client = MagicMock(spec=Minio)
self.model = MagicMock(spec=Module)
self.env_var_map = {
'MINIO_BUCKET_NAME': 'test-bucket',
}
# set bad arguments
self.bad_client = 'not-minio-type'
self.bad_model = 'not-image-type'
# populate env
for key, val in self.env_var_map.items():
os.environ[key] = val
def test_should_fail_on_wrong_input_type_client(self):
with self.assertRaises(AssertionError):
put_model(
client=self.bad_client,
model=self.model,
)
def test_should_fail_on_wrong_input_type_model(self):
with self.assertRaises(AssertionError):
put_model(
client=self.client,
model=self.bad_model,
)
-80
View File
@@ -1,80 +0,0 @@
"""Definition of unittests for put function."""
import unittest
from io import BytesIO
from unittest.mock import MagicMock
from minio import Minio
from PIL import Image
from shared.datastore import put
class TestPut(unittest.TestCase):
def setUp(self):
# set relevant variables
self.client = MagicMock(spec=Minio)
self.bucket_name = 'test-bucket'
self.object_name = '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX'
self.image = Image.new(mode='RGB', size=(480, 480))
self.buffer = BytesIO()
self.image.save(self.buffer, 'png')
# set bad arguments
self.bad_client = 'not-minio-type'
self.bad_string = float(0.0)
self.bad_buffer = 'not-buffer-type'
self.len_0_string = ''
def test_should_fail_on_wrong_input_type_client(self):
with self.assertRaises(AssertionError):
put(
client=self.bad_client,
buffer=self.buffer,
bucket_name=self.bucket_name,
object_name=self.object_name,
)
def test_should_fail_on_wrong_input_type_buffer(self):
with self.assertRaises(AssertionError):
put(
client=self.client,
buffer=self.bad_buffer,
bucket_name=self.bucket_name,
object_name=self.object_name,
)
def test_should_fail_on_wrong_input_type_bucket_name(self):
with self.assertRaises(AssertionError):
put(
client=self.client,
buffer=self.buffer,
bucket_name=self.bad_string,
object_name=self.object_name,
)
def test_should_fail_on_wrong_input_length_bucket_name(self):
with self.assertRaises(AssertionError):
put(
client=self.client,
buffer=self.buffer,
bucket_name=self.len_0_string,
object_name=self.object_name,
)
def test_should_fail_on_wrong_input_type_object_name(self):
with self.assertRaises(AssertionError):
put(
client=self.client,
buffer=self.buffer,
bucket_name=self.bucket_name,
object_name=self.bad_string,
)
def test_should_fail_on_wrong_input_length_object_name(self):
with self.assertRaises(AssertionError):
put(
client=self.client,
buffer=self.buffer,
bucket_name=self.bucket_name,
object_name=self.len_0_string,
)