Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Improvements - Autoscaling (#4985)

Browse Source
This commit is contained in:
Bert Blommers 2022-03-29 21:46:06 +00:00 committed by GitHub
parent c134afaf60
commit cf2690ca1e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 568 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
IMPLEMENTATION_COVERAGE.md
View File

@ -338,7 +338,7 @@
## autoscaling
<details>
<summary>47% implemented</summary>
<summary>49% implemented</summary>
- [X] attach_instances
- [X] attach_load_balancer_target_groups
@ -389,7 +389,7 @@
- [ ] get_predictive_scaling_forecast
- [ ] put_lifecycle_hook
- [ ] put_notification_configuration
- [ ] put_scaling_policy
- [X] put_scaling_policy
- [ ] put_scheduled_update_group_action
- [ ] put_warm_pool
- [ ] record_lifecycle_action_heartbeat
@ -1547,7 +1547,7 @@
- [ ] delete_ipam_pool
- [ ] delete_ipam_scope
- [X] delete_key_pair
- [ ] delete_launch_template
- [X] delete_launch_template
- [ ] delete_launch_template_versions
- [ ] delete_local_gateway_route
- [ ] delete_local_gateway_route_table_vpc_association

2
docs/docs/services/autoscaling.rst
View File

@ -79,7 +79,7 @@ autoscaling
- [ ] get_predictive_scaling_forecast
- [ ] put_lifecycle_hook
- [ ] put_notification_configuration
- [ ] put_scaling_policy
- [X] put_scaling_policy
- [ ] put_scheduled_update_group_action
- [ ] put_warm_pool
- [ ] record_lifecycle_action_heartbeat

2
docs/docs/services/ec2.rst
View File

@ -163,7 +163,7 @@ ec2
- [ ] delete_ipam_pool
- [ ] delete_ipam_scope
- [X] delete_key_pair
- [ ] delete_launch_template
- [X] delete_launch_template
- [ ] delete_launch_template_versions
- [ ] delete_local_gateway_route
- [ ] delete_local_gateway_route_table_vpc_association

34
moto/autoscaling/models.py
View File

@ -66,18 +66,24 @@ class FakeScalingPolicy(BaseModel):
self,
name,
policy_type,
metric_aggregation_type,
adjustment_type,
as_name,
min_adjustment_magnitude,
scaling_adjustment,
cooldown,
target_tracking_config,
step_adjustments,
estimated_instance_warmup,
predictive_scaling_configuration,
autoscaling_backend,
):
self.name = name
self.policy_type = policy_type
self.metric_aggregation_type = metric_aggregation_type
self.adjustment_type = adjustment_type
self.as_name = as_name
self.min_adjustment_magnitude = min_adjustment_magnitude
self.scaling_adjustment = scaling_adjustment
if cooldown is not None:
self.cooldown = cooldown
@ -85,6 +91,8 @@ class FakeScalingPolicy(BaseModel):
self.cooldown = DEFAULT_COOLDOWN
self.target_tracking_config = target_tracking_config
self.step_adjustments = step_adjustments
self.estimated_instance_warmup = estimated_instance_warmup
self.predictive_scaling_configuration = predictive_scaling_configuration
self.autoscaling_backend = autoscaling_backend
@property
@ -390,7 +398,7 @@ class FakeAutoScalingGroup(CloudFormationModel):
self.launch_template = self.ec2_backend.get_launch_template_by_name(
launch_template_name
)
self.launch_template_version = int(launch_template["version"])
self.launch_template_version = launch_template["version"]
@staticmethod
def __set_string_propagate_at_launch_booleans_on_tags(tags):
@ -963,27 +971,35 @@ class AutoScalingBackend(BaseBackend):
def delete_lifecycle_hook(self, as_name, name):
self.lifecycle_hooks.pop("%s_%s" % (as_name, name), None)
def create_autoscaling_policy(
def put_scaling_policy(
self,
name,
policy_type,
metric_aggregation_type,
adjustment_type,
as_name,
min_adjustment_magnitude,
scaling_adjustment,
cooldown,
target_tracking_config,
step_adjustments,
estimated_instance_warmup,
predictive_scaling_configuration,
):
policy = FakeScalingPolicy(
name,
policy_type,
adjustment_type,
as_name,
scaling_adjustment,
cooldown,
target_tracking_config,
step_adjustments,
self,
metric_aggregation_type,
adjustment_type=adjustment_type,
as_name=as_name,
min_adjustment_magnitude=min_adjustment_magnitude,
scaling_adjustment=scaling_adjustment,
cooldown=cooldown,
target_tracking_config=target_tracking_config,
step_adjustments=step_adjustments,
estimated_instance_warmup=estimated_instance_warmup,
predictive_scaling_configuration=predictive_scaling_configuration,
autoscaling_backend=self,
)
self.policies[name] = policy

174
moto/autoscaling/responses.py
View File

@ -268,15 +268,21 @@ class AutoScalingResponse(BaseResponse):
def put_scaling_policy(self):
params = self._get_params()
policy = self.autoscaling_backend.create_autoscaling_policy(
policy = self.autoscaling_backend.put_scaling_policy(
name=params.get("PolicyName"),
policy_type=params.get("PolicyType", "SimpleScaling"),
metric_aggregation_type=params.get("MetricAggregationType"),
adjustment_type=params.get("AdjustmentType"),
as_name=params.get("AutoScalingGroupName"),
min_adjustment_magnitude=params.get("MinAdjustmentMagnitude"),
scaling_adjustment=self._get_int_param("ScalingAdjustment"),
cooldown=self._get_int_param("Cooldown"),
target_tracking_config=params.get("TargetTrackingConfiguration", {}),
step_adjustments=params.get("StepAdjustments", []),
estimated_instance_warmup=params.get("EstimatedInstanceWarmup"),
predictive_scaling_configuration=params.get(
"PredictiveScalingConfiguration", {}
),
)
template = self.response_template(CREATE_SCALING_POLICY_TEMPLATE)
return template.render(policy=policy)
@ -441,7 +447,7 @@ DESCRIBE_LAUNCH_CONFIGURATIONS_TEMPLATE = """<DescribeLaunchConfigurationsRespon
<LaunchConfigurations>
{% for launch_configuration in launch_configurations %}
<member>
<AssociatePublicIpAddress>{{ launch_configuration.associate_public_ip_address }}</AssociatePublicIpAddress>
<AssociatePublicIpAddress>{{ 'true' if launch_configuration.associate_public_ip_address else 'false' }}</AssociatePublicIpAddress>
<SecurityGroups>
{% for security_group in launch_configuration.security_groups %}
<member>{{ security_group }}</member>
@ -805,38 +811,198 @@ DESCRIBE_SCALING_POLICIES_TEMPLATE = """<DescribePoliciesResponse xmlns="http://
{% for policy in policies %}
<member>
<PolicyARN>{{ policy.arn }}</PolicyARN>
{% if policy.adjustment_type %}
<AdjustmentType>{{ policy.adjustment_type }}</AdjustmentType>
{% endif %}
{% if policy.scaling_adjustment %}
<ScalingAdjustment>{{ policy.scaling_adjustment }}</ScalingAdjustment>
{% endif %}
{% if policy.min_adjustment_magnitude %}
<MinAdjustmentMagnitude>{{ policy.min_adjustment_magnitude }}</MinAdjustmentMagnitude>
{% endif %}
<PolicyName>{{ policy.name }}</PolicyName>
<PolicyType>{{ policy.policy_type }}</PolicyType>
<MetricAggregationType>{{ policy.metric_aggregation_type }}</MetricAggregationType>
<AutoScalingGroupName>{{ policy.as_name }}</AutoScalingGroupName>
{% if policy.policy_type == 'SimpleScaling' %}
<Cooldown>{{ policy.cooldown }}</Cooldown>
{% endif %}
{% if policy.policy_type == 'TargetTrackingScaling' %}
<TargetTrackingConfiguration>
{% if policy.target_tracking_config.get("PredefinedMetricSpecification") %}
<PredefinedMetricSpecification>
<PredefinedMetricType>{{ policy.target_tracking_config.get("PredefinedMetricSpecification", {}).get("PredefinedMetricType", "") }}</PredefinedMetricType>
{% if policy.target_tracking_config.get("PredefinedMetricSpecification", {}).get("ResourceLabel") %}
<ResourceLabel>{{ policy.target_tracking_config.get("PredefinedMetricSpecification", {}).get("ResourceLabel") }}</ResourceLabel>
{% endif %}
</PredefinedMetricSpecification>
{% endif %}
{% if policy.target_tracking_config.get("CustomizedMetricSpecification") %}
<CustomizedMetricSpecification>
<MetricName>{{ policy.target_tracking_config["CustomizedMetricSpecification"].get("MetricName") }}</MetricName>
<Namespace>{{ policy.target_tracking_config["CustomizedMetricSpecification"].get("Namespace") }}</Namespace>
<Dimensions>
{% for dim in policy.target_tracking_config["CustomizedMetricSpecification"].get("Dimensions", []) %}
<member>
<Name>{{ dim.get("Name") }}</Name>
<Value>{{ dim.get("Value") }}</Value>
</member>
{% endfor %}
</Dimensions>
<Statistic>{{ policy.target_tracking_config["CustomizedMetricSpecification"].get("Statistic") }}</Statistic>
{% if policy.target_tracking_config["CustomizedMetricSpecification"].get("Unit") %}
<Unit>{{ policy.target_tracking_config["CustomizedMetricSpecification"].get("Unit") }}</Unit>
{% endif %}
</CustomizedMetricSpecification>
{% endif %}
<TargetValue>{{ policy.target_tracking_config.get("TargetValue") }}</TargetValue>
</TargetTrackingConfiguration>
{% endif %}
{% if policy.policy_type == 'StepScaling' %}
<StepAdjustments>
{% for step in policy.step_adjustments %}
<entry>
<member>
{% if "MetricIntervalLowerBound" in step %}
<MetricIntervalLowerBound>{{ step.get("MetricIntervalLowerBound") }}</MetricIntervalLowerBound>
{% endif %}
{% if "MetricIntervalUpperBound" in step %}
<MetricIntervalUpperBound>{{ step.get("MetricIntervalUpperBound") }}</MetricIntervalUpperBound>
{% endif %}
{% if "ScalingAdjustment" in step %}
<ScalingAdjustment>{{ step.get("ScalingAdjustment") }}</ScalingAdjustment>
</entry>
{% endif %}
</member>
{% endfor %}
</StepAdjustments>
{% endif %}
{% if policy.estimated_instance_warmup %}
<EstimatedInstanceWarmup>{{ policy.estimated_instance_warmup }}</EstimatedInstanceWarmup>
{% endif %}
{% if policy.policy_type == 'PredictiveScaling' %}
<PredictiveScalingConfiguration>
<MetricSpecifications>
{% for config in policy.predictive_scaling_configuration.get("MetricSpecifications", []) %}
<member>
<TargetValue>{{ config.get("TargetValue") }}</TargetValue>
{% if config.get("PredefinedMetricPairSpecification", {}).get("PredefinedMetricType") %}
<PredefinedMetricPairSpecification>
<PredefinedMetricType>{{ config.get("PredefinedMetricPairSpecification", {}).get("PredefinedMetricType") }}</PredefinedMetricType>
<ResourceLabel>{{ config.get("PredefinedMetricPairSpecification", {}).get("ResourceLabel", "") }}</ResourceLabel>
</PredefinedMetricPairSpecification>
{% endif %}
{% if config.get("PredefinedScalingMetricSpecification", {}).get("PredefinedMetricType") %}
<PredefinedScalingMetricSpecification>
<PredefinedMetricType>{{ config.get("PredefinedScalingMetricSpecification", {}).get("PredefinedMetricType", "") }}</PredefinedMetricType>
<ResourceLabel>{{ config.get("PredefinedScalingMetricSpecification", {}).get("ResourceLabel", "") }}</ResourceLabel>
</PredefinedScalingMetricSpecification>
{% endif %}
{% if config.get("PredefinedLoadMetricSpecification", {}).get("PredefinedMetricType") %}
<PredefinedLoadMetricSpecification>
<PredefinedMetricType>{{ config.get("PredefinedLoadMetricSpecification", {}).get("PredefinedMetricType", "") }}</PredefinedMetricType>
<ResourceLabel>{{ config.get("PredefinedLoadMetricSpecification", {}).get("ResourceLabel", "") }}</ResourceLabel>
</PredefinedLoadMetricSpecification>
{% endif %}
{% if config.get("CustomizedScalingMetricSpecification", {}).get("MetricDataQueries") %}
<CustomizedScalingMetricSpecification>
<MetricDataQueries>
{% for query in config.get("CustomizedScalingMetricSpecification", {}).get("MetricDataQueries", []) %}
<member>
<Id>{{ query.get("Id") }}</Id>
<Expression>{{ query.get("Expression") }}</Expression>
<MetricStat>
<Metric>
<Namespace>{{ query.get("MetricStat", {}).get("Metric", {}).get("Namespace") }}</Namespace>
<MetricName>{{ query.get("MetricStat", {}).get("Metric", {}).get("MetricName") }}</MetricName>
<Dimensions>
{% for dim in query.get("MetricStat", {}).get("Metric", {}).get("Dimensions", []) %}
<Name>{{ dim.get("Name") }}</Name>
<Value>{{ dim.get("Value") }}</Value>
{% endfor %}
</Dimensions>
</Metric>
<Stat>{{ query.get("MetricStat", {}).get("Stat") }}</Stat>
<Unit>{{ query.get("MetricStat", {}).get("Unit") }}</Unit>
</MetricStat>
<Label>{{ query.get("Label") }}</Label>
<ReturnData>{{ 'true' if query.get("ReturnData") else 'false' }}</ReturnData>
</member>
{% endfor %}
</MetricDataQueries>
</CustomizedScalingMetricSpecification>
{% endif %}
{% if config.get("CustomizedLoadMetricSpecification", {}).get("MetricDataQueries") %}
<CustomizedLoadMetricSpecification>
<MetricDataQueries>
{% for query in config.get("CustomizedLoadMetricSpecification", {}).get("MetricDataQueries", []) %}
<member>
<Id>{{ query.get("Id") }}</Id>
<Expression>{{ query.get("Expression") }}</Expression>
<MetricStat>
<Metric>
<Namespace>{{ query.get("MetricStat", {}).get("Metric", {}).get("Namespace") }}</Namespace>
<MetricName>{{ query.get("MetricStat", {}).get("Metric", {}).get("MetricName") }}</MetricName>
<Dimensions>
{% for dim in query.get("MetricStat", {}).get("Metric", {}).get("Dimensions", []) %}
<Name>{{ dim.get("Name") }}</Name>
<Value>{{ dim.get("Value") }}</Value>
{% endfor %}
</Dimensions>
</Metric>
<Stat>{{ query.get("MetricStat", {}).get("Stat") }}</Stat>
<Unit>{{ query.get("MetricStat", {}).get("Unit") }}</Unit>
</MetricStat>
<Label>{{ query.get("Label") }}</Label>
<ReturnData>{{ 'true' if query.get("ReturnData") else 'false' }}</ReturnData>
</member>
{% endfor %}
</MetricDataQueries>
</CustomizedLoadMetricSpecification>
{% endif %}
{% if config.get("CustomizedCapacityMetricSpecification", {}).get("MetricDataQueries") %}
<CustomizedCapacityMetricSpecification>
<MetricDataQueries>
{% for query in config.get("CustomizedCapacityMetricSpecification", {}).get("MetricDataQueries", []) %}
<member>
<Id>{{ query.get("Id") }}</Id>
<Expression>{{ query.get("Expression") }}</Expression>
<MetricStat>
<Metric>
<Namespace>{{ query.get("MetricStat", {}).get("Metric", {}).get("Namespace") }}</Namespace>
<MetricName>{{ query.get("MetricStat", {}).get("Metric", {}).get("MetricName") }}</MetricName>
<Dimensions>
{% for dim in query.get("MetricStat", {}).get("Metric", {}).get("Dimensions", []) %}
<Name>{{ dim.get("Name") }}</Name>
<Value>{{ dim.get("Value") }}</Value>
{% endfor %}
</Dimensions>
</Metric>
<Stat>{{ query.get("MetricStat", {}).get("Stat") }}</Stat>
<Unit>{{ query.get("MetricStat", {}).get("Unit") }}</Unit>
</MetricStat>
<Label>{{ query.get("Label") }}</Label>
<ReturnData>{{ 'true' if query.get("ReturnData") else 'false' }}</ReturnData>
</member>
{% endfor %}
</MetricDataQueries>
</CustomizedCapacityMetricSpecification>
{% endif %}
</member>
{% endfor %}
</MetricSpecifications>
{% if "Mode" in policy.predictive_scaling_configuration %}
<Mode>{{ policy.predictive_scaling_configuration.get("Mode") }}</Mode>
{% endif %}
{% if "SchedulingBufferTime" in policy.predictive_scaling_configuration %}
<SchedulingBufferTime>{{ policy.predictive_scaling_configuration.get("SchedulingBufferTime") }}</SchedulingBufferTime>
{% endif %}
{% if "MaxCapacityBreachBehavior" in policy.predictive_scaling_configuration %}
<MaxCapacityBreachBehavior>{{ policy.predictive_scaling_configuration.get("MaxCapacityBreachBehavior") }}</MaxCapacityBreachBehavior>
{% endif %}
{% if "MaxCapacityBuffer" in policy.predictive_scaling_configuration %}
<MaxCapacityBuffer>{{ policy.predictive_scaling_configuration.get("MaxCapacityBuffer") }}</MaxCapacityBuffer>
{% endif %}
</PredictiveScalingConfiguration>
{% endif %}
<Alarms/>
</member>
{% endfor %}

120
moto/ec2/_models/launch_templates.py Normal file
View File

@ -0,0 +1,120 @@
from collections import OrderedDict
from .core import TaggedEC2Resource
from ..utils import generic_filter, random_launch_template_id, utc_date_and_time
from ..exceptions import InvalidLaunchTemplateNameError
class LaunchTemplateVersion(object):
def __init__(self, template, number, data, description):
self.template = template
self.number = number
self.data = data
self.description = description
self.create_time = utc_date_and_time()
@property
def image_id(self):
return self.data.get("ImageId", "")
@property
def instance_type(self):
return self.data.get("InstanceType", "")
@property
def security_groups(self):
return self.data.get("SecurityGroups", [])
@property
def user_data(self):
return self.data.get("UserData", "")
class LaunchTemplate(TaggedEC2Resource):
def __init__(self, backend, name, template_data, version_description):
self.ec2_backend = backend
self.name = name
self.id = random_launch_template_id()
self.create_time = utc_date_and_time()
self.versions = []
self.create_version(template_data, version_description)
self.default_version_number = 1
def create_version(self, data, description):
num = len(self.versions) + 1
version = LaunchTemplateVersion(self, num, data, description)
self.versions.append(version)
return version
def is_default(self, version):
return self.default_version == version.number
def get_version(self, num):
if str(num).lower() == "$latest":
return self.versions[-1]
if str(num).lower() == "$default":
return self.default_version()
return self.versions[int(num) - 1]
def default_version(self):
return self.versions[self.default_version_number - 1]
def latest_version(self):
return self.versions[-1]
@property
def latest_version_number(self):
return self.latest_version().number
def get_filter_value(self, filter_name):
if filter_name == "launch-template-name":
return self.name
else:
return super().get_filter_value(filter_name, "DescribeLaunchTemplates")
class LaunchTemplateBackend(object):
def __init__(self):
self.launch_template_name_to_ids = {}
self.launch_templates = OrderedDict()
self.launch_template_insert_order = []
super().__init__()
def create_launch_template(self, name, description, template_data):
if name in self.launch_template_name_to_ids:
raise InvalidLaunchTemplateNameError()
template = LaunchTemplate(self, name, template_data, description)
self.launch_templates[template.id] = template
self.launch_template_name_to_ids[template.name] = template.id
self.launch_template_insert_order.append(template.id)
return template
def get_launch_template(self, template_id):
return self.launch_templates[template_id]
def get_launch_template_by_name(self, name):
return self.get_launch_template(self.launch_template_name_to_ids[name])
def delete_launch_template(self, name, tid):
if name:
tid = self.launch_template_name_to_ids[name]
return self.launch_templates.pop(tid)
def describe_launch_templates(
self, template_names=None, template_ids=None, filters=None
):
if template_names and not template_ids:
template_ids = []
for name in template_names:
template_ids.append(self.launch_template_name_to_ids[name])
if template_ids:
templates = [
self.launch_templates[tid]
for tid in template_ids
if tid in self.launch_templates
]
else:
templates = list(self.launch_templates.values())
return generic_filter(filters, templates)

115
moto/ec2/models.py
View File

@ -70,7 +70,6 @@ from .exceptions import (
InvalidDependantParameterError,
InvalidDependantParameterTypeError,
InvalidFlowLogIdError,
InvalidLaunchTemplateNameError,
InvalidNetworkAclIdError,
InvalidNetworkAttachmentIdError,
InvalidNetworkInterfaceIdError,
@ -122,6 +121,7 @@ from .exceptions import (
InvalidCarrierGatewayID,
)
from ._models.core import TaggedEC2Resource
from ._models.launch_templates import LaunchTemplateBackend
from ._models.vpc_service_configuration import VPCServiceConfigurationBackend
from .utils import (
EC2_RESOURCE_TO_PREFIX,
@ -142,7 +142,6 @@ from .utils import (
random_transit_gateway_attachment_id,
random_transit_gateway_route_table_id,
random_vpc_ep_id,
random_launch_template_id,
random_nat_gateway_id,
random_transit_gateway_id,
random_key_pair,
@ -189,6 +188,7 @@ from .utils import (
rsa_public_key_parse,
rsa_public_key_fingerprint,
describe_tag_filter,
utc_date_and_time,
)
INSTANCE_TYPES = load_resource(__name__, "resources/instance_types.json")
@ -217,14 +217,6 @@ MAX_NUMBER_OF_ENDPOINT_SERVICES_RESULTS = 1000
DEFAULT_VPC_ENDPOINT_SERVICES = []
def utc_date_and_time():
x = datetime.utcnow()
# Better performing alternative to x.strftime("%Y-%m-%dT%H:%M:%S.000Z")
return "{}-{:02d}-{:02d}T{:02d}:{:02d}:{:02d}.000Z".format(
x.year, x.month, x.day, x.hour, x.minute, x.second
)
def validate_resource_ids(resource_ids):
if not resource_ids:
raise MissingParameterError(parameter="resourceIdSet")
@ -8415,109 +8407,6 @@ class NatGatewayBackend(object):
return nat_gw
class LaunchTemplateVersion(object):
def __init__(self, template, number, data, description):
self.template = template
self.number = number
self.data = data
self.description = description
self.create_time = utc_date_and_time()
@property
def image_id(self):
return self.data.get("ImageId", "")
@property
def instance_type(self):
return self.data.get("InstanceType", "")
@property
def security_groups(self):
return self.data.get("SecurityGroups", [])
@property
def user_data(self):
return self.data.get("UserData", "")
class LaunchTemplate(TaggedEC2Resource):
def __init__(self, backend, name, template_data, version_description):
self.ec2_backend = backend
self.name = name
self.id = random_launch_template_id()
self.create_time = utc_date_and_time()
self.versions = []
self.create_version(template_data, version_description)
self.default_version_number = 1
def create_version(self, data, description):
num = len(self.versions) + 1
version = LaunchTemplateVersion(self, num, data, description)
self.versions.append(version)
return version
def is_default(self, version):
return self.default_version == version.number
def get_version(self, num):
return self.versions[num - 1]
def default_version(self):
return self.versions[self.default_version_number - 1]
def latest_version(self):
return self.versions[-1]
@property
def latest_version_number(self):
return self.latest_version().number
def get_filter_value(self, filter_name):
if filter_name == "launch-template-name":
return self.name
else:
return super().get_filter_value(filter_name, "DescribeLaunchTemplates")
class LaunchTemplateBackend(object):
def __init__(self):
self.launch_template_name_to_ids = {}
self.launch_templates = OrderedDict()
self.launch_template_insert_order = []
super().__init__()
def create_launch_template(self, name, description, template_data):
if name in self.launch_template_name_to_ids:
raise InvalidLaunchTemplateNameError()
template = LaunchTemplate(self, name, template_data, description)
self.launch_templates[template.id] = template
self.launch_template_name_to_ids[template.name] = template.id
self.launch_template_insert_order.append(template.id)
return template
def get_launch_template(self, template_id):
return self.launch_templates[template_id]
def get_launch_template_by_name(self, name):
return self.get_launch_template(self.launch_template_name_to_ids[name])
def describe_launch_templates(
self, template_names=None, template_ids=None, filters=None
):
if template_names and not template_ids:
template_ids = []
for name in template_names:
template_ids.append(self.launch_template_name_to_ids[name])
if template_ids:
templates = [self.launch_templates[tid] for tid in template_ids]
else:
templates = list(self.launch_templates.values())
return generic_filter(filters, templates)
class IamInstanceProfileAssociation(CloudFormationModel):
def __init__(self, ec2_backend, association_id, instance, iam_instance_profile):
self.ec2_backend = ec2_backend

17
moto/ec2/resources/amis.json
View File

@ -644,5 +644,22 @@
"name": "amzn-ami-minimal-hvm-2018.03.0.20181129-x86_64-ebs",
"virtualization_type": "hvm",
"hypervisor": "xen"
},
{
"architecture": "x86_64",
"ami_id": "ami-04681a1dbd79675a5",
"image_location": "amazon/amzn2-ami-hvm-2.0.20180810-x86_64-gp2",
"image_type": "machine",
"public": true,
"owner_id": "137112412989",
"platform": "Linux/UNIX",
"state": "available",
"description": "Amazon Linux 2 AMI 2.0.20180810 x86_64 HVM gp2",
"hypervisor": "xen",
"name": "amzn2-ami-hvm-2.0.20180810-x86_64-gp2",
"root_device_name": "/dev/xvda",
"root_device_type": "ebs",
"sriov": "simple",
"virtualization_type": "hvm"
}
]

21
moto/ec2/responses/launch_templates.py
View File

@ -175,8 +175,25 @@ class LaunchTemplates(EC2BaseResponse):
)
return pretty_xml(tree)
# def delete_launch_template(self):
# pass
def delete_launch_template(self):
name = self._get_param("LaunchTemplateName")
tid = self._get_param("LaunchTemplateId")
if self.is_not_dryrun("DeleteLaunchTemplate"):
template = self.ec2_backend.delete_launch_template(name, tid)
tree = xml_root("DeleteLaunchTemplatesResponse")
xml_serialize(
tree,
"launchTemplate",
{
"defaultVersionNumber": template.default_version_number,
"launchTemplateId": template.id,
"launchTemplateName": template.name,
},
)
return pretty_xml(tree)
# def delete_launch_template_versions(self):
# pass

9
moto/ec2/utils.py
View File

@ -5,6 +5,7 @@ import random
import re
import ipaddress
from datetime import datetime
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
@ -297,6 +298,14 @@ def create_dns_entries(service_name, vpc_endpoint_id):
return dns_entries
def utc_date_and_time():
x = datetime.utcnow()
# Better performing alternative to x.strftime("%Y-%m-%dT%H:%M:%S.000Z")
return "{}-{:02d}-{:02d}T{:02d}:{:02d}:{:02d}.000Z".format(
x.year, x.month, x.day, x.hour, x.minute, x.second
)
def split_route_id(route_id):
values = route_id.split("~")
return values[0], values[1]

3
tests/terraform-tests.success.txt
View File

@ -10,6 +10,9 @@ TestAccAWSAPIGatewayV2RouteResponse
TestAccAWSAPIGatewayV2VpcLink
TestAccAWSAppsyncApiKey
TestAccAWSAppsyncGraphqlApi
TestAccAWSAutoscalingAttachment
TestAccAwsAutoScalingGroupDataSource
TestAccAWSAutoscalingPolicy
TestAccAWSAvailabilityZones
TestAccAWSBatchJobDefinition
TestAccAWSBatchJobQueue

141
tests/test_autoscaling/test_autoscaling.py
View File

@ -13,7 +13,7 @@ from tests import EXAMPLE_AMI_ID
@mock_autoscaling
@mock_ec2
@mock_elb
def test_create_autoscaling_group_boto3_within_elb():
def test_create_autoscaling_group_within_elb():
mocked_networking = setup_networking()
elb_client = boto3.client("elb", region_name="us-east-1")
elb_client.create_load_balancer(
@ -115,7 +115,7 @@ def test_create_autoscaling_group_boto3_within_elb():
@mock_autoscaling
def test_create_autoscaling_groups_defaults_boto3():
def test_create_autoscaling_groups_defaults():
"""Test with the minimum inputs and check that all of the proper defaults
are assigned for the other attributes"""
@ -223,7 +223,7 @@ def test_propogate_tags():
@mock_autoscaling
def test_autoscaling_group_delete_boto3():
def test_autoscaling_group_delete():
mocked_networking = setup_networking()
as_client = boto3.client("autoscaling", region_name="us-east-1")
as_client.create_launch_configuration(
@ -430,7 +430,7 @@ def test_detach_load_balancer():
@mock_autoscaling
def test_create_autoscaling_group_boto3():
def test_create_autoscaling_group():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
client.create_launch_configuration(
@ -566,6 +566,75 @@ def test_create_autoscaling_group_from_template():
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
@mock_ec2
@mock_autoscaling
def test_create_auto_scaling_from_template_version__latest():
ec2_client = boto3.client("ec2", region_name="us-west-1")
launch_template_name = "tester"
ec2_client.create_launch_template(
LaunchTemplateName=launch_template_name,
LaunchTemplateData={"ImageId": EXAMPLE_AMI_ID, "InstanceType": "t2.medium"},
)
asg_client = boto3.client("autoscaling", region_name="us-west-1")
asg_client.create_auto_scaling_group(
AutoScalingGroupName="name",
DesiredCapacity=1,
MinSize=1,
MaxSize=1,
LaunchTemplate={
"LaunchTemplateName": launch_template_name,
"Version": "$Latest",
},
AvailabilityZones=["us-west-1a"],
)
response = asg_client.describe_auto_scaling_groups(AutoScalingGroupNames=["name"])[
"AutoScalingGroups"
][0]
response.should.have.key("LaunchTemplate")
response["LaunchTemplate"].should.have.key("LaunchTemplateName").equals(
launch_template_name
)
response["LaunchTemplate"].should.have.key("Version").equals("$Latest")
@mock_ec2
@mock_autoscaling
def test_create_auto_scaling_from_template_version__default():
ec2_client = boto3.client("ec2", region_name="us-west-1")
launch_template_name = "tester"
ec2_client.create_launch_template(
LaunchTemplateName=launch_template_name,
LaunchTemplateData={"ImageId": EXAMPLE_AMI_ID, "InstanceType": "t2.medium"},
)
ec2_client.create_launch_template_version(
LaunchTemplateName=launch_template_name,
LaunchTemplateData={"ImageId": EXAMPLE_AMI_ID, "InstanceType": "t3.medium"},
VersionDescription="v2",
)
asg_client = boto3.client("autoscaling", region_name="us-west-1")
asg_client.create_auto_scaling_group(
AutoScalingGroupName="name",
DesiredCapacity=1,
MinSize=1,
MaxSize=1,
LaunchTemplate={
"LaunchTemplateName": launch_template_name,
"Version": "$Default",
},
AvailabilityZones=["us-west-1a"],
)
response = asg_client.describe_auto_scaling_groups(AutoScalingGroupNames=["name"])[
"AutoScalingGroups"
][0]
response.should.have.key("LaunchTemplate")
response["LaunchTemplate"].should.have.key("LaunchTemplateName").equals(
launch_template_name
)
response["LaunchTemplate"].should.have.key("Version").equals("$Default")
@mock_autoscaling
@mock_ec2
def test_create_autoscaling_group_no_template_ref():
@ -629,7 +698,7 @@ def test_create_autoscaling_group_multiple_template_ref():
@mock_autoscaling
def test_create_autoscaling_group_boto3_no_launch_configuration():
def test_create_autoscaling_group_no_launch_configuration():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
with pytest.raises(ClientError) as ex:
@ -651,7 +720,7 @@ def test_create_autoscaling_group_boto3_no_launch_configuration():
@mock_autoscaling
@mock_ec2
def test_create_autoscaling_group_boto3_multiple_launch_configurations():
def test_create_autoscaling_group_multiple_launch_configurations():
mocked_networking = setup_networking()
ec2_client = boto3.client("ec2", region_name="us-east-1")
@ -689,7 +758,7 @@ def test_create_autoscaling_group_boto3_multiple_launch_configurations():
@mock_autoscaling
def test_describe_autoscaling_groups_boto3_launch_config():
def test_describe_autoscaling_groups_launch_config():
mocked_networking = setup_networking(region_name="eu-north-1")
client = boto3.client("autoscaling", region_name="eu-north-1")
client.create_launch_configuration(
@ -729,7 +798,7 @@ def test_describe_autoscaling_groups_boto3_launch_config():
@mock_autoscaling
@mock_ec2
def test_describe_autoscaling_groups_boto3_launch_template():
def test_describe_autoscaling_groups_launch_template():
mocked_networking = setup_networking()
ec2_client = boto3.client("ec2", region_name="us-east-1")
template = ec2_client.create_launch_template(
@ -770,7 +839,7 @@ def test_describe_autoscaling_groups_boto3_launch_template():
@mock_autoscaling
def test_describe_autoscaling_instances_boto3_launch_config():
def test_describe_autoscaling_instances_launch_config():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
client.create_launch_configuration(
@ -801,7 +870,7 @@ def test_describe_autoscaling_instances_boto3_launch_config():
@mock_autoscaling
@mock_ec2
def test_describe_autoscaling_instances_boto3_launch_template():
def test_describe_autoscaling_instances_launch_template():
mocked_networking = setup_networking()
ec2_client = boto3.client("ec2", region_name="us-east-1")
template = ec2_client.create_launch_template(
@ -871,7 +940,7 @@ def test_describe_autoscaling_instances_instanceid_filter():
@mock_autoscaling
def test_update_autoscaling_group_boto3_launch_config():
def test_update_autoscaling_group_launch_config():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
client.create_launch_configuration(
@ -914,7 +983,7 @@ def test_update_autoscaling_group_boto3_launch_config():
@mock_autoscaling
@mock_ec2
def test_update_autoscaling_group_boto3_launch_template():
def test_update_autoscaling_group_launch_template():
mocked_networking = setup_networking()
ec2_client = boto3.client("ec2", region_name="us-east-1")
ec2_client.create_launch_template(
@ -1019,7 +1088,7 @@ def test_update_autoscaling_group_max_size_desired_capacity_change():
@mock_autoscaling
def test_autoscaling_describe_policies_boto3():
def test_autoscaling_describe_policies():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
@ -1048,6 +1117,7 @@ def test_autoscaling_describe_policies_boto3():
AutoScalingGroupName="test_asg",
PolicyName="test_policy_down",
PolicyType="SimpleScaling",
MetricAggregationType="Minimum",
AdjustmentType="PercentChangeInCapacity",
ScalingAdjustment=-10,
Cooldown=60,
@ -1080,6 +1150,7 @@ def test_autoscaling_describe_policies_boto3():
response["ScalingPolicies"].should.have.length_of(1)
policy = response["ScalingPolicies"][0]
policy["PolicyType"].should.equal("SimpleScaling")
policy["MetricAggregationType"].should.equal("Minimum")
policy["AdjustmentType"].should.equal("PercentChangeInCapacity")
policy["ScalingAdjustment"].should.equal(-10)
policy["Cooldown"].should.equal(60)
@ -1275,6 +1346,44 @@ def test_create_autoscaling_policy_with_policytype__stepscaling():
policy.shouldnt.have.key("Cooldown")
@mock_autoscaling
@mock_ec2
def test_create_autoscaling_policy_with_predictive_scaling_config():
mocked_networking = setup_networking(region_name="eu-west-1")
client = boto3.client("autoscaling", region_name="eu-west-1")
launch_config_name = "lg_name"
asg_name = "asg_test"
client.create_launch_configuration(
LaunchConfigurationName=launch_config_name,
ImageId=EXAMPLE_AMI_ID,
InstanceType="m1.small",
)
client.create_auto_scaling_group(
LaunchConfigurationName=launch_config_name,
AutoScalingGroupName=asg_name,
MinSize=1,
MaxSize=2,
VPCZoneIdentifier=mocked_networking["subnet1"],
)
client.put_scaling_policy(
AutoScalingGroupName=asg_name,
PolicyName=launch_config_name,
PolicyType="PredictiveScaling",
PredictiveScalingConfiguration={
"MetricSpecifications": [{"TargetValue": 5}],
"SchedulingBufferTime": 7,
},
)
resp = client.describe_policies(AutoScalingGroupName=asg_name)
policy = resp["ScalingPolicies"][0]
policy.should.have.key("PredictiveScalingConfiguration").equals(
{"MetricSpecifications": [{"TargetValue": 5.0}], "SchedulingBufferTime": 7}
)
@mock_elb
@mock_autoscaling
@mock_ec2
@ -2343,7 +2452,7 @@ def test_set_instance_protection():
@mock_autoscaling
def test_set_desired_capacity_up_boto3():
def test_set_desired_capacity_up():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
@ -2371,7 +2480,7 @@ def test_set_desired_capacity_up_boto3():
@mock_autoscaling
def test_set_desired_capacity_down_boto3():
def test_set_desired_capacity_down():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
@ -2640,7 +2749,7 @@ def test_autoscaling_lifecyclehook():
@pytest.mark.parametrize("original,new", [(2, 1), (2, 3), (1, 5), (1, 1)])
@mock_autoscaling
def test_set_desired_capacity_without_protection_boto3(original, new):
def test_set_desired_capacity_without_protection(original, new):
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
client.create_launch_configuration(

10
tests/test_autoscaling/test_launch_configurations.py
View File

@ -11,7 +11,7 @@ from tests import EXAMPLE_AMI_ID
@mock_autoscaling
def test_create_launch_configuration_boto3():
def test_create_launch_configuration():
client = boto3.client("autoscaling", region_name="us-east-1")
client.create_launch_configuration(
LaunchConfigurationName="tester",
@ -46,7 +46,7 @@ def test_create_launch_configuration_boto3():
@mock_autoscaling
def test_create_launch_configuration_with_block_device_mappings_boto3():
def test_create_launch_configuration_with_block_device_mappings():
client = boto3.client("autoscaling", region_name="us-east-1")
client.create_launch_configuration(
LaunchConfigurationName="tester",
@ -136,7 +136,7 @@ def test_create_launch_configuration_additional_params_default_to_false():
@mock_autoscaling
def test_create_launch_configuration_defaults_boto3():
def test_create_launch_configuration_defaults():
"""Test with the minimum inputs and check that all of the proper defaults
are assigned for the other attributes"""
client = boto3.client("autoscaling", region_name="us-east-1")
@ -158,7 +158,7 @@ def test_create_launch_configuration_defaults_boto3():
@mock_autoscaling
def test_launch_configuration_describe_filter_boto3():
def test_launch_configuration_describe_filter():
client = boto3.client("autoscaling", region_name="us-east-1")
for name in ["tester", "tester2", "tester3"]:
client.create_launch_configuration(
@ -200,7 +200,7 @@ def test_launch_configuration_describe_paginated():
@mock_autoscaling
def test_launch_configuration_delete_boto3():
def test_launch_configuration_delete():
client = boto3.client("autoscaling", region_name="us-east-1")
client.create_launch_configuration(
LaunchConfigurationName="tester",

68
tests/test_ec2/test_launch_templates.py
View File

@ -410,6 +410,74 @@ def test_create_launch_template_with_tag_spec():
)
@mock_ec2
def test_delete_launch_template__dryrun():
cli = boto3.client("ec2", region_name="us-east-1")
template_name = str(uuid4())
cli.create_launch_template(
LaunchTemplateName=template_name,
LaunchTemplateData={"ImageId": "ami-abc123"},
TagSpecifications=[
{"ResourceType": "instance", "Tags": [{"Key": "key", "Value": "value"}]}
],
)
cli.describe_launch_templates(LaunchTemplateNames=[template_name])[
"LaunchTemplates"
].should.have.length_of(1)
with pytest.raises(ClientError) as exc:
cli.delete_launch_template(DryRun=True, LaunchTemplateName=template_name)
err = exc.value.response["Error"]
err.should.have.key("Code").equals("DryRunOperation")
# Template still exists
cli.describe_launch_templates(LaunchTemplateNames=[template_name])[
"LaunchTemplates"
].should.have.length_of(1)
@mock_ec2
def test_delete_launch_template__by_name():
cli = boto3.client("ec2", region_name="us-east-1")
template_name = str(uuid4())
cli.create_launch_template(
LaunchTemplateName=template_name, LaunchTemplateData={"ImageId": "ami-abc123"}
)
cli.describe_launch_templates(LaunchTemplateNames=[template_name])[
"LaunchTemplates"
].should.have.length_of(1)
cli.delete_launch_template(LaunchTemplateName=template_name)
cli.describe_launch_templates(LaunchTemplateNames=[template_name])[
"LaunchTemplates"
].should.have.length_of(0)
@mock_ec2
def test_delete_launch_template__by_id():
cli = boto3.client("ec2", region_name="us-east-1")
template_name = str(uuid4())
template_id = cli.create_launch_template(
LaunchTemplateName=template_name, LaunchTemplateData={"ImageId": "ami-abc123"}
)["LaunchTemplate"]["LaunchTemplateId"]
cli.describe_launch_templates(LaunchTemplateNames=[template_name])[
"LaunchTemplates"
].should.have.length_of(1)
cli.delete_launch_template(LaunchTemplateId=template_id)
cli.describe_launch_templates(LaunchTemplateNames=[template_name])[
"LaunchTemplates"
].should.have.length_of(0)
def retrieve_all_templates(client, filters=[]): # pylint: disable=W0102
resp = client.describe_launch_templates(Filters=filters)
all_templates = resp["LaunchTemplates"]

2
tests/test_ec2/test_route_tables.py
View File

@ -204,7 +204,6 @@ def test_route_table_associations_boto3():
# Refresh
r = client.describe_route_tables(RouteTableIds=[route_table.id])["RouteTables"][0]
r["Associations"].should.have.length_of(1)
print(r)
# Associate
association_id = client.associate_route_table(
@ -915,7 +914,6 @@ def test_associate_route_table_by_subnet():
{"Name": "association.route-table-association-id", "Values": [assoc_id]}
]
)["RouteTables"]
print(verify[0]["Associations"])
# First assocation is the main
verify[0]["Associations"][0].should.have.key("Main").equals(True)