unquote key name multiple times until stable value

moto/s3/utils.py

@@ -68,11 +68,22 @@ def metadata_from_headers(headers):
     return metadata
 
 
-def clean_key_name(key_name):
+def clean_key_name(key_name, attempts=4):
     if six.PY2:
-        return unquote(key_name.encode('utf-8')).decode('utf-8')
+        def uq(k):
+            return unquote(k.encode('utf-8')).decode('utf-8')
+    else:
+        uq = unquote
 
-    return unquote(key_name)
+    original = cleaned = key_name
+    last_attempt = attempts - 1
+    for attempt in range(attempts):
+        cleaned = uq(key_name)
+        if cleaned == key_name:
+            return cleaned
+        if attempt != last_attempt:
+            key_name = cleaned
+    raise Exception('unable to fully clean name: original %s, last clean %s prior clean %s' % (original, cleaned, key_name))
 
 
 class _VersionedKeyStore(dict):

tests/test_s3/test_s3.py

@@ -21,6 +21,7 @@ from botocore.handlers import disable_signing
 from boto.s3.connection import S3Connection
 from boto.s3.key import Key
 from freezegun import freeze_time
+from parameterized import parameterized
 import six
 import requests
 import tests.backport_assert_raises  # noqa
@@ -3046,3 +3047,39 @@ def test_root_dir_with_empty_name_works():
     if os.environ.get('TEST_SERVER_MODE', 'false').lower() == 'true':
         raise SkipTest('Does not work in server mode due to error in Workzeug')
     store_and_read_back_a_key('/')
+
+
+@parameterized([
+    ('foo/bar/baz',),
+    ('foo',),
+    ('foo/run_dt%3D2019-01-01%252012%253A30%253A00',),
+])
+@mock_s3
+def test_delete_objects_with_url_encoded_key(key):
+    s3 = boto3.client('s3', region_name='us-east-1')
+    bucket_name = 'mybucket'
+    body = b'Some body'
+
+    s3.create_bucket(Bucket=bucket_name)
+
+    def put_object():
+        s3.put_object(
+            Bucket=bucket_name,
+            Key=key,
+            Body=body
+        )
+
+    def assert_deleted():
+        with assert_raises(ClientError) as e:
+            s3.get_object(Bucket=bucket_name, Key=key)
+
+        e.exception.response['Error']['Code'].should.equal('NoSuchKey')
+
+    put_object()
+    s3.delete_object(Bucket=bucket_name, Key=key)
+    assert_deleted()
+
+    put_object()
+    s3.delete_objects(Bucket=bucket_name, Delete={'Objects': [{'Key': key}]})
+    assert_deleted()
+

tests/test_s3/test_s3_utils.py

@@ -1,7 +1,8 @@
 from __future__ import unicode_literals
 import os
 from sure import expect
-from moto.s3.utils import bucket_name_from_url, _VersionedKeyStore, parse_region_from_url
+from moto.s3.utils import bucket_name_from_url, _VersionedKeyStore, parse_region_from_url, clean_key_name
+from parameterized import parameterized
 
 
 def test_base_url():
@@ -78,3 +79,15 @@ def test_parse_region_from_url():
                 'https://s3.amazonaws.com/bucket',
                 'https://bucket.s3.amazonaws.com']:
         parse_region_from_url(url).should.equal(expected)
+
+
+@parameterized([
+    ('foo/bar/baz',
+     'foo/bar/baz'),
+    ('foo',
+     'foo'),
+    ('foo/run_dt%3D2019-01-01%252012%253A30%253A00',
+     'foo/run_dt=2019-01-01 12:30:00'),
+])
+def test_clean_key_name(key, expected):
+    clean_key_name(key).should.equal(expected)