python - 模拟 boto3 客户端异常
问题描述
我正在尝试模拟对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
除此之外,使用 mixja 的这种方法,可以通过一些努力从模拟的 boto3 函数中引发特定的 boto3 异常。
在继续之前,请注意存根不是一个好的选择,因为它通常需要~/.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())
推荐阅读
- openthread - openthread/environment docker rsyslogd:imklog:无法打开内核日志(/proc/kmsg):不允许操作
- python - 如何从 .pb 转换为 .tflite?
- deep-learning - 操作数不能与形状一起广播
- r - 将 lm 输出列表转换为数据帧
- spring - 认证后重定向到白级错误页面
- node.js - 计算一系列地址的掩码
- mysql - InnoDB二级索引包含值而不是指向PK的指针,怎么够?
- haskell - 使用 Aeson 时如何使用 sum 类型作为 map 中的键?
- node.js - 无法通过 Google Cloud Vision API 处理 PDF
- python - 如何使用 gitpython repo.archive() 作为标签