'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 |
|---|
