Search code examples
pythonboto3python-unittestbotocore

UnrecognizedClientException when using botocore stubber


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.


Solution

  • 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 clienttomy_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.