test_s3_service

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()

Posted By