Include response headers when deleting objects. (#3313)

* Return delete meta.

* Add tests.

* Lint fixes.

Co-authored-by: Leo Sutic <leo.sutic@matterport.com>

moto/s3/models.py

@@ -1620,7 +1620,9 @@ class S3Backend(BaseBackend):
 
     def _set_delete_marker(self, bucket_name, key_name):
         bucket = self.get_bucket(bucket_name)
-        bucket.keys[key_name] = FakeDeleteMarker(key=bucket.keys[key_name])
+        delete_marker = FakeDeleteMarker(key=bucket.keys[key_name])
+        bucket.keys[key_name] = delete_marker
+        return delete_marker
 
     def delete_object_tagging(self, bucket_name, key_name, version_id=None):
         key = self.get_object(bucket_name, key_name, version_id=version_id)
@@ -1630,15 +1632,26 @@ class S3Backend(BaseBackend):
         key_name = clean_key_name(key_name)
         bucket = self.get_bucket(bucket_name)
 
+        response_meta = {}
+
         try:
             if not bucket.is_versioned:
                 bucket.keys.pop(key_name)
             else:
                 if version_id is None:
-                    self._set_delete_marker(bucket_name, key_name)
+                    delete_marker = self._set_delete_marker(bucket_name, key_name)
+                    response_meta["version-id"] = delete_marker.version_id
                 else:
                     if key_name not in bucket.keys:
                         raise KeyError
+
+                    response_meta["delete-marker"] = "false"
+                    for key in bucket.keys.getlist(key_name):
+                        if str(key.version_id) == str(version_id):
+                            if type(key) is FakeDeleteMarker:
+                                response_meta["delete-marker"] = "true"
+                            break
+
                     bucket.keys.setlist(
                         key_name,
                         [
@@ -1650,9 +1663,9 @@ class S3Backend(BaseBackend):
 
                     if not bucket.keys.getlist(key_name):
                         bucket.keys.pop(key_name)
-            return True
+            return True, response_meta
         except KeyError:
-            return False
+            return False, None
 
     def copy_key(
         self,

moto/s3/responses.py

@@ -902,7 +902,7 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
             key_name = object_["Key"]
             version_id = object_.get("VersionId", None)
 
-            success = self.backend.delete_object(
+            success, _ = self.backend.delete_object(
                 bucket_name, undo_clean_key_name(key_name), version_id=version_id
             )
             if success:
@@ -1666,8 +1666,14 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
             )
             template = self.response_template(S3_DELETE_KEY_TAGGING_RESPONSE)
             return 204, {}, template.render(version_id=version_id)
-        self.backend.delete_object(bucket_name, key_name, version_id=version_id)
-        return 204, {}, ""
+        success, response_meta = self.backend.delete_object(
+            bucket_name, key_name, version_id=version_id
+        )
+        response_headers = {}
+        if response_meta is not None:
+            for k in response_meta:
+                response_headers["x-amz-{}".format(k)] = response_meta[k]
+        return 204, response_headers, ""
 
     def _complete_multipart_body(self, body):
         ps = minidom.parseString(body).getElementsByTagName("Part")

tests/test_s3/test_s3.py

@@ -2315,6 +2315,30 @@ def test_boto3_delete_versioned_bucket():
     client.delete_bucket(Bucket="blah")
 
 
+@mock_s3
+def test_boto3_delete_versioned_bucket_returns_meta():
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+
+    client.create_bucket(Bucket="blah")
+    client.put_bucket_versioning(
+        Bucket="blah", VersioningConfiguration={"Status": "Enabled"}
+    )
+
+    put_resp = client.put_object(Bucket="blah", Key="test1", Body=b"test1")
+
+    # Delete the object
+    del_resp = client.delete_object(Bucket="blah", Key="test1")
+    assert "DeleteMarker" not in del_resp
+    assert del_resp["VersionId"] is not None
+
+    # Delete the delete marker
+    del_resp2 = client.delete_object(
+        Bucket="blah", Key="test1", VersionId=del_resp["VersionId"]
+    )
+    assert del_resp2["DeleteMarker"] == True
+    assert "VersionId" not in del_resp2
+
+
 @mock_s3
 def test_boto3_get_object_if_modified_since():
     s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME)