EC2: Add filter-options to describe_instance_types (#5382)

This commit is contained in:
Felipe Marinho 2022-08-23 11:56:57 -03:00 committed by GitHub
parent ccadde5d32
commit af5d3f9e30
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 134 additions and 4 deletions

View File

@ -1,9 +1,10 @@
import pathlib
from os import listdir
from ..utils import generic_filter
from moto.utilities.utils import load_resource
from ..exceptions import InvalidInstanceTypeError
from ..exceptions import FilterNotImplementedError, InvalidInstanceTypeError
INSTANCE_TYPES = load_resource(__name__, "../resources/instance_types.json")
INSTANCE_FAMILIES = list(set([i.split(".")[0] for i in INSTANCE_TYPES.keys()]))
@ -21,9 +22,42 @@ for location_type in listdir(root / offerings_path):
INSTANCE_TYPE_OFFERINGS[location_type][_region.replace(".json", "")] = res
class InstanceType(dict):
def __init__(self, name):
self.name = name
self.update(INSTANCE_TYPES[name])
def __getattr__(self, name):
return self[name]
def __setattr__(self, name, value):
self[name] = value
def __repr__(self):
return "<InstanceType: %s>" % self.name
def get_filter_value(self, filter_name):
if filter_name in ("instance-type"):
return self.get("InstanceType")
elif filter_name in ("vcpu-info.default-vcpus"):
return str(self.get("VCpuInfo").get("DefaultVCpus"))
elif filter_name in ("memory-info.size-in-mib"):
return str(self.get("MemoryInfo").get("SizeInMiB"))
elif filter_name in ("bare-metal"):
return str(self.get("BareMetal")).lower()
elif filter_name in ("burstable-performance-supported"):
return str(self.get("BurstablePerformanceSupported")).lower()
elif filter_name in ("current-generation"):
return str(self.get("CurrentGeneration")).lower()
else:
return FilterNotImplementedError(filter_name, "DescribeInstanceTypes")
class InstanceTypeBackend:
def describe_instance_types(self, instance_types=None):
matches = INSTANCE_TYPES.values()
instance_types = list(map(InstanceType, INSTANCE_TYPES.keys()))
def describe_instance_types(self, instance_types=None, filters=None):
matches = self.instance_types
if instance_types:
matches = [t for t in matches if t.get("InstanceType") in instance_types]
if len(instance_types) > len(matches):
@ -31,6 +65,8 @@ class InstanceTypeBackend:
t.get("InstanceType") for t in matches
)
raise InvalidInstanceTypeError(unknown_ids)
if filters:
matches = generic_filter(filters, matches)
return matches

View File

@ -158,7 +158,10 @@ class InstanceResponse(EC2BaseResponse):
def describe_instance_types(self):
instance_type_filters = self._get_multi_param("InstanceType")
instance_types = self.ec2_backend.describe_instance_types(instance_type_filters)
filter_dict = self._filters_from_querystring()
instance_types = self.ec2_backend.describe_instance_types(
instance_type_filters, filter_dict
)
template = self.response_template(EC2_DESCRIBE_INSTANCE_TYPES)
return template.render(instance_types=instance_types)

View File

@ -89,3 +89,94 @@ def test_describe_instance_types_unknown_type():
"The instance type '{'t1.non_existent'}' does not exist"
)
err.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
@mock_ec2
def test_describe_instance_types_filter_by_vcpus():
client = boto3.client("ec2", "us-east-1")
instance_types = client.describe_instance_types(
Filters=[{"Name": "vcpu-info.default-vcpus", "Values": ["1", "2"]}]
)
instance_types.should.have.key("InstanceTypes")
types = [
instance_type["InstanceType"]
for instance_type in instance_types["InstanceTypes"]
]
types.should.contain("t1.micro")
types.should.contain("t2.nano")
# not contain
types.should_not.contain("m5d.xlarge")
@mock_ec2
def test_describe_instance_types_filter_by_memory():
client = boto3.client("ec2", "us-east-1")
instance_types = client.describe_instance_types(
Filters=[{"Name": "memory-info.size-in-mib", "Values": ["512"]}]
)
instance_types.should.have.key("InstanceTypes")
types = [
instance_type["InstanceType"]
for instance_type in instance_types["InstanceTypes"]
]
types.should.contain("t4g.nano")
# not contain
types.should_not.contain("m5d.xlarge")
@mock_ec2
def test_describe_instance_types_filter_by_bare_metal():
client = boto3.client("ec2", "us-east-1")
instance_types = client.describe_instance_types(
Filters=[{"Name": "bare-metal", "Values": ["true"]}]
)
instance_types.should.have.key("InstanceTypes")
types = [
instance_type["InstanceType"]
for instance_type in instance_types["InstanceTypes"]
]
types.should.contain("a1.metal")
# not contain
types.should_not.contain("t1.micro")
@mock_ec2
def test_describe_instance_types_filter_by_burstable_performance_supported():
client = boto3.client("ec2", "us-east-1")
instance_types = client.describe_instance_types(
Filters=[{"Name": "burstable-performance-supported", "Values": ["true"]}]
)
instance_types.should.have.key("InstanceTypes")
types = [
instance_type["InstanceType"]
for instance_type in instance_types["InstanceTypes"]
]
types.should.contain("t2.micro")
# not contain
types.should_not.contain("t1.micro")
@mock_ec2
def test_describe_instance_types_filter_by_current_generation():
client = boto3.client("ec2", "us-east-1")
instance_types = client.describe_instance_types(
Filters=[{"Name": "current-generation", "Values": ["true"]}]
)
instance_types.should.have.key("InstanceTypes")
types = [
instance_type["InstanceType"]
for instance_type in instance_types["InstanceTypes"]
]
types.should.contain("t2.micro")
# not contain
types.should_not.contain("t1.micro")