aws_s3_client.py

from boto3 import client


class S3Client:
    __slots__ = ['boto_client', 'bucket_name', 'default_content_type', 'default_acl_type']
    boto_client: client
    bucket_name: str
    default_content_type: str
    default_acl_type: str

    def __init__(self, boto_client, bucket_name, default_content_type='text/plain', default_acl_type='private'):
        self.boto_client = boto_client
        self.bucket_name = bucket_name
        self.default_content_type = default_content_type
        self.default_acl_type = default_acl_type

    def upload(
        self, key: str, body: str, content_type: str = None, acl_type: str = None
    ):
        content_type = content_type if content_type else self.default_content_type
        acl_type = acl_type if acl_type else self.default_acl_type
        self.boto_client.put_object(
            Bucket=self.bucket_name,
            Body=body,
            ContentType=content_type,
            ACL=acl_type,
            Key=key,
        )

    def generate_presigned_url(self, object_name: str, expiration: str) -> str:
        params = {"Bucket": self.bucket_name, "Key": object_name}
        return self.boto_client.generate_presigned_url(
            "get_object", Params=params, ExpiresIn=expiration
        )

    def generate_url(self, object_key) -> str:
        return f"https://{self.bucket_name}.s3.amazonaws.com/{object_key}"

test_aws_s3_client.py

from unittest import TestCase
from unittest.mock import MagicMock

from app.clients.s3 import S3Client


class S3ClientTest(TestCase):
    def setUp(self):
        self.bucket_name = "bucket_name_for_tests"
        self.key = "name_of_object_for_tests"
        self.body_content = "content of file for tests"
        self.boto_client_mock = MagicMock()
        self.s3_client = S3Client(
            boto_client=self.boto_client_mock, bucket_name=self.bucket_name
        )

    def test_when_I_call_upload_with_correct_params_then_should_be_success(self):
        self.s3_client.upload(self.key, self.body_content)
        self.boto_client_mock.put_object.assert_called()

    def test_when_I_call_generate_presigned_url_with_correct_params_then_should_be_success(
        self
    ):
        self.s3_client.generate_presigned_url(self.key, 3600)
        self.boto_client_mock.generate_presigned_url.assert_called()

    def test_when_I_call_generate_url_with_correct_params_then_should_be_success(self):
        url = self.s3_client.generate_url(self.key)
        self.assertEquals(
            url,
            "https://bucket_name_for_tests.s3.amazonaws.com/name_of_object_for_tests",
        )