Издевательский ответ клиента boto3 ⇐ Python

[Anonymous]

Я пытаюсь создать модульные тесты для вызовов dynamodb. Мне нужно точно указать, что будет отвечать экземпляр клиента — чтобы реплицировать ответ об ошибке от AWS. Я могу принудительно отреагировать на сбой, который возникает в пакете boto, например, на недопустимые типы параметров. Однако я хочу, чтобы успешный ответ json был смоделирован.
Вот код
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

Я хочу проверить ошибку, которая возникает, когда HTTPStatusCode не равен 200.
Вот последняя попытка имитировать ответ< /p>
@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)

Я попробовал несколько разных методов из stackoverflow и даже использовал рекомендации GPT.
Одно из таких предложений — макетировать весь ресурс, например, вот так self.test_user_repo.table = table = Mock()
table.put_item.side_effect = Exception('Boto3 Exception')

Думаю, мне не удалось заставить это работать, поскольку я использую клиент boto3, а не ресурс, как в фрагменте решения.
Я просто нужно проверить это условие
if apps.get("ResponseMetadata", {}).get("HTTPStatusCode") != 200:
raise Exception(apps["ResponseMetadata"])


Подробнее здесь: https://stackoverflow.com/questions/791 ... t-response