dkirillov/s3-tests
1
0
Fork 0
forked from TrueCloudLab/s3-tests
Code Pull requests Activity

Add test case for getting version-id when upload objects to bucket

Browse source 
For default-bucket and versioning-suspended-bucket, no version-id should return when upload.
For versioning-enabled-bucket, an non-empty version-id should return and equal to what
bucket list shows.

Signed-off-by: Xinying Song <songxinying@cloudin.cn>
(cherry picked from commit 9506ac220f)
This commit is contained in:
Xinying Song 2017-11-16 14:23:36 +08:00 committed by Casey Bodley
parent 55e1c95b2f
commit 277c2993eb
1 changed files with 66 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

66
s3tests/functional/test_s3.py
View file

@ -34,6 +34,7 @@ from httplib import HTTPConnection, HTTPSConnection
from urlparse import urlparse
from nose.tools import eq_ as eq
from nose.tools import assert_is_none
from nose.plugins.attrib import attr
from nose.plugins.skip import SkipTest
from boto.s3.tagging import TagSet
@ -9302,3 +9303,68 @@ def test_delete_tags_obj_public():
tags = _get_obj_tags(bucket, key.name)
eq(len(tags),0)
#eq(input_tagset, res_tagset)
@attr(resource='object')
@attr(method='put')
@attr(operation='test whether a correct version-id returned')
@attr(assertion='version-id is same as bucket list')
def test_versioning_bucket_atomic_upload_return_version_id():
# for versioning-enabled-bucket, an non-empty version-id should return
bucket = get_new_bucket()
check_configure_versioning_retry(bucket, True, "Enabled")
key_name = 'bar'
key = bucket.new_key(key_name)
key.set_contents_from_string(key_name)
# list_verions will return an non-empty version-id
li = bucket.list_versions()
for k in li:
eq(key.version_id, k.version_id)
# for versioning-default-bucket, no version-id should return.
bucket = get_new_bucket()
key_name = 'baz'
key = bucket.new_key(key_name)
key.set_contents_from_string(key_name)
assert_is_none(key.version_id)
# for versioning-suspended-bucket, no version-id should return.
bucket = get_new_bucket()
check_configure_versioning_retry(bucket, False, "Suspended")
key_name = 'foo'
key = bucket.new_key(key_name)
key.set_contents_from_string(key_name)
assert_is_none(key.version_id)
@attr(resource='object')
@attr(method='put')
@attr(operation='test whether a correct version-id returned')
@attr(assertion='version-id is same as bucket list')
def test_versioning_bucket_multipart_upload_return_version_id():
content_type='text/bla'
objlen = 30 * 1024 * 1024
# for versioning-enabled-bucket, an non-empty version-id should return
bucket = get_new_bucket()
check_configure_versioning_retry(bucket, True, "Enabled")
key_name = 'bar'
(upload, data) = _multipart_upload(bucket, key_name, objlen, headers={'Content-Type': content_type}, metadata={'foo': 'baz'})
res = upload.complete_upload()
# list_verions will return an non-empty version-id
li = bucket.list_versions()
for k in li:
eq(k.version_id, res.version_id)
# for versioning-default-bucket, no version-id should return.
bucket = get_new_bucket()
key_name = 'baz'
(upload, data) = _multipart_upload(bucket, key_name, objlen, headers={'Content-Type': content_type}, metadata={'foo': 'baz'})
res = upload.complete_upload()
assert_is_none(res.version_id)
# for versioning-suspended-bucket, no version-id should return
bucket = get_new_bucket()
check_configure_versioning_retry(bucket, False, "Suspended")
key_name = 'foo'
(upload, data) = _multipart_upload(bucket, key_name, objlen, headers={'Content-Type': content_type}, metadata={'foo': 'baz'})
res = upload.complete_upload()
assert_is_none(res.version_id)