Tweak bucket.delete_keys for s3bucket_path.

This commit is contained in:
zeb 2015-09-04 18:48:48 +02:00
parent 8b0f42afed
commit b59a77d5bb
3 changed files with 52 additions and 2 deletions

View File

@ -22,10 +22,17 @@ def parse_key_name(pth):
class ResponseObject(_TemplateEnvironmentMixin):
def __init__(self, backend, bucket_name_from_url, parse_key_name):
def __init__(self, backend, bucket_name_from_url, parse_key_name,
is_delete_keys=None):
self.backend = backend
self.bucket_name_from_url = bucket_name_from_url
self.parse_key_name = parse_key_name
if is_delete_keys:
self.is_delete_keys = is_delete_keys
@staticmethod
def is_delete_keys(path, bucket_name):
return path == u'/?delete'
def all_buckets(self):
# No bucket specified. Listing all buckets
@ -209,7 +216,7 @@ class ResponseObject(_TemplateEnvironmentMixin):
return 409, headers, template.render(bucket=removed_bucket)
def _bucket_response_post(self, request, bucket_name, headers):
if request.path == u'/?delete':
if self.is_delete_keys(request.path, bucket_name):
return self._bucket_response_delete_keys(request, bucket_name, headers)
# POST to bucket-url should create file from form

View File

@ -9,8 +9,14 @@ from moto.s3.responses import ResponseObject
def parse_key_name(pth):
return "/".join(pth.rstrip("/").split("/")[2:])
def is_delete_keys(path, bucket_name):
return path == u'/' + bucket_name + u'/?delete'
S3BucketPathResponseInstance = ResponseObject(
s3bucket_path_backend,
bucket_name_from_url,
parse_key_name,
is_delete_keys,
)

View File

@ -281,3 +281,40 @@ def test_bucket_key_listing_order():
delimiter = '/'
keys = [x.name for x in bucket.list(prefix + 'x', delimiter)]
keys.should.equal(['toplevel/x/'])
@mock_s3bucket_path
def test_delete_keys():
conn = create_connection()
bucket = conn.create_bucket('foobar')
Key(bucket=bucket, name='file1').set_contents_from_string('abc')
Key(bucket=bucket, name='file2').set_contents_from_string('abc')
Key(bucket=bucket, name='file3').set_contents_from_string('abc')
Key(bucket=bucket, name='file4').set_contents_from_string('abc')
result = bucket.delete_keys(['file2', 'file3'])
result.deleted.should.have.length_of(2)
result.errors.should.have.length_of(0)
keys = bucket.get_all_keys()
keys.should.have.length_of(2)
keys[0].name.should.equal('file1')
@mock_s3bucket_path
def test_delete_keys_with_invalid():
conn = create_connection()
bucket = conn.create_bucket('foobar')
Key(bucket=bucket, name='file1').set_contents_from_string('abc')
Key(bucket=bucket, name='file2').set_contents_from_string('abc')
Key(bucket=bucket, name='file3').set_contents_from_string('abc')
Key(bucket=bucket, name='file4').set_contents_from_string('abc')
result = bucket.delete_keys(['abc', 'file3'])
result.deleted.should.have.length_of(1)
result.errors.should.have.length_of(1)
keys = bucket.get_all_keys()
keys.should.have.length_of(3)
keys[0].name.should.equal('file1')