mirror of
https://github.com/ceph/s3-tests.git
synced 2024-11-23 13:38:38 +00:00
sse: add test for default forced sse s3 encryption
Signed-off-by: Seena Fallah <seenafallah@gmail.com>
This commit is contained in:
parent
e29d6246fc
commit
2de06c9534
2 changed files with 56 additions and 0 deletions
|
@ -28,6 +28,7 @@ markers =
|
||||||
s3website_redirect_location
|
s3website_redirect_location
|
||||||
3website
|
3website
|
||||||
sse_s3
|
sse_s3
|
||||||
|
forced_sse_s3
|
||||||
storage_class
|
storage_class
|
||||||
tagging
|
tagging
|
||||||
test_of_iam
|
test_of_iam
|
||||||
|
|
|
@ -12703,6 +12703,61 @@ def test_delete_bucket_encryption_kms():
|
||||||
|
|
||||||
assert response_code == 'ServerSideEncryptionConfigurationNotFoundError'
|
assert response_code == 'ServerSideEncryptionConfigurationNotFoundError'
|
||||||
|
|
||||||
|
def _test_forced_sse_s3_default_upload(file_size):
|
||||||
|
"""
|
||||||
|
The cluster is forcing AES256 encryption if the encryption header is empty.
|
||||||
|
Create a file of A's of certain size, and use it to set_contents_from_file.
|
||||||
|
Re-read the contents, and confirm we get same content as input i.e., A's
|
||||||
|
"""
|
||||||
|
bucket_name = get_new_bucket()
|
||||||
|
client = get_client()
|
||||||
|
|
||||||
|
# make sure there is no bucket encryption in place
|
||||||
|
response_code = ""
|
||||||
|
try:
|
||||||
|
client.get_bucket_encryption(Bucket=bucket_name)
|
||||||
|
except ClientError as e:
|
||||||
|
response_code = e.response['Error']['Code']
|
||||||
|
|
||||||
|
assert response_code == 'ServerSideEncryptionConfigurationNotFoundError'
|
||||||
|
|
||||||
|
data = 'A'*file_size
|
||||||
|
response = client.put_object(Bucket=bucket_name, Key='testobj', Body=data)
|
||||||
|
assert response['ResponseMetadata']['HTTPHeaders']['x-amz-server-side-encryption'] == 'AES256'
|
||||||
|
|
||||||
|
response = client.get_object(Bucket=bucket_name, Key='testobj')
|
||||||
|
assert response['ResponseMetadata']['HTTPHeaders']['x-amz-server-side-encryption'] == 'AES256'
|
||||||
|
body = _get_body(response)
|
||||||
|
assert body == data
|
||||||
|
|
||||||
|
@pytest.mark.encryption
|
||||||
|
@pytest.mark.sse_s3
|
||||||
|
@pytest.mark.forced_sse_s3
|
||||||
|
@pytest.mark.fails_on_dbstore
|
||||||
|
def test_forced_sse_s3_default_upload_1b():
|
||||||
|
_test_forced_sse_s3_default_upload(1)
|
||||||
|
|
||||||
|
@pytest.mark.encryption
|
||||||
|
@pytest.mark.sse_s3
|
||||||
|
@pytest.mark.forced_sse_s3
|
||||||
|
@pytest.mark.fails_on_dbstore
|
||||||
|
def test_forced_sse_s3_default_upload_1kb():
|
||||||
|
_test_forced_sse_s3_default_upload(1024)
|
||||||
|
|
||||||
|
@pytest.mark.encryption
|
||||||
|
@pytest.mark.sse_s3
|
||||||
|
@pytest.mark.forced_sse_s3
|
||||||
|
@pytest.mark.fails_on_dbstore
|
||||||
|
def test_forced_sse_s3_default_upload_1mb():
|
||||||
|
_test_forced_sse_s3_default_upload(1024*1024)
|
||||||
|
|
||||||
|
@pytest.mark.encryption
|
||||||
|
@pytest.mark.sse_s3
|
||||||
|
@pytest.mark.forced_sse_s3
|
||||||
|
@pytest.mark.fails_on_dbstore
|
||||||
|
def test_forced_sse_s3_default_upload_8mb():
|
||||||
|
_test_forced_sse_s3_default_upload(8*1024*1024)
|
||||||
|
|
||||||
def _test_sse_s3_default_upload(file_size):
|
def _test_sse_s3_default_upload(file_size):
|
||||||
"""
|
"""
|
||||||
Test enables bucket encryption.
|
Test enables bucket encryption.
|
||||||
|
|
Loading…
Reference in a new issue