Add no overwrite flag for S3 copy

improved test case and format

Updated test cases

Fixed edge case for zero bytes

Update warning handler

Author: Rakshil Modi

awscli/customizations/s3/results.py

@@ -23,7 +23,6 @@
 from awscli.customizations.s3.subscribers import OnDoneFilteredSubscriber
 from awscli.customizations.s3.utils import (
     WarningResult,
-    create_warning,
     human_readable_size,
 )
 from awscli.customizations.utils import uni_print

awscli/customizations/s3/s3handler.py

@@ -13,6 +13,7 @@
 import logging
 import os
 
+from botocore.exceptions import ClientError
 from s3transfer.manager import TransferManager
 
 from awscli.compat import get_binary_stdin
@@ -481,7 +482,46 @@ def _submit_transfer_request(self, fileinfo, extra_args, subscribers):
         )
 
     def _get_warning_handlers(self):
-        return [self._warn_glacier]
+        return [
+            self._warn_glacier,
+            self._warn_if_zero_byte_file_exists_with_no_overwrite,
+        ]
+
+    def _warn_if_zero_byte_file_exists_with_no_overwrite(self, fileinfo):
+        """
+        Warning handler to skip zero-byte files when no_overwrite is set and file exists.
+
+        This method handles the transfer of zero-byte objects when the no-overwrite parameter is specified.
+        To prevent overwrite, it uses head_object to verify if the object exists at the destination:
+        If the object is present at destination: skip the file (return True)
+        If the object is not present at destination: allow transfer (return False)
+
+        :type fileinfo: FileInfo
+        :param fileinfo: The FileInfo object containing transfer details
+
+        :rtype: bool
+        :return: True if file should be skipped, False if transfer should proceed
+        """
+        if not self._cli_params.get('no_overwrite') or (
+            getattr(fileinfo, 'size') and fileinfo.size > 0
+        ):
+            return False
+
+        bucket, key = find_bucket_key(fileinfo.dest)
+        client = fileinfo.source_client
+        try:
+            client.head_object(Bucket=bucket, Key=key)
+            LOGGER.debug(
+                "Warning: Skipping file %s as it already exists on %s",
+                fileinfo.src,
+                fileinfo.dest,
+            )
+            return True
+        except ClientError as e:
+            if e.response['Error']['Code'] == '404':
+                return False
+            else:
+                raise
 
     def _format_src_dest(self, fileinfo):
         src = self._format_s3_path(fileinfo.src)

awscli/customizations/s3/utils.py

@@ -521,6 +521,7 @@ def map_copy_object_params(cls, request_params, cli_params):
         )
         cls._set_request_payer_param(request_params, cli_params)
         cls._set_checksum_algorithm_param(request_params, cli_params)
+        cls._set_no_overwrite_param(request_params, cli_params)
 
     @classmethod
     def map_head_object_params(cls, request_params, cli_params):

tests/functional/s3/__init__.py

@@ -257,6 +257,16 @@ def mp_copy_responses(self):
             self.complete_mpu_response(),
         ]
 
+    def precondition_failed_error_response(self, condition='If-None-Match'):
+        return {
+            'Error': {
+                'Code': 'PreconditionFailed',
+                'Message': 'At least one of the pre-conditions you '
+                'specified did not hold',
+                'Condition': condition,
+            }
+        }
+
 
 class BaseS3CLIRunnerTest(unittest.TestCase):
     def setUp(self):

tests/functional/s3/test_cp_command.py

@@ -335,7 +335,6 @@ def test_no_overwrite_flag_multipart_upload_when_object_not_exists_on_target(
         # Create a large file that will trigger multipart upload
         full_path = self.files.create_file('foo.txt', 'a' * 10 * (1024**2))
         cmdline = f'{self.prefix} {full_path} s3://bucket --no-overwrite'
-
         # Set up responses for multipart upload
         self.parsed_responses = [
             {'UploadId': 'foo'},  # CreateMultipartUpload response
@@ -397,6 +396,160 @@ def test_no_overwrite_flag_multipart_upload_when_object_exists_on_target(
         # Verify the IfNoneMatch condition was set in the CompleteMultipartUpload request
         self.assertEqual(self.operations_called[3][1]['IfNoneMatch'], '*')
 
+    def test_no_overwrite_flag_on_copy_when_small_object_does_not_exist_on_target(
+        self,
+    ):
+        cmdline = f'{self.prefix} s3://bucket1/key.txt s3://bucket/key1.txt --no-overwrite'
+        # Set up responses for multipart copy (since no-overwrite always uses multipart)
+        self.parsed_responses = [
+            self.head_object_response(),  # HeadObject to get source metadata
+            self.create_mpu_response('foo'),  # CreateMultipartUpload response
+            self.upload_part_copy_response(),  # UploadPartCopy response
+            {},  # CompleteMultipartUpload response
+        ]
+        self.run_cmd(cmdline, expected_rc=0)
+        # Verify all multipart operations were called
+        self.assertEqual(len(self.operations_called), 4)
+        self.assertEqual(self.operations_called[0][0].name, 'HeadObject')
+        self.assertEqual(
+            self.operations_called[1][0].name, 'CreateMultipartUpload'
+        )
+        self.assertEqual(self.operations_called[2][0].name, 'UploadPartCopy')
+        self.assertEqual(
+            self.operations_called[3][0].name, 'CompleteMultipartUpload'
+        )
+        # Verify the IfNoneMatch condition was set in the CompleteMultipartUpload request
+        self.assertEqual(self.operations_called[3][1]['IfNoneMatch'], '*')
+
+    def test_no_overwrite_flag_on_copy_when_small_object_exists_on_target(
+        self,
+    ):
+        cmdline = f'{self.prefix} s3://bucket1/key.txt s3://bucket/key.txt --no-overwrite'
+        # Set up responses for multipart copy (since no-overwrite always uses multipart)
+        self.parsed_responses = [
+            self.head_object_response(),  # HeadObject to get source metadata
+            self.create_mpu_response('foo'),  # CreateMultipartUpload response
+            self.upload_part_copy_response(),  # UploadPartCopy response
+            self.precondition_failed_error_response(),  # CompleteMultipartUpload
+            {},  # AbortMultipartUpload response
+        ]
+        self.run_cmd(cmdline, expected_rc=0)
+        # Verify all multipart operations were called
+        self.assertEqual(len(self.operations_called), 5)
+        self.assertEqual(self.operations_called[0][0].name, 'HeadObject')
+        self.assertEqual(
+            self.operations_called[1][0].name, 'CreateMultipartUpload'
+        )
+        self.assertEqual(self.operations_called[2][0].name, 'UploadPartCopy')
+        self.assertEqual(
+            self.operations_called[3][0].name, 'CompleteMultipartUpload'
+        )
+        self.assertEqual(
+            self.operations_called[4][0].name, 'AbortMultipartUpload'
+        )
+        # Verify the IfNoneMatch condition was set in the CompleteMultipartUpload request
+        self.assertEqual(self.operations_called[3][1]['IfNoneMatch'], '*')
+
+    def test_no_overwrite_flag_on_copy_when_zero_size_object_exists_at_destination(
+        self,
+    ):
+        cmdline = f'{self.prefix} s3://bucket1/file.txt s3://bucket2/file.txt --no-overwrite'
+        self.parsed_responses = [
+            self.head_object_response(
+                ContentLength=0
+            ),  # Source object (zero size)
+            self.head_object_response(),  # Checking the object at destination
+        ]
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assertEqual(len(self.operations_called), 2)
+        self.assertEqual(self.operations_called[0][0].name, 'HeadObject')
+        self.assertEqual(self.operations_called[1][0].name, 'HeadObject')
+
+    def test_no_overwrite_flag_on_copy_when_zero_size_object_not_exists_at_destination(
+        self,
+    ):
+        cmdline = f'{self.prefix} s3://bucket1/file.txt s3://bucket2/file1.txt --no-overwrite'
+        self.parsed_responses = [
+            self.head_object_response(
+                ContentLength=0
+            ),  # Source object (zero size)
+            {
+                'Error': {'Code': '404', 'Message': 'Not Found'}
+            },  # At destination object does not exists
+            self.copy_object_response(),  # Copy Request when object does not exists
+        ]
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assertEqual(len(self.operations_called), 3)
+        self.assertEqual(self.operations_called[0][0].name, 'HeadObject')
+        self.assertEqual(self.operations_called[1][0].name, 'HeadObject')
+        self.assertEqual(self.operations_called[2][0].name, 'CopyObject')
+
+    def test_no_overwrite_flag_on_copy_when_large_object_exists_on_target(
+        self,
+    ):
+        cmdline = f'{self.prefix} s3://bucket1/key.txt s3://bucket/key.txt --no-overwrite'
+        # Set up responses for multipart copy with large object
+        self.parsed_responses = [
+            self.head_object_response(
+                ContentLength=10 * (1024**2)
+            ),  # HeadObject with large content
+            self.get_object_tagging_response({}),  # GetObjectTagging response
+            self.create_mpu_response('foo'),  # CreateMultipartUpload response
+            self.upload_part_copy_response(),  # UploadPartCopy response part 1
+            self.upload_part_copy_response(),  # UploadPartCopy response part 2
+            self.precondition_failed_error_response(),  # CompleteMultipartUpload fails with PreconditionFailed
+            {},  # AbortMultipartUpload response
+        ]
+        self.run_cmd(cmdline, expected_rc=0)
+        # Verify all multipart operations were called
+        self.assertEqual(len(self.operations_called), 7)
+        self.assertEqual(self.operations_called[0][0].name, 'HeadObject')
+        self.assertEqual(self.operations_called[1][0].name, 'GetObjectTagging')
+        self.assertEqual(
+            self.operations_called[2][0].name, 'CreateMultipartUpload'
+        )
+        self.assertEqual(self.operations_called[3][0].name, 'UploadPartCopy')
+        self.assertEqual(self.operations_called[4][0].name, 'UploadPartCopy')
+        self.assertEqual(
+            self.operations_called[5][0].name, 'CompleteMultipartUpload'
+        )
+        self.assertEqual(
+            self.operations_called[6][0].name, 'AbortMultipartUpload'
+        )
+        # Verify the IfNoneMatch condition was set in the CompleteMultipartUpload request
+        self.assertEqual(self.operations_called[5][1]['IfNoneMatch'], '*')
+
+    def test_no_overwrite_flag_on_copy_when_large_object_does_not_exist_on_target(
+        self,
+    ):
+        cmdline = f'{self.prefix} s3://bucket1/key.txt s3://bucket/key1.txt --no-overwrite'
+        # Set up responses for multipart copy with large object
+        self.parsed_responses = [
+            self.head_object_response(
+                ContentLength=10 * (1024**2)
+            ),  # HeadObject with large content
+            self.get_object_tagging_response({}),  # GetObjectTagging response
+            self.create_mpu_response('foo'),  # CreateMultipartUpload response
+            self.upload_part_copy_response(),  # UploadPartCopy response part 1
+            self.upload_part_copy_response(),  # UploadPartCopy response part 2
+            {},  # CompleteMultipartUpload response
+        ]
+        self.run_cmd(cmdline, expected_rc=0)
+        # Verify all multipart operations were called
+        self.assertEqual(len(self.operations_called), 6)
+        self.assertEqual(self.operations_called[0][0].name, 'HeadObject')
+        self.assertEqual(self.operations_called[1][0].name, 'GetObjectTagging')
+        self.assertEqual(
+            self.operations_called[2][0].name, 'CreateMultipartUpload'
+        )
+        self.assertEqual(self.operations_called[3][0].name, 'UploadPartCopy')
+        self.assertEqual(self.operations_called[4][0].name, 'UploadPartCopy')
+        self.assertEqual(
+            self.operations_called[5][0].name, 'CompleteMultipartUpload'
+        )
+        # Verify the IfNoneMatch condition was set in the CompleteMultipartUpload request
+        self.assertEqual(self.operations_called[5][1]['IfNoneMatch'], '*')
+
     def test_dryrun_download(self):
         self.parsed_responses = [self.head_object_response()]
         target = self.files.full_path('file.txt')