Mocking boto3 client response

Question

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"])