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

s3tests: add test_versioning_obj_create_read_remove()

Browse source 
Signed-off-by: Yehuda Sadeh <yehuda@inktank.com>
This commit is contained in:
Yehuda Sadeh 2014-10-29 15:26:57 -07:00
parent 032f41f6f3
commit 0f0e833801
2 changed files with 58 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
s3tests/functional/__init__.py
View file

@ -72,7 +72,7 @@ def nuke_prefixed_buckets_on_conn(prefix, name, conn):
key=key,
)
# key.set_canned_acl('private')
key.delete()
bucket.delete_key(key.name, version_id = key.version_id)
bucket.delete()
except boto.exception.S3ResponseError as e:
if e.error_code != 'AccessDenied':

58
s3tests/functional/test_s3.py
View file

@ -5188,7 +5188,6 @@ def check_versioning(bucket, status):
@attr('versioning')
def test_versioning_bucket_create_suspend():
bucket = get_new_bucket()
print bucket.get_versioning_status()
check_versioning(bucket, None)
bucket.configure_versioning(False)
@ -5203,3 +5202,60 @@ def test_versioning_bucket_create_suspend():
bucket.configure_versioning(False)
check_versioning(bucket, "Suspended")
@attr(resource='object')
@attr(method='create')
@attr(operation='create versioned object')
@attr(assertion='can create access and remove appropriate versions')
@attr('versioning')
def test_versioning_obj_create_read_remove():
bucket = get_new_bucket()
objname = 'testobj'
total = 5
c = []
for i in xrange(total):
c.append('content-{i}'.format(i=i))
key = bucket.new_key(objname)
key.set_contents_from_string(c[i])
if i == 0:
bucket.configure_versioning(True)
check_versioning(bucket, "Enabled")
# check to see if object is pointing at correct version
key = bucket.get_key(objname)
eq(key.get_contents_as_string(), c[-1])
print 'total', total
k = []
for o in bucket.list_versions():
k.insert(0, o)
for j in xrange(total):
print j, k[j], k[j].version_id
for j in xrange(total):
# check by versioned key
rmkey = k.pop(-1)
eq(rmkey.get_contents_as_string(), c[-1])
# remove version
print 'removing version_id=', rmkey.version_id
bucket.delete_key(rmkey.name, version_id = rmkey.version_id)
c.pop(-1)
# check to see if object is pointing at correct version
key = bucket.get_key(objname)
if len(c) > 0:
print c[-1]
eq(key.get_contents_as_string(), c[-1])
i = len(c)
for key in bucket.list_versions():
i -= 1
eq(key.get_contents_as_string(), c[i])
else:
eq(key, None)