aws_s3: fix uploading from the target - backport #39023 (#40192)

* [aws_s3] Fix uploading src option on the target machine to a bucket (#39023)

* Fix backward compatibility for uploading src option on the target machine to a bucket

* Allow the module to handle errors for nonexistent files

(cherry picked from commit b8a93c12e2)

* changelog

changelogs/fragments/aws_s3_fix_uploading_src_on_target.yaml

@@ -0,0 +1,3 @@
+---
+bugfixes:
+  - Fix regression in aws_s3 to allow uploading files on the remote host to an S3 bucket

lib/ansible/plugins/action/aws_s3.py

@@ -20,7 +20,7 @@ __metaclass__ = type
 
 import os
 
-from ansible.errors import AnsibleError, AnsibleAction, AnsibleActionFail
+from ansible.errors import AnsibleError, AnsibleAction, AnsibleActionFail, AnsibleFileNotFound
 from ansible.module_utils._text import to_text
 from ansible.plugins.action import ActionBase
 
@@ -43,11 +43,17 @@ class ActionModule(ActionBase):
             new_module_args = self._task.args.copy()
             if source:
                 source = os.path.expanduser(source)
-                try:
-                    source = self._loader.get_real_file(self._find_needle('files', source))
-                    new_module_args['src'] = source
-                except AnsibleError as e:
-                    raise AnsibleActionFail(to_text(e))
+
+                # For backward compatibility check if the file exists on the remote; it should take precedence
+                if not self._remote_file_exists(source):
+                    try:
+                        source = self._loader.get_real_file(self._find_needle('files', source))
+                        new_module_args['src'] = source
+                    except AnsibleFileNotFound as e:
+                        # module handles error message for nonexistent files
+                        new_module_args['src'] = source
+                    except AnsibleError as e:
+                        raise AnsibleActionFail(to_text(e))
 
             # execute the aws_s3 module now, with the updated args
             result.update(self._execute_module(module_args=new_module_args, task_vars=task_vars))