Apply a unit test at the AWS S3 service `get_object`. #aws #boto3 #botocore

mock_s3.py

"""
This code doesn't belongs to @realFranco.

See: https://stackoverflow.com/questions/72253062/write-unit-tests-for-python-lambda-using-botocore-stub-stubber
"""
import json

import boto3


def data_import(bucket_name, url_key):
    s3_obj = boto3.client('s3')

    s3_clientobj = s3_obj.get_object(Bucket = bucket_name,
                                    Key = url_key)
    json_data = s3_clientobj['Body'].read().decode('utf-8')
    json_content = json.loads(json_data)

    return json_content

mock_s3_test.py

"""
This codes belongs to @realFranco.

# Instructions for installation:
> python3 -m venv env

> source env/bin/activate

> pip3 install -r requirements.txt

# Execute the test:
python3 -m pytest test/ -sv
"""
from unittest import TestCase, mock

import boto3
from botocore.stub import Stubber
from botocore.response import StreamingBody

from service.mock_s3 import data_import


class TestDataImport(TestCase):
    """Test the function `data_import`."""

    AwsRegion: str = 'us-east-1'
    AwsAccessKey: str = 'access-key'
    AwsSecretKey: str = 'secret-key'
    AwsServiceName: str = 'S3'

    # AWS S3 Resources.
    AwsS3BucketName: str = 'my-bucket'
    AwsS3UrlKey:str = 'my-url-key'

    def setUp(self) -> None:
        """Request resources before the unit test execution."""
        self.__botoClientS3 = boto3.client(
            self.AwsServiceName.lower(),
            aws_access_key_id=self.AwsAccessKey,
            aws_secret_access_key=self.AwsSecretKey,
            region_name=self.AwsRegion,
            endpoint_url=f'https://{self.AwsServiceName.lower()}.{self.AwsRegion}.amazonaws.com'
        )

        self.__botoClientStubberS3 = Stubber(self.__botoClientS3)
        self.__botoClientStubberS3.activate()

    def tearDown(self) -> None:
        """Release requested resoruces after the unit test execution."""
        self.__botoClientStubberS3.deactivate()

    def mocksS3GetObject(self) -> None:
        """
        Mocks the AWS S3 service `get_object`.

        :see: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Client.get_object
        """
        s3GetObjectResponse = {
            'Body': StreamingBody(raw_stream=None, content_length=None),
            # Some other properties.
            'Metadata': {'context': 'stackOv'}
        }

        self.__botoClientStubberS3.add_response(
            method='get_object',
            service_response=s3GetObjectResponse,
            expected_params={'Bucket': self.AwsS3BucketName, 'Key': self.AwsS3UrlKey}
        )

    @mock.patch('botocore.response.StreamingBody.read')
    @mock.patch('boto3.client')
    def testDataImport(self, mockBoto3: mock.MagicMock, mockStreamingBodyRead: mock.MagicMock) -> None:
        """Test `data_import` function."""
        mockBoto3.return_value = self.__botoClientS3

        # It's not required to mock the `StreamingBody.read` public method, but at this point in the time
        # I am not sure how to create an instance from the object `StreamingBody`, for that reason
        # I mock it, to return the bytes.
        mockStreamingBodyRead.return_value = b'{"bucketName": "my-bucket"}'

        self.mocksS3GetObject()

        dataImport = data_import(bucket_name=self.AwsS3BucketName,  url_key=self.AwsS3UrlKey)

        self.assertEqual(self.AwsS3BucketName, dataImport['bucketName'])