Fix HttpToS3Operator throws exception if s3_bucket parameter is not…

… passed (#43828)

* Code fix to extract bucket_name from airflow ui conn

* Additional test condition to check explicit bucket name precedence

providers/src/airflow/providers/amazon/aws/hooks/s3.py

@@ -86,7 +86,7 @@ def provide_bucket_name(func: Callable) -> Callable:
     async def maybe_add_bucket_name(*args, **kwargs):
         bound_args = function_signature.bind(*args, **kwargs)
 
-        if "bucket_name" not in bound_args.arguments:
+        if not bound_args.arguments.get("bucket_name"):
             self = args[0]
             if self.aws_conn_id:
                 connection = await sync_to_async(self.get_connection)(self.aws_conn_id)
@@ -116,7 +116,7 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
         def wrapper(*args, **kwargs) -> Callable:
             bound_args = function_signature.bind(*args, **kwargs)
 
-            if "bucket_name" not in bound_args.arguments:
+            if not bound_args.arguments.get("bucket_name"):
                 self = args[0]
 
                 if "bucket_name" in self.service_config:

providers/tests/amazon/aws/hooks/test_s3.py

@@ -1160,6 +1160,12 @@ def test_function(self, bucket_name=None):
             test_bucket_name = fake_s3_hook.test_function(bucket_name="bucket")
             assert test_bucket_name == "bucket"
 
+            # Test: `bucket_name` should use the explicitly provided value over `service_config`
+            test_bucket_name = fake_s3_hook.test_function(bucket_name="some_custom_bucket")
+            assert (
+                test_bucket_name == "some_custom_bucket"
+            ), "Expected provided bucket_name to take precedence over fallback"
+
     def test_delete_objects_key_does_not_exist(self, s3_bucket):
         # The behaviour of delete changed in recent version of s3 mock libraries.
         # It will succeed idempotently