started making unittest

shared/datastore/src/delete.py

@@ -1,22 +1,22 @@
 """Definition of delete function."""
 
-from __future__ import annotations
-
 import logging
-import os
+from traceback import print_exc
 
 from minio import Minio
 
 
 def delete(
     client: Minio,
+    bucket_name: str,
     object_name: str,
 ) -> None:
     """Delete object from MinIO."""
     assert isinstance(client, Minio)
-    assert isinstance(object_name, str)
+    assert isinstance(bucket_name, str)
-    bucket_name = os.getenv('MINIO_BUCKET_NAME', default='')
     assert len(bucket_name) > 0
+    assert isinstance(object_name, str)
+    assert len(object_name) > 0
     # remove object
     try:
         client.remove_object(
@@ -24,7 +24,8 @@ def delete(
             object_name=object_name,
         )
     except Exception as exc:
-        logging.debug(exc)
         logging.error('failed deleting %s', object_name)
+        print_exc()
+        raise exc
     else:
         logging.debug('deleted %s', object_name)

shared/datastore/src/get.py

@@ -14,9 +14,10 @@ def get(
 ) -> BytesIO:
     """Get buffer from bucket in MinIO."""
     assert isinstance(client, Minio)
-    assert isinstance(object_name, str)
     assert isinstance(bucket_name, str)
     assert len(bucket_name) > 0
+    assert isinstance(object_name, str)
+    assert len(object_name) > 0
     try:
         # make request
         response = client.get_object(

shared/datastore/tests/unit/connect_minio_test.py

@@ -0,0 +1,33 @@
+"""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()

shared/datastore/tests/unit/delete_test.py

@@ -0,0 +1,77 @@
+"""Definition of unittests for delete function."""
+
+import random
+import string
+import unittest
+from unittest.mock import Mock
+
+from minio import Minio
+
+from shared.datastore import delete
+
+
+class TestDelete(unittest.TestCase):
+
+    def setUp(self):
+        # mock minio client
+        self.client = Mock(spec=Minio)
+        # set bucket name
+        self.bucket_name = 'test-bucket'
+        # set object name
+        self.object_name = ''.join(
+            random.choices(
+                string.ascii_uppercase + string.digits,
+                k=24,
+            ),
+        )
+
+    def test_should_fail_on_wrong_input_type_client(self):
+        with self.assertRaises(AssertionError):
+            delete(
+                client='not-minio-type',
+                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=0.0,
+                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='',
+                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=0.0,
+            )
+
+    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='',
+            )
+
+    def test_should_call_client__remove_object(self):
+        delete(
+            client=self.client,
+            bucket_name=self.bucket_name,
+            object_name=self.object_name,
+        )
+        self.client.remove_object.assert_called_with(
+            bucket_name=self.bucket_name,
+            object_name=self.object_name,
+        )

shared/datastore/tests/unit/get_image_test.py

@@ -0,0 +1,62 @@
+# """Definition of unittests for get_image function."""
+
+# import os
+# import unittest
+# from unittest.mock import Mock
+# from minio import Minio
+# import random
+# import string
+# from PIL import Image
+# from io import BytesIO
+
+# from shared.datastore import get_image
+
+
+# class TestGetImage(unittest.TestCase):
+
+#     def setUp(self):
+#         # mock minio client
+#         self.client = Mock(spec=Minio)
+#         # mock response object
+#         image = Image.new(mode='RGB', size=(480,480))
+#         buffer = BytesIO()
+#         image.save(buffer, 'png')
+#         self.client.get_object.return_value.data = buffer.getvalue()
+#         # set object name
+#         self.object_name = ''.join(
+#             random.choices(
+#                 string.ascii_uppercase + string.digits,
+#                 k=24
+#             )
+#         )
+#         # populate env
+#         self.env_var_map = {
+#             'MINIO_BUCKET_NAME': 'test-bucket'
+#         }
+#         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_call_client__get_object(self):
+#         get_image(
+#             client=self.client,
+#             object_name=self.object_name,
+#         )
+#         self.client.get_object.assert_called_with(
+#             bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
+#             object_name=f'images/{self.object_name}',
+#         )

shared/datastore/tests/unit/get_model_test.py

@@ -0,0 +1,62 @@
+# """Definition of unittest for get_model function."""
+
+# import os
+# import unittest
+# from unittest.mock import Mock
+# from minio import Minio
+# import random
+# import string
+
+# from io import BytesIO
+
+# from shared.datastore import get_model
+
+
+# class TestGetModel(unittest.TestCase):
+
+#     def setUp(self):
+#         # mock minio client
+#         self.client = Mock(spec=Minio)
+#         # mock response object
+#         image = Image.new(mode='RGB', size=(480,480))
+#         buffer = BytesIO()
+#         image.save(buffer, 'png')
+#         self.client.get_object.return_value.data = buffer.getvalue()
+#         # set object name
+#         self.object_name = ''.join(
+#             random.choices(
+#                 string.ascii_uppercase + string.digits,
+#                 k=24
+#             )
+#         )
+#         # populate env
+#         self.env_var_map = {
+#             'MINIO_BUCKET_NAME': 'test-bucket'
+#         }
+#         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_call_client__get_object(self):
+#         get_image(
+#             client=self.client,
+#             object_name=self.object_name,
+#         )
+#         self.client.get_object.assert_called_with(
+#             bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
+#             object_name=f'images/{self.object_name}',
+#         )

shared/datastore/tests/unit/get_test.py

@@ -0,0 +1,83 @@
+"""Definition of unittests for get function."""
+
+import random
+import string
+import unittest
+from io import BytesIO
+from unittest.mock import Mock
+
+from minio import Minio
+
+from shared.datastore import get
+
+
+class TestGet(unittest.TestCase):
+
+    def setUp(self):
+        # mock minio client
+        self.client = Mock(spec=Minio)
+        # set bucket name
+        self.bucket_name = 'test-bucket'
+        # set object name
+        self.object_name = ''.join(
+            random.choices(
+                string.ascii_uppercase + string.digits,
+                k=24,
+            ),
+        )
+        # mock response object
+        self.client.get_object.return_value.status = 200
+        buffer = BytesIO(random.randbytes(2**20))
+        self.client.return_value.get_object = buffer
+        self.client.get_object.return_value.read = buffer.getvalue()
+
+    def test_should_fail_on_wrong_input_type_client(self):
+        with self.assertRaises(AssertionError):
+            get(
+                client='not-minio-type',
+                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=0.0,
+                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='',
+                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=0.0,
+            )
+
+    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='',
+            )
+
+    def test_should_call_client__get_object(self):
+        get(
+            client=self.client,
+            bucket_name=self.bucket_name,
+            object_name=self.object_name,
+        )
+        self.client.get_object.assert_called_with(
+            bucket_name=self.bucket_name,
+            object_name=self.object_name,
+        )