TrueCloudLab/s3-tests
7
0
Fork 1
mirror of https://github.com/ceph/s3-tests.git synced 2024-11-21 23:29:47 +00:00
Code Issues 1 Projects Releases Packages Wiki Activity

add functional tests for bucket HEAD

Browse source 
Signed-off-by: Yehuda Sadeh <yehuda.sadeh@dreamhost.com>
This commit is contained in:
Yehuda Sadeh 2012-04-19 13:33:08 -07:00
parent 1551c5b087
commit ebbafdb3e8
1 changed files with 106 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

106
s3tests/functional/test_s3.py
View file

@ -1021,6 +1021,16 @@ def _setup_request(bucket_acl=None, object_acl=None):
return (bucket, key)
def _setup_bucket_request(bucket_acl=None):
"""
set up a (new or existing) bucket with specified acl
"""
bucket = get_new_bucket()
if bucket_acl is not None:
bucket.set_acl(bucket_acl)
return bucket
def _make_request(method, bucket, key, body=None, authenticated=False):
"""
@ -1047,6 +1057,56 @@ def _make_request(method, bucket, key, body=None, authenticated=False):
print res.status, res.reason
return res
def _make_request(method, bucket, key, body=None, authenticated=False):
"""
issue a request for a specified method, on a specified <bucket,key>,
with a specified (optional) body (encrypted per the connection), and
return the response (status, reason)
"""
if authenticated:
url = key.generate_url(100000, method=method)
o = urlparse(url)
path = o.path + '?' + o.query
else:
path = '/{bucket}/{obj}'.format(bucket=key.bucket.name, obj=key.name)
if s3.main.is_secure:
class_ = HTTPSConnection
else:
class_ = HTTPConnection
c = class_(s3.main.host, s3.main.port, strict=True)
c.request(method, path, body=body)
res = c.getresponse()
print res.status, res.reason
return res
def _make_bucket_request(method, bucket, body=None, authenticated=False):
"""
issue a request for a specified method, on a specified <bucket,key>,
with a specified (optional) body (encrypted per the connection), and
return the response (status, reason)
"""
if authenticated:
url = bucket.generate_url(100000, method=method)
o = urlparse(url)
path = o.path + '?' + o.query
else:
path = '/{bucket}'.format(bucket=bucket.name)
if s3.main.is_secure:
class_ = HTTPSConnection
else:
class_ = HTTPConnection
c = class_(s3.main.host, s3.main.port, strict=True)
c.request(method, path, body=body)
res = c.getresponse()
print res.status, res.reason
return res
@attr(resource='object')
@attr(method='get')
@ -1085,6 +1145,52 @@ def test_object_raw_get_object_gone():
eq(res.status, 404)
eq(res.reason, 'Not Found')
def _head_bucket(bucket, authenticated=True):
res = _make_bucket_request('HEAD', bucket, authenticated=authenticated)
eq(res.status, 200)
eq(res.reason, 'OK')
obj_count = res.getheader('x-rgw-object-count')
assert obj_count is not None, "x-rgw-object-count wasn't returned"
bytes_used = res.getheader('x-rgw-bytes-used')
assert bytes_used is not None, "x-rgw-bytes-used wasn't returned"
return (int(obj_count), int(bytes_used))
@attr(resource='bucket')
@attr(method='head')
@attr(operation='head bucket')
@attr(assertion='succeeds')
@attr('fails_on_dho')
def test_bucket_head():
bucket = _setup_bucket_request('private')
_head_bucket(bucket)
@attr(resource='bucket')
@attr(method='head')
@attr(operation='read bucket extended information')
@attr(assertion='extended information is getting updated')
@attr('fails_on_dho')
def test_bucket_head_extended():
bucket = _setup_bucket_request('private')
(obj_count, bytes_used) = _head_bucket(bucket)
eq(obj_count, 0)
eq(bytes_used, 0)
_create_keys(bucket, keys=['foo', 'bar', 'baz'])
(obj_count, bytes_used) = _head_bucket(bucket)
eq(obj_count, 3)
assert bytes_used > 0
@attr(resource='bucket.acl')
@attr(method='get')