I'm attempting to write unit tests for a Python function that downloads a directory from S3 using Boto3. However, I'm facing a TypeError: 'Mock' object is not iterable issue when trying to simulate paginated behavior within my tests. Here's the relevant code:
S3Client Function (download_directory):
DOWNLOAD_DIRECTORY_OPERATION_NAME = "Download_Directory"
def download_directory(
self, s3_bucket_name: str, s3_directory_path: str, local_directory_path: str
) -> None:
if not s3_directory_path.endswith("/"):
s3_directory_path += "/"
# Ensure the local directory exists
os.makedirs(local_directory_path, exist_ok=True)
# List objects using paginator
paginator = self.s3_client.get_paginator("list_objects_v2").paginate(
Bucket=s3_bucket_name, Prefix=s3_directory_path
)
for page in paginator:
for obj in page.get("Contents", []):
s3_object_key = obj["Key"]
if not s3_object_key.endswith("/"): # Skip directories
relative_path = s3_object_key[len(s3_directory_path) :]
full_local_path = os.path.join(local_directory_path, relative_path)
os.makedirs(os.path.dirname(full_local_path), exist_ok=True)
# Use the operation wrapper for downloading each file
self._s3_operation_wrapper(
lambda: self.s3_client.download_file(
s3_bucket_name, s3_object_key, full_local_path
),
DOWNLOAD_DIRECTORY_OPERATION_NAME,
)
Unit Test:
@patch("boto3.client")
def test_download_directory_success(self, mock_boto3_client):
s3_directory_path = "sample-dir/"
local_directory_path = "/tmp/test"
bucket_name = "mock-bucket-name"
# Setup mock response for the paginator
mock_response = {
"Contents": [
{"Key": f"{s3_directory_path}file1.txt"},
{"Key": f"{s3_directory_path}file2.csv"},
]
}
# Setup mock paginator to return the mock response as an iterator
mock_paginator = MagicMock()
mock_paginator.paginate.return_value = iter([mock_response]) # Correctly returns an iterator
mock_boto3_client.return_value.get_paginator.return_value = mock_paginator
# Patch the _s3_operation_wrapper to prevent actual download operations
with patch.object(self.s3_client, '_s3_operation_wrapper', autospec=True) as mock_wrapper:
# Call the method under test
self.s3_client.download_directory(bucket_name, s3_directory_path, local_directory_path)
# Assertions to verify that _s3_operation_wrapper was called correctly
expected_calls = [
call(
lambda: self.s3_client.s3_client.download_file(
bucket_name, f"{s3_directory_path}file1.txt", f"{local_directory_path}/file1.txt"
),
'Download_Directory'
),
call(
lambda: self.s3_client.s3_client.download_file(
bucket_name, f"{s3_directory_path}file2.csv", f"{local_directory_path}/file2.csv"
),
'Download_Directory'
)
]
mock_wrapper.assert_has_calls(expected_calls, any_order=True)
Stack trace:
=================================== FAILURES ===================================
_________________ TestS3Client.test_download_directory_success _________________
self = <test.audio_video_processing_utilities.storage_access.test_s3_client.TestS3Client testMethod=test_download_directory_success>
mock_boto3_client = <MagicMock name='client' id='140601653634912'>
@patch('audio_video_processing_utilities.storage_access.s3_client.boto3.client')
def test_download_directory_success(self, mock_boto3_client):
s3_directory_path = "sample-dir/"
local_directory_path = "/tmp/test"
bucket_name = "mock-bucket-name"
# Setup mock response for the paginator
mock_response = {
"Contents": [
{"Key": f"{s3_directory_path}file1.txt"},
{"Key": f"{s3_directory_path}file2.csv"},
]
}
# Setup mock paginator to return the mock response as an iterator
mock_paginator = MagicMock()
mock_paginator.paginate.return_value = iter([mock_response]) # Correctly returns an iterator
mock_boto3_client.return_value.get_paginator.return_value = mock_paginator
# Patch the _s3_operation_wrapper to prevent actual download operations
with patch.object(self.s3_client, '_s3_operation_wrapper', autospec=True) as mock_wrapper:
# Call the method under test
> self.s3_client.download_directory(bucket_name, s3_directory_path, local_directory_path)
test/audio_video_processing_utilities/storage_access/test_s3_client.py:122:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <audio_video_processing_utilities.storage_access.s3_client.S3Client object at 0x7fe05f9ea5b0>
s3_bucket_name = 'mock-bucket-name', s3_directory_path = 'sample-dir/'
local_directory_path = '/tmp/test'
def download_directory(
self, s3_bucket_name: str, s3_directory_path: str, local_directory_path: str
) -> None:
"""
Downloads all files from a specified S3 path to a local directory.
Args:
s3_bucket_name: The name of the S3 bucket.
s3_path: The S3 directory path to download files from.
local_file_path: The local directory path to download the files to.
"""
"""
Downloads all files from a specified S3 path to a local directory.
Args:
s3_bucket_name: The name of the S3 bucket.
s3_path: The S3 directory path to download files from.
local_directory_path: The local directory path to download the files to.
"""
if not s3_directory_path.endswith("/"):
s3_directory_path += "/"
# Ensure the local directory exists
os.makedirs(local_directory_path, exist_ok=True)
# List objects using paginator
paginator = self.s3_client.get_paginator("list_objects_v2").paginate(
Bucket=s3_bucket_name, Prefix=s3_directory_path
)
> for page in paginator:
E TypeError: 'Mock' object is not iterable
Troubleshooting Attempts
- I've verified that my logic uses the paginator and iterates over the 'pages'.
- I've ensured my test patch setup utilizes
iter()directly to make the simulated response iterable inmock_paginator.paginate.return_value. - I've looked into mocking individual 'page' contents but faced challenges in getting these nested mocks to work correctly.
Could someone please assist in identifying where my mocking might be insufficient and provide the correct approach to simulate the Boto3 paginated response?