Raise an error on create bucket if use regional endpoint for us-east-…

…1 and region not set (#25945)

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

@@ -244,7 +244,13 @@ def create_bucket(self, bucket_name: Optional[str] = None, region_name: Optional
         :param region_name: The name of the aws region in which to create the bucket.
         """
         if not region_name:
-            region_name = self.get_conn().meta.region_name
+            if self.conn_region_name == "aws-global":
+                raise AirflowException(
+                    "Unable to create bucket if `region_name` not set "
+                    "and boto3 configured to use s3 regional endpoints."
+                )
+            region_name = self.conn_region_name
+
         if region_name == 'us-east-1':
             self.get_conn().create_bucket(Bucket=bucket_name)
         else:

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

@@ -26,6 +26,7 @@
 import pytest
 from botocore.exceptions import ClientError, NoCredentialsError
 
+from airflow.exceptions import AirflowException
 from airflow.models import Connection
 from airflow.providers.amazon.aws.hooks.s3 import S3Hook, provide_bucket_name, unify_bucket_name_and_key
 from airflow.utils.timezone import datetime
@@ -128,6 +129,43 @@ def test_create_bucket_other_region(self):
         region = bucket.meta.client.get_bucket_location(Bucket=bucket.name).get('LocationConstraint')
         assert region == 'us-east-2'
 
+    @mock_s3
+    @pytest.mark.parametrize("region_name", ["eu-west-1", "us-east-1"])
+    def test_create_bucket_regional_endpoint(self, region_name, monkeypatch):
+        conn = Connection(
+            conn_id="regional-endpoint",
+            conn_type="aws",
+            extra={
+                "config_kwargs": {"s3": {"us_east_1_regional_endpoint": "regional"}},
+            },
+        )
+        with mock.patch.dict("os.environ", values={f"AIRFLOW_CONN_{conn.conn_id.upper()}": conn.get_uri()}):
+            monkeypatch.delenv('AWS_DEFAULT_REGION', raising=False)
+            hook = S3Hook(aws_conn_id=conn.conn_id)
+            bucket_name = f"regional-{region_name}"
+            hook.create_bucket(bucket_name, region_name=region_name)
+            bucket = hook.get_bucket(bucket_name)
+            assert bucket is not None
+            assert bucket.name == bucket_name
+            region = bucket.meta.client.get_bucket_location(Bucket=bucket.name).get('LocationConstraint')
+            assert region == (region_name if region_name != "us-east-1" else None)
+
+    def test_create_bucket_no_region_regional_endpoint(self, monkeypatch):
+        conn = Connection(
+            conn_id="no-region-regional-endpoint",
+            conn_type="aws",
+            extra={"config_kwargs": {"s3": {"us_east_1_regional_endpoint": "regional"}}},
+        )
+        with mock.patch.dict("os.environ", values={f"AIRFLOW_CONN_{conn.conn_id.upper()}": conn.get_uri()}):
+            monkeypatch.delenv('AWS_DEFAULT_REGION', raising=False)
+            hook = S3Hook(aws_conn_id=conn.conn_id)
+            error_message = (
+                "Unable to create bucket if `region_name` not set and boto3 "
+                r"configured to use s3 regional endpoints\."
+            )
+            with pytest.raises(AirflowException, match=error_message):
+                hook.create_bucket("unable-to-create")
+
     def test_check_for_prefix(self, s3_bucket):
         hook = S3Hook()
         bucket = hook.get_bucket(s3_bucket)