text: # Copyright (c) 2014 OpenStack Foundation
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #    http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  # implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  import unittest
  import mock
  
  import six
  from six.moves.urllib.parse import quote, parse_qsl
  
  from swift.common import swob
  from swift.common.middleware.versioned_writes.object_versioning import \
      DELETE_MARKER_CONTENT_TYPE
  from swift.common.swob import Request
  from swift.common.utils import json
  
  from swift.common.middleware.s3api.etree import fromstring, tostring, \
      Element, SubElement
  from swift.common.middleware.s3api.subresource import Owner, encode_acl, \
      ACLPublicRead
  from swift.common.middleware.s3api.s3request import MAX_32BIT_INT
  
  from test.unit.common.middleware.helpers import normalize_path
  from test.unit.common.middleware.s3api import S3ApiTestCase
  from test.unit.common.middleware.s3api.test_s3_acl import s3acl
  from test.unit.common.middleware.s3api.helpers import UnreadableInput
  
  # Example etag from ProxyFS; note that it is already quote-wrapped
  PFS_ETAG = '"pfsv2/AUTH_test/01234567/89abcdef-32"'
  
  
  class TestS3ApiBucket(S3ApiTestCase):
      def setup_objects(self):
          self.objects = (('lily', '2011-01-05T02:19:14.275290', '0', '3909'),
                          (u'lily-\u062a', '2011-01-05T02:19:14.275290', 0, 390),
                          ('mu', '2011-01-05T02:19:14.275290',
                           'md5-of-the-manifest; s3_etag=0', '3909'),
                          ('pfs-obj', '2011-01-05T02:19:14.275290',
                           PFS_ETAG, '3909'),
                          ('rose', '2011-01-05T02:19:14.275290', 0, 303),
                          ('slo', '2011-01-05T02:19:14.275290',
                           'md5-of-the-manifest', '3909'),
                          ('viola', '2011-01-05T02:19:14.275290', '0', 3909),
                          ('with space', '2011-01-05T02:19:14.275290', 0, 390),
                          ('with%20space', '2011-01-05T02:19:14.275290', 0, 390))
  
          self.objects_list = [
              {'name': item[0], 'last_modified': str(item[1]),
               'content_type': 'application/octet-stream',
               'hash': str(item[2]), 'bytes': str(item[3])}
              for item in self.objects]
          self.objects_list[5]['slo_etag'] = '"0"'
          self.versioned_objects = [{
              'name': 'rose',
              'version_id': '2',
              'hash': '0',
              'bytes': '0',
              'last_modified': '2010-03-01T17:09:51.510928',
              'content_type': DELETE_MARKER_CONTENT_TYPE,
              'is_latest': False,
          }, {
              'name': 'rose',
              'version_id': '1',
              'hash': '1234',
              'bytes': '6',
              'last_modified': '2010-03-01T17:09:50.510928',
              'content_type': 'application/octet-stream',
              'is_latest': False,
          }]
  
          listing_body = json.dumps(self.objects_list)
          self.prefixes = ['rose', 'viola', 'lily']
          object_list_subdir = [{"subdir": p} for p in self.prefixes]
  
          self.swift.register('DELETE', '/v1/AUTH_test/bucket+segments',
                              swob.HTTPNoContent, {}, json.dumps([]))
          for name, _, _, _ in self.objects:
              self.swift.register(
                  'DELETE',
                  '/v1/AUTH_test/bucket+segments/' +
                  swob.bytes_to_wsgi(name.encode('utf-8')),
                  swob.HTTPNoContent, {}, json.dumps([]))
          self.swift.register(
              'GET',
              '/v1/AUTH_test/bucket+segments?format=json&marker=with%2520space',
              swob.HTTPOk,
              {'Content-Type': 'application/json; charset=utf-8'},
              json.dumps([]))
          self.swift.register(
              'GET', '/v1/AUTH_test/bucket+segments?format=json&marker=',
              swob.HTTPOk, {'Content-Type': 'application/json'}, listing_body)
          self.swift.register(
              'HEAD', '/v1/AUTH_test/junk', swob.HTTPNoContent, {}, None)
          self.swift.register(
              'HEAD', '/v1/AUTH_test/nojunk', swob.HTTPNotFound, {}, None)
          self.swift.register(
              'HEAD', '/v1/AUTH_test/unavailable', swob.HTTPServiceUnavailable,
              {}, None)
          self.swift.register(
              'GET', '/v1/AUTH_test/junk', swob.HTTPOk,
              {'Content-Type': 'application/json'}, listing_body)
          self.swift.register(
              'GET', '/v1/AUTH_test/junk-subdir', swob.HTTPOk,
              {'Content-Type': 'application/json; charset=utf-8'},
              json.dumps(object_list_subdir))
          self.swift.register(
              'GET',
              '/v1/AUTH_test/subdirs?delimiter=/&limit=3',
              swob.HTTPOk, {}, json.dumps([
                  {'subdir': 'nothing/'},
                  {'subdir': u'but-\u062a/'},
                  {'subdir': 'subdirs/'},
              ]))
  
      def setUp(self):
          super(TestS3ApiBucket, self).setUp()
          self.setup_objects()
  
      def test_bucket_HEAD(self):
          req = Request.blank('/junk',
                              environ={'REQUEST_METHOD': 'HEAD'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
  
      def test_bucket_HEAD_error(self):
          req = Request.blank('/nojunk',
                              environ={'REQUEST_METHOD': 'HEAD'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '404')
          self.assertEqual(body, b'')  # sanity
  
      def test_bucket_HEAD_503(self):
          req = Request.blank('/unavailable',
                              environ={'REQUEST_METHOD': 'HEAD'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '503')
          self.assertEqual(body, b'')  # sanity
  
      def test_bucket_HEAD_slash(self):
          req = Request.blank('/junk/',
                              environ={'REQUEST_METHOD': 'HEAD'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
  
      def test_bucket_HEAD_slash_error(self):
          req = Request.blank('/nojunk/',
                              environ={'REQUEST_METHOD': 'HEAD'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '404')
  
      @s3acl
      def test_bucket_GET_error(self):
          code = self._test_method_error('GET', '/bucket', swob.HTTPUnauthorized)
          self.assertEqual(code, 'SignatureDoesNotMatch')
          code = self._test_method_error('GET', '/bucket', swob.HTTPForbidden)
          self.assertEqual(code, 'AccessDenied')
          code = self._test_method_error('GET', '/bucket', swob.HTTPNotFound)
          self.assertEqual(code, 'NoSuchBucket')
          code = self._test_method_error('GET', '/bucket',
                                         swob.HTTPServiceUnavailable)
          self.assertEqual(code, 'ServiceUnavailable')
          code = self._test_method_error('GET', '/bucket', swob.HTTPServerError)
          self.assertEqual(code, 'InternalError')
  
      def test_bucket_GET(self):
          bucket_name = 'junk'
          req = Request.blank('/%s' % bucket_name,
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
  
          elem = fromstring(body, 'ListBucketResult')
          name = elem.find('./Name').text
          self.assertEqual(name, bucket_name)
  
          objects = elem.iterchildren('Contents')
  
          items = []
          for o in objects:
              items.append((o.find('./Key').text, o.find('./ETag').text))
              self.assertEqual('2011-01-05T02:19:15.000Z',
                               o.find('./LastModified').text)
          expected = [
              (i[0].encode('utf-8') if six.PY2 else i[0],
               PFS_ETAG if i[0] == 'pfs-obj' else
               '"0-N"' if i[0] == 'slo' else '"0"')
              for i in self.objects
          ]
          self.assertEqual(items, expected)
  
      def test_bucket_GET_last_modified_rounding(self):
          objects_list = [
              {'name': 'a', 'last_modified': '2011-01-05T02:19:59.275290',
               'content_type': 'application/octet-stream',
               'hash': 'ahash', 'bytes': '12345'},
              {'name': 'b', 'last_modified': '2011-01-05T02:19:59.000000',
               'content_type': 'application/octet-stream',
               'hash': 'ahash', 'bytes': '12345'},
          ]
          self.swift.register(
              'GET', '/v1/AUTH_test/junk',
              swob.HTTPOk, {'Content-Type': 'application/json'},
              json.dumps(objects_list))
          req = Request.blank('/junk',
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
  
          elem = fromstring(body, 'ListBucketResult')
          name = elem.find('./Name').text
          self.assertEqual(name, 'junk')
          objects = elem.iterchildren('Contents')
          actual = [(obj.find('./Key').text, obj.find('./LastModified').text)
                    for obj in objects]
          self.assertEqual(
              [('a', '2011-01-05T02:20:00.000Z'),
               ('b', '2011-01-05T02:19:59.000Z')],
              actual)
  
      def test_bucket_GET_url_encoded(self):
          bucket_name = 'junk'
          req = Request.blank('/%s?encoding-type=url' % bucket_name,
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
  
          elem = fromstring(body, 'ListBucketResult')
          name = elem.find('./Name').text
          self.assertEqual(name, bucket_name)
  
          objects = elem.iterchildren('Contents')
  
          items = []
          for o in objects:
              items.append((o.find('./Key').text, o.find('./ETag').text))
              self.assertEqual('2011-01-05T02:19:15.000Z',
                               o.find('./LastModified').text)
  
          self.assertEqual(items, [
              (quote(i[0].encode('utf-8')),
               PFS_ETAG if i[0] == 'pfs-obj' else
               '"0-N"' if i[0] == 'slo' else '"0"')
              for i in self.objects])
  
      def test_bucket_GET_subdir(self):
          bucket_name = 'junk-subdir'
          req = Request.blank('/%s' % bucket_name,
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListBucketResult')
          name = elem.find('./Name').text
          self.assertEqual(name, bucket_name)
  
          prefixes = elem.findall('CommonPrefixes')
  
          self.assertEqual(len(prefixes), len(self.prefixes))
          for p in prefixes:
              self.assertTrue(p.find('./Prefix').text in self.prefixes)
  
      def test_bucket_GET_is_truncated(self):
          bucket_name = 'junk'
  
          req = Request.blank(
              '/%s?max-keys=%d' % (bucket_name, len(self.objects)),
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./IsTruncated').text, 'false')
  
          req = Request.blank(
              '/%s?max-keys=%d' % (bucket_name, len(self.objects) - 1),
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./IsTruncated').text, 'true')
  
          req = Request.blank('/subdirs?delimiter=/&max-keys=2',
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./IsTruncated').text, 'true')
          if six.PY2:
              self.assertEqual(elem.find('./NextMarker').text,
                               u'but-\u062a/'.encode('utf-8'))
          else:
              self.assertEqual(elem.find('./NextMarker').text,
                               u'but-\u062a/')
  
      def test_bucket_GET_is_truncated_url_encoded(self):
          bucket_name = 'junk'
  
          req = Request.blank(
              '/%s?encoding-type=url&max-keys=%d' % (
                  bucket_name, len(self.objects)),
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./IsTruncated').text, 'false')
  
          req = Request.blank(
              '/%s?encoding-type=url&max-keys=%d' % (
                  bucket_name, len(self.objects) - 1),
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./IsTruncated').text, 'true')
  
          req = Request.blank('/subdirs?encoding-type=url&delimiter=/&'
                              'max-keys=2',
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./IsTruncated').text, 'true')
          self.assertEqual(elem.find('./NextMarker').text,
                           quote(u'but-\u062a/'.encode('utf-8')))
  
      def test_bucket_GET_v2_is_truncated(self):
          bucket_name = 'junk'
  
          req = Request.blank(
              '/%s?list-type=2&max-keys=%d' % (bucket_name, len(self.objects)),
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./KeyCount').text, str(len(self.objects)))
          self.assertEqual(elem.find('./IsTruncated').text, 'false')
  
          req = Request.blank(
              '/%s?list-type=2&max-keys=%d' % (bucket_name,
                                               len(self.objects) - 1),
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertIsNotNone(elem.find('./NextContinuationToken'))
          self.assertEqual(elem.find('./KeyCount').text,
                           str(len(self.objects) - 1))
          self.assertEqual(elem.find('./IsTruncated').text, 'true')
  
          req = Request.blank('/subdirs?list-type=2&delimiter=/&max-keys=2',
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertIsNotNone(elem.find('./NextContinuationToken'))
          self.assertEqual(elem.find('./KeyCount').text, '2')
          self.assertEqual(elem.find('./IsTruncated').text, 'true')
  
      def test_bucket_GET_max_keys(self):
          bucket_name = 'junk'
  
          req = Request.blank('/%s?max-keys=5' % bucket_name,
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./MaxKeys').text, '5')
          _, path = self.swift.calls[-1]
          _, query_string = path.split('?')
          args = dict(parse_qsl(query_string))
          self.assertEqual(args['limit'], '6')
  
          req = Request.blank('/%s?max-keys=5000' % bucket_name,
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./MaxKeys').text, '5000')
          _, path = self.swift.calls[-1]
          _, query_string = path.split('?')
          args = dict(parse_qsl(query_string))
          self.assertEqual(args['limit'], '1001')
  
      def test_bucket_GET_str_max_keys(self):
          bucket_name = 'junk'
  
          req = Request.blank('/%s?max-keys=invalid' % bucket_name,
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(self._get_error_code(body), 'InvalidArgument')
  
      def test_bucket_GET_negative_max_keys(self):
          bucket_name = 'junk'
  
          req = Request.blank('/%s?max-keys=-1' % bucket_name,
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(self._get_error_code(body), 'InvalidArgument')
  
      def test_bucket_GET_over_32bit_int_max_keys(self):
          bucket_name = 'junk'
  
          req = Request.blank('/%s?max-keys=%s' %
                              (bucket_name, MAX_32BIT_INT + 1),
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(self._get_error_code(body), 'InvalidArgument')
  
      def test_bucket_GET_passthroughs(self):
          bucket_name = 'junk'
          req = Request.blank('/%s?delimiter=a&marker=b&prefix=c' % bucket_name,
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./Prefix').text, 'c')
          self.assertEqual(elem.find('./Marker').text, 'b')
          self.assertEqual(elem.find('./Delimiter').text, 'a')
          _, path = self.swift.calls[-1]
          _, query_string = path.split('?')
          args = dict(parse_qsl(query_string))
          self.assertEqual(args['delimiter'], 'a')
          self.assertEqual(args['marker'], 'b')
          self.assertEqual(args['prefix'], 'c')
  
      def test_bucket_GET_v2_passthroughs(self):
          bucket_name = 'junk'
          req = Request.blank(
              '/%s?list-type=2&delimiter=a&start-after=b&prefix=c' % bucket_name,
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./Prefix').text, 'c')
          self.assertEqual(elem.find('./StartAfter').text, 'b')
          self.assertEqual(elem.find('./Delimiter').text, 'a')
          _, path = self.swift.calls[-1]
          _, query_string = path.split('?')
          args = dict(parse_qsl(query_string))
          self.assertEqual(args['delimiter'], 'a')
          # "start-after" is converted to "marker"
          self.assertEqual(args['marker'], 'b')
          self.assertEqual(args['prefix'], 'c')
  
      def test_bucket_GET_with_nonascii_queries(self):
          bucket_name = 'junk'
          req = Request.blank(
              '/%s?delimiter=\xef\xbc\xa1&marker=\xef\xbc\xa2&'
              'prefix=\xef\xbc\xa3' % bucket_name,
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./Prefix').text,
                           swob.wsgi_to_str('\xef\xbc\xa3'))
          self.assertEqual(elem.find('./Marker').text,
                           swob.wsgi_to_str('\xef\xbc\xa2'))
          self.assertEqual(elem.find('./Delimiter').text,
                           swob.wsgi_to_str('\xef\xbc\xa1'))
          _, path = self.swift.calls[-1]
          _, query_string = path.split('?')
          args = [part.partition('=')[::2] for part in query_string.split('&')]
          self.assertEqual(sorted(args), [
              ('delimiter', '%EF%BC%A1'),
              ('limit', '1001'),
              ('marker', '%EF%BC%A2'),
              ('prefix', '%EF%BC%A3'),
          ])
  
          req = Request.blank(
              '/%s?delimiter=\xef\xbc\xa1&marker=\xef\xbc\xa2&'
              'prefix=\xef\xbc\xa3&encoding-type=url' % bucket_name,
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./Prefix').text, '%EF%BC%A3')
          self.assertEqual(elem.find('./Marker').text, '%EF%BC%A2')
          self.assertEqual(elem.find('./Delimiter').text, '%EF%BC%A1')
          _, path = self.swift.calls[-1]
          _, query_string = path.split('?')
          args = [part.partition('=')[::2] for part in query_string.split('&')]
          self.assertEqual(sorted(args), [
              ('delimiter', '%EF%BC%A1'),
              ('limit', '1001'),
              ('marker', '%EF%BC%A2'),
              ('prefix', '%EF%BC%A3'),
          ])
  
      def test_bucket_GET_v2_with_nonascii_queries(self):
          bucket_name = 'junk'
          req = Request.blank(
              '/%s?list-type=2&delimiter=\xef\xbc\xa1&start-after=\xef\xbc\xa2&'
              'prefix=\xef\xbc\xa3' % bucket_name,
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./Prefix').text,
                           swob.wsgi_to_str('\xef\xbc\xa3'))
          self.assertEqual(elem.find('./StartAfter').text,
                           swob.wsgi_to_str('\xef\xbc\xa2'))
          self.assertEqual(elem.find('./Delimiter').text,
                           swob.wsgi_to_str('\xef\xbc\xa1'))
          _, path = self.swift.calls[-1]
          _, query_string = path.split('?')
          args = [part.partition('=')[::2] for part in query_string.split('&')]
          self.assertEqual(sorted(args), [
              ('delimiter', '%EF%BC%A1'),
              ('limit', '1001'),
              ('marker', '%EF%BC%A2'),
              ('prefix', '%EF%BC%A3'),
          ])
  
          req = Request.blank(
              '/%s?list-type=2&delimiter=\xef\xbc\xa1&start-after=\xef\xbc\xa2&'
              'prefix=\xef\xbc\xa3&encoding-type=url' % bucket_name,
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./Prefix').text, '%EF%BC%A3')
          self.assertEqual(elem.find('./StartAfter').text, '%EF%BC%A2')
          self.assertEqual(elem.find('./Delimiter').text, '%EF%BC%A1')
          _, path = self.swift.calls[-1]
          _, query_string = path.split('?')
          args = [part.partition('=')[::2] for part in query_string.split('&')]
          self.assertEqual(sorted(args), [
              ('delimiter', '%EF%BC%A1'),
              ('limit', '1001'),
              ('marker', '%EF%BC%A2'),
              ('prefix', '%EF%BC%A3'),
          ])
  
      def test_bucket_GET_with_delimiter_max_keys(self):
          bucket_name = 'junk'
          req = Request.blank('/%s?delimiter=a&max-keys=4' % bucket_name,
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./NextMarker').text,
                           self.objects_list[3]['name'])
          self.assertEqual(elem.find('./MaxKeys').text, '4')
          self.assertEqual(elem.find('./IsTruncated').text, 'true')
  
      def test_bucket_GET_v2_with_delimiter_max_keys(self):
          bucket_name = 'junk'
          req = Request.blank(
              '/%s?list-type=2&delimiter=a&max-keys=2' % bucket_name,
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListBucketResult')
          next_token = elem.find('./NextContinuationToken')
          self.assertIsNotNone(next_token)
          self.assertEqual(elem.find('./MaxKeys').text, '2')
          self.assertEqual(elem.find('./IsTruncated').text, 'true')
  
          req = Request.blank(
              '/%s?list-type=2&delimiter=a&max-keys=2&continuation-token=%s' %
              (bucket_name, next_token.text),
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListBucketResult')
          names = [o.find('./Key').text for o in elem.iterchildren('Contents')]
          self.assertEqual(names[0], 'lily')
  
      def test_bucket_GET_subdir_with_delimiter_max_keys(self):
          bucket_name = 'junk-subdir'
          req = Request.blank('/%s?delimiter=a&max-keys=1' % bucket_name,
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListBucketResult')
          self.assertEqual(elem.find('./NextMarker').text, 'rose')
          self.assertEqual(elem.find('./MaxKeys').text, '1')
          self.assertEqual(elem.find('./IsTruncated').text, 'true')
  
      def test_bucket_GET_v2_fetch_owner(self):
          bucket_name = 'junk'
          req = Request.blank('/%s?list-type=2' % bucket_name,
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
  
          elem = fromstring(body, 'ListBucketResult')
          name = elem.find('./Name').text
          self.assertEqual(name, bucket_name)
  
          objects = elem.iterchildren('Contents')
          for o in objects:
              self.assertIsNone(o.find('./Owner'))
  
          req = Request.blank('/%s?list-type=2&fetch-owner=true' % bucket_name,
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
  
          elem = fromstring(body, 'ListBucketResult')
          name = elem.find('./Name').text
          self.assertEqual(name, bucket_name)
  
          objects = elem.iterchildren('Contents')
          for o in objects:
              self.assertIsNotNone(o.find('./Owner'))
  
      def test_bucket_GET_with_versions_versioning_not_configured(self):
          for obj in self.objects:
              self.swift.register(
                  'HEAD', '/v1/AUTH_test/junk/%s' % quote(obj[0].encode('utf8')),
                  swob.HTTPOk, {}, None)
          # self.swift.register('HEAD', '/v1/AUTH_test/junk/viola',
          #                     swob.HTTPOk, {}, None)
  
          self._add_versions_request(versioned_objects=[])
          req = Request.blank('/junk?versions',
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
  
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListVersionsResult')
          self.assertEqual(elem.find('./Name').text, 'junk')
          self.assertIsNone(elem.find('./Prefix').text)
          self.assertIsNone(elem.find('./KeyMarker').text)
          self.assertIsNone(elem.find('./VersionIdMarker').text)
          self.assertEqual(elem.find('./MaxKeys').text, '1000')
          self.assertEqual(elem.find('./IsTruncated').text, 'false')
          self.assertEqual(elem.findall('./DeleteMarker'), [])
          versions = elem.findall('./Version')
          objects = list(self.objects)
          if six.PY2:
              expected = [v[0].encode('utf-8') for v in objects]
          else:
              expected = [v[0] for v in objects]
          self.assertEqual([v.find('./Key').text for v in versions], expected)
          self.assertEqual([v.find('./IsLatest').text for v in versions],
                           ['true' for v in objects])
          self.assertEqual([v.find('./VersionId').text for v in versions],
                           ['null' for v in objects])
          # Last modified in self.objects is 2011-01-05T02:19:14.275290 but
          # the returned value is rounded up to 2011-01-05T02:19:15Z
          self.assertEqual([v.find('./LastModified').text for v in versions],
                           ['2011-01-05T02:19:15.000Z'] * len(objects))
          self.assertEqual([v.find('./ETag').text for v in versions],
                           [PFS_ETAG if v[0] == 'pfs-obj' else
                            '"0-N"' if v[0] == 'slo' else '"0"'
                            for v in objects])
          self.assertEqual([v.find('./Size').text for v in versions],
                           [str(v[3]) for v in objects])
          self.assertEqual([v.find('./Owner/ID').text for v in versions],
                           ['test:tester' for v in objects])
          self.assertEqual([v.find('./Owner/DisplayName').text
                            for v in versions],
                           ['test:tester' for v in objects])
          self.assertEqual([v.find('./StorageClass').text for v in versions],
                           ['STANDARD' for v in objects])
  
      def _add_versions_request(self, orig_objects=None, versioned_objects=None,
                                bucket='junk'):
          if orig_objects is None:
              orig_objects = self.objects_list
          if versioned_objects is None:
              versioned_objects = self.versioned_objects
          all_versions = versioned_objects + [
              dict(i, version_id='null', is_latest=True)
              for i in orig_objects]
          all_versions.sort(key=lambda o: (
              o['name'], '' if o['version_id'] == 'null' else o['version_id']))
          self.swift.register(
              'GET', '/v1/AUTH_test/%s' % bucket, swob.HTTPOk,
              {'Content-Type': 'application/json'}, json.dumps(all_versions))
  
      def _assert_delete_markers(self, elem):
          delete_markers = elem.findall('./DeleteMarker')
          self.assertEqual(len(delete_markers), 1)
          self.assertEqual(delete_markers[0].find('./IsLatest').text, 'false')
          self.assertEqual(delete_markers[0].find('./VersionId').text, '2')
          self.assertEqual(delete_markers[0].find('./Key').text, 'rose')
  
      def test_bucket_GET_with_versions(self):
          self._add_versions_request()
          req = Request.blank('/junk?versions',
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
  
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListVersionsResult')
          self.assertEqual(elem.find('./Name').text, 'junk')
          self._assert_delete_markers(elem)
          versions = elem.findall('./Version')
          self.assertEqual(len(versions), len(self.objects) + 1)
  
          expected = []
          for o in self.objects_list:
              name = o['name']
              if six.PY2:
                  name = name.encode('utf8')
              expected.append((name, 'true', 'null'))
              if name == 'rose':
                  expected.append((name, 'false', '1'))
          discovered = [
              tuple(e.find('./%s' % key).text for key in (
                  'Key', 'IsLatest', 'VersionId'))
              for e in versions
          ]
          self.assertEqual(expected, discovered)
  
      def test_bucket_GET_with_versions_with_max_keys(self):
          self._add_versions_request()
          req = Request.blank('/junk?versions&max-keys=7',
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
  
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListVersionsResult')
          self.assertEqual(elem.find('./MaxKeys').text, '7')
          self.assertEqual(elem.find('./IsTruncated').text, 'true')
          self._assert_delete_markers(elem)
          versions = elem.findall('./Version')
          self.assertEqual(len(versions), 6)
  
          expected = []
          for o in self.objects_list[:5]:
              name = o['name']
              if six.PY2:
                  name = name.encode('utf8')
              expected.append((name, 'true', 'null'))
              if name == 'rose':
                  expected.append((name, 'false', '1'))
          discovered = [
              tuple(e.find('./%s' % key).text for key in (
                  'Key', 'IsLatest', 'VersionId'))
              for e in versions
          ]
          self.assertEqual(expected, discovered)
  
      def test_bucket_GET_with_versions_with_max_keys_and_key_marker(self):
          self._add_versions_request(orig_objects=self.objects_list[4:])
          req = Request.blank('/junk?versions&max-keys=3&key-marker=ros',
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
  
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListVersionsResult')
          self.assertEqual(elem.find('./MaxKeys').text, '3')
          self.assertEqual(elem.find('./IsTruncated').text, 'true')
          self._assert_delete_markers(elem)
          versions = elem.findall('./Version')
          self.assertEqual(len(versions), 2)
  
          expected = [
              ('rose', 'true', 'null'),
              ('rose', 'false', '1'),
          ]
          discovered = [
              tuple(e.find('./%s' % key).text for key in (
                  'Key', 'IsLatest', 'VersionId'))
              for e in versions
          ]
          self.assertEqual(expected, discovered)
  
      def test_bucket_GET_versions_with_key_marker_and_version_id_marker(self):
          container_listing = [{
              "bytes": 8192,
              "content_type": "binary/octet-stream",
              "hash": "221994040b14294bdf7fbc128e66633c",
              "last_modified": "2019-08-16T19:39:53.152780",
              "name": "subdir/foo",
          }]
          versions_listing = [{
              'bytes': 0,
              'content_type': DELETE_MARKER_CONTENT_TYPE,
              'hash': '0',
              "last_modified": "2019-08-19T19:05:33.565940",
              'name': 'subdir/bar',
              "version_id": "1565241533.55320",
              'is_latest': True,
          }, {
              "bytes": 8192,
              "content_type": "binary/octet-stream",
              "hash": "221994040b14294bdf7fbc128e66633c",
              "last_modified": "2019-08-16T19:39:53.508510",
              "name": "subdir/bar",
              "version_id": "1564984393.68962",
              'is_latest': False,
          }, {
              "bytes": 8192,
              "content_type": "binary/octet-stream",
              "hash": "221994040b14294bdf7fbc128e66633c",
              "last_modified": "2019-08-16T19:39:42.673260",
              "name": "subdir/foo",
              "version_id": "1565984382.67326",
              'is_latest': False,
          }]
          self._add_versions_request(container_listing, versions_listing,
                                     bucket='mybucket')
          req = Request.blank(
              '/mybucket?versions&key-marker=subdir/bar&'
              'version-id-marker=1566589611.065522',
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListVersionsResult')
          self.assertEqual(elem.find('./IsTruncated').text, 'false')
          delete_markers = elem.findall('./DeleteMarker')
          self.assertEqual(['subdir/bar'], [
              o.find('Key').text for o in delete_markers])
          expected = [
              ('subdir/bar', 'false', '1564984393.68962'),
              ('subdir/foo', 'true', 'null'),
              ('subdir/foo', 'false', '1565984382.67326'),
          ]
          discovered = [
              tuple(e.find('./%s' % key).text for key in (
                  'Key', 'IsLatest', 'VersionId'))
              for e in elem.findall('./Version')
          ]
          self.assertEqual(expected, discovered)
  
          self._add_versions_request(container_listing, versions_listing[1:],
                                     bucket='mybucket')
          req = Request.blank(
              '/mybucket?versions&key-marker=subdir/bar&'
              'version-id-marker=1565241533.55320',
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListVersionsResult')
          self.assertEqual(elem.find('./IsTruncated').text, 'false')
          delete_markers = elem.findall('./DeleteMarker')
          self.assertEqual(0, len(delete_markers))
          expected = [
              ('subdir/bar', 'false', '1564984393.68962'),
              ('subdir/foo', 'true', 'null'),
              ('subdir/foo', 'false', '1565984382.67326'),
          ]
          discovered = [
              tuple(e.find('./%s' % key).text for key in (
                  'Key', 'IsLatest', 'VersionId'))
              for e in elem.findall('./Version')
          ]
          self.assertEqual(expected, discovered)
  
          self._add_versions_request([], versions_listing[-1:],
                                     bucket='mybucket')
          req = Request.blank(
              '/mybucket?versions&key-marker=subdir/foo&'
              'version-id-marker=null',
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListVersionsResult')
          self.assertEqual(elem.find('./IsTruncated').text, 'false')
          delete_markers = elem.findall('./DeleteMarker')
          self.assertEqual(0, len(delete_markers))
          expected = [
              ('subdir/foo', 'false', '1565984382.67326'),
          ]
          discovered = [
              tuple(e.find('./%s' % key).text for key in (
                  'Key', 'IsLatest', 'VersionId'))
              for e in elem.findall('./Version')
          ]
          self.assertEqual(expected, discovered)
  
      def test_bucket_GET_versions_with_version_id_marker(self):
          self._add_versions_request()
          req = Request.blank(
              '/junk?versions',
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
  
          # sanity
          elem = fromstring(body, 'ListVersionsResult')
          expected = [('rose', 'false', '2')]
          discovered = [
              tuple(e.find('./%s' % key).text for key in (
                  'Key', 'IsLatest', 'VersionId'))
              for e in elem.findall('./DeleteMarker')
          ]
          self.assertEqual(expected, discovered)
          expected = [
              ('lily', 'true', 'null'),
              (b'lily-\xd8\xaa', 'true', 'null'),
              ('mu', 'true', 'null'),
              ('pfs-obj', 'true', 'null'),
              ('rose', 'true', 'null'),
              ('rose', 'false', '1'),
              ('slo', 'true', 'null'),
              ('viola', 'true', 'null'),
              ('with space', 'true', 'null'),
              ('with%20space', 'true', 'null'),
          ]
          if not six.PY2:
              item = list(expected[1])
              item[0] = item[0].decode('utf8')
              expected[1] = tuple(item)
  
          discovered = [
              tuple(e.find('./%s' % key).text for key in (
                  'Key', 'IsLatest', 'VersionId'))
              for e in elem.findall('./Version')
          ]
          self.assertEqual(expected, discovered)
  
          self._add_versions_request(self.objects_list[5:])
          req = Request.blank(
              '/junk?versions&key-marker=rose&version-id-marker=null',
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
  
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListVersionsResult')
          self.assertEqual(elem.find('./IsTruncated').text, 'false')
          delete_markers = elem.findall('./DeleteMarker')
          self.assertEqual(len(delete_markers), 1)
  
          expected = [
              ('rose', 'false', '1'),
              ('slo', 'true', 'null'),
              ('viola', 'true', 'null'),
              ('with space', 'true', 'null'),
              ('with%20space', 'true', 'null'),
          ]
          discovered = [
              tuple(e.find('./%s' % key).text for key in (
                  'Key', 'IsLatest', 'VersionId'))
              for e in elem.findall('./Version')
          ]
          self.assertEqual(expected, discovered)
  
          # N.B. versions are sorted most recent to oldest
          self._add_versions_request(self.objects_list[5:],
                                     self.versioned_objects[1:])
          req = Request.blank(
              '/junk?versions&key-marker=rose&version-id-marker=2',
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
  
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListVersionsResult')
          self.assertEqual(elem.find('./IsTruncated').text, 'false')
          delete_markers = elem.findall('./DeleteMarker')
          self.assertEqual(len(delete_markers), 0)
  
          expected = [
              ('rose', 'false', '1'),
              ('slo', 'true', 'null'),
              ('viola', 'true', 'null'),
              ('with space', 'true', 'null'),
              ('with%20space', 'true', 'null'),
          ]
          discovered = [
              tuple(e.find('./%s' % key).text for key in (
                  'Key', 'IsLatest', 'VersionId'))
              for e in elem.findall('./Version')
          ]
          self.assertEqual(expected, discovered)
  
          self._add_versions_request(self.objects_list[5:],
                                     self.versioned_objects[2:])
          req = Request.blank(
              '/junk?versions&key-marker=rose&version-id-marker=1',
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
  
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListVersionsResult')
          self.assertEqual(elem.find('./IsTruncated').text, 'false')
          delete_markers = elem.findall('./DeleteMarker')
          self.assertEqual(len(delete_markers), 0)
  
          expected = [
              ('slo', 'true', 'null'),
              ('viola', 'true', 'null'),
              ('with space', 'true', 'null'),
              ('with%20space', 'true', 'null'),
          ]
          discovered = [
              tuple(e.find('./%s' % key).text for key in (
                  'Key', 'IsLatest', 'VersionId'))
              for e in elem.findall('./Version')
          ]
          self.assertEqual(expected, discovered)
  
      def test_bucket_GET_versions_non_existent_version_id_marker(self):
          self._add_versions_request(orig_objects=self.objects_list[5:])
          req = Request.blank(
              '/junk?versions&key-marker=rose&'
              'version-id-marker=null',
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
  
          self.assertEqual(status.split()[0], '200', body)
          elem = fromstring(body, 'ListVersionsResult')
          self.assertEqual(elem.find('./Name').text, 'junk')
          delete_markers = elem.findall('./DeleteMarker')
          self.assertEqual(len(delete_markers), 1)
  
          expected = [
              ('rose', 'false', '1'),
              ('slo', 'true', 'null'),
              ('viola', 'true', 'null'),
              ('with space', 'true', 'null'),
              ('with%20space', 'true', 'null'),
          ]
          discovered = [
              tuple(e.find('./%s' % key).text for key in (
                  'Key', 'IsLatest', 'VersionId'))
              for e in elem.findall('./Version')
          ]
          self.assertEqual(expected, discovered)
          self.assertEqual(self.swift.calls, [
              ('GET', normalize_path('/v1/AUTH_test/junk?'
               'limit=1001&marker=rose&version_marker=null&versions=')),
          ])
  
      def test_bucket_GET_versions_prefix(self):
          container_listing = [{
              "bytes": 8192,
              "content_type": "binary/octet-stream",
              "hash": "221994040b14294bdf7fbc128e66633c",
              "last_modified": "2019-08-16T19:39:53.152780",
              "name": "subdir/foo",
          }]
          versions_listing = [{
              "bytes": 8192,
              "content_type": "binary/octet-stream",
              "hash": "221994040b14294bdf7fbc128e66633c",
              "last_modified": "2019-08-16T19:39:53.508510",
              "name": "subdir/bar",
              "version_id": "1565984393.68962",
              "is_latest": True,
          }, {
              'bytes': 0,
              'content_type': DELETE_MARKER_CONTENT_TYPE,
              'hash': '0',
              "last_modified": "2019-08-19T19:05:33.565940",
              'name': 'subdir/bar',
              'version_id': '1566241533.55320',
              'is_latest': False,
          }, {
              "bytes": 8192,
              "content_type": "binary/octet-stream",
              "hash": "221994040b14294bdf7fbc128e66633c",
              "last_modified": "2019-08-16T19:39:42.673260",
              "name": "subdir/foo",
              "version_id": "1565984382.67326",
              'is_latest': False,
          }]
          self._add_versions_request(container_listing, versions_listing)
          req = Request.blank(
              '/junk?versions&prefix=subdir/',
              environ={'REQUEST_METHOD': 'GET'},
              headers={'Authorization': 'AWS test:tester:hmac',
                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
  
          self.assertEqual(status.split()[0], '200')
          elem = fromstring(body, 'ListVersionsResult')
          self.assertEqual(elem.find('./Name').text, 'junk')
          delete_markers = elem.findall('./DeleteMarker')
          self.assertEqual(len(delete_markers), 1)
  
          expected = [
              ('subdir/bar', 'true', '1565984393.68962'),
              ('subdir/foo', 'true', 'null'),
              ('subdir/foo', 'false', '1565984382.67326'),
          ]
          discovered = [
              tuple(e.find('./%s' % key).text for key in (
                  'Key', 'IsLatest', 'VersionId'))
              for e in elem.findall('./Version')
          ]
          self.assertEqual(expected, discovered)
  
          self.assertEqual(self.swift.calls, [
              ('GET', normalize_path('/v1/AUTH_test/junk'
               '?limit=1001&prefix=subdir/&versions=')),
          ])
  
      @s3acl
      def test_bucket_PUT_error(self):
          code = self._test_method_error('PUT', '/bucket', swob.HTTPCreated,
                                         headers={'Content-Length': 'a'})
          self.assertEqual(code, 'InvalidArgument')
          code = self._test_method_error('PUT', '/bucket', swob.HTTPCreated,
                                         headers={'Content-Length': '-1'})
          self.assertEqual(code, 'InvalidArgument')
          code = self._test_method_error('PUT', '/bucket', swob.HTTPUnauthorized)
          self.assertEqual(code, 'SignatureDoesNotMatch')
          code = self._test_method_error('PUT', '/bucket', swob.HTTPForbidden)
          self.assertEqual(code, 'AccessDenied')
          code = self._test_method_error('PUT', '/bucket', swob.HTTPAccepted)
          self.assertEqual(code, 'BucketAlreadyOwnedByYou')
          with mock.patch(
                  'swift.common.middleware.s3api.s3request.get_container_info',
                  return_value={'sysmeta': {'s3api-acl': '{"Owner": "nope"}'}}):
              code = self._test_method_error(
                  'PUT', '/bucket', swob.HTTPAccepted)
          self.assertEqual(code, 'BucketAlreadyExists')
          code = self._test_method_error('PUT', '/bucket', swob.HTTPServerError)
          self.assertEqual(code, 'InternalError')
          code = self._test_method_error(
              'PUT', '/bucket', swob.HTTPServiceUnavailable)
          self.assertEqual(code, 'ServiceUnavailable')
          code = self._test_method_error(
              'PUT', '/bucket+bucket', swob.HTTPCreated)
          self.assertEqual(code, 'InvalidBucketName')
          code = self._test_method_error(
              'PUT', '/192.168.11.1', swob.HTTPCreated)
          self.assertEqual(code, 'InvalidBucketName')
          code = self._test_method_error(
              'PUT', '/bucket.-bucket', swob.HTTPCreated)
          self.assertEqual(code, 'InvalidBucketName')
          code = self._test_method_error(
              'PUT', '/bucket-.bucket', swob.HTTPCreated)
          self.assertEqual(code, 'InvalidBucketName')
          code = self._test_method_error('PUT', '/bucket*', swob.HTTPCreated)
          self.assertEqual(code, 'InvalidBucketName')
          code = self._test_method_error('PUT', '/b', swob.HTTPCreated)
          self.assertEqual(code, 'InvalidBucketName')
          code = self._test_method_error(
              'PUT', '/%s' % ''.join(['b' for x in range(64)]),
              swob.HTTPCreated)
          self.assertEqual(code, 'InvalidBucketName')
  
      @s3acl(s3acl_only=True)
      def test_bucket_PUT_error_non_swift_owner(self):
          code = self._test_method_error('PUT', '/bucket', swob.HTTPAccepted,
                                         env={'swift_owner': False})
          self.assertEqual(code, 'AccessDenied')
  
      @s3acl
      def test_bucket_PUT_bucket_already_owned_by_you(self):
          self.swift.register(
              'PUT', '/v1/AUTH_test/bucket', swob.HTTPAccepted,
              {'X-Container-Object-Count': 0}, None)
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'PUT'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status, '409 Conflict')
          self.assertIn(b'BucketAlreadyOwnedByYou', body)
  
      @s3acl
      def test_bucket_PUT_first_put_fail(self):
          self.swift.register(
              'PUT', '/v1/AUTH_test/bucket',
              swob.HTTPServiceUnavailable,
              {'X-Container-Object-Count': 0}, None)
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'PUT'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status, '503 Service Unavailable')
          # The last call was PUT not POST for acl set
          self.assertEqual(self.swift.calls, [
              ('PUT', '/v1/AUTH_test/bucket'),
          ])
  
      @s3acl
      def test_bucket_PUT(self):
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'PUT'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(body, b'')
          self.assertEqual(status.split()[0], '200')
          self.assertEqual(headers['Location'], '/bucket')
  
          # Apparently some clients will include a chunked transfer-encoding
          # even with no body
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'PUT'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header(),
                                       'Transfer-Encoding': 'chunked'})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(body, b'')
          self.assertEqual(status.split()[0], '200')
          self.assertEqual(headers['Location'], '/bucket')
  
          with UnreadableInput(self) as fake_input:
              req = Request.blank(
                  '/bucket',
                  environ={'REQUEST_METHOD': 'PUT',
                           'wsgi.input': fake_input},
                  headers={'Authorization': 'AWS test:tester:hmac',
                           'Date': self.get_date_header()})
              status, headers, body = self.call_s3api(req)
          self.assertEqual(body, b'')
          self.assertEqual(status.split()[0], '200')
          self.assertEqual(headers['Location'], '/bucket')
  
      def _test_bucket_PUT_with_location(self, root_element):
          elem = Element(root_element)
          SubElement(elem, 'LocationConstraint').text = 'us-east-1'
          xml = tostring(elem)
  
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'PUT'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()},
                              body=xml)
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
  
      @s3acl
      def test_bucket_PUT_with_location(self):
          self._test_bucket_PUT_with_location('CreateBucketConfiguration')
  
      @s3acl
      def test_bucket_PUT_with_ami_location(self):
          # ec2-ami-tools apparently uses CreateBucketConstraint instead?
          self._test_bucket_PUT_with_location('CreateBucketConstraint')
  
      @s3acl
      def test_bucket_PUT_with_strange_location(self):
          # Even crazier: it doesn't seem to matter
          self._test_bucket_PUT_with_location('foo')
  
      def test_bucket_PUT_with_mixed_case_location(self):
          self.s3api.conf.location = 'RegionOne'
          elem = Element('CreateBucketConfiguration')
          # We've observed some clients (like aws-sdk-net) shift regions
          # to lower case
          SubElement(elem, 'LocationConstraint').text = 'regionone'
          headers = {
              'Authorization': 'AWS4-HMAC-SHA256 ' + ', '.join([
                  'Credential=test:tester/%s/regionone/s3/aws4_request' %
                  self.get_v4_amz_date_header().split('T', 1)[0],
                  'SignedHeaders=host',
                  'Signature=X',
              ]),
              'Date': self.get_date_header(),
              'x-amz-content-sha256': 'UNSIGNED-PAYLOAD',
          }
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'PUT'},
                              headers=headers,
                              body=tostring(elem))
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200', body)
  
      def test_bucket_PUT_with_canned_acl(self):
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'PUT'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header(),
                                       'X-Amz-Acl': 'public-read'})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
          _, _, headers = self.swift.calls_with_headers[-1]
          self.assertTrue('X-Container-Read' in headers)
          self.assertEqual(headers.get('X-Container-Read'), '.r:*,.rlistings')
          self.assertNotIn('X-Container-Sysmeta-S3api-Acl', headers)
  
      @s3acl(s3acl_only=True)
      def test_bucket_PUT_with_canned_s3acl(self):
          account = 'test:tester'
          acl = \
              encode_acl('container', ACLPublicRead(Owner(account, account)))
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'PUT'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header(),
                                       'X-Amz-Acl': 'public-read'})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '200')
          _, _, headers = self.swift.calls_with_headers[-1]
          self.assertNotIn('X-Container-Read', headers)
          self.assertIn('X-Container-Sysmeta-S3api-Acl', headers)
          self.assertEqual(headers.get('X-Container-Sysmeta-S3api-Acl'),
                           acl['x-container-sysmeta-s3api-acl'])
  
      @s3acl
      def test_bucket_PUT_with_location_error(self):
          elem = Element('CreateBucketConfiguration')
          SubElement(elem, 'LocationConstraint').text = 'XXX'
          xml = tostring(elem)
  
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'PUT'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()},
                              body=xml)
          status, headers, body = self.call_s3api(req)
          self.assertEqual(self._get_error_code(body),
                           'InvalidLocationConstraint')
  
      @s3acl
      def test_bucket_PUT_with_location_invalid_xml(self):
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'PUT'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()},
                              body='invalid_xml')
          status, headers, body = self.call_s3api(req)
          self.assertEqual(self._get_error_code(body), 'MalformedXML')
  
      def _test_method_error_delete(self, path, sw_resp):
          self.swift.register('HEAD', '/v1/AUTH_test' + path, sw_resp, {}, None)
          return self._test_method_error('DELETE', path, sw_resp)
  
      @s3acl
      def test_bucket_DELETE_error(self):
          code = self._test_method_error_delete('/bucket', swob.HTTPUnauthorized)
          self.assertEqual(code, 'SignatureDoesNotMatch')
          code = self._test_method_error_delete('/bucket', swob.HTTPForbidden)
          self.assertEqual(code, 'AccessDenied')
          code = self._test_method_error_delete('/bucket', swob.HTTPNotFound)
          self.assertEqual(code, 'NoSuchBucket')
          code = self._test_method_error_delete('/bucket', swob.HTTPServerError)
          self.assertEqual(code, 'InternalError')
  
          # bucket not empty is now validated at s3api
          self.swift._responses.get(('HEAD', '/v1/AUTH_test/bucket'))
          self.swift.register('HEAD', '/v1/AUTH_test/bucket', swob.HTTPNoContent,
                              {'X-Container-Object-Count': '1'}, None)
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'DELETE'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, _headers, body = self.call_s3api(req)
          self.assertEqual('409 Conflict', status)
          self.assertEqual('BucketNotEmpty', self._get_error_code(body))
          self.assertNotIn('You must delete all versions in the bucket',
                           self._get_error_message(body))
  
      @s3acl
      def test_bucket_DELETE_error_with_enabled_versioning(self):
          self.swift.register('HEAD', '/v1/AUTH_test/bucket', swob.HTTPNoContent,
                              {'X-Container-Object-Count': '1',
                               'X-Container-Sysmeta-Versions-Enabled': 'True'},
                              None)
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'DELETE'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, _headers, body = self.call_s3api(req)
          self.assertEqual('409 Conflict', status)
          self.assertEqual('BucketNotEmpty', self._get_error_code(body))
          self.assertIn('You must delete all versions in the bucket',
                        self._get_error_message(body))
  
      @s3acl
      def test_bucket_DELETE_error_with_suspended_versioning(self):
          self.swift.register('HEAD', '/v1/AUTH_test/bucket', swob.HTTPNoContent,
                              {'X-Container-Object-Count': '1',
                               'X-Container-Sysmeta-Versions-Enabled': 'False'},
                              None)
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'DELETE'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, _headers, body = self.call_s3api(req)
          self.assertEqual('409 Conflict', status)
          self.assertEqual('BucketNotEmpty', self._get_error_code(body))
          self.assertIn('You must delete all versions in the bucket',
                        self._get_error_message(body))
  
      @s3acl
      def test_bucket_DELETE(self):
          # overwrite default HEAD to return x-container-object-count
          self.swift.register(
              'HEAD', '/v1/AUTH_test/bucket', swob.HTTPNoContent,
              {'X-Container-Object-Count': 0}, None)
  
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'DELETE'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '204')
  
      @s3acl
      def test_bucket_DELETE_with_empty_versioning(self):
          self.swift.register('HEAD', '/v1/AUTH_test/bucket+versioning',
                              swob.HTTPNoContent, {}, None)
          self.swift.register('DELETE', '/v1/AUTH_test/bucket+versioning',
                              swob.HTTPNoContent, {}, None)
          # overwrite default HEAD to return x-container-object-count
          self.swift.register(
              'HEAD', '/v1/AUTH_test/bucket', swob.HTTPNoContent,
              {'X-Container-Object-Count': 0}, None)
  
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'DELETE'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '204')
  
      @s3acl
      def test_bucket_DELETE_error_while_segment_bucket_delete(self):
          # An error occurred while deleting segment objects
          self.swift.register('DELETE', '/v1/AUTH_test/bucket+segments/lily',
                              swob.HTTPServiceUnavailable, {}, json.dumps([]))
          # overwrite default HEAD to return x-container-object-count
          self.swift.register(
              'HEAD', '/v1/AUTH_test/bucket', swob.HTTPNoContent,
              {'X-Container-Object-Count': 0}, None)
  
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': 'DELETE'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
          status, headers, body = self.call_s3api(req)
          self.assertEqual(status.split()[0], '503')
          called = [(method, path) for method, path, _ in
                    self.swift.calls_with_headers]
          # Don't delete original bucket when error occurred in segment container
          self.assertNotIn(('DELETE', '/v1/AUTH_test/bucket'), called)
  
      def _test_bucket_for_s3acl(self, method, account):
          req = Request.blank('/bucket',
                              environ={'REQUEST_METHOD': method},
                              headers={'Authorization': 'AWS %s:hmac' % account,
                                       'Date': self.get_date_header()})
  
          return self.call_s3api(req)
  
      @s3acl(s3acl_only=True)
      def test_bucket_GET_without_permission(self):
          status, headers, body = self._test_bucket_for_s3acl('GET',
                                                              'test:other')
          self.assertEqual(self._get_error_code(body), 'AccessDenied')
  
      @s3acl(s3acl_only=True)
      def test_bucket_GET_with_read_permission(self):
          status, headers, body = self._test_bucket_for_s3acl('GET',
                                                              'test:read')
          self.assertEqual(status.split()[0], '200')
  
      @s3acl(s3acl_only=True)
      def test_bucket_GET_with_fullcontrol_permission(self):
          status, headers, body = \
              self._test_bucket_for_s3acl('GET', 'test:full_control')
          self.assertEqual(status.split()[0], '200')
  
      @s3acl(s3acl_only=True)
      def test_bucket_GET_with_owner_permission(self):
          status, headers, body = self._test_bucket_for_s3acl('GET',
                                                              'test:tester')
          self.assertEqual(status.split()[0], '200')
  
      def _test_bucket_GET_canned_acl(self, bucket):
          req = Request.blank('/%s' % bucket,
                              environ={'REQUEST_METHOD': 'GET'},
                              headers={'Authorization': 'AWS test:tester:hmac',
                                       'Date': self.get_date_header()})
  
          return self.call_s3api(req)
  
      @s3acl(s3acl_only=True)
      def test_bucket_GET_authenticated_users(self):
          status, headers, body = \
              self._test_bucket_GET_canned_acl('authenticated')
          self.assertEqual(status.split()[0], '200')
  
      @s3acl(s3acl_only=True)
      def test_bucket_GET_all_users(self):
          status, headers, body = self._test_bucket_GET_canned_acl('public')
          self.assertEqual(status.split()[0], '200')
  
      @s3acl(s3acl_only=True)
      def test_bucket_DELETE_without_permission(self):
          status, headers, body = self._test_bucket_for_s3acl('DELETE',
                                                              'test:other')
          self.assertEqual(self._get_error_code(body), 'AccessDenied')
          # Don't delete anything in backend Swift
          called = [method for method, _, _ in self.swift.calls_with_headers]
          self.assertNotIn('DELETE', called)
  
      @s3acl(s3acl_only=True)
      def test_bucket_DELETE_with_write_permission(self):
          status, headers, body = self._test_bucket_for_s3acl('DELETE',
                                                              'test:write')
          self.assertEqual(self._get_error_code(body), 'AccessDenied')
          # Don't delete anything in backend Swift
          called = [method for method, _, _ in self.swift.calls_with_headers]
          self.assertNotIn('DELETE', called)
  
      @s3acl(s3acl_only=True)
      def test_bucket_DELETE_with_fullcontrol_permission(self):
          status, headers, body = \
              self._test_bucket_for_s3acl('DELETE', 'test:full_control')
          self.assertEqual(self._get_error_code(body), 'AccessDenied')
          # Don't delete anything in backend Swift
          called = [method for method, _, _ in self.swift.calls_with_headers]
          self.assertNotIn('DELETE', called)
  
  
  if __name__ == '__main__':
      unittest.main()
  