I am using unittest
to test a function that makes a call to AWS using boto3
.
The function looks like this:
import boto3
def my_function():
client = boto3.client('athena')
res = client.start_query_exeuction(
QueryString='SELECT * FROM logs',
ResultConfiguration={'OutputLocation': 's3://mybucket'}
)
return res['QueryExecutionId']
I am using botocore stubber to stub this request in my unit tests like this:
from botocore.stub import Stubber
import botocore.session
def test_my_function():
client = botocore.session.get_session().create_client('athena')
client_res = {'QueryExecutionId': 'testid'}
exp_params = {
'QueryString': 'SELECT * FROM logs',
'ResultConfiguration': {
'OutputLocation': 's3://mybucket'
}
}
with Stubber(client) as stubber:
stubber.add_response('start_query_execution', client_res, exp_params)
res = my_function()
self.assertEqual(res, 'testid')
This test is failing with
botocore.exceptions.ClientError: An error occurred (UnrecognizedClientException) when calling the StartQueryExecution operation: The security token included in the request is invalid.
Why would this be failing? Is it because I am creating a new client in my_function()
which is different from the client used in the stubber? If so, how can I test this?
Any help is much appreciated.
Currently, my_function()
is creating a new client, and using that instead of stubber
.
One option would be to alter my_function
to take _client
as an argument.
def my_function(_client=None):
if _client is not None:
client = _client
else:
client = boto3.client('athena')
res = client.start_query_exeuction(
QueryString='SELECT * FROM logs',
ResultConfiguration={'OutputLocation': 's3://mybucket'}
)
return res['QueryExecutionId']
Then pass the stubbed clientto
my_function`.
with Stubber(client) as stubber:
stubber.add_response('start_query_execution', client_res, exp_params)
res = my_function(_client=client)
Another option would be to use mock to patch boto.client
to return your stubber.