Feature: Change ARN's to use correct partition (#7699)

moto/acm/utils.py

@@ -1,7 +1,8 @@
 from moto.moto_api._internal import mock_random
+from moto.utilities.utils import get_partition
 
 
 def make_arn_for_certificate(account_id: str, region_name: str) -> str:
     # Example
     # arn:aws:acm:eu-west-2:764371465172:certificate/c4b738b8-56fe-4b3a-b841-1c047654780b
-    return f"arn:aws:acm:{region_name}:{account_id}:certificate/{mock_random.uuid4()}"
+    return f"arn:{get_partition(region_name)}:acm:{region_name}:{account_id}:certificate/{mock_random.uuid4()}"

moto/amp/models.py

@@ -8,6 +8,7 @@
 from moto.moto_api._internal import mock_random
 from moto.utilities.paginator import paginate
 from moto.utilities.tagging_service import TaggingService
+from moto.utilities.utils import get_partition
 
 from .exceptions import RuleGroupNamespaceNotFound, WorkspaceNotFound
 from .utils import PAGINATION_MODEL
@@ -26,7 +27,7 @@ def __init__(
         self.name = name
         self.data = data
         self.tag_fn = tag_fn
-        self.arn = f"arn:aws:aps:{region}:{account_id}:rulegroupsnamespace/{workspace_id}/{self.name}"
+        self.arn = f"arn:{get_partition(region)}:aps:{region}:{account_id}:rulegroupsnamespace/{workspace_id}/{self.name}"
         self.created_at = unix_time()
         self.modified_at = self.created_at
 
@@ -56,7 +57,7 @@ def __init__(
     ):
         self.alias = alias
         self.workspace_id = f"ws-{mock_random.uuid4()}"
-        self.arn = f"arn:aws:aps:{region}:{account_id}:workspace/{self.workspace_id}"
+        self.arn = f"arn:{get_partition(region)}:aps:{region}:{account_id}:workspace/{self.workspace_id}"
         self.endpoint = f"https://aps-workspaces.{region}.amazonaws.com/workspaces/{self.workspace_id}/"
         self.status = {"statusCode": "ACTIVE"}
         self.created_at = unix_time()

moto/apigatewayv2/models.py

@@ -11,6 +11,7 @@
 from moto.core.utils import unix_time
 from moto.moto_api._internal import mock_random as random
 from moto.utilities.tagging_service import TaggingService
+from moto.utilities.utils import get_partition
 
 from .exceptions import (
     ApiMappingNotFound,
@@ -594,7 +595,9 @@ def __init__(
         self.routes: Dict[str, Route] = dict()
         self.stages: Dict[str, Stage] = dict()
 
-        self.arn = f"arn:aws:apigateway:{region}::/apis/{self.api_id}"
+        self.arn = (
+            f"arn:{get_partition(region)}:apigateway:{region}::/apis/{self.api_id}"
+        )
         self.backend.tag_resource(self.arn, tags)
 
     def clear(self) -> None:
@@ -1061,7 +1064,7 @@ def __init__(
         self.sg_ids = sg_ids
         self.subnet_ids = subnet_ids
 
-        self.arn = f"arn:aws:apigateway:{backend.region_name}::/vpclinks/{self.id}"
+        self.arn = f"arn:{get_partition(backend.region_name)}:apigateway:{backend.region_name}::/vpclinks/{self.id}"
         self.backend = backend
         self.backend.tag_resource(self.arn, tags)
 

moto/appconfig/models.py

@@ -4,6 +4,7 @@
 from moto.core.common_models import BaseModel
 from moto.moto_api._internal import mock_random
 from moto.utilities.tagging_service import TaggingService
+from moto.utilities.utils import get_partition
 
 from .exceptions import (
     AppNotFoundException,
@@ -56,7 +57,7 @@ def __init__(
         _type: str,
     ):
         self.id = mock_random.get_random_hex(7)
-        self.arn = f"arn:aws:appconfig:{region}:{account_id}:application/{application_id}/configurationprofile/{self.id}"
+        self.arn = f"arn:{get_partition(region)}:appconfig:{region}:{account_id}:application/{application_id}/configurationprofile/{self.id}"
         self.application_id = application_id
         self.name = name
         self.description = description
@@ -116,7 +117,7 @@ def __init__(
         self, name: str, description: Optional[str], region: str, account_id: str
     ):
         self.id = mock_random.get_random_hex(7)
-        self.arn = f"arn:aws:appconfig:{region}:{account_id}:application/{self.id}"
+        self.arn = f"arn:{get_partition(region)}:appconfig:{region}:{account_id}:application/{self.id}"
         self.name = name
         self.description = description
 

moto/applicationautoscaling/models.py

@@ -7,6 +7,7 @@
 from moto.core.common_models import BaseModel
 from moto.ecs import ecs_backends
 from moto.moto_api._internal import mock_random
+from moto.utilities.utils import get_partition
 
 from .exceptions import AWSValidationException
 
@@ -463,7 +464,7 @@ def __init__(
         self.policy_name = policy_name
         self.policy_type = policy_type
         self._guid = mock_random.uuid4()
-        self.policy_arn = f"arn:aws:autoscaling:{region_name}:{account_id}:scalingPolicy:{self._guid}:resource/{self.service_namespace}/{self.resource_id}:policyName/{self.policy_name}"
+        self.policy_arn = f"arn:{get_partition(region_name)}:autoscaling:{region_name}:{account_id}:scalingPolicy:{self._guid}:resource/{self.service_namespace}/{self.resource_id}:policyName/{self.policy_name}"
         self.creation_time = time.time()
         self.alarms: List["FakeAlarm"] = []
 
@@ -638,7 +639,7 @@ def __init__(
         account_id: str,
         region: str,
     ) -> None:
-        self.arn = f"arn:aws:autoscaling:{region}:{account_id}:scheduledAction:{service_namespace}:scheduledActionName/{scheduled_action_name}"
+        self.arn = f"arn:{get_partition(region)}:autoscaling:{region}:{account_id}:scheduledAction:{service_namespace}:scheduledActionName/{scheduled_action_name}"
         self.service_namespace = service_namespace
         self.schedule = schedule
         self.timezone = timezone

moto/appsync/models.py

@@ -8,6 +8,7 @@
 from moto.core.utils import unix_time
 from moto.moto_api._internal import mock_random
 from moto.utilities.tagging_service import TaggingService
+from moto.utilities.utils import get_partition
 
 from .exceptions import BadRequestException, GraphqlAPINotFound, GraphQLSchemaException
 
@@ -155,7 +156,7 @@ def __init__(
         self.user_pool_config = user_pool_config
         self.xray_enabled = xray_enabled
 
-        self.arn = f"arn:aws:appsync:{self.region}:{account_id}:apis/{self.api_id}"
+        self.arn = f"arn:{get_partition(self.region)}:appsync:{self.region}:{account_id}:apis/{self.api_id}"
         self.graphql_schema: Optional[GraphqlSchema] = None
 
         self.api_keys: Dict[str, GraphqlAPIKey] = dict()

moto/athena/models.py

@@ -6,6 +6,7 @@
 from moto.core.common_models import BaseModel
 from moto.moto_api._internal import mock_random
 from moto.utilities.paginator import paginate
+from moto.utilities.utils import get_partition
 
 
 class TaggableResourceMixin:
@@ -22,7 +23,7 @@ def __init__(
         self.region = region_name
         self.resource_name = resource_name
         self.tags = tags or []
-        self.arn = f"arn:aws:athena:{region_name}:{account_id}:{resource_name}"
+        self.arn = f"arn:{get_partition(region_name)}:athena:{region_name}:{account_id}:{resource_name}"
 
     def create_tags(self, tags: List[Dict[str, str]]) -> List[Dict[str, str]]:
         new_keys = [tag_set["Key"] for tag_set in tags]

moto/autoscaling/models.py

@@ -18,6 +18,7 @@
     BlockDeviceMapping,
     BlockDeviceType,
 )
+from moto.utilities.utils import get_partition
 
 from .exceptions import (
     AutoscalingClientError,
@@ -112,7 +113,7 @@ def __init__(
 
     @property
     def arn(self) -> str:
-        return f"arn:aws:autoscaling:{self.autoscaling_backend.region_name}:{self.autoscaling_backend.account_id}:scalingPolicy:c322761b-3172-4d56-9a21-0ed9d6161d67:autoScalingGroupName/{self.as_name}:policyName/{self.name}"
+        return f"arn:{get_partition(self.autoscaling_backend.region_name)}:autoscaling:{self.autoscaling_backend.region_name}:{self.autoscaling_backend.account_id}:scalingPolicy:c322761b-3172-4d56-9a21-0ed9d6161d67:autoScalingGroupName/{self.as_name}:policyName/{self.name}"
 
     def execute(self) -> None:
         if self.adjustment_type == "ExactCapacity":
@@ -174,7 +175,7 @@ def __init__(
         self.metadata_options = metadata_options
         self.classic_link_vpc_id = classic_link_vpc_id
         self.classic_link_vpc_security_groups = classic_link_vpc_security_groups
-        self.arn = f"arn:aws:autoscaling:{region_name}:{account_id}:launchConfiguration:9dbbbf87-6141-428a-a409-0752edbe6cad:launchConfigurationName/{self.name}"
+        self.arn = f"arn:{get_partition(region_name)}:autoscaling:{region_name}:{account_id}:launchConfiguration:9dbbbf87-6141-428a-a409-0752edbe6cad:launchConfigurationName/{self.name}"
 
     @classmethod
     def create_from_instance(
@@ -502,7 +503,7 @@ def tags(self, tags: List[Dict[str, str]]) -> None:
 
     @property
     def arn(self) -> str:
-        return f"arn:aws:autoscaling:{self.region}:{self.account_id}:autoScalingGroup:{self._id}:autoScalingGroupName/{self.name}"
+        return f"arn:{get_partition(self.region)}:autoscaling:{self.region}:{self.account_id}:autoScalingGroup:{self._id}:autoScalingGroupName/{self.name}"
 
     def active_instances(self) -> List[InstanceState]:
         return [x for x in self.instance_states if x.lifecycle_state == "InService"]

moto/awslambda/models.py

@@ -40,7 +40,7 @@
 from moto.s3.models import FakeKey, s3_backends
 from moto.sqs.models import sqs_backends
 from moto.utilities.docker_utilities import DockerModel
-from moto.utilities.utils import load_resource_as_bytes
+from moto.utilities.utils import get_partition, load_resource_as_bytes
 
 from .exceptions import (
     ConflictException,
@@ -529,9 +529,7 @@ def __init__(
         description: str,
         routing_config: str,
     ):
-        self.arn = (
-            f"arn:aws:lambda:{region}:{account_id}:function:{function_name}:{name}"
-        )
+        self.arn = f"arn:{get_partition(region)}:lambda:{region}:{account_id}:function:{function_name}:{name}"
         self.name = name
         self.function_version = function_version
         self.description = description
@@ -1451,7 +1449,7 @@ def get_alias(self, name: str, function_name: str) -> LambdaAlias:
         if name in aliases:
             return aliases[name]
 
-        arn = f"arn:aws:lambda:{self.region_name}:{self.account_id}:function:{function_name}:{name}"
+        arn = f"arn:{get_partition(self.region_name)}:lambda:{self.region_name}:{self.account_id}:function:{function_name}:{name}"
         raise UnknownAliasException(arn)
 
     def put_alias(
@@ -1467,7 +1465,7 @@ def put_alias(
         )
         aliases = self._get_function_aliases(function_name)
         if name in aliases:
-            arn = f"arn:aws:lambda:{self.region_name}:{self.account_id}:function:{function_name}:{name}"
+            arn = f"arn:{get_partition(self.region_name)}:lambda:{self.region_name}:{self.account_id}:function:{function_name}:{name}"
             raise ConflictException(f"Alias already exists: {arn}")
 
         alias = LambdaAlias(

moto/awslambda/utils.py

@@ -2,6 +2,8 @@
 from functools import partial
 from typing import TYPE_CHECKING, Any, Callable
 
+from moto.utilities.utils import get_partition
+
 if TYPE_CHECKING:
     from .models import LambdaBackend
 
@@ -10,7 +12,9 @@
 
 
 def make_arn(resource_type: str, region: str, account: str, name: str) -> str:
-    return f"arn:aws:lambda:{region}:{account}:{resource_type}:{name}"
+    return (
+        f"arn:{get_partition(region)}:lambda:{region}:{account}:{resource_type}:{name}"
+    )
 
 
 make_function_arn = partial(make_arn, "function")

moto/backup/models.py

@@ -6,6 +6,7 @@
 from moto.core.utils import unix_time
 from moto.moto_api._internal import mock_random
 from moto.utilities.tagging_service import TaggingService
+from moto.utilities.utils import get_partition
 
 from .exceptions import AlreadyExistsException, ResourceNotFoundException
 
@@ -18,7 +19,7 @@ def __init__(
         backend: "BackupBackend",
     ):
         self.backup_plan_id = str(mock_random.uuid4())
-        self.backup_plan_arn = f"arn:aws:backup:{backend.region_name}:{backend.account_id}:backup-plan:{self.backup_plan_id}"
+        self.backup_plan_arn = f"arn:{get_partition(backend.region_name)}:backup:{backend.region_name}:{backend.account_id}:backup-plan:{self.backup_plan_id}"
         self.creation_date = unix_time()
         ran_str = mock_random.get_random_string(length=48)
         self.version_id = ran_str
@@ -84,7 +85,7 @@ def __init__(
         backend: "BackupBackend",
     ):
         self.backup_vault_name = backup_vault_name
-        self.backup_vault_arn = f"arn:aws:backup:{backend.region_name}:{backend.account_id}:backup-vault:{backup_vault_name}"
+        self.backup_vault_arn = f"arn:{get_partition(backend.region_name)}:backup:{backend.region_name}:{backend.account_id}:backup-vault:{backup_vault_name}"
         self.creation_date = unix_time()
         self.encryption_key_arn = encryption_key_arn
         self.creator_request_id = creator_request_id

moto/batch/models.py

@@ -27,6 +27,7 @@
 from moto.moto_api._internal.managed_state_model import ManagedState
 from moto.utilities.docker_utilities import DockerModel
 from moto.utilities.tagging_service import TaggingService
+from moto.utilities.utils import get_partition
 
 from .exceptions import ClientException, InvalidParameterValueException, ValidationError
 from .utils import (
@@ -977,7 +978,7 @@ def __init__(
         tags: Dict[str, str],
     ):
         self.name = name
-        self.arn = f"arn:aws:batch:{region}:{account_id}:scheduling-policy/{name}"
+        self.arn = f"arn:{get_partition(region)}:batch:{region}:{account_id}:scheduling-policy/{name}"
         self.fairshare_policy = {
             "computeReservation": fairshare_policy.get("computeReservation") or 0,
             "shareDecaySeconds": fairshare_policy.get("shareDecaySeconds") or 0,

moto/batch/utils.py

@@ -1,25 +1,27 @@
 from enum import Enum
 from typing import Any, Dict, List, Optional, Tuple
 
+from moto.utilities.utils import get_partition
+
 from .exceptions import ValidationError
 
 
 def make_arn_for_compute_env(account_id: str, name: str, region_name: str) -> str:
-    return f"arn:aws:batch:{region_name}:{account_id}:compute-environment/{name}"
+    return f"arn:{get_partition(region_name)}:batch:{region_name}:{account_id}:compute-environment/{name}"
 
 
 def make_arn_for_job_queue(account_id: str, name: str, region_name: str) -> str:
-    return f"arn:aws:batch:{region_name}:{account_id}:job-queue/{name}"
+    return f"arn:{get_partition(region_name)}:batch:{region_name}:{account_id}:job-queue/{name}"
 
 
 def make_arn_for_job(account_id: str, job_id: str, region_name: str) -> str:
-    return f"arn:aws:batch:{region_name}:{account_id}:job/{job_id}"
+    return f"arn:{get_partition(region_name)}:batch:{region_name}:{account_id}:job/{job_id}"
 
 
 def make_arn_for_task_def(
     account_id: str, name: str, revision: int, region_name: str
 ) -> str:
-    return f"arn:aws:batch:{region_name}:{account_id}:job-definition/{name}:{revision}"
+    return f"arn:{get_partition(region_name)}:batch:{region_name}:{account_id}:job-definition/{name}:{revision}"
 
 
 def lowercase_first_key(some_dict: Dict[str, Any]) -> Dict[str, Any]:

moto/bedrock/models.py

@@ -14,6 +14,7 @@
 from moto.core.common_models import BaseModel
 from moto.utilities.paginator import paginate
 from moto.utilities.tagging_service import TaggingService
+from moto.utilities.utils import get_partition
 
 
 class ModelCustomizationJob(BaseModel):
@@ -80,16 +81,16 @@ def __init__(
         self.vpc_config = vpc_config
         self.region_name = region_name
         self.account_id = account_id
-        self.job_arn = f"arn:aws:bedrock:{self.region_name}:{self.account_id}:model-customization-job/{self.job_name}"
+        self.job_arn = f"arn:{get_partition(self.region_name)}:bedrock:{self.region_name}:{self.account_id}:model-customization-job/{self.job_name}"
         self.output_model_name = f"{self.custom_model_name}-{self.job_name}"
-        self.output_model_arn = f"arn:aws:bedrock:{self.region_name}:{self.account_id}:custom-model/{self.output_model_name}"
+        self.output_model_arn = f"arn:{get_partition(self.region_name)}:bedrock:{self.region_name}:{self.account_id}:custom-model/{self.output_model_name}"
         self.status = "InProgress"
         self.failure_message = "Failure Message"
         self.creation_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
         self.last_modified_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
         self.end_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-        self.base_model_arn = f"arn:aws:bedrock:{self.region_name}::foundation-model/{self.base_model_identifier}"
-        self.output_model_kms_key_arn = f"arn:aws:kms:{self.region_name}:{self.account_id}:key/{self.output_model_name}-kms-key"
+        self.base_model_arn = f"arn:{get_partition(self.region_name)}:bedrock:{self.region_name}::foundation-model/{self.base_model_identifier}"
+        self.output_model_kms_key_arn = f"arn:{get_partition(self.region_name)}:kms:{self.region_name}:{self.account_id}:key/{self.output_model_name}-kms-key"
         self.training_metrics = {"trainingLoss": 0.0}  # hard coded
         self.validation_metrics = [{"validationLoss": 0.0}]  # hard coded
 
@@ -153,7 +154,7 @@ def __init__(
         self.validation_metrics = validation_metrics
         self.region_name = region_name
         self.account_id = account_id
-        self.model_arn = f"arn:aws:bedrock:{self.region_name}:{self.account_id}:custom-model/{self.model_name}"
+        self.model_arn = f"arn:{get_partition(self.region_name)}:bedrock:{self.region_name}:{self.account_id}:custom-model/{self.model_name}"
         self.creation_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
         self.base_model_name = base_model_name
 

moto/bedrockagent/models.py

@@ -13,6 +13,7 @@
 from moto.moto_api._internal import mock_random
 from moto.utilities.paginator import paginate
 from moto.utilities.tagging_service import TaggingService
+from moto.utilities.utils import get_partition
 
 
 class Agent(BaseModel):
@@ -46,7 +47,7 @@ def __init__(
         self.prepared_at = unix_time()
         self.agent_status = "PREPARED"
         self.agent_id = self.agent_name + str(mock_random.uuid4())[:8]
-        self.agent_arn = f"arn:aws:bedrock:{self.region_name}:{self.account_id}:agent/{self.agent_id}"
+        self.agent_arn = f"arn:{get_partition(self.region_name)}:bedrock:{self.region_name}:{self.account_id}:agent/{self.agent_id}"
         self.agent_version = "1.0"
         self.failure_reasons: List[str] = []
         self.recommended_actions = ["action"]
@@ -124,7 +125,7 @@ def __init__(
         self.region_name = region_name
         self.account_id = account_id
         self.knowledge_base_id = self.name + str(mock_random.uuid4())[:8]
-        self.knowledge_base_arn = f"arn:aws:bedrock:{self.region_name}:{self.account_id}:knowledge-base/{self.knowledge_base_id}"
+        self.knowledge_base_arn = f"arn:{get_partition(self.region_name)}:bedrock:{self.region_name}:{self.account_id}:knowledge-base/{self.knowledge_base_id}"
         self.created_at = unix_time()
         self.updated_at = unix_time()
         self.status = "Active"