Add support for multiple delete markers on an s3 object

moto/s3/models.py

@@ -27,8 +27,14 @@ class FakeDeleteMarker(BaseModel):
 
     def __init__(self, key):
         self.key = key
+        self.name = key.name
+        self.last_modified = datetime.datetime.utcnow()
         self._version_id = key.version_id + 1
 
+    @property
+    def last_modified_ISO8601(self):
+        return iso_8601_datetime_with_milliseconds(self.last_modified)
+
     @property
     def version_id(self):
         return self._version_id
@@ -630,9 +636,6 @@ class S3Backend(BaseBackend):
         latest_versions = {}
 
         for version in versions:
-            if isinstance(version, FakeDeleteMarker):
-                name = version.key.name
-            else:
             name = version.name
             version_id = version.version_id
             maximum_version_per_key[name] = max(

moto/s3/responses.py

@@ -1273,10 +1273,10 @@ S3_BUCKET_GET_VERSIONS = """<?xml version="1.0" encoding="UTF-8"?>
     {% endfor %}
     {% for marker in delete_marker_list %}
     <DeleteMarker>
-        <Key>{{ marker.key.name }}</Key>
+        <Key>{{ marker.name }}</Key>
         <VersionId>{{ marker.version_id }}</VersionId>
-        <IsLatest>{% if latest_versions[marker.key.name] == marker.version_id %}true{% else %}false{% endif %}</IsLatest>
-        <LastModified>{{ marker.key.last_modified_ISO8601 }}</LastModified>
+        <IsLatest>{% if latest_versions[marker.name] == marker.version_id %}true{% else %}false{% endif %}</IsLatest>
+        <LastModified>{{ marker.last_modified_ISO8601 }}</LastModified>
         <Owner>
             <ID>75aa57f09aa0c8caeab4f8c24e99d10f8e7faeebf76c078efc7c6caea54ba06a</ID>
             <DisplayName>webfile</DisplayName>

tests/test_s3/test_s3.py

@@ -2471,6 +2471,72 @@ def test_boto3_delete_markers():
     oldest['Key'].should.equal('key-with-versions-and-unicode-ó')
 
 
+@mock_s3
+def test_boto3_multiple_delete_markers():
+    s3 = boto3.client('s3', region_name='us-east-1')
+    bucket_name = 'mybucket'
+    key = u'key-with-versions-and-unicode-ó'
+    s3.create_bucket(Bucket=bucket_name)
+    s3.put_bucket_versioning(
+        Bucket=bucket_name,
+        VersioningConfiguration={
+            'Status': 'Enabled'
+        }
+    )
+    items = (six.b('v1'), six.b('v2'))
+    for body in items:
+        s3.put_object(
+            Bucket=bucket_name,
+            Key=key,
+            Body=body
+        )
+
+    # Delete the object twice to add multiple delete markers
+    s3.delete_object(Bucket=bucket_name, Key=key)
+    s3.delete_object(Bucket=bucket_name, Key=key)
+
+    response = s3.list_object_versions(Bucket=bucket_name)
+    response['DeleteMarkers'].should.have.length_of(2)
+
+    with assert_raises(ClientError) as e:
+        s3.get_object(
+            Bucket=bucket_name,
+            Key=key
+        )
+        e.response['Error']['Code'].should.equal('404')
+
+    # Remove both delete markers to restore the object
+    s3.delete_object(
+        Bucket=bucket_name,
+        Key=key,
+        VersionId='2'
+    )
+    s3.delete_object(
+        Bucket=bucket_name,
+        Key=key,
+        VersionId='3'
+    )
+
+    response = s3.get_object(
+        Bucket=bucket_name,
+        Key=key
+    )
+    response['Body'].read().should.equal(items[-1])
+    response = s3.list_object_versions(Bucket=bucket_name)
+    response['Versions'].should.have.length_of(2)
+
+    # We've asserted there is only 2 records so one is newest, one is oldest
+    latest = list(filter(lambda item: item['IsLatest'], response['Versions']))[0]
+    oldest = list(filter(lambda item: not item['IsLatest'], response['Versions']))[0]
+
+    # Double check ordering of version ID's
+    latest['VersionId'].should.equal('1')
+    oldest['VersionId'].should.equal('0')
+
+    # Double check the name is still unicode
+    latest['Key'].should.equal('key-with-versions-and-unicode-ó')
+    oldest['Key'].should.equal('key-with-versions-and-unicode-ó')
+
 @mock_s3
 def test_get_stream_gzipped():
     payload = b"this is some stuff here"