import time
import operator
import nose
+import pytest
import os
import string
import random
@attr(operation='list')
@attr(assertion='distinct buckets have different contents')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_list_distinct():
bucket1 = get_new_bucket_resource()
bucket2 = get_new_bucket_resource()
@attr(operation='list all keys')
@attr(assertion='pagination w/max_keys=2, no marker')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_list_many():
bucket_name = _create_objects(keys=['foo', 'bar', 'baz'])
client = get_client()
@attr(operation='list all keys')
@attr(assertion='pagination w/max_keys=2, no marker')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_listv2_many():
bucket_name = _create_objects(keys=['foo', 'bar', 'baz'])
client = get_client()
@attr(operation='list')
@attr(assertion='keycount in listobjectsv2')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_basic_key_count():
client = get_client()
bucket_names = []
@attr(operation='list')
@attr(assertion='prefixes in multi-component object names')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_delimiter_basic():
bucket_name = _create_objects(keys=['foo/bar', 'foo/bar/xyzzy', 'quux/thud', 'asdf'])
client = get_client()
@attr(operation='list')
@attr(assertion='test url encoding')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_encoding_basic():
bucket_name = _create_objects(keys=['foo+1/bar', 'foo/bar/xyzzy', 'quux ab/thud', 'asdf+b'])
client = get_client()
@attr(operation='list')
@attr(assertion='prefixes in multi-component object names')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_list_delimiter_prefix():
bucket_name = _create_objects(keys=['asdf', 'boo/bar', 'boo/baz/xyzzy', 'cquux/thud', 'cquux/bla'])
@attr(operation='list')
@attr(assertion='prefixes in multi-component object names')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_listv2_delimiter_prefix():
bucket_name = _create_objects(keys=['asdf', 'boo/bar', 'boo/baz/xyzzy', 'cquux/thud', 'cquux/bla'])
@attr(operation='list')
@attr(assertion='prefix and delimiter handling when object ends with delimiter')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_delimiter_prefix_ends_with_delimiter():
bucket_name = _create_objects(keys=['asdf/'])
validate_bucket_listv2(bucket_name, 'asdf/', '/', None, 1000, False, ['asdf/'], [], last=True)
@attr(method='get')
@attr(assertion='non-slash delimiter characters')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_delimiter_alt():
bucket_name = _create_objects(keys=['bar', 'baz', 'cab', 'foo'])
client = get_client()
@attr(operation='list')
@attr(assertion='prefixes starting with underscore')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_list_delimiter_prefix_underscore():
bucket_name = _create_objects(keys=['_obj1_','_under1/bar', '_under1/baz/xyzzy', '_under2/thud', '_under2/bla'])
@attr(operation='list')
@attr(assertion='prefixes starting with underscore')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_listv2_delimiter_prefix_underscore():
bucket_name = _create_objects(keys=['_obj1_','_under1/bar', '_under1/baz/xyzzy', '_under2/thud', '_under2/bla'])
@attr(method='get')
@attr(assertion='percentage delimiter characters')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_delimiter_percentage():
bucket_name = _create_objects(keys=['b%ar', 'b%az', 'c%ab', 'foo'])
client = get_client()
@attr(method='get')
@attr(assertion='whitespace delimiter characters')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_delimiter_whitespace():
bucket_name = _create_objects(keys=['b ar', 'b az', 'c ab', 'foo'])
client = get_client()
@attr(method='get')
@attr(assertion='dot delimiter characters')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_delimiter_dot():
bucket_name = _create_objects(keys=['b.ar', 'b.az', 'c.ab', 'foo'])
client = get_client()
@attr(method='get')
@attr(assertion='non-printable delimiter can be specified')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_delimiter_unreadable():
key_names=['bar', 'baz', 'cab', 'foo']
bucket_name = _create_objects(keys=key_names)
@attr(method='get')
@attr(assertion='empty delimiter can be specified')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_delimiter_empty():
key_names = ['bar', 'baz', 'cab', 'foo']
bucket_name = _create_objects(keys=key_names)
@attr(method='get')
@attr(assertion='unspecified delimiter defaults to none')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_delimiter_none():
key_names = ['bar', 'baz', 'cab', 'foo']
bucket_name = _create_objects(keys=key_names)
eq(prefixes, [])
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_fetchowner_notempty():
key_names = ['foo/bar', 'foo/baz', 'quux']
bucket_name = _create_objects(keys=key_names)
eq('Owner' in objs_list[0], True)
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_fetchowner_defaultempty():
key_names = ['foo/bar', 'foo/baz', 'quux']
bucket_name = _create_objects(keys=key_names)
eq('Owner' in objs_list[0], False)
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_fetchowner_empty():
key_names = ['foo/bar', 'foo/baz', 'quux']
bucket_name = _create_objects(keys=key_names)
@attr(method='get')
@attr(assertion='unused delimiter is not found')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_delimiter_not_exist():
key_names = ['bar', 'baz', 'cab', 'foo']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list')
@attr(assertion='list with delimiter not skip special keys')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_list_delimiter_not_skip_special():
key_names = ['0/'] + ['0/%s' % i for i in range(1000, 1999)]
key_names2 = ['1999', '1999#', '1999+', '2000']
@attr(operation='list under prefix with list-objects-v2')
@attr(assertion='returns only objects under prefix')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_prefix_basic():
key_names = ['foo/bar', 'foo/baz', 'quux']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list under prefix with list-objects-v2')
@attr(assertion='prefixes w/o delimiters')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_prefix_alt():
key_names = ['bar', 'baz', 'foo']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list under prefix with list-objects-v2')
@attr(assertion='empty prefix returns everything')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_prefix_empty():
key_names = ['foo/bar', 'foo/baz', 'quux']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list under prefix with list-objects-v2')
@attr(assertion='unspecified prefix returns everything')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_prefix_none():
key_names = ['foo/bar', 'foo/baz', 'quux']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list under prefix with list-objects-v2')
@attr(assertion='nonexistent prefix returns nothing')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_prefix_not_exist():
key_names = ['foo/bar', 'foo/baz', 'quux']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list under prefix with list-objects-v2')
@attr(assertion='non-printable prefix can be specified')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_prefix_unreadable():
key_names = ['foo/bar', 'foo/baz', 'quux']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list-objects-v2 under prefix w/delimiter')
@attr(assertion='returns only objects directly under prefix')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_prefix_delimiter_basic():
key_names = ['foo/bar', 'foo/baz/xyzzy', 'quux/thud', 'asdf']
bucket_name = _create_objects(keys=key_names)
eq(prefixes, ['baza'])
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_prefix_delimiter_alt():
key_names = ['bar', 'bazar', 'cab', 'foo']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list-objects-v2 under prefix w/delimiter')
@attr(assertion='finds nothing w/unmatched prefix')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_prefix_delimiter_prefix_not_exist():
key_names = ['b/a/r', 'b/a/c', 'b/a/g', 'g']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list-objects-v2 under prefix w/delimiter')
@attr(assertion='over-ridden slash ceases to be a delimiter')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_prefix_delimiter_delimiter_not_exist():
key_names = ['b/a/c', 'b/a/g', 'b/a/r', 'g']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list-objects-v2 under prefix w/delimiter')
@attr(assertion='finds nothing w/unmatched prefix and delimiter')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_prefix_delimiter_prefix_delimiter_not_exist():
key_names = ['b/a/c', 'b/a/g', 'b/a/r', 'g']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list all keys')
@attr(assertion='pagination w/max_keys=1, marker')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_list_maxkeys_one():
key_names = ['bar', 'baz', 'foo', 'quxx']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list all keys with list-objects-v2')
@attr(assertion='pagination w/max_keys=1, marker')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_listv2_maxkeys_one():
key_names = ['bar', 'baz', 'foo', 'quxx']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list all keys with list-objects-v2')
@attr(assertion='pagination w/max_keys=0')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_maxkeys_zero():
key_names = ['bar', 'baz', 'foo', 'quxx']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list all keys with list-objects-v2')
@attr(assertion='pagination w/o max_keys')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_maxkeys_none():
key_names = ['bar', 'baz', 'foo', 'quxx']
bucket_name = _create_objects(keys=key_names)
@attr(operation='get usage by client')
@attr(assertion='account usage api')
@attr('fails_on_aws') # allow-unordered is a non-standard extension
+@pytest.mark.fails_on_aws
def test_account_usage():
# boto3.set_stream_logger(name='botocore')
client = get_client()
@attr(operation='get usage by client')
@attr(assertion='account usage by head bucket')
@attr('fails_on_aws') # allow-unordered is a non-standard extension
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_head_bucket_usage():
# boto3.set_stream_logger(name='botocore')
client = get_client()
@attr(operation='list all keys')
@attr(assertion='bucket list unordered')
@attr('fails_on_aws') # allow-unordered is a non-standard extension
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_list_unordered():
# boto3.set_stream_logger(name='botocore')
keys_in = ['ado', 'bot', 'cob', 'dog', 'emu', 'fez', 'gnu', 'hex',
@attr(operation='list all keys with list-objects-v2')
@attr(assertion='bucket list unordered')
@attr('fails_on_aws') # allow-unordered is a non-standard extension
+@pytest.mark.fails_on_aws
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_listv2_unordered():
# boto3.set_stream_logger(name='botocore')
keys_in = ['ado', 'bot', 'cob', 'dog', 'emu', 'fez', 'gnu', 'hex',
@attr(operation='list all keys with list-objects-v2')
@attr(assertion='no pagination, empty continuationtoken')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_continuationtoken_empty():
key_names = ['bar', 'baz', 'foo', 'quxx']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list keys with list-objects-v2')
@attr(assertion='no pagination, non-empty continuationtoken')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_continuationtoken():
key_names = ['bar', 'baz', 'foo', 'quxx']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list keys with list-objects-v2')
@attr(assertion='no pagination, non-empty continuationtoken and startafter')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_listv2_both_continuationtoken_startafter():
key_names = ['bar', 'baz', 'foo', 'quxx']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list all keys with list-objects-v2')
@attr(assertion='non-printing startafter')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_startafter_unreadable():
key_names = ['bar', 'baz', 'foo', 'quxx']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list all keys with list-objects-v2')
@attr(assertion='startafter not-in-list')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_startafter_not_in_list():
key_names = ['bar', 'baz', 'foo', 'quxx']
bucket_name = _create_objects(keys=key_names)
@attr(operation='list all keys with list-objects-v2')
@attr(assertion='startafter after list')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_startafter_after_list():
key_names = ['bar', 'baz', 'foo', 'quxx']
bucket_name = _create_objects(keys=key_names)
@attr(operation='compare w/bucket list')
@attr(assertion='return same metadata')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_list_return_data():
key_names = ['bar', 'baz', 'foo']
bucket_name = _create_objects(keys=key_names)
@attr(assertion='return same metadata')
@attr('versioning')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_list_return_data_versioning():
bucket_name = get_new_bucket()
check_configure_versioning_retry(bucket_name, "Enabled", "Enabled")
@attr(operation='list all objects (anonymous) with list-objects-v2')
@attr(assertion='succeeds')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_objects_anonymous():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='list all objects (anonymous) with list-objects-v2')
@attr(assertion='fails')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucket_listv2_objects_anonymous_fail():
bucket_name = get_new_bucket()
@attr(operation='non-existant bucket with list-objects-v2')
@attr(assertion='fails 404')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucketv2_notexist():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(operation='read contents that were never written to raise one error response')
@attr(assertion='RequestId appears in the error response')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_requestid_matches_header_on_error():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='delete multiple objects with list-objects-v2')
@attr(assertion='deletes multiple objects with a single call')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_multi_objectv2_delete():
key_names = ['key0', 'key1', 'key2']
bucket_name = _create_objects(keys=key_names)
@attr(assertion='UTF-8 values passed through')
# TODO: the decoding of this unicode metadata is not happening properly for unknown reasons
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
def test_object_set_get_unicode_metadata():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='get w/ If-Modified-Since: after')
@attr(assertion='fails 304')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_get_object_ifmodifiedsince_failed():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='get w/ If-Unmodified-Since: before')
@attr(assertion='fails 412')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_get_object_ifunmodifiedsince_good():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='data re-write w/ If-Match: the latest ETag')
@attr(assertion='replaces previous data and metadata')
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
def test_put_object_ifmatch_good():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='get w/ If-Match: bogus ETag')
@attr(assertion='fails 412')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_put_object_ifmatch_failed():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='overwrite existing object w/ If-Match: *')
@attr(assertion='replaces previous data and metadata')
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
def test_put_object_ifmatch_overwrite_existed_good():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='overwrite non-existing object w/ If-Match: *')
@attr(assertion='fails 412')
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_put_object_ifmatch_nonexisted_failed():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='overwrite existing object w/ If-None-Match: outdated ETag')
@attr(assertion='replaces previous data and metadata')
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
def test_put_object_ifnonmatch_good():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='overwrite existing object w/ If-None-Match: the latest ETag')
@attr(assertion='fails 412')
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_put_object_ifnonmatch_failed():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='overwrite non-existing object w/ If-None-Match: *')
@attr(assertion='succeeds')
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
def test_put_object_ifnonmatch_nonexisted_good():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='overwrite existing object w/ If-None-Match: *')
@attr(assertion='fails 412')
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_put_object_ifnonmatch_overwrite_existed_failed():
bucket_name = get_new_bucket()
client = get_client()
#eq(error_code, 'NoSuchKey')
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr(resource='bucket')
@attr(method='head')
@attr(operation='read bucket extended information')
@attr(assertion='extended information is getting updated')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_head_extended():
bucket_name = get_new_bucket()
client = get_client()
# AWS does not enforce all documented bucket restrictions.
# http://docs.amazonwebservices.com/AmazonS3/2006-03-01/dev/index.html?BucketRestrictions.html
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
# Breaks DNS with SubdomainCallingFormat
@attr('fails_with_subdomain')
@attr(resource='bucket')
@attr(operation='create w/60 byte name')
@attr(assertion='fails with subdomain')
@attr('fails_on_aws') # <Error><Code>InvalidBucketName</Code><Message>The specified bucket is not valid.</Message>...</Error>
+@pytest.mark.fails_on_aws
# Should now pass on AWS even though it has 'fails_on_aws' attr.
def test_bucket_create_naming_good_long_60():
_test_bucket_create_naming_good_long(60)
@attr(operation='create w/61 byte name')
@attr(assertion='fails with subdomain')
@attr('fails_on_aws') # <Error><Code>InvalidBucketName</Code><Message>The specified bucket is not valid.</Message>...</Error>
+@pytest.mark.fails_on_aws
# Should now pass on AWS even though it has 'fails_on_aws' attr.
def test_bucket_create_naming_good_long_61():
_test_bucket_create_naming_good_long(61)
@attr(operation='create w/62 byte name')
@attr(assertion='fails with subdomain')
@attr('fails_on_aws') # <Error><Code>InvalidBucketName</Code><Message>The specified bucket is not valid.</Message>...</Error>
+@pytest.mark.fails_on_aws
# Should now pass on AWS even though it has 'fails_on_aws' attr.
def test_bucket_create_naming_good_long_62():
_test_bucket_create_naming_good_long(62)
@attr(operation='list w/61 byte name')
@attr(assertion='fails with subdomain')
@attr('fails_on_aws') # <Error><Code>InvalidBucketName</Code><Message>The specified bucket is not valid.</Message>...</Error>
+@pytest.mark.fails_on_aws
# Should now pass on AWS even though it has 'fails_on_aws' attr.
def test_bucket_list_long_name():
prefix = get_new_bucket_name()
# AWS does not enforce all documented bucket restrictions.
# http://docs.amazonwebservices.com/AmazonS3/2006-03-01/dev/index.html?BucketRestrictions.html
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr(resource='bucket')
@attr(method='put')
@attr(operation='create w/ip address for name')
@attr(operation='create w/underscore in name')
@attr(assertion='fails')
@attr('fails_on_aws') # <Error><Code>InvalidBucketName</Code><Message>The specified bucket is not valid.</Message>...</Error>
+@pytest.mark.fails_on_aws
# Should now pass on AWS even though it has 'fails_on_aws' attr.
def test_bucket_create_naming_dns_underscore():
invalid_bucketname = 'foo_bar'
@attr(operation='create w/100 byte name')
@attr(assertion='fails with subdomain')
@attr('fails_on_aws') # <Error><Code>InvalidBucketName</Code><Message>The specified bucket is not valid.</Message>...</Error>
+@pytest.mark.fails_on_aws
def test_bucket_create_naming_dns_long():
prefix = get_prefix()
assert len(prefix) < 50
@attr(operation='create w/dash at end of name')
@attr(assertion='fails')
@attr('fails_on_aws') # <Error><Code>InvalidBucketName</Code><Message>The specified bucket is not valid.</Message>...</Error>
+@pytest.mark.fails_on_aws
# Should now pass on AWS even though it has 'fails_on_aws' attr.
def test_bucket_create_naming_dns_dash_at_end():
invalid_bucketname = 'foo-'
@attr(operation='create w/.. in name')
@attr(assertion='fails')
@attr('fails_on_aws') # <Error><Code>InvalidBucketName</Code><Message>The specified bucket is not valid.</Message>...</Error>
+@pytest.mark.fails_on_aws
# Should now pass on AWS even though it has 'fails_on_aws' attr.
def test_bucket_create_naming_dns_dot_dot():
invalid_bucketname = 'foo..bar'
@attr(operation='create w/.- in name')
@attr(assertion='fails')
@attr('fails_on_aws') # <Error><Code>InvalidBucketName</Code><Message>The specified bucket is not valid.</Message>...</Error>
+@pytest.mark.fails_on_aws
# Should now pass on AWS even though it has 'fails_on_aws' attr.
def test_bucket_create_naming_dns_dot_dash():
invalid_bucketname = 'foo.-bar'
@attr(operation='create w/-. in name')
@attr(assertion='fails')
@attr('fails_on_aws') # <Error><Code>InvalidBucketName</Code><Message>The specified bucket is not valid.</Message>...</Error>
+@pytest.mark.fails_on_aws
# Should now pass on AWS even though it has 'fails_on_aws' attr.
def test_bucket_create_naming_dns_dash_dot():
invalid_bucketname = 'foo-.bar'
@attr(method='get')
@attr(operation='get location')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_get_location():
location_constraint = get_main_api_name()
if not location_constraint:
@attr(operation='re-create by non-owner')
@attr(assertion='fails 409')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_create_exists_nonowner():
# Names are shared across a global namespace. As such, no two
# users can create a bucket with that same name.
@attr(operation='re-create with existing acl')
@attr(assertion='fails 409')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_recreate_overwrite_acl():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(operation='re-create with new acl')
@attr(assertion='fails 409')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_recreate_new_acl():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(operation='public-read acl')
@attr(assertion='read back expected defaults')
@attr('fails_on_aws') # <Error><Code>IllegalLocationConstraintException</Code><Message>The unspecified location constraint is incompatible for the region specific endpoint this request was sent to.</Message>
+@pytest.mark.fails_on_aws
def test_bucket_acl_canned_during_create():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(operation='set write-acp')
@attr(assertion='does not modify owner')
@attr('fails_on_aws') # <Error><Code>InvalidArgument</Code><Message>Invalid id</Message><ArgumentName>CanonicalUser/ID</ArgumentName><ArgumentValue>${ALTUSER}</ArgumentValue>
+@pytest.mark.fails_on_aws
def test_object_acl_full_control_verify_owner():
bucket_name = get_new_bucket_name()
main_client = get_client()
@attr(operation='set acl FULL_CONTRO')
@attr(assertion='reads back correctly')
@attr('fails_on_aws') # <Error><Code>InvalidArgument</Code><Message>Invalid id</Message><ArgumentName>CanonicalUser/ID</ArgumentName><ArgumentValue>${USER}</ArgumentValue>
+@pytest.mark.fails_on_aws
def test_object_acl():
_check_object_acl('FULL_CONTROL')
@attr(operation='set acl WRITE')
@attr(assertion='reads back correctly')
@attr('fails_on_aws') # <Error><Code>InvalidArgument</Code><Message>Invalid id</Message><ArgumentName>CanonicalUser/ID</ArgumentName><ArgumentValue>${USER}</ArgumentValue>
+@pytest.mark.fails_on_aws
def test_object_acl_write():
_check_object_acl('WRITE')
@attr(operation='set acl WRITE_ACP')
@attr(assertion='reads back correctly')
@attr('fails_on_aws') # <Error><Code>InvalidArgument</Code><Message>Invalid id</Message><ArgumentName>CanonicalUser/ID</ArgumentName><ArgumentValue>${USER}</ArgumentValue>
+@pytest.mark.fails_on_aws
def test_object_acl_writeacp():
_check_object_acl('WRITE_ACP')
@attr(operation='set acl READ')
@attr(assertion='reads back correctly')
@attr('fails_on_aws') # <Error><Code>InvalidArgument</Code><Message>Invalid id</Message><ArgumentName>CanonicalUser/ID</ArgumentName><ArgumentValue>${USER}</ArgumentValue>
+@pytest.mark.fails_on_aws
def test_object_acl_read():
_check_object_acl('READ')
@attr(operation='set acl READ_ACP')
@attr(assertion='reads back correctly')
@attr('fails_on_aws') # <Error><Code>InvalidArgument</Code><Message>Invalid id</Message><ArgumentName>CanonicalUser/ID</ArgumentName><ArgumentValue>${USER}</ArgumentValue>
+@pytest.mark.fails_on_aws
def test_object_acl_readacp():
_check_object_acl('READ_ACP')
@attr(operation='set acl w/userid FULL_CONTROL')
@attr(assertion='can read/write data/acls')
@attr('fails_on_aws') # <Error><Code>InvalidArgument</Code><Message>Invalid id</Message><ArgumentName>CanonicalUser/ID</ArgumentName><ArgumentValue>${USER}</ArgumentValue>
+@pytest.mark.fails_on_aws
def test_bucket_acl_grant_userid_fullcontrol():
bucket_name = _bucket_acl_grant_userid('FULL_CONTROL')
@attr(operation='set acl w/userid READ')
@attr(assertion='can read data, no other r/w')
@attr('fails_on_aws') # <Error><Code>InvalidArgument</Code><Message>Invalid id</Message><ArgumentName>CanonicalUser/ID</ArgumentName><ArgumentValue>${ALTUSER}</ArgumentValue>
+@pytest.mark.fails_on_aws
def test_bucket_acl_grant_userid_read():
bucket_name = _bucket_acl_grant_userid('READ')
@attr(operation='set acl w/userid READ_ACP')
@attr(assertion='can read acl, no other r/w')
@attr('fails_on_aws') # <Error><Code>InvalidArgument</Code><Message>Invalid id</Message><ArgumentName>CanonicalUser/ID</ArgumentName><ArgumentValue>${ALTUSER}</ArgumentValue>
+@pytest.mark.fails_on_aws
def test_bucket_acl_grant_userid_readacp():
bucket_name = _bucket_acl_grant_userid('READ_ACP')
@attr(operation='set acl w/userid WRITE')
@attr(assertion='can write data, no other r/w')
@attr('fails_on_aws') # <Error><Code>InvalidArgument</Code><Message>Invalid id</Message><ArgumentName>CanonicalUser/ID</ArgumentName><ArgumentValue>${ALTUSER}</ArgumentValue>
+@pytest.mark.fails_on_aws
def test_bucket_acl_grant_userid_write():
bucket_name = _bucket_acl_grant_userid('WRITE')
@attr(operation='set acl w/userid WRITE_ACP')
@attr(assertion='can write acls, no other r/w')
@attr('fails_on_aws') # <Error><Code>InvalidArgument</Code><Message>Invalid id</Message><ArgumentName>CanonicalUser/ID</ArgumentName><ArgumentValue>${ALTUSER}</ArgumentValue>
+@pytest.mark.fails_on_aws
def test_bucket_acl_grant_userid_writeacp():
bucket_name = _bucket_acl_grant_userid('WRITE_ACP')
@attr(operation='add all grants to user through headers')
@attr(assertion='adds all grants individually to second user')
@attr('fails_on_dho')
+@pytest.mark.fails_on_dho
@attr('fails_on_aws') # <Error><Code>InvalidArgument</Code><Message>Invalid id</Message><ArgumentName>CanonicalUser/ID</ArgumentName><ArgumentValue>${ALTUSER}</ArgumentValue>
+@pytest.mark.fails_on_aws
def test_object_header_acl_grants():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='add all grants to user through headers')
@attr(assertion='adds all grants individually to second user')
@attr('fails_on_dho')
+@pytest.mark.fails_on_dho
@attr('fails_on_aws') # <Error><Code>InvalidArgument</Code><Message>Invalid id</Message><ArgumentName>CanonicalUser/ID</ArgumentName><ArgumentValue>${ALTUSER}</ArgumentValue>
+@pytest.mark.fails_on_aws
def test_bucket_header_acl_grants():
headers = _get_acl_header()
bucket_name = get_new_bucket_name()
@attr(operation='add second FULL_CONTROL user')
@attr(assertion='works for S3, fails for DHO')
@attr('fails_on_aws') # <Error><Code>AmbiguousGrantByEmailAddress</Code><Message>The e-mail address you provided is associated with more than one account. Please retry your request using a different identification method or after resolving the ambiguity.</Message>
+@pytest.mark.fails_on_aws
def test_bucket_acl_grant_email():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='set/enable/disable logging target')
@attr(assertion='operations succeed')
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
def test_logging_toggle():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='set bucket/object acls: private/private with list-objects-v2')
@attr(assertion='public has no access to bucket or objects')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_access_bucket_private_objectv2_private():
# all the test_access_* tests follow this template
bucket_name, key1, key2, newkey = _setup_access(bucket_acl='private', object_acl='private')
@attr(operation='set bucket/object acls: private/public-read with list-objects-v2')
@attr(assertion='public can only read readable object')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_access_bucket_private_objectv2_publicread():
bucket_name, key1, key2, newkey = _setup_access(bucket_acl='private', object_acl='public-read')
@attr(operation='set bucket/object acls: private/public-read/write with list-objects-v2')
@attr(assertion='public can only read the readable object')
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_access_bucket_private_objectv2_publicreadwrite():
bucket_name, key1, key2, newkey = _setup_access(bucket_acl='private', object_acl='public-read-write')
alt_client = get_alt_client()
@attr(operation='list all buckets (anonymous)')
@attr(assertion='succeeds')
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
def test_list_buckets_anonymous():
# Get a connection with bad authorization, then change it to be our new Anonymous auth mechanism,
# emulating standard HTTP access.
@attr(operation='create and list objects with special names')
@attr(assertion='special names work')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_create_special_key_names():
key_names = [
' ',
@attr(operation='copy zero sized object in same bucket')
@attr(assertion='works')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_copy_zero_size():
key = 'foo123bar'
bucket_name = _create_objects(keys=[key])
@attr(operation='copy 16mb object in same bucket')
@attr(assertion='works')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_copy_16m():
bucket_name = get_new_bucket()
key1 = 'obj1'
@attr(operation='copy object in same bucket')
@attr(assertion='works')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_copy_same_bucket():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='copy object with content-type')
@attr(assertion='works')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_copy_verify_contenttype():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='modify object metadata by copying')
@attr(assertion='fails')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_copy_to_itself_with_metadata():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='copy object from different bucket')
@attr(assertion='works')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_copy_diff_bucket():
bucket_name1 = get_new_bucket()
bucket_name2 = get_new_bucket()
@attr(operation='copy object and change acl')
@attr(assertion='works')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_copy_canned_acl():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='copy object and retain metadata')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_copy_retaining_metadata():
for size in [3, 1024 * 1024]:
bucket_name = get_new_bucket()
@attr(method='put')
@attr(operation='copy object and replace metadata')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_copy_replacing_metadata():
for size in [3, 1024 * 1024]:
bucket_name = get_new_bucket()
@attr(assertion='works')
@attr('versioning')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_copy_versioned_bucket():
bucket_name = get_new_bucket()
client = get_client()
@attr(assertion='works')
@attr('versioning')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_copy_versioned_url_encoding():
bucket = get_new_bucket_resource()
check_configure_versioning_retry(bucket.name, "Enabled", "Enabled")
@attr(assertion='successful')
@attr('versioning')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_copy_versioning_multipart_upload():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='check multipart uploads with single small part')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_multipart_upload_small():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='check multipart copies with single small part')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_multipart_copy_small():
src_key = 'foo'
src_bucket_name = _create_key_with_random_content(src_key)
@attr(operation='check multipart copy with an improperly formatted range')
# TODO: remove fails_on_rgw when https://tracker.ceph.com/issues/40795 is resolved
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
def test_multipart_copy_improper_range():
client = get_client()
src_key = 'source'
@attr(method='put')
@attr(operation='check multipart copies with single small part')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_multipart_copy_special_names():
src_bucket_name = get_new_bucket()
@attr(operation='complete multi-part upload')
@attr(assertion='successful')
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_multipart_upload():
bucket_name = get_new_bucket()
key="mymultipart"
@attr(operation='check multipart copies of versioned objects')
@attr('versioning')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_multipart_copy_versioned():
src_bucket_name = get_new_bucket()
dest_bucket_name = get_new_bucket()
@attr(operation='complete multi-part upload')
@attr(assertion='successful')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_multipart_upload_resend_part():
bucket_name = get_new_bucket()
key="mymultipart"
@attr(assertion='successful')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_multipart_copy_multiple_sizes():
src_key = 'foo'
src_bucket_name = _create_key_with_random_content(src_key, 12*1024*1024)
@attr(operation='check contents of multi-part upload')
@attr(assertion='successful')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_multipart_upload_contents():
bucket_name = get_new_bucket()
_do_test_multipart_upload_contents(bucket_name, 'mymultipart', 3)
@attr(operation='concurrent multi-part uploads')
@attr(assertion='successful')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_list_multipart_upload():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='list multipart uploads with different owners')
@attr(assertion='successful')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_list_multipart_upload_owner():
bucket_name = get_new_bucket()
@attr(operation='put tags')
@attr(assertion='succeeds')
@attr('tagging')
+@pytest.mark.tagging
def test_set_bucket_tagging():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='write atomicity')
@attr(assertion='1MB successful')
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
def test_atomic_conditional_write_1mb():
_test_atomic_conditional_write(1024*1024)
@attr(operation='write one or the other')
@attr(assertion='1MB successful')
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
# TODO: test not passing with SSL, fix this
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
def test_atomic_dual_conditional_write_1mb():
_test_atomic_dual_conditional_write(1024*1024)
@attr(operation='write file in deleted bucket')
@attr(assertion='fail 404')
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
# TODO: test not passing with SSL, fix this
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
def test_atomic_write_bucket_gone():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='range')
@attr(assertion='returns correct data, 206')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_ranged_request_response_code():
content = 'testcontent'
@attr(operation='range')
@attr(assertion='returns correct data, 206')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_ranged_big_request_response_code():
content = _generate_random_string(8*1024*1024)
@attr(operation='range')
@attr(assertion='returns correct data, 206')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_ranged_request_skip_leading_bytes_response_code():
content = 'testcontent'
@attr(operation='range')
@attr(assertion='returns correct data, 206')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_ranged_request_return_trailing_bytes_response_code():
content = 'testcontent'
@attr(assertion='everything works')
@attr('versioning')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_versioning_obj_create_overwrite_multipart():
bucket_name = get_new_bucket()
client = get_client()
@attr(assertion='everything works')
@attr('versioning')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_versioning_copy_obj_version():
bucket_name = get_new_bucket()
client = get_client()
@attr(assertion='returns correct marker version id')
@attr('versioning')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_versioning_multi_object_delete_with_marker_create():
bucket_name = get_new_bucket()
client = get_client()
@attr(assertion='works')
@attr('versioning')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_versioned_object_acl_no_version_specified():
bucket_name = get_new_bucket()
client = get_client()
@attr(assertion='works')
# TODO: remove fails_on_rgw when https://tracker.ceph.com/issues/39142 is resolved
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
@attr('versioning')
def test_versioned_concurrent_object_create_concurrent_remove():
bucket_name = get_new_bucket()
@attr(method='put')
@attr(operation='set lifecycle config')
@attr('lifecycle')
+@pytest.mark.lifecycle
def test_lifecycle_set():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='get')
@attr(operation='get lifecycle config')
@attr('lifecycle')
+@pytest.mark.lifecycle
def test_lifecycle_get():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='get')
@attr(operation='get lifecycle config no id')
@attr('lifecycle')
+@pytest.mark.lifecycle
def test_lifecycle_get_no_id():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='test lifecycle expiration')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecycle_expiration():
bucket_name = _create_objects(keys=['expire1/foo', 'expire1/bar', 'keep2/foo',
'keep2/bar', 'expire3/foo', 'expire3/bar'])
@attr(method='put')
@attr(operation='test lifecycle expiration with list-objects-v2')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecyclev2_expiration():
bucket_name = _create_objects(keys=['expire1/foo', 'expire1/bar', 'keep2/foo',
'keep2/bar', 'expire3/foo', 'expire3/bar'])
@attr(method='put')
@attr(operation='test lifecycle expiration on versioning enabled bucket')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
def test_lifecycle_expiration_versioning_enabled():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='test lifecycle expiration with 1 tag')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
def test_lifecycle_expiration_tags1():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='test lifecycle expiration with 2 tags')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecycle_expiration_tags2():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='test lifecycle expiration with versioning and 2 tags')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecycle_expiration_versioned_tags2():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='test lifecycle noncurrent expiration with 1 tag filter')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecycle_expiration_noncur_tags1():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='id too long in lifecycle rule')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr(assertion='fails 400')
def test_lifecycle_id_too_long():
bucket_name = get_new_bucket()
@attr(method='put')
@attr(operation='same id')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr(assertion='fails 400')
def test_lifecycle_same_id():
bucket_name = get_new_bucket()
@attr(method='put')
@attr(operation='invalid status in lifecycle rule')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr(assertion='fails 400')
def test_lifecycle_invalid_status():
bucket_name = get_new_bucket()
@attr(method='put')
@attr(operation='set lifecycle config with expiration date')
@attr('lifecycle')
+@pytest.mark.lifecycle
def test_lifecycle_set_date():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='set lifecycle config with not iso8601 date')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr(assertion='fails 400')
def test_lifecycle_set_invalid_date():
bucket_name = get_new_bucket()
@attr(method='put')
@attr(operation='test lifecycle expiration with date')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecycle_expiration_date():
bucket_name = _create_objects(keys=['past/foo', 'future/bar'])
client = get_client()
@attr(method='put')
@attr(operation='test lifecycle expiration days 0')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
def test_lifecycle_expiration_days0():
bucket_name = _create_objects(keys=['days0/foo', 'days0/bar'])
client = get_client()
@attr(method='put')
@attr(operation='test lifecycle expiration header put')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
def test_lifecycle_expiration_header_put():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='head')
@attr(operation='test lifecycle expiration header head')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecycle_expiration_header_head():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='head')
@attr(operation='test lifecycle expiration header head with tags')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecycle_expiration_header_tags_head():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='head')
@attr(operation='test lifecycle expiration header head with tags and And')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
def test_lifecycle_expiration_header_and_tags_head():
now = datetime.datetime.now(None)
bucket_name = get_new_bucket()
@attr(method='put')
@attr(operation='set lifecycle config with noncurrent version expiration')
@attr('lifecycle')
+@pytest.mark.lifecycle
def test_lifecycle_set_noncurrent():
bucket_name = _create_objects(keys=['past/foo', 'future/bar'])
client = get_client()
@attr(method='put')
@attr(operation='test lifecycle non-current version expiration')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecycle_noncur_expiration():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='set lifecycle config with delete marker expiration')
@attr('lifecycle')
+@pytest.mark.lifecycle
def test_lifecycle_set_deletemarker():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='set lifecycle config with Filter')
@attr('lifecycle')
+@pytest.mark.lifecycle
def test_lifecycle_set_filter():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='set lifecycle config with empty Filter')
@attr('lifecycle')
+@pytest.mark.lifecycle
def test_lifecycle_set_empty_filter():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='test lifecycle delete marker expiration')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecycle_deletemarker_expiration():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='set lifecycle config with multipart expiration')
@attr('lifecycle')
+@pytest.mark.lifecycle
def test_lifecycle_set_multipart():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='test lifecycle multipart expiration')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecycle_multipart_expiration():
bucket_name = get_new_bucket()
client = get_client()
@attr(method='put')
@attr(operation='set lifecycle config transition with not iso8601 date')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr(assertion='fails 400')
def test_lifecycle_transition_set_invalid_date():
bucket_name = get_new_bucket()
@attr(method='put')
@attr(operation='test lifecycle transition')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_transition')
+@pytest.mark.lifecycle_transition
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
def test_lifecycle_transition():
sc = configured_storage_classes()
if len(sc) < 3:
@attr(method='put')
@attr(operation='test lifecycle expiration')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_transition')
+@pytest.mark.lifecycle_transition
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
def test_lifecycle_transition_single_rule_multi_trans():
sc = configured_storage_classes()
if len(sc) < 3:
@attr(method='put')
@attr(operation='set lifecycle config with noncurrent version expiration')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_transition')
+@pytest.mark.lifecycle_transition
def test_lifecycle_set_noncurrent_transition():
sc = configured_storage_classes()
if len(sc) < 3:
@attr(method='put')
@attr(operation='test lifecycle non-current version expiration')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('lifecycle_transition')
+@pytest.mark.lifecycle_transition
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
def test_lifecycle_noncur_transition():
sc = configured_storage_classes()
if len(sc) < 3:
@attr(method='put')
@attr(operation='test lifecycle transition for cloud')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_transition')
+@pytest.mark.lifecycle_transition
@attr('cloud_transition')
+@pytest.mark.cloud_transition
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecycle_cloud_transition():
cloud_sc = get_cloud_storage_class()
if cloud_sc == None:
@attr(method='put')
@attr(operation='test lifecycle transition for cloud')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_transition')
+@pytest.mark.lifecycle_transition
@attr('cloud_transition')
+@pytest.mark.cloud_transition
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecycle_cloud_multiple_transition():
cloud_sc = get_cloud_storage_class()
if cloud_sc == None:
@attr(method='put')
@attr(operation='test lifecycle non-current version expiration on cloud transition')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_expiration')
+@pytest.mark.lifecycle_expiration
@attr('lifecycle_transition')
+@pytest.mark.lifecycle_transition
@attr('cloud_transition')
+@pytest.mark.cloud_transition
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecycle_noncur_cloud_transition():
cloud_sc = get_cloud_storage_class()
if cloud_sc == None:
@attr(method='put')
@attr(operation='test lifecycle transition for cloud')
@attr('lifecycle')
+@pytest.mark.lifecycle
@attr('lifecycle_transition')
+@pytest.mark.lifecycle_transition
@attr('cloud_transition')
+@pytest.mark.cloud_transition
@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_lifecycle_cloud_transition_large_obj():
cloud_sc = get_cloud_storage_class()
if cloud_sc == None:
@attr(operation='Test SSE-C encrypted transfer 1 byte')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_encrypted_transfer_1b():
_test_encryption_sse_customer_write(1)
@attr(operation='Test SSE-C encrypted transfer 1KB')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_encrypted_transfer_1kb():
_test_encryption_sse_customer_write(1024)
@attr(operation='Test SSE-C encrypted transfer 1MB')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_encrypted_transfer_1MB():
_test_encryption_sse_customer_write(1024*1024)
@attr(operation='Test SSE-C encrypted transfer 13 bytes')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_encrypted_transfer_13b():
_test_encryption_sse_customer_write(13)
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
def test_encryption_sse_c_method_head():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='write encrypted with SSE-C and read without SSE-C')
@attr(assertion='operation fails')
@attr('encryption')
+@pytest.mark.encryption
def test_encryption_sse_c_present():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='write encrypted with SSE-C but read with other key')
@attr(assertion='operation fails')
@attr('encryption')
+@pytest.mark.encryption
def test_encryption_sse_c_other_key():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='write encrypted with SSE-C, but md5 is bad')
@attr(assertion='operation fails')
@attr('encryption')
+@pytest.mark.encryption
def test_encryption_sse_c_invalid_md5():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='write encrypted with SSE-C, but dont provide MD5')
@attr(assertion='operation fails')
@attr('encryption')
+@pytest.mark.encryption
def test_encryption_sse_c_no_md5():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='declare SSE-C but do not provide key')
@attr(assertion='operation fails')
@attr('encryption')
+@pytest.mark.encryption
def test_encryption_sse_c_no_key():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='Do not declare SSE-C but provide key and MD5')
@attr(assertion='operation successfull, no encryption')
@attr('encryption')
+@pytest.mark.encryption
def test_encryption_key_no_sse_c():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='complete multi-part upload')
@attr(assertion='successful')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_aws') # allow-unordered is a non-standard extension
+@pytest.mark.fails_on_aws
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_encryption_sse_c_multipart_upload():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='multipart upload with bad key for uploading chunks')
@attr(assertion='successful')
@attr('encryption')
+@pytest.mark.encryption
# TODO: remove this fails_on_rgw when I fix it
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
def test_encryption_sse_c_multipart_invalid_chunks_1():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='multipart upload with bad md5 for chunks')
@attr(assertion='successful')
@attr('encryption')
+@pytest.mark.encryption
# TODO: remove this fails_on_rgw when I fix it
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
def test_encryption_sse_c_multipart_invalid_chunks_2():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='complete multi-part upload and download with bad key')
@attr(assertion='successful')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_encryption_sse_c_multipart_bad_download():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='authenticated browser based upload via POST request')
@attr(assertion='succeeds and returns written data')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_encryption_sse_c_post_object_authenticated_request():
bucket_name = get_new_bucket()
client = get_client()
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def _test_sse_kms_customer_write(file_size, key_id = 'testkey-1'):
"""
Tests Create a file of A's, use it to set_contents_from_file.
@attr(operation='Test SSE-KMS encrypted does perform head properly')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_method_head():
kms_keyid = get_main_kms_keyid()
bucket_name = get_new_bucket()
@attr(operation='write encrypted with SSE-KMS and read without SSE-KMS')
@attr(assertion='operation success')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_present():
kms_keyid = get_main_kms_keyid()
bucket_name = get_new_bucket()
@attr(operation='declare SSE-KMS but do not provide key_id')
@attr(assertion='operation fails')
@attr('encryption')
+@pytest.mark.encryption
def test_sse_kms_no_key():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='Do not declare SSE-KMS but provide key_id')
@attr(assertion='operation successfull, no encryption')
@attr('encryption')
+@pytest.mark.encryption
def test_sse_kms_not_declared():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='complete KMS multi-part upload')
@attr(assertion='successful')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_multipart_upload():
kms_keyid = get_main_kms_keyid()
bucket_name = get_new_bucket()
@attr(operation='multipart KMS upload with bad key_id for uploading chunks')
@attr(assertion='successful')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_multipart_invalid_chunks_1():
kms_keyid = get_main_kms_keyid()
kms_keyid2 = get_secondary_kms_keyid()
@attr(operation='multipart KMS upload with unexistent key_id for chunks')
@attr(assertion='successful')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_multipart_invalid_chunks_2():
kms_keyid = get_main_kms_keyid()
bucket_name = get_new_bucket()
@attr(operation='authenticated KMS browser based upload via POST request')
@attr(assertion='succeeds and returns written data')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_post_object_authenticated_request():
kms_keyid = get_main_kms_keyid()
bucket_name = get_new_bucket()
@attr(operation='Test SSE-KMS encrypted transfer 1 byte')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_transfer_1b():
kms_keyid = get_main_kms_keyid()
if kms_keyid is None:
@attr(operation='Test SSE-KMS encrypted transfer 1KB')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_transfer_1kb():
kms_keyid = get_main_kms_keyid()
if kms_keyid is None:
@attr(operation='Test SSE-KMS encrypted transfer 1MB')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_transfer_1MB():
kms_keyid = get_main_kms_keyid()
if kms_keyid is None:
@attr(operation='Test SSE-KMS encrypted transfer 13 bytes')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_transfer_13b():
kms_keyid = get_main_kms_keyid()
if kms_keyid is None:
@attr(operation='write encrypted with SSE-KMS and read with SSE-KMS')
@attr(assertion='operation fails')
@attr('encryption')
+@pytest.mark.encryption
def test_sse_kms_read_declare():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='Test Bucket Policy')
@attr(assertion='succeeds')
@attr('bucket-policy')
+@pytest.mark.bucket_policy
def test_bucket_policy():
bucket_name = get_new_bucket()
client = get_client()
eq(len(response['Contents']), 1)
@attr('bucket-policy')
+@pytest.mark.bucket_policy
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucketv2_policy():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='Test Bucket Policy and ACL')
@attr(assertion='fails')
@attr('bucket-policy')
+@pytest.mark.bucket_policy
def test_bucket_policy_acl():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='Test Bucket Policy and ACL with list-objects-v2')
@attr(assertion='fails')
@attr('bucket-policy')
+@pytest.mark.bucket_policy
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucketv2_policy_acl():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='Test Bucket Policy for a user belonging to a different tenant')
@attr(assertion='succeeds')
@attr('bucket-policy')
+@pytest.mark.bucket_policy
# TODO: remove this fails_on_rgw when I fix it
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
def test_bucket_policy_different_tenant():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='Test Bucket Policy for a user belonging to a different tenant')
@attr(assertion='succeeds')
@attr('bucket-policy')
+@pytest.mark.bucket_policy
# TODO: remove this fails_on_rgw when I fix it
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucketv2_policy_different_tenant():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='Test Bucket Policy on another bucket')
@attr(assertion='succeeds')
@attr('bucket-policy')
+@pytest.mark.bucket_policy
def test_bucket_policy_another_bucket():
bucket_name = get_new_bucket()
bucket_name2 = get_new_bucket()
@attr(operation='Test Bucket Policy on another bucket with list-objects-v2')
@attr(assertion='succeeds')
@attr('bucket-policy')
+@pytest.mark.bucket_policy
@attr('list-objects-v2')
+@pytest.mark.list_objects_v2
def test_bucketv2_policy_another_bucket():
bucket_name = get_new_bucket()
bucket_name2 = get_new_bucket()
@attr(method='put')
@attr(operation='Test put condition operator end with ifExists')
@attr('bucket-policy')
+@pytest.mark.bucket_policy
# TODO: remove this fails_on_rgw when I fix it
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
def test_bucket_policy_set_condition_operator_end_with_IfExists():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='Test Get/PutObjTagging output')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_get_obj_tagging():
key = 'testputtags'
bucket_name = _create_key_with_random_content(key)
@attr(operation='Test HEAD obj tagging output')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
def test_get_obj_head_tagging():
key = 'testputtags'
bucket_name = _create_key_with_random_content(key)
@attr(operation='Test Put max allowed tags')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_put_max_tags():
key = 'testputmaxtags'
bucket_name = _create_key_with_random_content(key)
@attr(operation='Test Put max allowed tags')
@attr(assertion='fails')
@attr('tagging')
+@pytest.mark.tagging
def test_put_excess_tags():
key = 'testputmaxtags'
bucket_name = _create_key_with_random_content(key)
@attr(operation='Test Put max allowed k-v size')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
def test_put_max_kvsize_tags():
key = 'testputmaxkeysize'
bucket_name = _create_key_with_random_content(key)
@attr(operation='Test exceed key size')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
def test_put_excess_key_tags():
key = 'testputexcesskeytags'
bucket_name = _create_key_with_random_content(key)
@attr(operation='Test exceed val size')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
def test_put_excess_val_tags():
key = 'testputexcesskeytags'
bucket_name = _create_key_with_random_content(key)
@attr(operation='Test PUT modifies existing tags')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_put_modify_tags():
key = 'testputmodifytags'
bucket_name = _create_key_with_random_content(key)
@attr(operation='Test Delete tags')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_put_delete_tags():
key = 'testputmodifytags'
bucket_name = _create_key_with_random_content(key)
@attr(method='post')
@attr(operation='anonymous browser based upload via POST request')
@attr('tagging')
+@pytest.mark.tagging
@attr(assertion='succeeds and returns written data')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_post_object_tags_anonymous_request():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(method='post')
@attr(operation='authenticated browser based upload via POST request')
@attr('tagging')
+@pytest.mark.tagging
@attr(assertion='succeeds and returns written data')
def test_post_object_tags_authenticated_request():
bucket_name = get_new_bucket()
@attr(operation='Test PutObj with tagging headers')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_put_obj_with_tags():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='Test GetObjTagging public read')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('bucket-policy')
+@pytest.mark.bucket_policy
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_get_tags_acl_public():
key = 'testputtagsacl'
bucket_name = _create_key_with_random_content(key)
@attr(operation='Test PutObjTagging public wrote')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('bucket-policy')
+@pytest.mark.bucket_policy
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_put_tags_acl_public():
key = 'testputtagsacl'
bucket_name = _create_key_with_random_content(key)
@attr(operation='test deleteobjtagging public')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('bucket-policy')
+@pytest.mark.bucket_policy
def test_delete_tags_obj_public():
key = 'testputtagsacl'
bucket_name = _create_key_with_random_content(key)
@attr(operation='Test ExistingObjectTag conditional on get object')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('bucket-policy')
+@pytest.mark.bucket_policy
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_policy_get_obj_existing_tag():
bucket_name = _create_objects(keys=['publictag', 'privatetag', 'invalidtag'])
client = get_client()
@attr(operation='Test ExistingObjectTag conditional on get object tagging')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('bucket-policy')
+@pytest.mark.bucket_policy
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_policy_get_obj_tagging_existing_tag():
bucket_name = _create_objects(keys=['publictag', 'privatetag', 'invalidtag'])
client = get_client()
@attr(operation='Test ExistingObjectTag conditional on put object tagging')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('bucket-policy')
+@pytest.mark.bucket_policy
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_policy_put_obj_tagging_existing_tag():
bucket_name = _create_objects(keys=['publictag', 'privatetag', 'invalidtag'])
client = get_client()
@attr(operation='Test copy-source conditional on put obj')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('bucket-policy')
+@pytest.mark.bucket_policy
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_policy_put_obj_copy_source():
bucket_name = _create_objects(keys=['public/foo', 'public/bar', 'private/foo'])
client = get_client()
@attr(operation='Test copy-source conditional on put obj')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('bucket-policy')
+@pytest.mark.bucket_policy
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_policy_put_obj_copy_source_meta():
src_bucket_name = _create_objects(keys=['public/foo', 'public/bar'])
client = get_client()
@attr(operation='Test put obj with canned-acl not to be public')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('bucket-policy')
+@pytest.mark.bucket_policy
def test_bucket_policy_put_obj_acl():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='Test put obj with amz-grant back to bucket-owner')
@attr(assertion='success')
@attr('bucket-policy')
+@pytest.mark.bucket_policy
def test_bucket_policy_put_obj_grant():
bucket_name = get_new_bucket()
@attr(operation='Deny put obj specifying both sse-c and sse-s3')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
def test_put_obj_enc_conflict_c_s3():
bucket_name = get_new_bucket()
client = get_v2_client()
@attr(operation='Deny put obj specifying both sse-c and sse-kms')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
def test_put_obj_enc_conflict_c_kms():
kms_keyid = get_main_kms_keyid()
if kms_keyid is None:
@attr(operation='Deny put obj specifying sse-s3 with kms key id')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
def test_put_obj_enc_conflict_s3_kms():
kms_keyid = get_main_kms_keyid()
if kms_keyid is None:
@attr(operation='Deny put obj specifying invalid algorithm' )
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
def test_put_obj_enc_conflict_bad_enc_kms():
kms_keyid = get_main_kms_keyid()
if kms_keyid is None:
@attr(operation='Deny put obj requests if not sse-s3: without encryption')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-policy')
+@pytest.mark.bucket_policy
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_policy_put_obj_s3_noenc():
bucket_name = get_new_bucket()
client = get_v2_client()
@attr(operation='Deny put obj requests if not sse-s3: kms')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-policy')
+@pytest.mark.bucket_policy
@attr('sse-s3')
+@pytest.mark.sse_s3
def test_bucket_policy_put_obj_s3_kms():
kms_keyid = get_main_kms_keyid()
if kms_keyid is None:
@attr(operation='Deny put obj requests if not sse-kms: without encryption')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
@attr('bucket-policy')
+@pytest.mark.bucket_policy
def test_bucket_policy_put_obj_kms_noenc():
kms_keyid = get_main_kms_keyid()
if kms_keyid is None:
@attr(operation='Deny put obj requests if not sse-kms: s3')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-policy')
+@pytest.mark.bucket_policy
def test_bucket_policy_put_obj_kms_s3():
bucket_name = get_new_bucket()
client = get_v2_client()
@attr(operation='put obj with RequestObjectTag')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('bucket-policy')
+@pytest.mark.bucket_policy
# TODO: remove this fails_on_rgw when I fix it
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
def test_bucket_policy_put_obj_request_obj_tag():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='Test ExistingObjectTag conditional on get object acl')
@attr(assertion='success')
@attr('tagging')
+@pytest.mark.tagging
@attr('bucket-policy')
+@pytest.mark.bucket_policy
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_bucket_policy_get_obj_acl_existing_tag():
bucket_name = _create_objects(keys=['publictag', 'privatetag', 'invalidtag'])
client = get_client()
@attr(assertion='success')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_obj_lock():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='fails')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_obj_lock_with_days_and_years():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='fails')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_obj_lock_invalid_days():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='fails')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_obj_lock_invalid_years():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='fails')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_obj_lock_invalid_mode():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='fails')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_obj_lock_invalid_status():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='fails')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_suspend_versioning():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='success')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_get_obj_lock():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='success')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_obj_retention():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='fails')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_obj_retention_invalid_mode():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='success')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_get_obj_retention():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='success')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_get_obj_retention_iso8601():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='success')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_obj_retention_versionid():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='success')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_obj_retention_override_default_retention():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='success')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_obj_retention_increase_period():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='fails')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_obj_retention_shorten_period():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='success')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_obj_retention_shorten_period_bypass():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='retention period make effects')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_delete_object_with_retention():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='retention period make effects')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_delete_object_with_retention_and_marker():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='retention period make effects')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_multi_delete_object_with_retention():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='success')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_legal_hold():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='fails')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_put_legal_hold_invalid_status():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='success')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_get_legal_hold():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='fails')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_delete_object_with_legal_hold_on():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='fails')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_delete_object_with_legal_hold_off():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='success')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_get_obj_metadata():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='success')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_uploading_obj():
bucket_name = get_new_bucket_name()
client = get_client()
@attr(assertion='succeeds')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_changing_mode_from_governance_with_bypass():
bucket_name = get_new_bucket_name()
key = 'file1'
@attr(assertion='fails')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_changing_mode_from_governance_without_bypass():
bucket_name = get_new_bucket_name()
key = 'file1'
@attr(assertion='fails')
@attr('object-lock')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_object_lock_changing_mode_from_compliance():
bucket_name = get_new_bucket_name()
key = 'file1'
@attr(operation='copy w/ x-amz-copy-source-if-match: the latest ETag')
@attr(assertion='succeeds')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_copy_object_ifmatch_good():
bucket_name = get_new_bucket()
client = get_client()
@attr(assertion='fails 412')
# TODO: remove fails_on_rgw when https://tracker.ceph.com/issues/40808 is resolved
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
def test_copy_object_ifmatch_failed():
bucket_name = get_new_bucket()
client = get_client()
@attr(assertion='fails 412')
# TODO: remove fails_on_rgw when https://tracker.ceph.com/issues/40808 is resolved
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
def test_copy_object_ifnonematch_good():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='copy w/ x-amz-copy-source-if-none-match: bogus ETag')
@attr(assertion='succeeds')
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_copy_object_ifnonematch_failed():
bucket_name = get_new_bucket()
client = get_client()
@attr(assertion='fails 400')
# TODO: results in a 404 instead of 400 on the RGW
@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
def test_object_read_unreadable():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='put bucket encryption on bucket - s3')
@attr(assertion='succeeds')
@attr('sse-s3')
+@pytest.mark.sse_s3
def test_put_bucket_encryption_s3():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='put bucket encryption on bucket - kms')
@attr(assertion='succeeds')
@attr('encryption')
+@pytest.mark.encryption
def test_put_bucket_encryption_kms():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='get bucket encryption on bucket - s3')
@attr(assertion='succeeds')
@attr('sse-s3')
+@pytest.mark.sse_s3
def test_get_bucket_encryption_s3():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='get bucket encryption on bucket - kms')
@attr(assertion='succeeds')
@attr('encryption')
+@pytest.mark.encryption
def test_get_bucket_encryption_kms():
kms_keyid = get_main_kms_keyid()
if kms_keyid is None:
@attr(operation='delete bucket encryption on bucket - s3')
@attr(assertion='succeeds')
@attr('sse-s3')
+@pytest.mark.sse_s3
def test_delete_bucket_encryption_s3():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='delete bucket encryption on bucket - kms')
@attr(assertion='succeeds')
@attr('encryption')
+@pytest.mark.encryption
def test_delete_bucket_encryption_kms():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='Test 1 byte upload to SSE-S3 default-encrypted bucket')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-encryption')
+@pytest.mark.bucket_encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_s3_default_upload_1b():
_test_sse_s3_default_upload(1)
@attr(operation='Test 1KB upload to SSE-S3 default-encrypted bucket')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-encryption')
+@pytest.mark.bucket_encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_s3_default_upload_1kb():
_test_sse_s3_default_upload(1024)
@attr(operation='Test 1MB upload to SSE-S3 default-encrypted bucket')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-encryption')
+@pytest.mark.bucket_encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_s3_default_upload_1mb():
_test_sse_s3_default_upload(1024*1024)
@attr(operation='Test 8MB upload to SSE-S3 default-encrypted bucket')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-encryption')
+@pytest.mark.bucket_encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_s3_default_upload_8mb():
_test_sse_s3_default_upload(8*1024*1024)
@attr(operation='Test 1 byte upload to SSE-KMS default-encrypted bucket')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-encryption')
+@pytest.mark.bucket_encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_default_upload_1b():
_test_sse_kms_default_upload(1)
@attr(operation='Test 1KB upload to SSE-KMS default-encrypted bucket')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-encryption')
+@pytest.mark.bucket_encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_default_upload_1kb():
_test_sse_kms_default_upload(1024)
@attr(operation='Test 1MB upload to SSE-KMS default-encrypted bucket')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-encryption')
+@pytest.mark.bucket_encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_default_upload_1mb():
_test_sse_kms_default_upload(1024*1024)
@attr(operation='Test 8MB upload to SSE-KMS default-encrypted bucket')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-encryption')
+@pytest.mark.bucket_encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_default_upload_8mb():
_test_sse_kms_default_upload(8*1024*1024)
@attr(operation='Test head operation on SSE-S3 default-encrypted object')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-encryption')
+@pytest.mark.bucket_encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_s3_default_method_head():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='complete SSE-S3 multi-part upload')
@attr(assertion='successful')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-encryption')
+@pytest.mark.bucket_encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_s3_default_multipart_upload():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='authenticated SSE-S3 browser based upload via POST request')
@attr(assertion='succeeds and returns written data')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-encryption')
+@pytest.mark.bucket_encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_s3_default_post_object_authenticated_request():
bucket_name = get_new_bucket()
client = get_client()
@attr(operation='authenticated SSE-kMS browser based upload via POST request')
@attr(assertion='succeeds and returns written data')
@attr('encryption')
+@pytest.mark.encryption
@attr('bucket-encryption')
-@attr('encryption')
+@pytest.mark.bucket_encryption
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_kms_default_post_object_authenticated_request():
kms_keyid = get_main_kms_keyid()
if kms_keyid is None:
@attr(operation='Test 1 byte upload with SSE-S3 encryption')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_s3_encrypted_upload_1b():
_test_sse_s3_encrypted_upload(1)
@attr(operation='Test 1Kb upload with SSE-S3 encryption')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_s3_encrypted_upload_1kb():
_test_sse_s3_encrypted_upload(1024)
@attr(operation='Test 1MB upload with SSE-S3 encryption')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_s3_encrypted_upload_1mb():
_test_sse_s3_encrypted_upload(1024*1024)
@attr(operation='Test 8MB upload with SSE-S3 encryption')
@attr(assertion='success')
@attr('encryption')
+@pytest.mark.encryption
@attr('sse-s3')
+@pytest.mark.sse_s3
@attr('fails_on_dbstore')
+@pytest.mark.fails_on_dbstore
def test_sse_s3_encrypted_upload_8mb():
_test_sse_s3_encrypted_upload(8*1024*1024)
projects / s3-tests.git / commitdiff