pythonpython-unittestpatchmagicmock

Python unit testing with patch not working for s3.head_object


@patch('scripts.s3_objects.boto3.client')  
# Patch where boto3.client is used 
def test_check_s3_uri_exists(self, mock_boto_client): 
    # Create a mock S3 client 
    mock_s3 = MagicMock() 
    mock_boto_client.return_value = mock_s3  
    # Test case where the S3 object exists
    mock_s3.head_object.return_value = {}
    result = check_s3_uri_exists('bucket-name', 'prefix/file')
    self.assertTrue(result, "Expected True when S3 object exists")
    
    # Test case where the S3 object does not exist (404 error)
    mock_s3.head_object.side_effect = ClientError(
        {'Error': {'Code': '404', 'Message': 'Not Found'}},
            'HeadObject'
    )
    result = check_s3_uri_exists('bucket-name', 'prefix/file')
    self.assertFalse(result, "Expected False when S3 object does not exist")


def check_s3_uri_exists(bucket_name, key):
    """Check if the S3 URI exists."""
    s3_client = boto3.client('s3')
    try:
        print(f"Checking bucket: {bucket_name}, key: {key}")
        s3_client.head_object(Bucket=bucket_name, Key=key)
        return True
    except s3_client.exceptions.ClientError as e:
        print(f"Exception caught: {e}")
        if e.response['Error']['Code'] == '404':
            return False
        raise

scripts is the folder and s3_objects is the python code where check_s3_uri_exists defined.

The unit testing not working as expected. side_effect has no effect and getting "True" for second call to check_s3_uri_exists.

Tried all options and no luck


Solution

  • The issue arises from using the incorrect exception class for ClientError. Instead of s3_client.exceptions.ClientError, you should use botocore.exceptions.ClientError. This change is needed in both your test and the check_s3_uri_exists function.

    Updated Code

    from unittest.mock import patch, MagicMock
    from botocore.exceptions import ClientError
    
    @patch('scripts.s3_objects.boto3.client')  
    def test_check_s3_uri_exists(self, mock_boto_client): 
        # Create a mock S3 client 
        mock_s3 = MagicMock() 
        mock_boto_client.return_value = mock_s3  
        
        # Test case where the S3 object exists
        mock_s3.head_object.return_value = {}
        result = check_s3_uri_exists('bucket-name', 'prefix/file')
        self.assertTrue(result, "Expected True when S3 object exists")
        
        # Test case where the S3 object does not exist (404 error)
        mock_s3.head_object.side_effect = ClientError(
            {'Error': {'Code': '404', 'Message': 'Not Found'}},
                'HeadObject'
        )
        result = check_s3_uri_exists('bucket-name', 'prefix/file')
        self.assertFalse(result, "Expected False when S3 object does not exist")
    

    Function to be Tested

    import boto3
    from botocore.exceptions import ClientError
    
    def check_s3_uri_exists(bucket_name, key):
        """Check if the S3 URI exists."""
        s3_client = boto3.client('s3')
        try:
            print(f"Checking bucket: {bucket_name}, key: {key}")
            s3_client.head_object(Bucket=bucket_name, Key=key)
            return True
        except ClientError as e:
            print(f"Exception caught: {e}")
            if e.response['Error']['Code'] == '404':
                return False
            raise