title: How to mock functionality of boto3 module using pytest
body: <p>I have a custom module written called sqs.py. The script will do the following:</p>
  
  <ol>
  <li>Get a message from AWS SQS</li>
  <li>Get the AWS S3 path to delete</li>
  <li>Delete the path</li>
  <li>Send a confirmation email to the user</li>
  </ol>
  
  <p>I'm trying to write unit tests for this module that will verify the code will execute as expected and that it will raise exceptions when they do occur.</p>
  
  <p>This means I will need to mock the response from Boto3 calls that I make. My problem is that the code will first establish the SQS client to obtain the message and then a second call to establish the S3 client. I'm not sure how to mock these 2 independent calls and be able to fake a response so I can test my script's functionality. Perhaps my approach is incorrect. At any case, any advice on how to do this properly is appreciated.</p>
  
  <p>Here's how the code looks like:</p>
  
  <pre><code>import boto3
  import json
  import os
  import pprint
  import time
  import asyncio
  import logging
  from send_email import send_email
  
  queue_url = 'https://xxxx.queue.amazonaws.com/1234567890/queue'
  
  def shutdown(message):
      """ Sends shutdown command to OS """
      os.system(f'shutdown +5 "{message}"')
  
  def send_failure_email(email_config: dict, error_message: str):
      """ Sends email notification to user with error message attached. """
      recipient_name = email_config['recipient_name']
      email_config['subject'] = 'Subject: Restore Failed'
      email_config['message'] = f'Hello {recipient_name},\n\n' \
                             + 'We regret that an error has occurred during the restore process. ' \
                             + 'Please try again in a few minutes.\n\n' \
                             + f'Error: {error_message}.\n\n' \
      try:
          send_email(email_config)
      except RuntimeError as error_message:
          logging.error(f'ERROR: cannot send email to user. {error_message}')
  
  async def restore_s3_objects(s3_client: object, p_bucket_name: str, p_prefix: str):
      """Attempts to restore objects specified by p_bucket_name and p_prefix.
  
      Returns True if restore took place, false otherwise.
      """
  
      is_truncated = True
      key_marker = None
      key = ''
      number_of_items_restored = 0
      has_restore_occured = False
      logging.info(f'performing restore for {p_bucket_name}/{p_prefix}')
      try:
          while is_truncated == True:
              if not key_marker:
                  version_list = s3_client.list_object_versions(
                      Bucket = p_bucket_name,
                      Prefix = p_prefix)
              else:
                  version_list = s3_client.list_object_versions(
                      Bucket = p_bucket_name,
                      Prefix = p_prefix,
                      KeyMarker = key_marker)
  
              if 'DeleteMarkers' in version_list:
                  logging.info('found delete markers')
                  delete_markers = version_list['DeleteMarkers']
                  for d in delete_markers:
                      if d['IsLatest'] == True:
                          key = d['Key']
                          version_id = d['VersionId']
  
                          s3_client.delete_object(
                              Bucket = p_bucket_name,
                              Key = key,
                              VersionId = version_id
                          )
                          number_of_items_restored = number_of_items_restored + 1
  
              is_truncated = version_list['IsTruncated']
              logging.info(f'is_truncated: {is_truncated}')
  
              if 'NextKeyMarker' in version_list:
                  key_marker = version_list['NextKeyMarker']
  
          if number_of_items_restored &gt; 0:
              has_restore_occured = True
  
          return has_restore_occured
  
      except Exception as error_message:
          raise RuntimeError(error_message)
  
  async def main():
      if 'AWS_ACCESS_KEY_ID' in os.environ \
              and 'AWS_SECRET_ACCESS_KEY' in os.environ \
              and os.environ['AWS_ACCESS_KEY_ID'] != '' \
              and os.environ['AWS_SECRET_ACCESS_KEY'] != '':
                  sqs_client = boto3.client(
                      'sqs',
                      aws_access_key_id=os.environ['AWS_ACCESS_KEY_ID'],
                      aws_secret_access_key=os.environ['AWS_SECRET_ACCESS_KEY'],
                      verify=False
                  )
                  s3_client = boto3.client(
                      's3',
                      aws_access_key_id=os.environ['AWS_ACCESS_KEY_ID'],
                      aws_secret_access_key=os.environ['AWS_SECRET_ACCESS_KEY'],
                      verify=False
                  )
      else:
          sqs_client = boto3.client(
              'sqs',
              verify=False,
          )
          s3_client = boto3.client(
              's3',
              verify=False,
          )
  
      received_message = sqs_client.receive_message(
          QueueUrl=queue_url,
          AttributeNames=['All'],
          VisibilityTimeout=10,
          WaitTimeSeconds=20,    # Wait up to 20 seconds for a message to arrive
      )
  
      if 'Messages' in received_message \
          and len(received_message['Messages']) &gt; 0:
          # NOTE: Initialize email configuration
          receipient_email = 'support@example.com'
          username = receipient_email.split('@')[0]
          fullname_length = len(username.split('.'))
          fullname = f"{username.split('.')[0]}" # Group name / First name only
  
          if (fullname_length == 2): # First name and last name available
              fullname = f"{username.split('.')[0]} {username.split('.')[1]}"
  
          fullname = fullname.title()
  
          email_config = {
              'destination': receipient_email,
              'recipient_name': fullname,
              'subject': 'Subject: Restore Complete',
              'message': ''
          }
  
          try:
              receipt_handle = received_message['Messages'][0]['ReceiptHandle']
          except Exception as error_message:
              logging.error(error_message)
              send_failure_email(email_config, error_message)
              shutdown(f'{error_message}')
  
          try:
              data = received_message['Messages'][0]['Body']
              data = json.loads(data)
              logging.info('A SQS message for a restore has been received.')
          except Exception as error_message:
              message = f'Unable to obtain and parse message body. {error_message}'
              logging.error(message)
              send_failure_email(email_config, message)
              shutdown(f'{error_message}')
  
          try:
              bucket = data['bucket']
              prefix = data['prefix']
          except Exception as error_message:
              message = f'Retrieving bucket name and prefix failed. {error_message}'
              logging.error(message)
              send_failure_email(email_config, message)
              shutdown(f'{error_message}')
  
          try:
              logging.info(f'Initiating restore for path: {bucket}/{prefix}')
              restore_was_performed = await asyncio.create_task(restore_s3_objects(s3_client, bucket, prefix))
  
              if restore_was_performed is True:
                  email_config['message'] = f'Hello {fullname},\n\n' \
                                      + f'The files in the path \'{bucket}/{prefix}\' have been restored. ' \
  
                  send_email(email_config)
                  logging.info('Restore complete. Shutting down.')
              else:
                  logging.info('Path does not require restore. Shutting down.')
              shutdown(f'shutdown +5 "Restore successful! System will shutdown in 5 mins"')
  
          except Exception as error_message:
              message = f'File restoration failed. {error_message}'
              logging.error(message)
              send_failure_email(email_config, message)
              shutdown(f'{error_message}')
  
          try:
              sqs_client.delete_message(
                  QueueUrl=queue_url,
                  ReceiptHandle=receipt_handle,
              )
          except Exception as error_message:
              message = f'Deleting restore session from SQS failed. {error_message}'
              logging.error(message)
              send_failure_email(email_config, message)
              shutdown(f'{error_message}')
  
  if __name__ == '__main__':
      logging.basicConfig(filename='restore.log',level=logging.INFO)
      loop = asyncio.get_event_loop()
      loop.run_until_complete(main())
      loop.close()
  </code></pre>