Merge pull request #2919 from antoinewdg-mt/fix-s3-delete-objects-with-versioning

Handle VersionId in S3:delete_objects

moto/s3/responses.py

@@ -838,27 +838,35 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
 
     def _bucket_response_delete_keys(self, request, body, bucket_name):
         template = self.response_template(S3_DELETE_KEYS_RESPONSE)
+        body_dict = xmltodict.parse(body)
 
-        keys = minidom.parseString(body).getElementsByTagName("Key")
-        deleted_names = []
-        error_names = []
-        if len(keys) == 0:
+        objects = body_dict["Delete"].get("Object", [])
+        if not isinstance(objects, list):
+            # We expect a list of objects, but when there is a single <Object> node xmltodict does not
+            # return a list.
+            objects = [objects]
+        if len(objects) == 0:
             raise MalformedXML()
 
-        for k in keys:
-            key_name = k.firstChild.nodeValue
+        deleted_objects = []
+        error_names = []
+
+        for object_ in objects:
+            key_name = object_["Key"]
+            version_id = object_.get("VersionId", None)
+
             success = self.backend.delete_key(
-                bucket_name, undo_clean_key_name(key_name)
+                bucket_name, undo_clean_key_name(key_name), version_id=version_id
             )
             if success:
-                deleted_names.append(key_name)
+                deleted_objects.append((key_name, version_id))
             else:
                 error_names.append(key_name)
 
         return (
             200,
             {},
-            template.render(deleted=deleted_names, delete_errors=error_names),
+            template.render(deleted=deleted_objects, delete_errors=error_names),
         )
 
     def _handle_range_header(self, request, headers, response_content):
@@ -1852,9 +1860,10 @@ S3_BUCKET_GET_VERSIONS = """<?xml version="1.0" encoding="UTF-8"?>
 
 S3_DELETE_KEYS_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
 <DeleteResult xmlns="http://s3.amazonaws.com/doc/2006-03-01">
-{% for k in deleted %}
+{% for k, v in deleted %}
 <Deleted>
 <Key>{{k}}</Key>
+{% if v %}<VersionId>{{v}}</VersionId>{% endif %}
 </Deleted>
 {% endfor %}
 {% for k in delete_errors %}

tests/test_s3/test_s3.py

@@ -2218,6 +2218,29 @@ def test_boto3_deleted_versionings_list():
     assert len(listed["Contents"]) == 1
 
 
+@mock_s3
+def test_boto3_delete_objects_for_specific_version_id():
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    client.create_bucket(Bucket="blah")
+    client.put_bucket_versioning(
+        Bucket="blah", VersioningConfiguration={"Status": "Enabled"}
+    )
+
+    client.put_object(Bucket="blah", Key="test1", Body=b"test1a")
+    client.put_object(Bucket="blah", Key="test1", Body=b"test1b")
+
+    response = client.list_object_versions(Bucket="blah", Prefix="test1")
+    id_to_delete = [v["VersionId"] for v in response["Versions"] if v["IsLatest"]][0]
+
+    response = client.delete_objects(
+        Bucket="blah", Delete={"Objects": [{"Key": "test1", "VersionId": id_to_delete}]}
+    )
+    assert response["Deleted"] == [{"Key": "test1", "VersionId": id_to_delete}]
+
+    listed = client.list_objects_v2(Bucket="blah")
+    assert len(listed["Contents"]) == 1
+
+
 @mock_s3
 def test_boto3_delete_versioned_bucket():
     client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)