Autoscaling - Describe Tags (#4634)

moto/autoscaling/models.py

@@ -1,4 +1,6 @@
+import itertools
 import random
+from uuid import uuid4
 
 from moto.packages.boto.ec2.blockdevicemapping import (
     BlockDeviceType,
@@ -282,6 +284,8 @@ class FakeAutoScalingGroup(CloudFormationModel):
         self.autoscaling_backend = autoscaling_backend
         self.ec2_backend = ec2_backend
         self.name = name
+        self._id = str(uuid4())
+        self.region = self.autoscaling_backend.region
 
         self._set_azs_and_vpcs(availability_zones, vpc_zone_identifier)
 
@@ -308,9 +312,26 @@ class FakeAutoScalingGroup(CloudFormationModel):
 
         self.suspended_processes = []
         self.instance_states = []
-        self.tags = tags if tags else []
+        self.tags = tags or []
         self.set_desired_capacity(desired_capacity)
 
+    @property
+    def tags(self):
+        return self._tags
+
+    @tags.setter
+    def tags(self, tags):
+        for tag in tags:
+            if "resource_id" not in tag or not tag["resource_id"]:
+                tag["resource_id"] = self.name
+            if "resource_type" not in tag or not tag["resource_type"]:
+                tag["resource_type"] = "auto-scaling-group"
+        self._tags = tags
+
+    @property
+    def arn(self):
+        return f"arn:aws:autoscaling:{self.region}:{ACCOUNT_ID}:autoScalingGroup:{self._id}:autoScalingGroupName/{self.name}"
+
     def active_instances(self):
         return [x for x in self.instance_states if x.lifecycle_state == "InService"]
 
@@ -1204,6 +1225,25 @@ class AutoScalingBackend(BaseBackend):
         self.ec2_backend.terminate_instances([instance.id])
         return instance_state, original_size, group.desired_capacity
 
+    def describe_tags(self, filters):
+        """
+        Pagination is not yet implemented.
+        Only the `auto-scaling-group` and `propagate-at-launch` filters are implemented.
+        """
+        resources = self.autoscaling_groups.values()
+        tags = list(itertools.chain(*[r.tags for r in resources]))
+        for f in filters:
+            if f["Name"] == "auto-scaling-group":
+                tags = [t for t in tags if t["resource_id"] in f["Values"]]
+            if f["Name"] == "propagate-at-launch":
+                values = [v.lower() for v in f["Values"]]
+                tags = [
+                    t
+                    for t in tags
+                    if t.get("propagate_at_launch", "").lower() in values
+                ]
+        return tags
+
 
 autoscaling_backends = {}
 for region, ec2_backend in ec2_backends.items():

moto/autoscaling/responses.py

@@ -429,6 +429,12 @@ class AutoScalingResponse(BaseResponse):
             timestamp=iso_8601_datetime_with_milliseconds(datetime.datetime.utcnow()),
         )
 
+    def describe_tags(self):
+        filters = self._get_params().get("Filters", [])
+        tags = self.autoscaling_backend.describe_tags(filters=filters)
+        template = self.response_template(DESCRIBE_TAGS_TEMPLATE)
+        return template.render(tags=tags, next_token=None)
+
 
 CREATE_LAUNCH_CONFIGURATION_TEMPLATE = """<CreateLaunchConfigurationResponse xmlns="http://autoscaling.amazonaws.com/doc/2011-01-01/">
 <ResponseMetadata>
@@ -682,7 +688,7 @@ DESCRIBE_AUTOSCALING_GROUPS_TEMPLATE = """<DescribeAutoScalingGroupsResponse xml
         {% endif %}
         <HealthCheckGracePeriod>{{ group.health_check_period }}</HealthCheckGracePeriod>
         <DefaultCooldown>{{ group.default_cooldown }}</DefaultCooldown>
-        <AutoScalingGroupARN>arn:aws:autoscaling:us-east-1:803981987763:autoScalingGroup:ca861182-c8f9-4ca7-b1eb-cd35505f5ebb:autoScalingGroupName/{{ group.name }}</AutoScalingGroupARN>
+        <AutoScalingGroupARN>{{ group.arn }}</AutoScalingGroupARN>
         {% if group.termination_policies %}
         <TerminationPolicies>
           {% for policy in group.termination_policies %}
@@ -990,3 +996,25 @@ TERMINATE_INSTANCES_TEMPLATE = """<TerminateInstanceInAutoScalingGroupResponse x
     <RequestId>a1ba8fb9-31d6-4d9a-ace1-a7f76749df11EXAMPLE</RequestId>
   </ResponseMetadata>
 </TerminateInstanceInAutoScalingGroupResponse>"""
+
+DESCRIBE_TAGS_TEMPLATE = """<DescribeTagsResponse xmlns="http://autoscaling.amazonaws.com/doc/2011-01-01/">
+  <ResponseMetadata>
+    <RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId>
+  </ResponseMetadata>
+  <DescribeTagsResult>
+    <Tags>
+{% for tag in tags %}
+      <member>
+        <ResourceId>{{ tag.resource_id }}</ResourceId>
+        <ResourceType>{{ tag.resource_type }}</ResourceType>
+        <Key>{{ tag.key }}</Key>
+        <Value>{{ tag.value }}</Value>
+        <PropagateAtLaunch>{{ tag.propagate_at_launch }}</PropagateAtLaunch>
+      </member>
+{% endfor %}
+    </Tags>
+    {% if next_token %}
+    <NextToken>{{ next_token }}</NextToken>
+    {% endif %}
+  </DescribeTagsResult>
+</DescribeTagsResponse>"""

scripts/scaffold.py

@@ -396,9 +396,9 @@ def get_response_query_template(service, operation):  # pylint: disable=too-many
 
         start_tag = f"<{name}>"
         iter_name = f"{prefix[-1]}.{name.lower()}" if prefix else name.lower()
-        loop_start = f"{{%% for {singular_name.lower()} in {iter_name} %%}}"
+        loop_start = f"{{% for {singular_name.lower()} in {iter_name} %}}"
         end_tag = f"</{name}>"
-        loop_end = "{{ endfor }}"
+        loop_end = "{% endfor %}"
 
         start_tag_indexes = [i for i, l in enumerate(xml_body_lines) if start_tag in l]
         if len(start_tag_indexes) != 1:

tests/test_autoscaling/test_autoscaling.py

@@ -1166,8 +1166,8 @@ def test_create_autoscaling_group_boto3_multiple_launch_configurations():
 
 @mock_autoscaling
 def test_describe_autoscaling_groups_boto3_launch_config():
-    mocked_networking = setup_networking()
+    mocked_networking = setup_networking(region_name="eu-north-1")
-    client = boto3.client("autoscaling", region_name="us-east-1")
+    client = boto3.client("autoscaling", region_name="eu-north-1")
     client.create_launch_configuration(
         LaunchConfigurationName="test_launch_configuration",
         InstanceType="t2.micro",
@@ -1186,16 +1186,19 @@ def test_describe_autoscaling_groups_boto3_launch_config():
     response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
     response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
     group = response["AutoScalingGroups"][0]
+    group["AutoScalingGroupARN"].should.match(
+        f"arn:aws:autoscaling:eu-north-1:{ACCOUNT_ID}:autoScalingGroup:"
+    )
     group["AutoScalingGroupName"].should.equal("test_asg")
     group["LaunchConfigurationName"].should.equal("test_launch_configuration")
     group.should_not.have.key("LaunchTemplate")
-    group["AvailabilityZones"].should.equal(["us-east-1a"])
+    group["AvailabilityZones"].should.equal(["eu-north-1a"])
     group["VPCZoneIdentifier"].should.equal(mocked_networking["subnet1"])
     group["NewInstancesProtectedFromScaleIn"].should.equal(True)
     for instance in group["Instances"]:
         instance["LaunchConfigurationName"].should.equal("test_launch_configuration")
         instance.should_not.have.key("LaunchTemplate")
-        instance["AvailabilityZone"].should.equal("us-east-1a")
+        instance["AvailabilityZone"].should.equal("eu-north-1a")
         instance["ProtectedFromScaleIn"].should.equal(True)
         instance["InstanceType"].should.equal("t2.micro")
 
@@ -1491,53 +1494,6 @@ def test_update_autoscaling_group_max_size_desired_capacity_change():
     group["Instances"].should.have.length_of(5)
 
 
-@mock_autoscaling
-def test_autoscaling_tags_update_boto3():
-    mocked_networking = setup_networking()
-    client = boto3.client("autoscaling", region_name="us-east-1")
-    _ = client.create_launch_configuration(
-        LaunchConfigurationName="test_launch_configuration",
-        ImageId=EXAMPLE_AMI_ID,
-        InstanceType="t2.medium",
-    )
-    _ = client.create_auto_scaling_group(
-        AutoScalingGroupName="test_asg",
-        LaunchConfigurationName="test_launch_configuration",
-        MinSize=0,
-        MaxSize=20,
-        DesiredCapacity=5,
-        Tags=[
-            {
-                "ResourceId": "test_asg",
-                "Key": "test_key",
-                "Value": "test_value",
-                "PropagateAtLaunch": True,
-            }
-        ],
-        VPCZoneIdentifier=mocked_networking["subnet1"],
-    )
-
-    client.create_or_update_tags(
-        Tags=[
-            {
-                "ResourceId": "test_asg",
-                "Key": "test_key",
-                "Value": "updated_test_value",
-                "PropagateAtLaunch": True,
-            },
-            {
-                "ResourceId": "test_asg",
-                "Key": "test_key2",
-                "Value": "test_value2",
-                "PropagateAtLaunch": False,
-            },
-        ]
-    )
-
-    response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
-    response["AutoScalingGroups"][0]["Tags"].should.have.length_of(2)
-
-
 @mock_autoscaling
 def test_autoscaling_describe_policies_boto3():
     mocked_networking = setup_networking()
@@ -3074,58 +3030,6 @@ def test_terminate_instance_in_auto_scaling_group_no_decrement():
     )
 
 
-@mock_autoscaling
-@mock_ec2
-def test_delete_tags_by_key():
-    mocked_networking = setup_networking()
-    client = boto3.client("autoscaling", region_name="us-east-1")
-    client.create_launch_configuration(
-        LaunchConfigurationName="TestLC",
-        ImageId=EXAMPLE_AMI_ID,
-        InstanceType="t2.medium",
-    )
-    tag_to_delete = {
-        "ResourceId": "tag_test_asg",
-        "ResourceType": "auto-scaling-group",
-        "PropagateAtLaunch": True,
-        "Key": "TestDeleteTagKey1",
-        "Value": "TestTagValue1",
-    }
-    tag_to_keep = {
-        "ResourceId": "tag_test_asg",
-        "ResourceType": "auto-scaling-group",
-        "PropagateAtLaunch": True,
-        "Key": "TestTagKey1",
-        "Value": "TestTagValue1",
-    }
-    client.create_auto_scaling_group(
-        AutoScalingGroupName="tag_test_asg",
-        MinSize=1,
-        MaxSize=2,
-        LaunchConfigurationName="TestLC",
-        Tags=[tag_to_delete, tag_to_keep],
-        VPCZoneIdentifier=mocked_networking["subnet1"],
-    )
-
-    client.delete_tags(
-        Tags=[
-            {
-                "ResourceId": "tag_test_asg",
-                "ResourceType": "auto-scaling-group",
-                "PropagateAtLaunch": True,
-                "Key": "TestDeleteTagKey1",
-            }
-        ]
-    )
-    response = client.describe_auto_scaling_groups(
-        AutoScalingGroupNames=["tag_test_asg"]
-    )
-    group = response["AutoScalingGroups"][0]
-    tags = group["Tags"]
-    tags.should.contain(tag_to_keep)
-    tags.should_not.contain(tag_to_delete)
-
-
 @mock_ec2
 @mock_autoscaling
 def test_attach_instances():

tests/test_autoscaling/test_autoscaling_tags.py

@@ -0,0 +1,295 @@
+import boto3
+
+from moto import mock_autoscaling, mock_ec2
+
+from .utils import setup_networking
+from tests import EXAMPLE_AMI_ID
+
+
+@mock_autoscaling
+def test_autoscaling_tags_update():
+    mocked_networking = setup_networking()
+    client = boto3.client("autoscaling", region_name="us-east-1")
+    _ = client.create_launch_configuration(
+        LaunchConfigurationName="test_launch_configuration",
+        ImageId=EXAMPLE_AMI_ID,
+        InstanceType="t2.medium",
+    )
+    _ = client.create_auto_scaling_group(
+        AutoScalingGroupName="test_asg",
+        LaunchConfigurationName="test_launch_configuration",
+        MinSize=0,
+        MaxSize=20,
+        DesiredCapacity=5,
+        Tags=[
+            {
+                "ResourceId": "test_asg",
+                "Key": "test_key",
+                "Value": "test_value",
+                "PropagateAtLaunch": True,
+            }
+        ],
+        VPCZoneIdentifier=mocked_networking["subnet1"],
+    )
+
+    client.create_or_update_tags(
+        Tags=[
+            {
+                "ResourceId": "test_asg",
+                "Key": "test_key",
+                "Value": "updated_test_value",
+                "PropagateAtLaunch": True,
+            },
+            {
+                "ResourceId": "test_asg",
+                "Key": "test_key2",
+                "Value": "test_value2",
+                "PropagateAtLaunch": False,
+            },
+        ]
+    )
+
+    response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
+    response["AutoScalingGroups"][0]["Tags"].should.have.length_of(2)
+
+
+@mock_autoscaling
+@mock_ec2
+def test_delete_tags_by_key():
+    mocked_networking = setup_networking()
+    client = boto3.client("autoscaling", region_name="us-east-1")
+    client.create_launch_configuration(
+        LaunchConfigurationName="TestLC",
+        ImageId=EXAMPLE_AMI_ID,
+        InstanceType="t2.medium",
+    )
+    tag_to_delete = {
+        "ResourceId": "tag_test_asg",
+        "ResourceType": "auto-scaling-group",
+        "PropagateAtLaunch": True,
+        "Key": "TestDeleteTagKey1",
+        "Value": "TestTagValue1",
+    }
+    tag_to_keep = {
+        "ResourceId": "tag_test_asg",
+        "ResourceType": "auto-scaling-group",
+        "PropagateAtLaunch": True,
+        "Key": "TestTagKey1",
+        "Value": "TestTagValue1",
+    }
+    client.create_auto_scaling_group(
+        AutoScalingGroupName="tag_test_asg",
+        MinSize=1,
+        MaxSize=2,
+        LaunchConfigurationName="TestLC",
+        Tags=[tag_to_delete, tag_to_keep],
+        VPCZoneIdentifier=mocked_networking["subnet1"],
+    )
+
+    client.delete_tags(
+        Tags=[
+            {
+                "ResourceId": "tag_test_asg",
+                "ResourceType": "auto-scaling-group",
+                "PropagateAtLaunch": True,
+                "Key": "TestDeleteTagKey1",
+            }
+        ]
+    )
+    response = client.describe_auto_scaling_groups(
+        AutoScalingGroupNames=["tag_test_asg"]
+    )
+    group = response["AutoScalingGroups"][0]
+    tags = group["Tags"]
+    tags.should.contain(tag_to_keep)
+    tags.should_not.contain(tag_to_delete)
+
+
+@mock_autoscaling
+def test_describe_tags_without_resources():
+    client = boto3.client("autoscaling", region_name="us-east-2")
+    resp = client.describe_tags()
+    resp.should.have.key("Tags").equals([])
+    resp.shouldnt.have.key("NextToken")
+
+
+@mock_autoscaling
+def test_describe_tags_no_filter():
+    subnet = setup_networking()["subnet1"]
+    client = boto3.client("autoscaling", region_name="us-east-1")
+    create_asgs(client, subnet)
+
+    response = client.describe_tags()
+    response.should.have.key("Tags").length_of(4)
+    response["Tags"].should.contain(
+        {
+            "ResourceId": "test_asg",
+            "ResourceType": "auto-scaling-group",
+            "Key": "test_key",
+            "Value": "updated_test_value",
+            "PropagateAtLaunch": True,
+        }
+    )
+    response["Tags"].should.contain(
+        {
+            "ResourceId": "test_asg",
+            "ResourceType": "auto-scaling-group",
+            "Key": "test_key2",
+            "Value": "test_value2",
+            "PropagateAtLaunch": False,
+        }
+    )
+    response["Tags"].should.contain(
+        {
+            "ResourceId": "test_asg2",
+            "ResourceType": "auto-scaling-group",
+            "Key": "asg2tag1",
+            "Value": "val",
+            "PropagateAtLaunch": False,
+        }
+    )
+    response["Tags"].should.contain(
+        {
+            "ResourceId": "test_asg2",
+            "ResourceType": "auto-scaling-group",
+            "Key": "asg2tag2",
+            "Value": "diff",
+            "PropagateAtLaunch": False,
+        }
+    )
+
+
+@mock_autoscaling
+def test_describe_tags_filter_by_name():
+    subnet = setup_networking()["subnet1"]
+    client = boto3.client("autoscaling", region_name="us-east-1")
+    create_asgs(client, subnet)
+
+    response = client.describe_tags(
+        Filters=[{"Name": "auto-scaling-group", "Values": ["test_asg"]}]
+    )
+    response.should.have.key("Tags").length_of(2)
+    response["Tags"].should.contain(
+        {
+            "ResourceId": "test_asg",
+            "ResourceType": "auto-scaling-group",
+            "Key": "test_key",
+            "Value": "updated_test_value",
+            "PropagateAtLaunch": True,
+        }
+    )
+    response["Tags"].should.contain(
+        {
+            "ResourceId": "test_asg",
+            "ResourceType": "auto-scaling-group",
+            "Key": "test_key2",
+            "Value": "test_value2",
+            "PropagateAtLaunch": False,
+        }
+    )
+
+    response = client.describe_tags(
+        Filters=[{"Name": "auto-scaling-group", "Values": ["test_asg", "test_asg2"]}]
+    )
+    response.should.have.key("Tags").length_of(4)
+    response["Tags"].should.contain(
+        {
+            "ResourceId": "test_asg",
+            "ResourceType": "auto-scaling-group",
+            "Key": "test_key",
+            "Value": "updated_test_value",
+            "PropagateAtLaunch": True,
+        }
+    )
+    response["Tags"].should.contain(
+        {
+            "ResourceId": "test_asg",
+            "ResourceType": "auto-scaling-group",
+            "Key": "test_key2",
+            "Value": "test_value2",
+            "PropagateAtLaunch": False,
+        }
+    )
+    response["Tags"].should.contain(
+        {
+            "ResourceId": "test_asg2",
+            "ResourceType": "auto-scaling-group",
+            "Key": "asg2tag1",
+            "Value": "val",
+            "PropagateAtLaunch": False,
+        }
+    )
+    response["Tags"].should.contain(
+        {
+            "ResourceId": "test_asg2",
+            "ResourceType": "auto-scaling-group",
+            "Key": "asg2tag2",
+            "Value": "diff",
+            "PropagateAtLaunch": False,
+        }
+    )
+
+
+@mock_autoscaling
+def test_describe_tags_filter_by_propgateatlaunch():
+    subnet = setup_networking()["subnet1"]
+    client = boto3.client("autoscaling", region_name="us-east-1")
+    create_asgs(client, subnet)
+
+    response = client.describe_tags(
+        Filters=[{"Name": "propagate-at-launch", "Values": ["True"]}]
+    )
+    response.should.have.key("Tags").length_of(1)
+    response["Tags"].should.contain(
+        {
+            "ResourceId": "test_asg",
+            "ResourceType": "auto-scaling-group",
+            "Key": "test_key",
+            "Value": "updated_test_value",
+            "PropagateAtLaunch": True,
+        }
+    )
+
+
+def create_asgs(client, subnet):
+    _ = client.create_launch_configuration(
+        LaunchConfigurationName="test_launch_configuration",
+        ImageId=EXAMPLE_AMI_ID,
+        InstanceType="t2.medium",
+    )
+    client.create_auto_scaling_group(
+        AutoScalingGroupName="test_asg",
+        LaunchConfigurationName="test_launch_configuration",
+        MinSize=0,
+        MaxSize=20,
+        DesiredCapacity=5,
+        VPCZoneIdentifier=subnet,
+    )
+    client.create_auto_scaling_group(
+        AutoScalingGroupName="test_asg2",
+        LaunchConfigurationName="test_launch_configuration",
+        MinSize=0,
+        MaxSize=20,
+        DesiredCapacity=5,
+        Tags=[
+            {"Key": "asg2tag1", "Value": "val"},
+            {"Key": "asg2tag2", "Value": "diff"},
+        ],
+        VPCZoneIdentifier=subnet,
+    )
+    client.create_or_update_tags(
+        Tags=[
+            {
+                "ResourceId": "test_asg",
+                "Key": "test_key",
+                "Value": "updated_test_value",
+                "PropagateAtLaunch": True,
+            },
+            {
+                "ResourceId": "test_asg",
+                "Key": "test_key2",
+                "Value": "test_value2",
+                "PropagateAtLaunch": False,
+            },
+        ]
+    )