s3: test listing when object has lots of versions

s3tests_boto3/functional/__init__.py

@@ -79,28 +79,18 @@ def get_objects_list(bucket, client=None, prefix=None):
 
     return objects_list
 
-# generator function that returns object listings in batches, where each
-# batch is a list of dicts compatible with delete_objects()
-def list_versions(client, bucket, batch_size):
-    kwargs = {'Bucket': bucket, 'MaxKeys': batch_size}
-    truncated = True
-    while truncated:
-        listing = client.list_object_versions(**kwargs)
-
-        kwargs['KeyMarker'] = listing.get('NextKeyMarker')
-        kwargs['VersionIdMarker'] = listing.get('NextVersionIdMarker')
-        truncated = listing['IsTruncated']
-
-        objs = listing.get('Versions', []) + listing.get('DeleteMarkers', [])
-        if len(objs):
-            yield [{'Key': o['Key'], 'VersionId': o['VersionId']} for o in objs]
-
 def nuke_bucket(client, bucket):
     batch_size = 128
     max_retain_date = None
 
     # list and delete objects in batches
-    for objects in list_versions(client, bucket, batch_size):
+    truncated = True
+    while truncated:
+        listing = client.list_object_versions(Bucket=bucket, MaxKeys=batch_size)
+        truncated = listing['IsTruncated']
+        objs = listing.get('Versions', []) + listing.get('DeleteMarkers', [])
+        objects = [{'Key': o['Key'], 'VersionId': o['VersionId']} for o in objs]
+
         delete = client.delete_objects(Bucket=bucket,
                 Delete={'Objects': objects, 'Quiet': True},
                 BypassGovernanceRetention=True)
@@ -131,7 +121,13 @@ def nuke_bucket(client, bucket):
                     'seconds for object locks to expire')
             time.sleep(delta.total_seconds())
 
-        for objects in list_versions(client, bucket, batch_size):
+        truncated = True
+        while truncated:
+            listing = client.list_object_versions(Bucket=bucket, MaxKeys=batch_size)
+            truncated = listing['IsTruncated']
+            objs = listing.get('Versions', []) + listing.get('DeleteMarkers', [])
+            objects = [{'Key': o['Key'], 'VersionId': o['VersionId']} for o in objs]
+
             client.delete_objects(Bucket=bucket,
                     Delete={'Objects': objects, 'Quiet': True},
                     BypassGovernanceRetention=True)

s3tests_boto3/functional/test_s3.py

@@ -1343,6 +1343,48 @@ def test_bucket_list_marker_not_in_list():
     keys = _get_keys(response)
     assert keys == [ 'foo','quxx']
 
+def test_versioned_bucket_listing():
+    client = get_client()
+    bucket_name = get_new_bucket(client)
+    count = 15
+    all_versions = set()
+
+    # write one object before enabling versioning
+    client.put_object(Bucket=bucket_name, Key='A')
+    # this gets converted to a "null" version
+    all_versions.add('null')
+
+    check_configure_versioning_retry(bucket_name, "Enabled", "Enabled")
+
+    for i in range(count):
+        response = client.put_object(Bucket=bucket_name, Key='A')
+        all_versions.add(response['VersionId'])
+
+    response = client.put_object(Bucket=bucket_name, Key='B')
+    all_versions.add(response['VersionId'])
+
+    last_marker = None
+    last_version = None
+    paginator = client.get_paginator('list_object_versions')
+    for page in paginator.paginate(Bucket=bucket_name, MaxKeys=1):
+        next_marker = page.get('NextKeyMarker')
+        next_version = page.get('NextVersionIdMarker')
+        assert last_marker != next_marker or last_version != next_version
+        last_marker = next_marker
+        last_version = next_version
+        for v in page['Versions']:
+            all_versions.remove(v['VersionId']) # fails if already removed
+
+    assert not all_versions, 'list_object_versions() did not visit every version'
+
+    response = client.list_objects(Bucket=bucket_name, Marker='A', MaxKeys=1)
+    assert response['Contents'][0]['Key'] == 'B'
+    assert not response['IsTruncated']
+
+    response = client.list_objects_v2(Bucket=bucket_name, StartAfter='A', MaxKeys=1)
+    assert response['Contents'][0]['Key'] == 'B'
+    assert not response['IsTruncated']
+
 @pytest.mark.list_objects_v2
 def test_bucket_listv2_startafter_not_in_list():
     key_names = ['bar', 'baz', 'foo', 'quxx']