Skip to content

Commit 6fd036c

Browse files
Edward AlmondEdward Almond
authored andcommitted
Add S3 ListObjectsV2 pagination integration test
- Test with 50 objects and MaxKeys=10 - Verify IsTruncated and NextContinuationToken returned - Test multiple pages using continuation tokens - Verify all objects retrieved across all pages
1 parent 811219b commit 6fd036c

1 file changed

Lines changed: 54 additions & 0 deletions

File tree

tests/integration/test_docker.py

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -190,6 +190,60 @@ def test_list_objects(self, client):
190190
result = client.list_objects_v2(Bucket=bucket_name)
191191
assert len(result["Contents"]) == 3
192192

193+
def test_list_objects_pagination(self, client):
194+
"""Test ListObjectsV2 pagination with continuation token."""
195+
bucket_name = "test-pagination-bucket"
196+
197+
# Create bucket and objects
198+
try:
199+
client.create_bucket(Bucket=bucket_name)
200+
except ClientError:
201+
pass
202+
203+
# Put more objects than default max-keys (typically 1000)
204+
num_objects = 50
205+
for i in range(num_objects):
206+
client.put_object(Bucket=bucket_name, Key=f"file{i:04d}.txt", Body=f"content {i}")
207+
208+
# First request with small max-keys
209+
result = client.list_objects_v2(Bucket=bucket_name, MaxKeys=10)
210+
211+
assert len(result["Contents"]) == 10
212+
assert result["IsTruncated"] is True
213+
assert "NextContinuationToken" in result
214+
215+
# Second request with continuation token
216+
token = result["NextContinuationToken"]
217+
result2 = client.list_objects_v2(Bucket=bucket_name, ContinuationToken=token)
218+
219+
assert len(result2["Contents"]) == 10
220+
# Should have more objects after second batch
221+
total = 10 + len(result2["Contents"])
222+
223+
# Third request
224+
if result2["IsTruncated"]:
225+
token2 = result2["NextContinuationToken"]
226+
result3 = client.list_objects_v2(Bucket=bucket_name, ContinuationToken=token2)
227+
total += len(result3["Contents"])
228+
229+
# Verify we can get all objects by continuing
230+
all_keys = []
231+
token = None
232+
while True:
233+
if token:
234+
result = client.list_objects_v2(Bucket=bucket_name, ContinuationToken=token)
235+
else:
236+
result = client.list_objects_v2(Bucket=bucket_name)
237+
238+
all_keys.extend([obj["Key"] for obj in result.get("Contents", [])])
239+
240+
if result.get("IsTruncated"):
241+
token = result.get("NextContinuationToken")
242+
else:
243+
break
244+
245+
assert len(all_keys) == num_objects
246+
193247

194248
class TestSecretsManager:
195249
"""Secrets Manager integration tests."""

0 commit comments

Comments
 (0)