Fixed validation on custom-resource in applicationautoscaling (#4026)

* Added ResourceTypeExceptions

* Added test for custom-resource

Co-authored-by: Phil Sheets <p.sheets@fetchrewards.com>

moto/applicationautoscaling/models.py

@@ -8,6 +8,11 @@ import time
 import uuid
 
 
+@unique
+class ResourceTypeExceptionValueSet(Enum):
+    RESOURCE_TYPE = "ResourceType"
+
+
 @unique
 class ServiceNamespaceValueSet(Enum):
     APPSTREAM = "appstream"
@@ -74,7 +79,7 @@ class ApplicationAutoscalingBackend(BaseBackend):
     def describe_scalable_targets(
         self, namespace, r_ids=None, dimension=None,
     ):
-        """ Describe scalable targets. """
+        """Describe scalable targets."""
         if r_ids is None:
             r_ids = []
         targets = self._flatten_scalable_targets(namespace)
@@ -85,7 +90,7 @@ class ApplicationAutoscalingBackend(BaseBackend):
         return targets
 
     def _flatten_scalable_targets(self, namespace):
-        """ Flatten scalable targets for a given service namespace down to a list. """
+        """Flatten scalable targets for a given service namespace down to a list."""
         targets = []
         for dimension in self.targets.keys():
             for resource_id in self.targets[dimension].keys():
@@ -94,7 +99,7 @@ class ApplicationAutoscalingBackend(BaseBackend):
         return targets
 
     def register_scalable_target(self, namespace, r_id, dimension, **kwargs):
-        """ Registers or updates a scalable target. """
+        """Registers or updates a scalable target."""
         _ = _target_params_are_valid(namespace, r_id, dimension)
         if namespace == ServiceNamespaceValueSet.ECS.value:
             _ = self._ecs_service_exists_for_target(r_id)
@@ -127,7 +132,7 @@ class ApplicationAutoscalingBackend(BaseBackend):
         return target
 
     def deregister_scalable_target(self, namespace, r_id, dimension):
-        """ Registers or updates a scalable target. """
+        """Registers or updates a scalable target."""
         if self._scalable_target_exists(r_id, dimension):
             del self.targets[dimension][r_id]
         else:
@@ -222,7 +227,7 @@ class ApplicationAutoscalingBackend(BaseBackend):
 
 
 def _target_params_are_valid(namespace, r_id, dimension):
-    """ Check whether namespace, resource_id and dimension are valid and consistent with each other. """
+    """Check whether namespace, resource_id and dimension are valid and consistent with each other."""
     is_valid = True
     valid_namespaces = [n.value for n in ServiceNamespaceValueSet]
     if namespace not in valid_namespaces:
@@ -230,8 +235,12 @@ def _target_params_are_valid(namespace, r_id, dimension):
     if dimension is not None:
         try:
             valid_dimensions = [d.value for d in ScalableDimensionValueSet]
+            resource_type_exceptions = [r.value for r in ResourceTypeExceptionValueSet]
             d_namespace, d_resource_type, scaling_property = dimension.split(":")
-            resource_type = _get_resource_type_from_resource_id(r_id)
+            if d_resource_type not in resource_type_exceptions:
+                resource_type = _get_resource_type_from_resource_id(r_id)
+            else:
+                resource_type = d_resource_type
             if (
                 dimension not in valid_dimensions
                 or d_namespace != namespace

moto/applicationautoscaling/responses.py

@@ -33,7 +33,7 @@ class ApplicationAutoScalingResponse(BaseResponse):
         return json.dumps({"ScalableTargets": targets, "NextToken": next_token})
 
     def register_scalable_target(self):
-        """ Registers or updates a scalable target. """
+        """Registers or updates a scalable target."""
         self._validate_params()
         self.applicationautoscaling_backend.register_scalable_target(
             self._get_param("ServiceNamespace"),
@@ -47,7 +47,7 @@ class ApplicationAutoScalingResponse(BaseResponse):
         return json.dumps({})
 
     def deregister_scalable_target(self):
-        """ Deregisters a scalable target. """
+        """Deregisters a scalable target."""
         self._validate_params()
         self.applicationautoscaling_backend.deregister_scalable_target(
             self._get_param("ServiceNamespace"),

moto/core/models.py

@@ -547,7 +547,7 @@ class Model(type):
 
     @staticmethod
     def prop(model_name):
-        """ decorator to mark a class method as returning model values """
+        """decorator to mark a class method as returning model values"""
 
         def dec(f):
             f.__returns_model__ = model_name

moto/ec2/models.py

@@ -2105,7 +2105,7 @@ class SecurityGroup(TaggedEC2Resource, CloudFormationModel):
             security_group.delete(region_name)
 
     def delete(self, region_name):
-        """ Not exposed as part of the ELB API - used for CloudFormation. """
+        """Not exposed as part of the ELB API - used for CloudFormation."""
         self.ec2_backend.delete_security_group(group_id=self.id)
 
     @property

moto/ecs/models.py

@@ -1646,7 +1646,7 @@ class EC2ContainerServiceBackend(BaseBackend):
         return task_set_obj
 
     def update_service_primary_task_set(self, cluster, service, primary_task_set):
-        """ Updates task sets be PRIMARY or ACTIVE for given cluster:service task sets """
+        """Updates task sets be PRIMARY or ACTIVE for given cluster:service task sets"""
         cluster_name = cluster.split("/")[-1]
         service_name = service.split("/")[-1]
         task_set_obj = self.describe_task_sets(

moto/elb/models.py

@@ -265,7 +265,7 @@ class FakeLoadBalancer(CloudFormationModel):
             del self.tags[key]
 
     def delete(self, region):
-        """ Not exposed as part of the ELB API - used for CloudFormation. """
+        """Not exposed as part of the ELB API - used for CloudFormation."""
         elb_backends[region].delete_load_balancer(self.name)
 
 

moto/elbv2/models.py

@@ -480,7 +480,7 @@ class FakeLoadBalancer(CloudFormationModel):
             self.state = "active"
 
     def delete(self, region):
-        """ Not exposed as part of the ELB API - used for CloudFormation. """
+        """Not exposed as part of the ELB API - used for CloudFormation."""
         elbv2_backends[region].delete_load_balancer(self.arn)
 
     @staticmethod

moto/packages/httpretty/core.py

@@ -212,7 +212,7 @@ class HTTPrettyRequest(BaseHTTPRequestHandler, BaseClass):
         return result
 
     def parse_request_body(self, body):
-        """ Attempt to parse the post based on the content-type passed. Return the regular body if not """
+        """Attempt to parse the post based on the content-type passed. Return the regular body if not"""
 
         PARSING_FUNCTIONS = {
             "application/json": json.loads,

moto/route53/models.py

@@ -208,7 +208,7 @@ class RecordSet(CloudFormationModel):
         return template.render(record_set=self)
 
     def delete(self, *args, **kwargs):
-        """ Not exposed as part of the Route 53 API - used for CloudFormation. args are ignored """
+        """Not exposed as part of the Route 53 API - used for CloudFormation. args are ignored"""
         hosted_zone = route53_backend.get_hosted_zone_by_name(self.hosted_zone_name)
         if not hosted_zone:
             hosted_zone = route53_backend.get_hosted_zone(self.hosted_zone_id)

tests/test_applicationautoscaling/test_applicationautoscaling.py

@@ -244,6 +244,11 @@ def test_register_scalable_target_resource_id_variations():
             "keyspace/mykeyspace/table/mytable",
             "cassandra:table:ReadCapacityUnits",
         ),
+        (
+            "custom-resource",
+            "https://test-endpoint.amazon.com/ScalableDimension/test-resource",
+            "custom-resource:ResourceType:Property",
+        ),
     ]
 
     client = boto3.client("application-autoscaling", region_name=DEFAULT_REGION)