ASG: support filters in describe_autoscaling_groups (#7383)

This commit is contained in:
Satish Lakkoju 2024-02-22 18:08:55 -05:00 committed by GitHub
parent 25bc983df0
commit e2529697c7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 152 additions and 6 deletions

View File

@ -1190,13 +1190,40 @@ class AutoScalingBackend(BaseBackend):
return group
def describe_auto_scaling_groups(
self, names: List[str]
self, names: List[str], filters: Optional[List[Dict[str, str]]] = None
) -> List[FakeAutoScalingGroup]:
groups = self.autoscaling_groups.values()
groups = list(self.autoscaling_groups.values())
if filters:
for f in filters:
if f["Name"] == "tag-key":
groups = [
group
for group in groups
if any(tag["Key"] in f["Values"] for tag in group.tags)
]
elif f["Name"] == "tag-value":
groups = [
group
for group in groups
if any(tag["Value"] in f["Values"] for tag in group.tags)
]
elif f["Name"].startswith("tag:"):
tag_key = f["Name"][4:]
groups = [
group
for group in groups
if any(
tag["Key"] == tag_key and tag["Value"] in f["Values"]
for tag in group.tags
)
]
if names:
return [group for group in groups if group.name in names]
else:
return list(groups)
groups = [group for group in groups if group.name in names]
return groups
def delete_auto_scaling_group(self, group_name: str) -> None:
self.set_desired_capacity(group_name, 0)

View File

@ -228,7 +228,10 @@ class AutoScalingResponse(BaseResponse):
def describe_auto_scaling_groups(self) -> str:
names = self._get_multi_param("AutoScalingGroupNames.member")
token = self._get_param("NextToken")
all_groups = self.autoscaling_backend.describe_auto_scaling_groups(names)
filters = self._get_params().get("Filters", [])
all_groups = self.autoscaling_backend.describe_auto_scaling_groups(
names, filters=filters
)
all_names = [group.name for group in all_groups]
if token:
start = all_names.index(token) + 1

View File

@ -0,0 +1,116 @@
import boto3
from moto import mock_aws
from tests import EXAMPLE_AMI_ID
from .utils import setup_networking
@mock_aws
def test_describe_autoscaling_groups_filter_by_tag_key():
subnet = setup_networking()["subnet1"]
client = boto3.client("autoscaling", region_name="us-east-1")
create_asgs(client, subnet)
response = client.describe_auto_scaling_groups(
Filters=[{"Name": "tag-key", "Values": ["test_key1"]}]
)
group = response["AutoScalingGroups"][0]
tags = group["Tags"]
assert len(response["AutoScalingGroups"]) == 1
assert response["AutoScalingGroups"][0]["AutoScalingGroupName"] == "test_asg1"
assert {
"Key": "test_key1",
"PropagateAtLaunch": False,
"ResourceId": "test_asg1",
"ResourceType": "auto-scaling-group",
"Value": "test_value1",
} in tags
@mock_aws
def test_describe_autoscaling_groups_filter_by_tag_value():
subnet = setup_networking()["subnet1"]
client = boto3.client("autoscaling", region_name="us-east-1")
create_asgs(client, subnet)
response = client.describe_auto_scaling_groups(
Filters=[{"Name": "tag-value", "Values": ["test_value1"]}]
)
group = response["AutoScalingGroups"][0]
tags = group["Tags"]
assert len(response["AutoScalingGroups"]) == 1
assert response["AutoScalingGroups"][0]["AutoScalingGroupName"] == "test_asg1"
assert {
"Key": "test_key1",
"PropagateAtLaunch": False,
"ResourceId": "test_asg1",
"ResourceType": "auto-scaling-group",
"Value": "test_value1",
} in tags
@mock_aws
def test_describe_autoscaling_groups_filter_by_tag_key_value():
subnet = setup_networking()["subnet1"]
client = boto3.client("autoscaling", region_name="us-east-1")
create_asgs(client, subnet)
response = client.describe_auto_scaling_groups(
Filters=[{"Name": "tag:test_key1", "Values": ["test_value1"]}]
)
group = response["AutoScalingGroups"][0]
tags = group["Tags"]
assert len(response["AutoScalingGroups"]) == 1
assert response["AutoScalingGroups"][0]["AutoScalingGroupName"] == "test_asg1"
assert {
"Key": "test_key1",
"PropagateAtLaunch": False,
"ResourceId": "test_asg1",
"ResourceType": "auto-scaling-group",
"Value": "test_value1",
} in tags
@mock_aws
def test_describe_autoscaling_groups_no_filter():
subnet = setup_networking()["subnet1"]
client = boto3.client("autoscaling", region_name="us-east-1")
create_asgs(client, subnet)
response = client.describe_auto_scaling_groups()
assert len(response["AutoScalingGroups"]) == 2
assert response["AutoScalingGroups"][0]["AutoScalingGroupName"] == "test_asg1"
assert response["AutoScalingGroups"][1]["AutoScalingGroupName"] == "test_asg2"
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_asg1",
LaunchConfigurationName="test_launch_configuration",
MinSize=0,
MaxSize=20,
DesiredCapacity=5,
Tags=[
{"Key": "test_key1", "Value": "test_value1"},
{"Key": "test_key2", "Value": "test_value2"},
],
VPCZoneIdentifier=subnet,
)
client.create_auto_scaling_group(
AutoScalingGroupName="test_asg2",
LaunchConfigurationName="test_launch_configuration",
MinSize=0,
MaxSize=20,
DesiredCapacity=5,
VPCZoneIdentifier=subnet,
)