From 2de06c95349ac9e3dd787b517e4f9bea8aa4c6d1 Mon Sep 17 00:00:00 2001
From: Seena Fallah <seenafallah@gmail.com>
Date: Thu, 2 Nov 2023 16:27:07 +0100
Subject: [PATCH] sse: add test for default forced sse s3 encryption

Signed-off-by: Seena Fallah <seenafallah@gmail.com>
---
 pytest.ini                          |  1 +
 s3tests_boto3/functional/test_s3.py | 55 +++++++++++++++++++++++++++++
 2 files changed, 56 insertions(+)

diff --git a/pytest.ini b/pytest.ini
index 0e3bcba..2971ea4 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -28,6 +28,7 @@ markers =
     s3website_redirect_location
     3website
     sse_s3
+    forced_sse_s3
     storage_class
     tagging
     test_of_iam
diff --git a/s3tests_boto3/functional/test_s3.py b/s3tests_boto3/functional/test_s3.py
index 1273d8a..8bc492e 100644
--- a/s3tests_boto3/functional/test_s3.py
+++ b/s3tests_boto3/functional/test_s3.py
@@ -12703,6 +12703,61 @@ def test_delete_bucket_encryption_kms():
 
     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):
     """
     Test enables bucket encryption.