python - 我如何为使用的 EventStream 创建模拟，但 S3 Select API 用于基于 S3 Select 查询获取内容？

Question

我正在为使用 boto3 的 s3 客户端函数“select_object_content”从 S3 存储桶读取对象的函数创建单元测试。我想要模拟的响应是

{
    'Payload': EventStream({
        'Records': {
            'Payload': b'bytes'
        },
        'Stats': {
            'Details': {
                'BytesScanned': 123,
                'BytesProcessed': 123,
                'BytesReturned': 123
            }
        },
        'Progress': {
            'Details': {
                'BytesScanned': 123,
                'BytesProcessed': 123,
                'BytesReturned': 123
            }
        },
        'Cont': {},
        'End': {}
    })
}

Payload 是一个 EventStream 对象，创建为 EventStream(self, raw_stream, output_shape, parser, operation_name) 并有 4 个参数。我将 raw_stream 作为用“utf-8”编码的字节字符串，但我无法找到有关如何分配其他参数的更多信息。

我正在使用 MagicMock 模拟 s3_client.select_object_content。

我希望能够将 athena 结果(以 CSV 格式保存在 S3 中)作为流传递，并确保代码具有处理某些场景的单元测试。

编辑:我可以使用以下结构模拟响应:

我的 mock 函数的返回类型是 Dict[str, Any]

return {'Payload': [{
        'Records': {
            'Payload': b"some utf8 encoded byte stream"
        }},{
        'Records': {
            'Payload': b"some utf8 encoded byte stream"
        }}]}

Answer 1

有一个 open bug for this反对botocore。此时最好的选择是模拟整个函数而不依赖于 Stubber:

with patch.object(self.s3_client, 'select_object_content') as mock_select:

    mock_select.return_value = {
        "ResponseMetadata": ANY,
        "Payload": [{
            "Records": {
                "Payload": json.dumps(MANIFEST_DATA).encode()
            },
            "Stats": {},
            "End": {}
        }]
    }


    mock_select.assert_called_once_with(
        Bucket="test-bucket",
        Key=manifest_key,
        Expression=i"Select * from s3Object o",
        ExpressionType="SQL",
        InputSerialization={"JSON": {"Type": "LINES"}},
        OutputSerialization={"JSON": {}}
    )