模拟boto3客户端异常

2024-03-29 12:34:28 发布

您现在位置:Python中文网/ 问答频道 /正文

我试图模拟对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没有处理异常


Tags: nameselfclientreturnadminresponsecreateidp
1条回答
网友
1楼 · 发布于 2024-03-29 12:34:28

^{}除此之外,通过一些努力,可以使用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())

相关问题 更多 >