Reputation: 11
I am trying to build unit tests for my calls to dynamodb. I need to specify exactly what the client instance will respond with - to replicate a failure response from AWS. I can force failure responses that occur within the boto package- such as invalid parameter types. What I want though is the successful json response to be mocked out.
Here is the code
session = boto3.Session(profile_name=os.getenv("profile_name"))
client = session.client("dynamodb")
def get_owned_apps(owner) -> list:
try:
apps = client.query(
TableName="pingone-developer-self-service-ownership",
KeyConditionExpression="PersonnelIdentifier = :PersonnelIdentifier",
ExpressionAttributeValues={":PersonnelIdentifier": {"S": owner}},
)
if apps.get("ResponseMetadata", {}).get("HTTPStatusCode") != 200:
raise Exception(apps["ResponseMetadata"])
except Exception as e:
raise DBReadFailure(f"Failed to query owned applications: {e}")
cleaned_apps = []
for item in apps["Items"]:
item_dict = {k: item[k]["S"] for k, v in item.items()}
cleaned_apps.append(item_dict)
return cleaned_apps
I am wanting to test against the failure that is raised when HTTPStatusCode is not 200.
Here is the latest attempt to mock out the response
@mock_aws
class TestGetOwnedApps(TestCase):
def setUp(self):
self.conn = boto3.client('dynamodb')
self.conn.create_table(
TableName='pingone-developer-self-service-ownership',
AttributeDefinitions=[
{
'AttributeName': 'AppId',
'AttributeType': 'S',
},
{
'AttributeName': 'PersonnelIdentifier',
'AttributeType': 'S',
},
],
KeySchema=[
{
'AttributeName': 'PersonnelIdentifier',
'KeyType': 'HASH',
},
{
'AttributeName': 'AppId',
'KeyType': 'RANGE',
},
],
ProvisionedThroughput={
'ReadCapacityUnits': 5,
'WriteCapacityUnits': 5,
},
)
self.conn.batch_write_item(
RequestItems=table_data_bulk_add
)
def test_get_owned_apps(self):
pid = "12345678"
owned_app_ids = ["973182465", "987654321"]
apps = db.get_owned_apps(pid)
self.assertEqual(len(apps), 2)
for app in apps:
self.assertIsInstance(app, dict)
self.assertEqual(app.get('PersonnelIdentifier'), pid)
self.assertIn(app.get('AppId'), owned_app_ids)
@patch('src.dss.helpers.db.boto3.session.Session.client')
def test_get_owned_apps_fail(self, mock_boto_client):
mock_client_instance = mock_boto_client.return_value
mock_client_instance.query.side_effect = ClientError(
{
'Error': {
'Code': 'InternalServerError',
'Message': 'An internal server error occurred.'
}
},
'Query'
)
with self.assertRaises(DBReadFailure):
db.get_owned_apps(12345678)
I have tried a few different methods from stackoverflow and even used guideance from GPT. One such suggestion is to mock out the entire resource such as like this
self.test_user_repo.table = table = Mock()
table.put_item.side_effect = Exception('Boto3 Exception')
I think I could not get this to work, as I am using a boto3 client and not a resource as in the solution snippet.
I just need to test this condition
if apps.get("ResponseMetadata", {}).get("HTTPStatusCode") != 200:
raise Exception(apps["ResponseMetadata"])
Upvotes: 1
Views: 84