模拟 boto3 客户端异常

Question

我试图模拟对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不会处理异常。

Answer 1

撇开moto不谈，使用 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())