Updated s3 handler

Adding Warning handler for no_overwrite

Author: Rakshil Modi

awscli/customizations/s3/s3handler.py

@@ -438,7 +438,38 @@ def _get_fileout(self, fileinfo):
         return fileinfo.dest
 
     def _get_warning_handlers(self):
-        return [self._warn_glacier, self._warn_parent_reference]
+        return [
+            self._warn_glacier,
+            self._warn_parent_reference,
+            self._warn_if_file_exists_with_no_overwrite,
+        ]
+
+    def _warn_if_file_exists_with_no_overwrite(self, fileinfo):
+        """
+        Warning handler to skip downloads when no-overwrite is set and local file exists.
+
+        This method prevents overwriting existing local files during S3 download operations
+        when the --no-overwrite flag is specified. It checks if the destination file already
+        exists on the local filesystem and skips the download if found.
+
+        :type fileinfo: awscli.customizations.s3.fileinfo.FileInfo
+        :param fileinfo: The FileInfo object containing source and destination details
+
+        :rtype: bool
+        :returns: True if the file should be skipped (exists and no-overwrite is set),
+                False if the download should proceed
+        """
+        if not self._cli_params.get('no_overwrite'):
+            return False
+        fileout = self._get_fileout(fileinfo)
+        if os.path.exists(fileout):
+            LOGGER.debug(
+                "Warning: Skipping file %s as it already exists on %s",
+                fileinfo.src,
+                fileinfo.dest,
+            )
+            return True
+        return False
 
     def _format_src_dest(self, fileinfo):
         src = self._format_s3_path(fileinfo.src)

tests/unit/customizations/s3/test_s3handler.py

@@ -679,6 +679,39 @@ def test_submit_move_adds_delete_source_subscriber(self):
         for i, actual_subscriber in enumerate(actual_subscribers):
             self.assertIsInstance(actual_subscriber, ref_subscribers[i])
 
+    def test_submit_with_no_overwrite_flag_when_file_exists_at_destination(
+        self,
+    ):
+        # Setting up the CLI params with no_overwrite flag as True
+        self.cli_params['no_overwrite'] = True
+        fileinfo = self.create_file_info(self.key)
+        # Mocking os.path.exists to simulate that file already exists
+        with mock.patch('os.path.exists', return_value=True):
+            # Submitting download request
+            future = self.transfer_request_submitter.submit(fileinfo)
+            # Asserting that the future is None, as the file already exists
+            self.assertIsNone(future)
+            # Asserting that no download happened
+            self.assert_no_downloads_happened()
+
+    def test_submit_with_no_overwrite_flag_when_file_does_not_exist_at_destination(
+        self,
+    ):
+        # Setting up the CLI params with no_overwrite flag as True
+        self.cli_params['no_overwrite'] = True
+        fileinfo = self.create_file_info(self.key)
+        # Mocking os.path.exists to return False, to simulate that file does not exist
+        with mock.patch('os.path.exists', return_value=False):
+            # Submitting download request
+            future = self.transfer_request_submitter.submit(fileinfo)
+            # Asserting that the future is the same object returned by transfer_manager.download
+            # This confirms that download was actually initiated
+            self.assertIs(self.transfer_manager.download.return_value, future)
+            # Asserting that download happened
+            self.assertEqual(
+                len(self.transfer_manager.download.call_args_list), 1
+            )
+
 
 class TestCopyRequestSubmitter(BaseTransferRequestSubmitterTest):
     def setUp(self):