模拟 boto3 客户端异常答案

【问题标题】：Mock a boto3 client exception模拟 boto3 客户端异常
【发布时间】：2020-10-06 05:41:30
【问题描述】：

我正在尝试模拟对boto3.client("cognito-idp") 中的函数的调用。我已经能够成功地模拟正常的代码流，但事实证明异常更加棘手。我有类似的东西：

@patch("boto3.client")
def test_failure_cannot_create_duplicate_user_in_cognito(self, mock_client):
  client = boto3.client("cognito-idp")
  boto3_mock = MagicMock()
  boto3_mock.admin_create_user.side_effect = client.exceptions.UsernameExistsException()
  mock_client.return_value = boto3_mock
  response = handler(event, None)
  body = json.loads(response["body"])
  self.assertEqual(HTTPStatus.CONFLICT, response["statusCode"])

然后在处理程序中：

client = boto3.client("cognito-idp")
try:
    create_user_response = client.admin_create_user(
      UserPoolId=user_pool,
      Username=user_name,
      UserAttributes=[
        {
           "Name": user_attribute,
           "Value": user_name
        },
        {
            "Name": verify,
            "Value": "True"
        },
      ],
      TemporaryPassword=temp_pass,
            DesiredDeliveryMediums=[delivery],
    )
    cognito_user_name = create_user_response["User"]["Username"]
except client.exceptions.UsernameExistsException:
   return conflict(Errors.DUPLICATE_USER)

except 没有处理异常。

【问题讨论】：

或许使用moto可以帮到你

标签： python mocking boto3 amazon-cognito

【解决方案1】：

除了moto，还可以通过使用this approach by mixja从模拟的boto3函数引发特定的boto3异常。

在继续之前，请注意 stubber 不是一个好的选择，因为它通常需要存在 ~/.aws/ 配置文件，并且它们通常不存在于 CI（持续集成）环境中。

这是使用上述方法的boto3.client('glue') 的示例。它可以适应其他 AWS 服务和例外情况。

功能：

def ensure_crawler(**kwargs: Any) -> None:
    client = boto3.client("glue")
    try:
        response = client.create_crawler(**kwargs)
    except client.exceptions.AlreadyExistsException:
        response = client.update_crawler(**kwargs)
    assert response["ResponseMetadata"]["HTTPStatusCode"] == 200

测试：

import botocore.errorfactory
import botocore.session

_GLUE_MODEL = botocore.session.get_session().get_service_model("glue")
_GLUE_FACTORY = botocore.errorfactory.ClientExceptionsFactory()
_GLUE_EXCEPTIONS = _GLUE_FACTORY.create_client_exceptions(_GLUE_MODEL)


class TestEnsureCrawler(unittest.TestCase):
    def setUp(self) -> None:
        self.crawler = "abc-crawler"
        self.mock_boto3_client = patch("boto3.client").start()

    def test_create_crawler(self):
        self.mock_boto3_client.return_value.create_crawler.return_value = {"ResponseMetadata": {"HTTPStatusCode": 200}}
        kwargs = dict(Name=self.crawler, Targets={"S3Targets": [{"Path": "s3://foo/bar"}]})
        ensure_crawler(**kwargs)

        self.mock_boto3_client.return_value.create_crawler.assert_called_once()
        call_args = self.mock_boto3_client.return_value.create_crawler.call_args
        self.assertEqual(call_args.args, ())
        self.assertLessEqual(kwargs.items(), call_args.kwargs.items())

    def test_update_crawler(self):
        self.mock_boto3_client.return_value.exceptions.AlreadyExistsException = _GLUE_EXCEPTIONS.AlreadyExistsException
        self.mock_boto3_client.return_value.create_crawler.side_effect = _GLUE_EXCEPTIONS.AlreadyExistsException({}, "")
        self.mock_boto3_client.return_value.update_crawler.return_value = {"ResponseMetadata": {"HTTPStatusCode": 200}}
        kwargs = dict(Name=self.crawler, Targets={"S3Targets": [{"Path": "s3://foo/bar"}]})
        ensure_crawler(**kwargs)

        self.mock_boto3_client.return_value.update_crawler.assert_called_once()
        call_args = self.mock_boto3_client.return_value.update_crawler.call_args
        self.assertEqual(call_args.args, ())
        self.assertLessEqual(kwargs.items(), call_args.kwargs.items())

【讨论】：