Как я могу создать макет для EventStream, который используется, но S3 Select API для получения содержимого на основе запроса S3 Select? - PullRequest
Новая
       22

Как я могу создать макет для EventStream, который используется, но S3 Select API для получения содержимого на основе запроса S3 Select?

1 голос
/ 25 марта 2019

Я создаю модульный тест для функции, которая читает объекты из сегментов S3, используя клиентскую функцию s3 в boto3 'select_object_content'. Ответ, который я ищу, чтобы высмеять, 

{
    '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 (которые находятся в S3 как CSV) как поток и убедиться, что в коде есть модульные тесты для обработки определенных сценариев.

Редактировать: Я мог бы смоделировать ответ со следующей структурой:

Возвращаемым типом моей фиктивной функции является Dict [str, Any] 

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