Skip to content

Instantly share code, notes, and snippets.

@realFranco
Created December 3, 2022 17:55
Show Gist options
  • Save realFranco/7c3110c4efd2660f21d887b3b42b94a9 to your computer and use it in GitHub Desktop.
Save realFranco/7c3110c4efd2660f21d887b3b42b94a9 to your computer and use it in GitHub Desktop.
Apply a unit test at the AWS S3 service `get_object`. #aws #boto3 #botocore
"""
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
"""
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'])
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment