s3tests: test list bucket versions

Signed-off-by: Yehuda Sadeh <yehuda@inktank.com>
This commit is contained in:
Yehuda Sadeh 2015-01-08 16:11:23 -08:00
parent 460e3f1f1e
commit 7a0932fe02

View file

@ -22,6 +22,7 @@ import pytz
import json import json
import httplib2 import httplib2
import threading import threading
import itertools
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
@ -5250,6 +5251,9 @@ def check_obj_versions(bucket, objname, keys, contents):
check_head_obj_content(key, contents[-1]) check_head_obj_content(key, contents[-1])
i = len(contents) i = len(contents)
for key in bucket.list_versions(): for key in bucket.list_versions():
if key.name != objname:
continue
i -= 1 i -= 1
eq(keys[i].version_id or 'null', key.version_id) eq(keys[i].version_id or 'null', key.version_id)
print 'testing obj version-id=', key.version_id print 'testing obj version-id=', key.version_id
@ -5272,6 +5276,8 @@ def create_multiple_versions(bucket, objname, num_versions, k = None, c = None):
k_pos = len(k) k_pos = len(k)
i = 0 i = 0
for o in bucket.list_versions(): for o in bucket.list_versions():
if o.name != objname:
continue
i += 1 i += 1
if i > num_versions: if i > num_versions:
break break
@ -5538,6 +5544,43 @@ def test_versioning_obj_create_overwrite_multipart():
@attr(resource='object')
@attr(method='multipart')
@attr(operation='list versioned objects')
@attr(assertion='everything works')
@attr('versioning')
def test_versioning_obj_list_marker():
bucket = get_new_bucket()
check_configure_versioning_retry(bucket, True, "Enabled")
objname = 'testobj'
objname2 = 'testobj-1'
num_vers = 5
(k, c) = create_multiple_versions(bucket, objname, num_vers)
(k2, c2) = create_multiple_versions(bucket, objname2, num_vers)
k.reverse()
k2.reverse()
allkeys = k + k2
names = []
for key1, key2 in itertools.izip_longest(bucket.list_versions(), allkeys):
eq(key1.version_id, key2.version_id)
names.append(key1.name)
for i in xrange(len(allkeys)):
for key1, key2 in itertools.izip_longest(bucket.list_versions(key_marker=names[i], version_id_marker=allkeys[i].version_id), allkeys[i+1:]):
eq(key1.version_id, key2.version_id)
# with nonexisting version id, skip to next object
for key1, key2 in itertools.izip_longest(bucket.list_versions(key_marker=objname, version_id_marker='nosuchversion'), allkeys[5:]):
eq(key1.version_id, key2.version_id)
@attr(resource='object') @attr(resource='object')
@attr(method='multipart') @attr(method='multipart')
@attr(operation='create and test versioned object copying') @attr(operation='create and test versioned object copying')