|
1 |
| -import io |
2 | 1 | from csv import DictReader
|
3 | 2 | from gzip import GzipFile
|
4 | 3 |
|
5 |
| -import boto3 |
6 |
| -import pytest |
7 |
| -from botocore import stub |
8 |
| -from botocore.response import StreamingBody |
9 |
| - |
10 | 4 | from aws_lambda_powertools.utilities.streaming import S3Object
|
11 | 5 | from aws_lambda_powertools.utilities.streaming._s3_seekable_io import _S3SeekableIO
|
12 | 6 | from aws_lambda_powertools.utilities.streaming.transformations import GzipTransform
|
@@ -44,24 +38,3 @@ def test_s3_transform_in_place():
|
44 | 38 |
|
45 | 39 | new_obj = obj.transform(GzipTransform(), in_place=True)
|
46 | 40 | assert new_obj is None
|
47 |
| - |
48 |
| - |
49 |
| -def test_s3_transform_after_read(): |
50 |
| - # GIVEN a S3 Object with a "hello world" payload |
51 |
| - payload = b"hello world" |
52 |
| - |
53 |
| - s3_client = boto3.client("s3") |
54 |
| - s3_stub = stub.Stubber(s3_client) |
55 |
| - s3_stub.add_response( |
56 |
| - "get_object", {"Body": StreamingBody(raw_stream=io.BytesIO(payload), content_length=len(payload))} |
57 |
| - ) |
58 |
| - s3_stub.activate() |
59 |
| - obj = S3Object(bucket="bucket", key="key", boto3_client=s3_client) |
60 |
| - |
61 |
| - # WHEN you read some part of the object and then apply a transformation |
62 |
| - assert obj.read(5) == b"hello" |
63 |
| - |
64 |
| - # THEN it raises ValueError |
65 |
| - with pytest.raises(ValueError) as exc: |
66 |
| - obj.transform(GzipTransform()) |
67 |
| - assert str(exc.value) == "Cannot add transformations to a read object. Already read 5 bytes" |
0 commit comments