Adding s3_list_objects_encoding_type_url handler to ListObjectsV2

.changes/next-release/enhancement-s3-12341.json

@@ -0,0 +1,5 @@
+{
+  "category": "s3", 
+  "type": "enhancement", 
+  "description": "Adds encoding and decoding handlers for ListObjectsV2 `#1552 <https://github.com/boto/botocore/issues/1552>`__"
+}

botocore/handlers.py

@@ -730,21 +730,38 @@ def decode_list_object(parsed, context, **kwargs):
     # Amazon S3 includes this element in the response, and returns encoded key
     # name values in the following response elements:
     # Delimiter, Marker, Prefix, NextMarker, Key.
+    _decode_list_object(
+        top_level_keys=['Delimiter', 'Marker', 'NextMarker'],
+        nested_keys=[('Contents', 'Key'), ('CommonPrefixes', 'Prefix')],
+        parsed=parsed,
+        context=context
+    )
+
+def decode_list_object_v2(parsed, context, **kwargs):
+    # From the documentation: If you specify encoding-type request parameter,
+    # Amazon S3 includes this element in the response, and returns encoded key
+    # name values in the following response elements:
+    # Delimiter, Prefix, ContinuationToken, Key, and StartAfter.
+    _decode_list_object(
+        top_level_keys=['Delimiter', 'Prefix', 'ContinuationToken', 'StartAfter'],
+        nested_keys=[('Contents', 'Key'), ('CommonPrefixes', 'Prefix')],
+        parsed=parsed,
+        context=context
+    )
+
+def _decode_list_object(top_level_keys, nested_keys, parsed, context):
     if parsed.get('EncodingType') == 'url' and \
                     context.get('encoding_type_auto_set'):
         # URL decode top-level keys in the response if present.
-        top_level_keys = ['Delimiter', 'Marker', 'NextMarker']
         for key in top_level_keys:
             if key in parsed:
                 parsed[key] = unquote_str(parsed[key])
         # URL decode nested keys from the response if present.
-        nested_keys = [('Contents', 'Key'), ('CommonPrefixes', 'Prefix')]
         for (top_key, child_key) in nested_keys:
             if top_key in parsed:
                 for member in parsed[top_key]:
                     member[child_key] = unquote_str(member[child_key])
 
-
 def convert_body_to_file_like_object(params, **kwargs):
     if 'Body' in params:
         if isinstance(params['Body'], six.string_types):
@@ -880,6 +897,8 @@ def remove_subscribe_to_shard(class_attributes, **kwargs):
 
     ('before-parameter-build.s3.ListObjects',
      set_list_objects_encoding_type_url),
+    ('before-parameter-build.s3.ListObjectsV2',
+     set_list_objects_encoding_type_url),
     ('before-call.s3.PutBucketTagging', calculate_md5),
     ('before-call.s3.PutBucketLifecycle', calculate_md5),
     ('before-call.s3.PutBucketLifecycleConfiguration', calculate_md5),
@@ -943,6 +962,7 @@ def remove_subscribe_to_shard(class_attributes, **kwargs):
     ('before-parameter-build.route53', fix_route53_ids),
     ('before-parameter-build.glacier', inject_account_id),
     ('after-call.s3.ListObjects', decode_list_object),
+    ('after-call.s3.ListObjectsV2', decode_list_object_v2),
 
     # Cloudsearchdomain search operation will be sent by HTTP POST
     ('request-created.cloudsearchdomain.Search',

tests/integration/test_s3.py

@@ -416,6 +416,21 @@ def test_unicode_system_character(self):
         self.assertEqual(len(parsed['Contents']), 1)
         self.assertEqual(parsed['Contents'][0]['Key'], 'foo%08')
 
+    def test_unicode_system_character_with_list_v2(self):
+        # Verify we can use a unicode system character which would normally
+        # break the xml parser
+        key_name = 'foo\x08'
+        self.create_object(key_name)
+        self.addCleanup(self.delete_object, key_name, self.bucket_name)
+        parsed = self.client.list_objects_v2(Bucket=self.bucket_name)
+        self.assertEqual(len(parsed['Contents']), 1)
+        self.assertEqual(parsed['Contents'][0]['Key'], key_name)
+
+        parsed = self.client.list_objects_v2(Bucket=self.bucket_name,
+                                          EncodingType='url')
+        self.assertEqual(len(parsed['Contents']), 1)
+        self.assertEqual(parsed['Contents'][0]['Key'], 'foo%08')
+
     def test_thread_safe_auth(self):
         self.auth_paths = []
         self.session.register('before-sign', self.increment_auth)

tests/unit/test_handlers.py

@@ -873,6 +873,69 @@ def test_decode_list_objects_with_delimiter(self):
         handlers.decode_list_object(parsed, context=context)
         self.assertEqual(parsed['Delimiter'], u'\xe7\xf6s% asd\x08 c')
 
+    def test_decode_list_objects_v2(self):
+        parsed = {
+            'Contents': [{'Key': "%C3%A7%C3%B6s%25asd%08"}],
+            'EncodingType': 'url',
+        }
+        context = {'encoding_type_auto_set': True}
+        handlers.decode_list_object_v2(parsed, context=context)
+        self.assertEqual(parsed['Contents'][0]['Key'], u'\xe7\xf6s%asd\x08')
+
+    def test_decode_list_objects_v2_does_not_decode_without_context(self):
+        parsed = {
+            'Contents': [{'Key': "%C3%A7%C3%B6s%25asd"}],
+            'EncodingType': 'url',
+        }
+        handlers.decode_list_object_v2(parsed, context={})
+        self.assertEqual(parsed['Contents'][0]['Key'], u'%C3%A7%C3%B6s%25asd')
+
+    def test_decode_list_objects_v2_with_delimiter(self):
+        parsed = {
+            'Delimiter': "%C3%A7%C3%B6s%25%20asd%08+c",
+            'EncodingType': 'url',
+        }
+        context = {'encoding_type_auto_set': True}
+        handlers.decode_list_object_v2(parsed, context=context)
+        self.assertEqual(parsed['Delimiter'], u'\xe7\xf6s% asd\x08 c')
+
+    def test_decode_list_objects_v2_with_prefix(self):
+        parsed = {
+            'Prefix': "%C3%A7%C3%B6s%25%20asd%08+c",
+            'EncodingType': 'url',
+        }
+        context = {'encoding_type_auto_set': True}
+        handlers.decode_list_object_v2(parsed, context=context)
+        self.assertEqual(parsed['Prefix'], u'\xe7\xf6s% asd\x08 c')
+
+    def test_decode_list_objects_v2_with_continuationtoken(self):
+        parsed = {
+            'ContinuationToken': "%C3%A7%C3%B6s%25%20asd%08+c",
+            'EncodingType': 'url',
+        }
+        context = {'encoding_type_auto_set': True}
+        handlers.decode_list_object_v2(parsed, context=context)
+        self.assertEqual(parsed['ContinuationToken'], u'\xe7\xf6s% asd\x08 c')
+
+    def test_decode_list_objects_v2_with_startafter(self):
+        parsed = {
+            'StartAfter': "%C3%A7%C3%B6s%25%20asd%08+c",
+            'EncodingType': 'url',
+        }
+        context = {'encoding_type_auto_set': True}
+        handlers.decode_list_object_v2(parsed, context=context)
+        self.assertEqual(parsed['StartAfter'], u'\xe7\xf6s% asd\x08 c')
+
+    def test_decode_list_objects_v2_with_common_prefixes(self):
+        parsed = {
+            'CommonPrefixes': [{'Prefix': "%C3%A7%C3%B6s%25%20asd%08+c"}],
+            'EncodingType': 'url',
+        }
+        context = {'encoding_type_auto_set': True}
+        handlers.decode_list_object_v2(parsed, context=context)
+        self.assertEqual(parsed['CommonPrefixes'][0]['Prefix'],
+                         u'\xe7\xf6s% asd\x08 c')
+
     def test_get_bucket_location_optional(self):
         # This handler should no-op if another hook (i.e. stubber) has already
         # filled in response