@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
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.
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")
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