S3 - ListObjectVersions: Limit DeleteMarkers by Prefix

moto/s3/models.py

@@ -1488,10 +1488,6 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
             version.is_latest = name != last_name
             if version.is_latest:
                 last_name = name
-            # Differentiate between FakeKey and FakeDeleteMarkers
-            if not isinstance(version, FakeKey):
-                delete_markers.append(version)
-                continue
             # skip all keys that alphabetically come before keymarker
             if key_marker and name < key_marker:
                 continue
@@ -1505,6 +1501,11 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
                 common_prefixes.append(prefix_including_delimiter)
                 continue
 
+            # Differentiate between FakeKey and FakeDeleteMarkers
+            if not isinstance(version, FakeKey):
+                delete_markers.append(version)
+                continue
+
             requested_versions.append(version)
 
         common_prefixes = sorted(set(common_prefixes))

tests/test_s3/test_s3.py

@@ -2402,6 +2402,55 @@ def test_list_object_versions_with_delimiter():
     )
 
 
+@mock_s3
+def test_list_object_versions_with_delimiter_for_deleted_objects():
+    bucket_name = "tests_bucket"
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    # Create bucket with versioning
+    client.create_bucket(Bucket=bucket_name)
+    client.put_bucket_versioning(
+        Bucket=bucket_name,
+        VersioningConfiguration={"MFADelete": "Disabled", "Status": "Enabled"},
+    )
+
+    # Create a history of objects
+    for pos in range(2):
+        client.put_object(
+            Bucket=bucket_name, Key=f"obj_{pos}", Body=f"object {pos}".encode("utf-8")
+        )
+
+    for pos in range(2):
+        client.put_object(
+            Bucket=bucket_name,
+            Key=f"hist_obj_{pos}",
+            Body=f"history object {pos}".encode("utf-8"),
+        )
+        for hist_pos in range(2):
+            client.put_object(
+                Bucket=bucket_name,
+                Key=f"hist_obj_{pos}",
+                Body=f"object {pos} {hist_pos}".encode("utf-8"),
+            )
+
+    for pos in range(2):
+        client.put_object(
+            Bucket=bucket_name,
+            Key=f"del_obj_{pos}",
+            Body=f"deleted object {pos}".encode("utf-8"),
+        )
+        client.delete_object(Bucket=bucket_name, Key=f"del_obj_{pos}")
+
+    # Verify we only retrieve the DeleteMarkers that have this prefix
+    objs = client.list_object_versions(Bucket=bucket_name)
+    [dm["Key"] for dm in objs["DeleteMarkers"]].should.equal(["del_obj_0", "del_obj_1"])
+
+    hist_objs = client.list_object_versions(Bucket=bucket_name, Prefix="hist_obj")
+    hist_objs.shouldnt.have.key("DeleteMarkers")
+
+    del_objs = client.list_object_versions(Bucket=bucket_name, Prefix="del_obj_0")
+    [dm["Key"] for dm in del_objs["DeleteMarkers"]].should.equal(["del_obj_0"])
+
+
 @mock_s3
 def test_list_object_versions_with_versioning_disabled():
     s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME)