0

我正在尝试设置对 kinesis firehose 的模拟服务调用。我正在从 moto 导入 mock_firehose 并将其引用为 @mock_firehose。在测试方法中，我使用 boto3 创建了一个客户端。

@mock_firehose
def test_push_to_stream(push_record, stream):
    ret = app.push_to_stream(push_record, stream)
    client = boto3.client('firehose', region_name='us-west-2)

我已导出要使用的 AWS_PROFILE 并检查了凭据是否正确。我遇到的错误是：

botocore.errorfactory.ResourceNotFoundException: An error occurred (ResourceNotFoundException) when calling the PutRecord operation: Firehose sample-name under account 123456789012 not found.

显然，虚拟账户 123456789012 是针对模拟 AWS 服务运行测试的默认测试账户。我不确定是否需要为测试帐户创建一个流，但这是有道理的。如果我注释掉 boto3.client 行并且在方法上方仅使用 @mock_firehose 似乎会失败。是否有我缺少的设置步骤要求我在调用@mock_firehose 之前初始化流？

4

1 回答 1

0

Moto 用于拦截对 AWS 的任何调用。Moto 对请求进行解码，找出您要执行的操作，并在内存中保留所需基础架构的副本。因此，使用 Moto 应该让您看起来像是在与 AWS 交谈，但没有与之相关的成本。

虚拟123456789012帐户用于表明我们正在针对 Moto 运行，并确保它不会意外改变任何真实的基础设施。

ResourceNotFound-exception 实际上来自这里的 Moto 。它知道您可以访问此测试帐户，但它确实知道该流 - 因为尚未创建任何内容。
（AWS 可能会返回 AccessDenied 错误，表示您无权访问虚拟帐号。）

考虑到这一点：

我已导出要使用的 AWS_PROFILE 并检查了凭据是否正确。我遇到的错误是：

凭证不必正确，因为永远无法访问 AWS。
更进一步：凭证不应该是正确的，以确保永远无法访问 AWS。

我不确定是否需要为测试帐户创建流

是的你应该。
使用 Moto 进行单元测试的一般流程如下所示：

  1. 在 Moto 中设置您期望在 AWS 中存在的等效基础设施
  2. 针对 Moto 运行业务逻辑
  3. 根据需要断言

因此，对于您的用例，它大致如下所示：

@mock_firehose
def test_push_to_stream(push_record, stream):
    # Setup architecture that exists in Prod
    client = boto3.client('firehose', region_name='us-west-2)
    client.create_delivery_stream(...)

    # Run business logic
    ret = app.push_to_stream(push_record, stream)

    # Verify that the records exist
    ...

作为实现提示：
目前，Moto 仅在端点是 S3 或 HTTP 时发送 Firehose 记录。到其他端点的记录，例如Elasticsearch和Redshift尚未处理。

因此，将您的交付流设置为交付到 S3 是有意义的，因为这样可以轻松验证业务逻辑是否发送了正确的记录。

该文档有助于查看有关如何正确安全地使用 Moto 的更多使用示例/想法：http ://docs.getmoto.org/en/latest/docs/getting_started.html#moto-usage

于 2022-02-04T10:59:51.887 回答