]> git-server-git.apps.pok.os.sepia.ceph.com Git - s3-tests.git/commitdiff
CompleteMultipartUpload: test to verify sse-c headers
authorSeena Fallah <seenafallah@gmail.com>
Sat, 21 Jun 2025 17:10:33 +0000 (19:10 +0200)
committerCasey Bodley <cbodley@redhat.com>
Tue, 2 Dec 2025 18:50:02 +0000 (13:50 -0500)
Signed-off-by: Seena Fallah <seenafallah@gmail.com>
(cherry picked from commit 0086db16fc57a84e0ff5d8dba4a096636088bb29)

s3tests/functional/test_s3.py

index 9af637c8f090e47c9d84c7685ac4c31cd0cd2018..6c88f77460f55028038835e65776ac553d046c0c 100644 (file)
@@ -19976,6 +19976,25 @@ def _test_copy_part_enc(file_size, source_mode_key, dest_mode_key, source_sc=Non
         'PartNumber': 2
     })
 
+    if dest_mode_key == 'sse-c':
+        # make sure api is verifying the SSE-C headers
+        e = assert_raises(ClientError, client.complete_multipart_upload,
+                          Bucket=dest_bucket_name, Key='testobj2',
+                          UploadId=upload_id, MultipartUpload={'Parts': parts})
+        status, _ = _get_status_and_error_code(e.response)
+        assert status == 400
+
+        # and the key would be the same as the one used in upload part
+        # use the source key to complete the upload
+        # this is not allowed, so we expect an error
+        source_sse_c_args = _copy_enc_source_modes['sse-c']['args']
+        e = assert_raises(ClientError, client.complete_multipart_upload,
+                          Bucket=dest_bucket_name, Key='testobj2',
+                          UploadId=upload_id, MultipartUpload={'Parts': parts},
+                          **source_sse_c_args)
+        status, _ = _get_status_and_error_code(e.response)
+        assert status == 400
+
     # complete the multipart upload
     response = client.complete_multipart_upload(
         Bucket=dest_bucket_name,
@@ -19985,6 +20004,7 @@ def _test_copy_part_enc(file_size, source_mode_key, dest_mode_key, source_sc=Non
         **complete_args
     )
     assert dest_args.get('assert', lambda r: True)(response)
+
     # verify the copy is encrypted
     get_args = dest_args.get('get_args', {})
     response = client.get_object(Bucket=dest_bucket_name, Key='testobj2', **get_args)