'write unit tests for python lambda using botocore.stub.Stubber()

I have the python lambda function which import data from s3 bucket with bucketname and key as parameter input. I want to use botocore.stub.Stubber() to mock S3 client and write unit test on this method.

this following code is my lambda function that i want to write unit test on:

Question 1: Now this test case is expecting me to input a real S3 bucket name and key, which i don't want, because the purpose of using mock s3 is to avoid hitting real resources and use real bucket name and key. but i don't understand why it's expecting me to have real bucketname/key defined? coz otherwise it's throwing me "no such bucket" error. which part of this mocking s3 client of this botocore stub usage is wrong?

Question 2: is this the right way to test data_import()? i'm only testing the get_object() method, not the data_import() itself, if i want to test data_import(), how should I modify the add_response() part of this code?

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

my unit test that I wrote is as following:

    def test_data_import(self):

    # mock S3 client:
    s3 = botocore.session.get_session().create_client('s3')
    stubber = Stubber(s3)

    # mock client returned response:

    r = {...} # example json 
    mocked_response_payload = json.dumps(r).encode("utf-8")

    response = {
        'Body': StreamingBody(io.BytesIO(mocked_response_payload), len(mocked_response_payload))
    }

    expected_params = {'Bucket': self.bucket_name, 'Key': self.url_key}

    # stubber.add_response('get_object', response, expected_params)
    self.stubber.add_response('get_object', response, expected_params)
    self.stubber.activate()

    with self.stubber:
        service_response = self.s3.get_object(Bucket = self.bucket_name, Key = self.url_key)

    assert service_response == response


    try:
        data_response = handler.data_import(self.bucket_name, self.url_key)
    except:
        self.fail("Lambda called failed with errors/exceptions.")


    # 1.
    self.assertFalse(data_response == r)

    # 2.
    # Asserts that all expected calls were made:
    self.stubber.assert_no_pending_responses()

    # self.assertRaises(IOError, data_response['Body'].read())
    # self.assertRaises(UnicodeDecodeError, data_response['Body'].read().decode('utf-8'))

    # 3. assert response not empty:
    self.assertNotEqual(data_response, [])
    # self.assertEqual(errors, False)


Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source