Avoid infinite loop with bucket listing and encoding_type='url'

stanhu · Aug 26, 2014 · 7378329 · 7378329
1 parent 2b87583
commit 7378329
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 0 deletions.
diff --git a/boto/s3/bucketlistresultset.py b/boto/s3/bucketlistresultset.py
@@ -19,6 +19,8 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 # IN THE SOFTWARE.
 
+from boto.compat import urllib, six
+
 def bucket_lister(bucket, prefix='', delimiter='', marker='', headers=None,
                   encoding_type=None):
     """
@@ -34,6 +36,10 @@ def bucket_lister(bucket, prefix='', delimiter='', marker='', headers=None,
             yield k
         if k:
             marker = rs.next_marker or k.name
+        if marker and encoding_type == "url":
+            if isinstance(marker, six.text_type):
+                marker = marker.encode('utf-8')
+            marker = urllib.parse.unquote(marker)
         more_results= rs.is_truncated
 
 class BucketListResultSet(object):

diff --git a/tests/integration/s3/test_bucket.py b/tests/integration/s3/test_bucket.py
@@ -26,6 +26,7 @@
 Some unit tests for the S3 Bucket
 """
 
+from mock import patch, Mock
 import unittest
 import time
 
@@ -39,6 +40,7 @@
 from boto.s3.acl import Grant
 from boto.s3.tagging import Tags, TagSet
 from boto.s3.website import RedirectLocation
+from boto.compat import urllib
 
 
 class S3BucketTest (unittest.TestCase):
@@ -86,6 +88,25 @@ def test_next_marker(self):
             self.assertEqual(element.name, expected.pop(0))
         self.assertEqual(expected, [])
 
+    def test_list_with_url_encoding(self):
+        expected = ["α", "β", "γ"]
+        for key_name in expected:
+            key = self.bucket.new_key(key_name)
+            key.set_contents_from_string(key_name)
+
+        # ensure bucket.list() still works by just
+        # popping elements off the front of expected.
+        orig_getall = self.bucket._get_all
+        getall = lambda *a, **k: orig_getall(*a, max_keys=2, **k)
+        with patch.object(self.bucket, '_get_all', getall):
+            rs = self.bucket.list(encoding_type="url")
+            import pdb
+            pdb.set_trace()
+            for element in rs:
+                name = urllib.parse.unquote(element.name.encode('utf-8'))
+                self.assertEqual(name, expected.pop(0))
+            self.assertEqual(expected, [])
+
     def test_logging(self):
         # use self.bucket as the target bucket so that teardown
         # will delete any log files that make it into the bucket