I am trying to mock a call to the get_paginator function on the Python AWS S3 Client. Here my production code:
handler.py
import boto3
class RealClass:
def __init__(self):
self.s3_client = boto3.client("s3")
def get_unprocessed_files(self) -> list[str]:
paginator = self.s3_client.get_paginator("list_objects_v2")
operation_parameters = {"Bucket": self.bronze_bucket, "Prefix": self.prefix}
page_iterator = paginator.paginate(**operation_parameters)
un_processed_files = []
for page in page_iterator:
for obj in page.get("Contents", []):
key = obj["Key"]
if key.endswith(".content.txt") or key.endswith(".metadata.json"):
un_processed_files.append(key)
return un_processed_files
test_handler.py
import unittest
from unittest.mock import patch, MagicMock
from handler import RealClass
class TestRealClass(unittest.TestCase):
def setUp(self) -> None:
self.real = RealClass()
@patch("boto3.client")
def test_get_unprocessed_files(self, mock_boto3_client):
response = [
{
"Contents": [
{"Key": "files/1001000284.txt"},
]
}
]
# What to do here?
result = self.pii.get_unprocessed_files()
self.assertIsInstance(result, list)
self.assertTrue(len(result) > 0)
self.assertTrue(result[0].find("1001000284") > -1)
All I get is The provided token has expired, which means I guess the real functions aren’t being mocked.
Thank you!
Mock Setup
You are missing the setup of your mock. Almost in all languages, after mock an object you need setup the desired behavior, in your case:
When get_paginator method is called, return this fake response
Maybe in more descriptive languages, you can understand it.
JAVA : When getBook() method is execute, return a fake book
C# In the following example the mock is being setup to return a list of Person (instead a real database) only when GetPeople() method is executed
mock_boto3_client.get_paginator.return_value
I didn’t test this code, but I think is what you need
import unittest
from datetime import datetime
from unittest.mock import patch
# real handler
import real_code
class TestRealClass(unittest.TestCase):
@patch('boto3.client')
def test_get_unprocessed_files(self, mock_boto3_client):
desired_response = {...}
mock_boto3_client.return_value = mock_boto3_client
mock_boto3_client.get_paginator.return_value = desired_response
result_value = real_code.get_unprocessed_files("MyTestParameterName")
References
- /a/39553581/3957754
- How to properly patch boto3 calls in unit test
Advice
If you can share us a reproducible hello world code to help you