[SPEC-7931] Filter list_buckets results based on region during import (#2950)

Filter s3 list_buckets results based on region correctly to improve resource import.

Signed-off-by: onecent1101 <liug@amazon.com>

Gems/AWSCore/Code/Tools/ResourceMappingTool/tests/unit/utils/test_aws_utils.py

@@ -98,16 +98,52 @@ class TestAWSUtils(TestCase):
         mocked_s3_client.list_buckets.assert_called_once()
         assert not actual_buckets
 
-    def test_list_s3_buckets_return_expected_buckets(self) -> None:
+    def test_list_s3_buckets_return_empty_list_with_no_matching_region(self) -> None:
+        expected_region: str = "us-east-1"
         mocked_s3_client: MagicMock = self._mock_client.return_value
         expected_buckets: List[str] = [f"{TestAWSUtils._expected_bucket}1", f"{TestAWSUtils._expected_bucket}2"]
         mocked_s3_client.list_buckets.return_value = {"Buckets": [{"Name": expected_buckets[0]},
                                                                   {"Name": expected_buckets[1]}]}
+        mocked_s3_client.get_bucket_location.side_effect = [{"LocationConstraint": "us-east-2"},
+                                                            {"LocationConstraint": "us-west-1"}]
 
-        actual_buckets: List[str] = aws_utils.list_s3_buckets()
-        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.S3_SERVICE_NAME)
+        actual_buckets: List[str] = aws_utils.list_s3_buckets(expected_region)
+        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.S3_SERVICE_NAME,
+                                                  region_name=expected_region)
+        mocked_s3_client.list_buckets.assert_called_once()
+        assert not actual_buckets
+
+    def test_list_s3_buckets_return_expected_buckets_matching_region(self) -> None:
+        expected_region: str = "us-west-2"
+        mocked_s3_client: MagicMock = self._mock_client.return_value
+        expected_buckets: List[str] = [f"{TestAWSUtils._expected_bucket}1", f"{TestAWSUtils._expected_bucket}2"]
+        mocked_s3_client.list_buckets.return_value = {"Buckets": [{"Name": expected_buckets[0]},
+                                                                  {"Name": expected_buckets[1]}]}
+        mocked_s3_client.get_bucket_location.side_effect = [{"LocationConstraint": "us-west-2"},
+                                                            {"LocationConstraint": "us-west-1"}]
+
+        actual_buckets: List[str] = aws_utils.list_s3_buckets(expected_region)
+        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.S3_SERVICE_NAME,
+                                                  region_name=expected_region)
+        mocked_s3_client.list_buckets.assert_called_once()
+        assert len(actual_buckets) == 1
+        assert actual_buckets[0] == expected_buckets[0]
+
+    def test_list_s3_buckets_return_expected_iad_buckets(self) -> None:
+        expected_region: str = "us-east-1"
+        mocked_s3_client: MagicMock = self._mock_client.return_value
+        expected_buckets: List[str] = [f"{TestAWSUtils._expected_bucket}1", f"{TestAWSUtils._expected_bucket}2"]
+        mocked_s3_client.list_buckets.return_value = {"Buckets": [{"Name": expected_buckets[0]},
+                                                                  {"Name": expected_buckets[1]}]}
+        mocked_s3_client.get_bucket_location.side_effect = [{"LocationConstraint": None},
+                                                            {"LocationConstraint": "us-west-1"}]
+
+        actual_buckets: List[str] = aws_utils.list_s3_buckets(expected_region)
+        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.S3_SERVICE_NAME,
+                                                  region_name=expected_region)
         mocked_s3_client.list_buckets.assert_called_once()
-        assert actual_buckets == expected_buckets
+        assert len(actual_buckets) == 1
+        assert actual_buckets[0] == expected_buckets[0]
 
     def test_list_lambda_functions_return_empty_list(self) -> None:
         mocked_lambda_client: MagicMock = self._mock_client.return_value

Gems/AWSCore/Code/Tools/ResourceMappingTool/utils/aws_utils.py

@@ -103,7 +103,17 @@ def list_s3_buckets(region: str = "") -> List[str]:
     bucket_names: List[str] = []
     bucket: Dict[str, any]
     for bucket in response["Buckets"]:
-        bucket_names.append(bucket["Name"])
+        try:
+            bucket_name: str = bucket["Name"]
+            location_response: Dict[str, any] = s3_client.get_bucket_location(Bucket=bucket_name)
+            # Buckets in Region us-east-1 have a LocationConstraint of null .
+            # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Client.get_bucket_location
+            if ((location_response["LocationConstraint"] == region) or
+                    (not location_response["LocationConstraint"] and region == "us-east-1")):
+                bucket_names.append(bucket_name)
+        except ClientError as error:
+            raise RuntimeError(error_messages.AWS_SERVICE_REQUEST_CLIENT_ERROR_MESSAGE.format(
+                "get_bucket_location", error.response['Error']['Code'], error.response['Error']['Message']))
     return bucket_names