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

Add CloudFormation support to EC2 Launch Templates (#5477)

Browse Source
This commit is contained in:
Timothy Klopotoski 2022-09-16 06:01:43 -04:00 committed by GitHub
parent 837ad2cbb7
commit c648c0f957
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 298 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
moto/ec2/exceptions.py
View File

@ -718,6 +718,14 @@ class InvalidLaunchTemplateNameNotFoundError(EC2ClientError):
)
class InvalidLaunchTemplateNameNotFoundWithNameError(EC2ClientError):
def __init__(self, name):
super().__init__(
"InvalidLaunchTemplateName.NotFoundException",
f"The specified launch template, with template name {name}, does not exist",
)
class InvalidParameterDependency(EC2ClientError):
def __init__(self, param, param_needed):
super().__init__(

91
moto/ec2/models/launch_templates.py
View File

@ -1,9 +1,17 @@
from collections import OrderedDict
from moto.core import CloudFormationModel
from .core import TaggedEC2Resource
from ..utils import generic_filter, random_launch_template_id, utc_date_and_time
from ..utils import (
generic_filter,
random_launch_template_id,
utc_date_and_time,
convert_tag_spec,
)
from ..exceptions import (
InvalidLaunchTemplateNameAlreadyExistsError,
InvalidLaunchTemplateNameNotFoundError,
InvalidLaunchTemplateNameNotFoundWithNameError,
)
@ -32,7 +40,7 @@ class LaunchTemplateVersion(object):
return self.data.get("UserData", "")
class LaunchTemplate(TaggedEC2Resource):
class LaunchTemplate(TaggedEC2Resource, CloudFormationModel):
def __init__(self, backend, name, template_data, version_description, tag_spec):
self.ec2_backend = backend
self.name = name
@ -72,12 +80,89 @@ class LaunchTemplate(TaggedEC2Resource):
def latest_version_number(self):
return self.latest_version().number
@property
def physical_resource_id(self):
return self.id
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")
@staticmethod
def cloudformation_name_type():
return "LaunchTemplateName"
@staticmethod
def cloudformation_type():
# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-ec2-launchtemplate.html
return "AWS::EC2::LaunchTemplate"
@classmethod
def create_from_cloudformation_json(
cls, resource_name, cloudformation_json, account_id, region_name, **kwargs
):
from ..models import ec2_backends
backend = ec2_backends[account_id][region_name]
properties = cloudformation_json["Properties"]
name = properties.get("LaunchTemplateName")
data = properties.get("LaunchTemplateData")
description = properties.get("VersionDescription")
tag_spec = convert_tag_spec(
properties.get("TagSpecifications", {}), tag_key="Tags"
)
launch_template = backend.create_launch_template(
name, description, data, tag_spec
)
return launch_template
@classmethod
def update_from_cloudformation_json(
cls,
original_resource,
new_resource_name,
cloudformation_json,
account_id,
region_name,
):
from ..models import ec2_backends
backend = ec2_backends[account_id][region_name]
properties = cloudformation_json["Properties"]
name = properties.get("LaunchTemplateName")
data = properties.get("LaunchTemplateData")
description = properties.get("VersionDescription")
launch_template = backend.get_launch_template_by_name(name)
launch_template.create_version(data, description)
return launch_template
@classmethod
def delete_from_cloudformation_json(
cls, resource_name, cloudformation_json, account_id, region_name
):
from ..models import ec2_backends
backend = ec2_backends[account_id][region_name]
properties = cloudformation_json["Properties"]
name = properties.get("LaunchTemplateName")
backend.delete_launch_template(name, None)
class LaunchTemplateBackend:
def __init__(self):
@ -98,6 +183,8 @@ class LaunchTemplateBackend:
return self.launch_templates[template_id]
def get_launch_template_by_name(self, name):
if name not in self.launch_template_name_to_ids:
raise InvalidLaunchTemplateNameNotFoundWithNameError(name)
return self.get_launch_template(self.launch_template_name_to_ids[name])
def delete_launch_template(self, name, tid):

9
moto/ec2/utils.py
View File

@ -787,14 +787,15 @@ def gen_moto_amis(described_images, drop_images_missing_keys=True):
return result
def convert_tag_spec(tag_spec_set):
# IN: [{"ResourceType": _type, "Tag": [{"Key": k, "Value": v}, ..]}]
# OUT: {_type: {k: v, ..}}
def convert_tag_spec(tag_spec_set, tag_key="Tag"):
# IN: [{"ResourceType": _type, "Tag": [{"Key": k, "Value": v}, ..]}]
# (or) [{"ResourceType": _type, "Tags": [{"Key": k, "Value": v}, ..]}] <-- special cfn case
# OUT: {_type: {k: v, ..}}
tags = {}
for tag_spec in tag_spec_set:
if tag_spec["ResourceType"] not in tags:
tags[tag_spec["ResourceType"]] = {}
tags[tag_spec["ResourceType"]].update(
{tag["Key"]: tag["Value"] for tag in tag_spec["Tag"]}
{tag["Key"]: tag["Value"] for tag in tag_spec[tag_key]}
)
return tags

181
tests/test_ec2/test_ec2_cloudformation.py
View File

@ -760,3 +760,184 @@ def test_vpc_endpoint_creation():
endpoint.should.have.key("State").equals("available")
endpoint.should.have.key("SubnetIds").equals([subnet1.id])
endpoint.should.have.key("VpcEndpointType").equals("GatewayLoadBalancer")
@mock_cloudformation
@mock_ec2
def test_launch_template_create():
cf = boto3.client("cloudformation", region_name="us-west-1")
ec2 = boto3.client("ec2", region_name="us-west-1")
launch_template_name = str(uuid4())[0:6]
logical_id = str(uuid4())[0:6]
stack_name = str(uuid4())[0:6]
lt_tags = [{"Key": "lt-tag-key", "Value": "lt-tag-value"}]
template_json = json.dumps(
{
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
logical_id: {
"Type": "AWS::EC2::LaunchTemplate",
"Properties": {
"LaunchTemplateName": launch_template_name,
"VersionDescription": "some template",
"LaunchTemplateData": {
"TagSpecifications": [
{
"ResourceType": "instance",
"Tags": [
{"Key": "i-tag-key", "Value": "i-tag-value"}
],
}
]
},
"TagSpecifications": [
{
"ResourceType": "launch-template",
"Tags": lt_tags,
}
],
},
}
},
"Outputs": {
"LaunchTemplateId": {
"Description": "The ID of the created launch template",
"Value": {"Ref": logical_id},
},
},
}
)
cf.create_stack(StackName=stack_name, TemplateBody=template_json)
resources = cf.list_stack_resources(StackName=stack_name)["StackResourceSummaries"]
resources.should.have.length_of(1)
resources[0].should.have.key("LogicalResourceId").equals(logical_id)
launch_template_id = resources[0]["PhysicalResourceId"]
outputs = cf.describe_stacks(StackName=stack_name)["Stacks"][0]["Outputs"]
outputs.should.have.length_of(1)
outputs[0].should.equal(
{"OutputKey": "LaunchTemplateId", "OutputValue": launch_template_id}
)
launch_template = ec2.describe_launch_templates(
LaunchTemplateNames=[launch_template_name]
)["LaunchTemplates"][0]
launch_template.should.have.key("LaunchTemplateName").equals(launch_template_name)
launch_template.should.have.key("LaunchTemplateId").equals(launch_template_id)
launch_template.should.have.key("Tags").equals(lt_tags)
launch_template_version = ec2.describe_launch_template_versions(
LaunchTemplateName=launch_template_name
)["LaunchTemplateVersions"][0]
launch_template_version.should.have.key("LaunchTemplateName").equals(
launch_template_name
)
launch_template_version.should.have.key("LaunchTemplateId").equals(
launch_template_id
)
launch_template_version["LaunchTemplateData"]["TagSpecifications"][
0
].should.have.key("ResourceType").equals("instance")
@mock_cloudformation
@mock_ec2
def test_launch_template_update():
cf = boto3.client("cloudformation", region_name="us-west-1")
ec2 = boto3.client("ec2", region_name="us-west-1")
launch_template_name = str(uuid4())[0:6]
logical_id = str(uuid4())[0:6]
stack_name = str(uuid4())[0:6]
template_json = json.dumps(
{
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
logical_id: {
"Type": "AWS::EC2::LaunchTemplate",
"Properties": {
"LaunchTemplateName": launch_template_name,
"VersionDescription": "some template",
"LaunchTemplateData": {"UserData": ""},
},
}
},
}
)
cf.create_stack(StackName=stack_name, TemplateBody=template_json)
template_json = json.dumps(
{
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
logical_id: {
"Type": "AWS::EC2::LaunchTemplate",
"Properties": {
"LaunchTemplateName": launch_template_name,
"VersionDescription": "a better template",
"LaunchTemplateData": {"UserData": ""},
},
}
},
}
)
cf.update_stack(StackName=stack_name, TemplateBody=template_json)
resources = cf.list_stack_resources(StackName=stack_name)["StackResourceSummaries"]
resources.should.have.length_of(1)
launch_template_versions = ec2.describe_launch_template_versions(
LaunchTemplateName=launch_template_name
)["LaunchTemplateVersions"]
launch_template_versions.should.have.length_of(2)
launch_template_versions[0].should.have.key("VersionDescription").equals(
"some template"
)
launch_template_versions[1].should.have.key("VersionDescription").equals(
"a better template"
)
@mock_cloudformation
@mock_ec2
def test_launch_template_delete():
cf = boto3.client("cloudformation", region_name="us-west-1")
ec2 = boto3.client("ec2", region_name="us-west-1")
launch_template_name = str(uuid4())[0:6]
logical_id = str(uuid4())[0:6]
stack_name = str(uuid4())[0:6]
template_json = json.dumps(
{
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
logical_id: {
"Type": "AWS::EC2::LaunchTemplate",
"Properties": {
"LaunchTemplateName": launch_template_name,
"VersionDescription": "some template",
"LaunchTemplateData": {"UserData": ""},
},
}
},
}
)
cf.create_stack(StackName=stack_name, TemplateBody=template_json)
cf.delete_stack(StackName=stack_name)
ec2.describe_launch_templates(LaunchTemplateNames=[launch_template_name])[
"LaunchTemplates"
].should.have.length_of(0)

15
tests/test_ec2/test_launch_templates.py
View File

@ -86,6 +86,21 @@ def test_describe_launch_template_versions():
templ.should.equal(template_data)
@mock_ec2
def test_describe_launch_template_versions_by_name_when_absent():
cli = boto3.client("ec2", region_name="us-east-1")
template_name = "foo"
# test using name
with pytest.raises(
ClientError,
match=f"The specified launch template, with template name {template_name}, does not exist",
):
cli.describe_launch_template_versions(LaunchTemplateName=template_name)
@mock_ec2
def test_create_launch_template_version():
cli = boto3.client("ec2", region_name="us-east-1")