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

test consistency of bucket acls, exercises cache

Browse source
This commit is contained in:
Yehuda Sadeh 2011-07-21 14:09:21 -07:00
parent 338728ecd2
commit 9b3cdf753f
1 changed files with 17 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
s3tests/functional/test_s3.py
View file

@ -713,11 +713,12 @@ def _make_acl_xml(acl):
return '<?xml version="1.0" encoding="UTF-8"?><AccessControlPolicy xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Owner><ID>' + config.main.user_id + '</ID></Owner>' + acl.to_xml() + '</AccessControlPolicy>' return '<?xml version="1.0" encoding="UTF-8"?><AccessControlPolicy xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Owner><ID>' + config.main.user_id + '</ID></Owner>' + acl.to_xml() + '</AccessControlPolicy>'
def _build_bucket_acl_xml(permission): def _build_bucket_acl_xml(permission, bucket=None):
acl = boto.s3.acl.ACL() acl = boto.s3.acl.ACL()
acl.add_user_grant(permission=permission, user_id=config.main.user_id) acl.add_user_grant(permission=permission, user_id=config.main.user_id)
XML = _make_acl_xml(acl) XML = _make_acl_xml(acl)
bucket = get_new_bucket() if bucket is None:
bucket = get_new_bucket()
bucket.set_xml_acl(XML) bucket.set_xml_acl(XML)
policy = bucket.get_acl() policy = bucket.get_acl()
print repr(policy) print repr(policy)
@ -1420,6 +1421,20 @@ def test_100_continue():
status = _simple_http_req_100_cont(s3.main.host, s3.main.port, s3.main.is_secure, 'PUT', resource) status = _simple_http_req_100_cont(s3.main.host, s3.main.port, s3.main.is_secure, 'PUT', resource)
eq(status, '100') eq(status, '100')
def _test_bucket_acls_changes_persistent(bucket):
perms = ('FULL_CONTROL', 'WRITE', 'WRITE_ACP', 'READ', 'READ_ACP')
for p in perms:
_build_bucket_acl_xml(p, bucket)
def test_bucket_acls_changes_persistent():
bucket = get_new_bucket()
_test_bucket_acls_changes_persistent(bucket);
def test_stress_bucket_acls_changes():
bucket = get_new_bucket()
for i in xrange(10):
_test_bucket_acls_changes_persistent(bucket);
class FakeFile(object): class FakeFile(object):
def __init__(self, size, char='A', interrupt=None): def __init__(self, size, char='A', interrupt=None):
self.offset = 0 self.offset = 0