|
1 | 1 | from __future__ import annotations |
2 | 2 |
|
| 3 | +import gzip |
| 4 | +import io |
3 | 5 | from csv import DictReader |
4 | 6 | from gzip import GzipFile |
5 | 7 |
|
| 8 | +import boto3 |
| 9 | +import pytest |
| 10 | +from botocore import stub |
| 11 | + |
6 | 12 | from aws_lambda_powertools.utilities.streaming import S3Object |
7 | 13 | from aws_lambda_powertools.utilities.streaming._s3_seekable_io import _S3SeekableIO |
| 14 | +from aws_lambda_powertools.utilities.streaming.compat import PowertoolsStreamingBody |
8 | 15 | from aws_lambda_powertools.utilities.streaming.transformations import GzipTransform |
9 | 16 |
|
10 | 17 |
|
| 18 | +@pytest.fixture |
| 19 | +def s3_client(): |
| 20 | + return boto3.client("s3") |
| 21 | + |
| 22 | + |
| 23 | +@pytest.fixture |
| 24 | +def s3_client_stub(s3_client): |
| 25 | + with stub.Stubber(s3_client) as s3_stub: |
| 26 | + yield s3_stub |
| 27 | + |
| 28 | + |
11 | 29 | def test_s3_basic_stream(): |
12 | 30 | obj = S3Object(bucket="bucket", key="key") |
13 | 31 | assert type(obj.transformed_stream) is _S3SeekableIO |
@@ -40,3 +58,36 @@ def test_s3_transform_in_place(): |
40 | 58 |
|
41 | 59 | new_obj = obj.transform(GzipTransform(), in_place=True) |
42 | 60 | assert new_obj is None |
| 61 | + |
| 62 | + |
| 63 | +@pytest.mark.parametrize("is_gzip", [False, True], ids=["raw", "gzip"]) |
| 64 | +@pytest.mark.parametrize( |
| 65 | + "readline_kwargs, expected", |
| 66 | + [ |
| 67 | + pytest.param({"size": 0}, b"", id="zero"), |
| 68 | + pytest.param({"size": 2}, b"he", id="short"), |
| 69 | + pytest.param({}, b"hello world\n", id="default"), |
| 70 | + pytest.param({"size": -1}, b"hello world\n", id="negative"), |
| 71 | + pytest.param({"size": None}, b"hello world\n", id="none"), |
| 72 | + pytest.param({"size": 100}, b"hello world\n", id="past-newline"), |
| 73 | + ], |
| 74 | +) |
| 75 | +def test_s3_readline_respects_size(s3_client, s3_client_stub, is_gzip, readline_kwargs, expected): |
| 76 | + payload = b"hello world\nnext line\n" |
| 77 | + wire_payload = gzip.compress(payload) if is_gzip else payload |
| 78 | + streaming_body = PowertoolsStreamingBody(raw_stream=io.BytesIO(wire_payload), content_length=len(wire_payload)) |
| 79 | + s3_client_stub.add_response( |
| 80 | + "get_object", |
| 81 | + {"Body": streaming_body}, |
| 82 | + {"Bucket": "bucket", "Key": "key", "Range": "bytes=0-"}, |
| 83 | + ) |
| 84 | + |
| 85 | + with S3Object(bucket="bucket", key="key", boto3_client=s3_client, is_gzip=is_gzip) as obj: |
| 86 | + assert obj.readline(**readline_kwargs) == expected |
| 87 | + assert obj.tell() == len(expected) |
| 88 | + assert obj.read() == payload[len(expected) :] |
| 89 | + assert obj.tell() == len(payload) |
| 90 | + assert obj.readline(**readline_kwargs) == b"" |
| 91 | + assert obj.tell() == len(payload) |
| 92 | + |
| 93 | + s3_client_stub.assert_no_pending_responses() |
0 commit comments