python
1 week ago
import unittest
from unittest.mock import MagicMock, patch
from s3service import S3Service
from minio import S3Error
from io import BytesIO
class TestS3Service(unittest.TestCase):
def setUp(self):
self.mock_config = {
"s3": {
"DEV": {
"BUCKET_NAME": "dev-bucket",
"ENDPOINT": "localhost:9000",
"ACCESS_KEY_ID": "dev-access-key",
"SECRET_ACCESS_KEY": "dev-secret-key",
"NUM_POOLS": 10,
"MAXSIZE": 50
},
"QA": {
"BUCKET_NAME": "qa-bucket",
"ENDPOINT": "localhost:9001",
"ACCESS_KEY_ID": "qa-access-key",
"SECRET_ACCESS_KEY": "qa-secret-key",
"NUM_POOLS": 10,
"MAXSIZE": 50
},
"PROD": {
"BUCKET_NAME": "prod-bucket",
"ENDPOINT": "localhost:9002",
"ACCESS_KEY_ID": "prod-access-key",
"SECRET_ACCESS_KEY": "prod-secret-key",
"NUM_POOLS": 10,
"MAXSIZE": 50
}
}
}
self.env = "DEV"
self.trace_id = "trace123"
with patch("s3service.Minio") as mock_minio:
self.mock_minio_client = MagicMock()
mock_minio.return_value = self.mock_minio_client
self.s3_service = S3Service(self.mock_config, env=self.env)
def test_initialize_minio_instance(self):
with patch("s3service.Minio") as mock_minio:
s3_service = S3Service(self.mock_config, env="DEV")
self.assertIsNotNone(s3_service.minio_client)
mock_minio.assert_called_with(
"localhost:9000",
access_key="dev-access-key",
secret_key="dev-secret-key",
http_client=MagicMock()
)
def test_get_bucket_list(self):
self.mock_minio_client.list_buckets.return_value = ["bucket1", "bucket2"]
buckets = self.s3_service.get_bucket_list()
self.assertEqual(buckets, ["bucket1", "bucket2"])
def test_upload_files_to_bucket(self):
self.mock_minio_client.bucket_exists.return_value = False
response = self.s3_service.upload_files_to_bucket(
"test-bucket", "file_path.txt", "file_name.txt", self.trace_id
)
self.mock_minio_client.make_bucket.assert_called_once_with("test-bucket")
self.mock_minio_client.fput_object.assert_called_once_with(
"test-bucket", "file_name.txt", "file_path.txt"
)
self.assertIn("Successfully uploaded", response["Status"])
def test_file_exists_in_bucket(self):
self.mock_minio_client.stat_object.return_value = MagicMock()
exists = self.s3_service.file_exists_in_bucket("path/to/file.txt")
self.assertTrue(exists)
self.mock_minio_client.stat_object.side_effect = S3Error("Error", "Mock error")
exists = self.s3_service.file_exists_in_bucket("path/to/missing_file.txt")
self.assertFalse(exists)
def test_list_files_in_bucket(self):
mock_objects = [
MagicMock(object_name="file1.txt"),
MagicMock(object_name="file2.txt")
]
self.mock_minio_client.list_objects.return_value = mock_objects
files = self.s3_service.list_files_in_bucket("prefix/", self.trace_id)
self.assertEqual(files, {"file1.txt", "file2.txt"})
def test_download_video_from_bucket(self):
mock_data_stream = MagicMock()
mock_data_stream.stream.return_value = [b"data1", b"data2"]
self.mock_minio_client.get_object.return_value = mock_data_stream
response = self.s3_service.download_video_from_bucket("path/to/video.mp4", self.trace_id)
self.assertIn("Successfully downloaded", response["Status"])
self.assertIsInstance(response["Data"], BytesIO)
def test_stream_object_to_bucket(self):
mock_io_object = BytesIO(b"test data")
response = self.s3_service.stream_object_to_bucket(
mock_io_object, "test_object.txt", self.trace_id
)
self.mock_minio_client.put_object.assert_called_once()
self.assertIn("Successfully uploaded", response["Status"])
def test_delete_from_bucket(self):
mock_objects = [
MagicMock(object_name="file1.txt"),
MagicMock(object_name="file2.txt")
]
self.mock_minio_client.list_objects.return_value = mock_objects
deleted_count = self.s3_service.delete_from_bucket("folder/", self.trace_id)
self.assertEqual(deleted_count, 2)
self.mock_minio_client.remove_object.assert_any_call(self.mock_config["s3"][self.env]["BUCKET_NAME"], "file1.txt")
self.mock_minio_client.remove_object.assert_any_call(self.mock_config["s3"][self.env]["BUCKET_NAME"], "file2.txt")
def test_merge_strings(self):
result = self.s3_service.merge_strings("prefix/", "/suffix")
self.assertEqual(result, "prefix/suffix")
result = self.s3_service.merge_strings("prefix123", "123suffix")
self.assertEqual(result, "prefix123suffix")
if __name__ == "__main__":
unittest.main()
0 Comments
Please Login to Comment Here