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

EC2: Simplify DryRun-handling (#5926)

Browse Source
This commit is contained in:
Bert Blommers 2023-02-14 12:43:28 -01:00 committed by GitHub
parent c50737d027
commit 78840fd71c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
25 changed files with 810 additions and 725 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
moto/core/responses.py
View File

@ -838,14 +838,10 @@ class BaseResponse(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
return "JSON" in self.querystring.get("ContentType", [])
def error_on_dryrun(self) -> None:
self.is_not_dryrun()
def is_not_dryrun(self, action: Optional[str] = None) -> bool:
if "true" in self.querystring.get("DryRun", ["false"]):
a = action or self._get_param("Action")
a = self._get_param("Action")
message = f"An error occurred (DryRunOperation) when calling the {a} operation: Request would have succeeded, but DryRun flag is set"
raise DryRunClientError(error_type="DryRunOperation", message=message)
return True
class _RecursiveDictRef(object):

83
moto/ec2/responses/amis.py
View File

@ -7,34 +7,40 @@ class AmisResponse(EC2BaseResponse):
description = self._get_param("Description", if_none="")
instance_id = self._get_param("InstanceId")
tag_specifications = self._get_multi_param("TagSpecification")
if self.is_not_dryrun("CreateImage"):
image = self.ec2_backend.create_image(
instance_id,
name,
description,
tag_specifications=tag_specifications,
)
template = self.response_template(CREATE_IMAGE_RESPONSE)
return template.render(image=image)
self.error_on_dryrun()
image = self.ec2_backend.create_image(
instance_id,
name,
description,
tag_specifications=tag_specifications,
)
template = self.response_template(CREATE_IMAGE_RESPONSE)
return template.render(image=image)
def copy_image(self):
source_image_id = self._get_param("SourceImageId")
source_region = self._get_param("SourceRegion")
name = self._get_param("Name")
description = self._get_param("Description")
if self.is_not_dryrun("CopyImage"):
image = self.ec2_backend.copy_image(
source_image_id, source_region, name, description
)
template = self.response_template(COPY_IMAGE_RESPONSE)
return template.render(image=image)
self.error_on_dryrun()
image = self.ec2_backend.copy_image(
source_image_id, source_region, name, description
)
template = self.response_template(COPY_IMAGE_RESPONSE)
return template.render(image=image)
def deregister_image(self):
ami_id = self._get_param("ImageId")
if self.is_not_dryrun("DeregisterImage"):
self.ec2_backend.deregister_image(ami_id)
template = self.response_template(DEREGISTER_IMAGE_RESPONSE)
return template.render(success="true")
self.error_on_dryrun()
self.ec2_backend.deregister_image(ami_id)
template = self.response_template(DEREGISTER_IMAGE_RESPONSE)
return template.render(success="true")
def describe_images(self):
self.error_on_dryrun()
@ -60,30 +66,33 @@ class AmisResponse(EC2BaseResponse):
operation_type = self._get_param("OperationType")
group = self._get_param("UserGroup.1")
user_ids = self._get_multi_param("UserId")
if self.is_not_dryrun("ModifyImageAttribute"):
if operation_type == "add":
self.ec2_backend.add_launch_permission(
ami_id, user_ids=user_ids, group=group
)
elif operation_type == "remove":
self.ec2_backend.remove_launch_permission(
ami_id, user_ids=user_ids, group=group
)
return MODIFY_IMAGE_ATTRIBUTE_RESPONSE
self.error_on_dryrun()
if operation_type == "add":
self.ec2_backend.add_launch_permission(
ami_id, user_ids=user_ids, group=group
)
elif operation_type == "remove":
self.ec2_backend.remove_launch_permission(
ami_id, user_ids=user_ids, group=group
)
return MODIFY_IMAGE_ATTRIBUTE_RESPONSE
def register_image(self):
name = self.querystring.get("Name")[0]
description = self._get_param("Description", if_none="")
if self.is_not_dryrun("RegisterImage"):
image = self.ec2_backend.register_image(name, description)
template = self.response_template(REGISTER_IMAGE_RESPONSE)
return template.render(image=image)
self.error_on_dryrun()
image = self.ec2_backend.register_image(name, description)
template = self.response_template(REGISTER_IMAGE_RESPONSE)
return template.render(image=image)
def reset_image_attribute(self):
if self.is_not_dryrun("ResetImageAttribute"):
raise NotImplementedError(
"AMIs.reset_image_attribute is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError("AMIs.reset_image_attribute is not yet implemented")
CREATE_IMAGE_RESPONSE = """<CreateImageResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">

180
moto/ec2/responses/elastic_block_store.py
View File

@ -6,12 +6,12 @@ class ElasticBlockStore(EC2BaseResponse):
volume_id = self._get_param("VolumeId")
instance_id = self._get_param("InstanceId")
device_path = self._get_param("Device")
if self.is_not_dryrun("AttachVolume"):
attachment = self.ec2_backend.attach_volume(
volume_id, instance_id, device_path
)
template = self.response_template(ATTACHED_VOLUME_RESPONSE)
return template.render(attachment=attachment)
self.error_on_dryrun()
attachment = self.ec2_backend.attach_volume(volume_id, instance_id, device_path)
template = self.response_template(ATTACHED_VOLUME_RESPONSE)
return template.render(attachment=attachment)
def copy_snapshot(self):
source_snapshot_id = self._get_param("SourceSnapshotId")
@ -19,24 +19,28 @@ class ElasticBlockStore(EC2BaseResponse):
description = self._get_param("Description")
tags = self._parse_tag_specification()
snapshot_tags = tags.get("snapshot", {})
if self.is_not_dryrun("CopySnapshot"):
snapshot = self.ec2_backend.copy_snapshot(
source_snapshot_id, source_region, description
)
snapshot.add_tags(snapshot_tags)
template = self.response_template(COPY_SNAPSHOT_RESPONSE)
return template.render(snapshot=snapshot)
self.error_on_dryrun()
snapshot = self.ec2_backend.copy_snapshot(
source_snapshot_id, source_region, description
)
snapshot.add_tags(snapshot_tags)
template = self.response_template(COPY_SNAPSHOT_RESPONSE)
return template.render(snapshot=snapshot)
def create_snapshot(self):
volume_id = self._get_param("VolumeId")
description = self._get_param("Description")
tags = self._parse_tag_specification()
snapshot_tags = tags.get("snapshot", {})
if self.is_not_dryrun("CreateSnapshot"):
snapshot = self.ec2_backend.create_snapshot(volume_id, description)
snapshot.add_tags(snapshot_tags)
template = self.response_template(CREATE_SNAPSHOT_RESPONSE)
return template.render(snapshot=snapshot)
self.error_on_dryrun()
snapshot = self.ec2_backend.create_snapshot(volume_id, description)
snapshot.add_tags(snapshot_tags)
template = self.response_template(CREATE_SNAPSHOT_RESPONSE)
return template.render(snapshot=snapshot)
def create_snapshots(self):
params = self._get_params()
@ -45,12 +49,13 @@ class ElasticBlockStore(EC2BaseResponse):
tags = self._parse_tag_specification()
snapshot_tags = tags.get("snapshot", {})
if self.is_not_dryrun("CreateSnapshots"):
snapshots = self.ec2_backend.create_snapshots(
instance_spec, description, snapshot_tags
)
template = self.response_template(CREATE_SNAPSHOTS_RESPONSE)
return template.render(snapshots=snapshots)
self.error_on_dryrun()
snapshots = self.ec2_backend.create_snapshots(
instance_spec, description, snapshot_tags
)
template = self.response_template(CREATE_SNAPSHOTS_RESPONSE)
return template.render(snapshots=snapshots)
def create_volume(self):
size = self._get_param("Size")
@ -62,31 +67,34 @@ class ElasticBlockStore(EC2BaseResponse):
encrypted = self._get_bool_param("Encrypted", if_none=False)
kms_key_id = self._get_param("KmsKeyId")
iops = self._get_param("Iops")
if self.is_not_dryrun("CreateVolume"):
volume = self.ec2_backend.create_volume(
size=size,
zone_name=zone,
snapshot_id=snapshot_id,
encrypted=encrypted,
kms_key_id=kms_key_id,
volume_type=volume_type,
iops=iops,
)
volume.add_tags(volume_tags)
template = self.response_template(CREATE_VOLUME_RESPONSE)
return template.render(volume=volume)
self.error_on_dryrun()
volume = self.ec2_backend.create_volume(
size=size,
zone_name=zone,
snapshot_id=snapshot_id,
encrypted=encrypted,
kms_key_id=kms_key_id,
volume_type=volume_type,
iops=iops,
)
volume.add_tags(volume_tags)
template = self.response_template(CREATE_VOLUME_RESPONSE)
return template.render(volume=volume)
def modify_volume(self):
volume_id = self._get_param("VolumeId")
target_size = self._get_param("Size")
target_volume_type = self._get_param("VolumeType")
if self.is_not_dryrun("ModifyVolume"):
volume = self.ec2_backend.modify_volume(
volume_id, target_size, target_volume_type
)
template = self.response_template(MODIFY_VOLUME_RESPONSE)
return template.render(volume=volume)
self.error_on_dryrun()
volume = self.ec2_backend.modify_volume(
volume_id, target_size, target_volume_type
)
template = self.response_template(MODIFY_VOLUME_RESPONSE)
return template.render(volume=volume)
def describe_volumes_modifications(self):
filters = self._filters_from_querystring()
@ -99,15 +107,19 @@ class ElasticBlockStore(EC2BaseResponse):
def delete_snapshot(self):
snapshot_id = self._get_param("SnapshotId")
if self.is_not_dryrun("DeleteSnapshot"):
self.ec2_backend.delete_snapshot(snapshot_id)
return DELETE_SNAPSHOT_RESPONSE
self.error_on_dryrun()
self.ec2_backend.delete_snapshot(snapshot_id)
return DELETE_SNAPSHOT_RESPONSE
def delete_volume(self):
volume_id = self._get_param("VolumeId")
if self.is_not_dryrun("DeleteVolume"):
self.ec2_backend.delete_volume(volume_id)
return DELETE_VOLUME_RESPONSE
self.error_on_dryrun()
self.ec2_backend.delete_volume(volume_id)
return DELETE_VOLUME_RESPONSE
def describe_snapshots(self):
filters = self._filters_from_querystring()
@ -141,24 +153,26 @@ class ElasticBlockStore(EC2BaseResponse):
volume_id = self._get_param("VolumeId")
instance_id = self._get_param("InstanceId")
device_path = self._get_param("Device")
if self.is_not_dryrun("DetachVolume"):
attachment = self.ec2_backend.detach_volume(
volume_id, instance_id, device_path
)
template = self.response_template(DETATCH_VOLUME_RESPONSE)
return template.render(attachment=attachment)
self.error_on_dryrun()
attachment = self.ec2_backend.detach_volume(volume_id, instance_id, device_path)
template = self.response_template(DETATCH_VOLUME_RESPONSE)
return template.render(attachment=attachment)
def enable_volume_io(self):
if self.is_not_dryrun("EnableVolumeIO"):
raise NotImplementedError(
"ElasticBlockStore.enable_volume_io is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"ElasticBlockStore.enable_volume_io is not yet implemented"
)
def import_volume(self):
if self.is_not_dryrun("ImportVolume"):
raise NotImplementedError(
"ElasticBlockStore.import_volume is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"ElasticBlockStore.import_volume is not yet implemented"
)
def describe_snapshot_attribute(self):
snapshot_id = self._get_param("SnapshotId")
@ -172,28 +186,32 @@ class ElasticBlockStore(EC2BaseResponse):
operation_type = self._get_param("OperationType")
groups = self._get_multi_param("UserGroup")
user_ids = self._get_multi_param("UserId")
if self.is_not_dryrun("ModifySnapshotAttribute"):
if operation_type == "add":
self.ec2_backend.add_create_volume_permission(
snapshot_id, user_ids=user_ids, groups=groups
)
elif operation_type == "remove":
self.ec2_backend.remove_create_volume_permission(
snapshot_id, user_ids=user_ids, groups=groups
)
return MODIFY_SNAPSHOT_ATTRIBUTE_RESPONSE
self.error_on_dryrun()
if operation_type == "add":
self.ec2_backend.add_create_volume_permission(
snapshot_id, user_ids=user_ids, groups=groups
)
elif operation_type == "remove":
self.ec2_backend.remove_create_volume_permission(
snapshot_id, user_ids=user_ids, groups=groups
)
return MODIFY_SNAPSHOT_ATTRIBUTE_RESPONSE
def modify_volume_attribute(self):
if self.is_not_dryrun("ModifyVolumeAttribute"):
raise NotImplementedError(
"ElasticBlockStore.modify_volume_attribute is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"ElasticBlockStore.modify_volume_attribute is not yet implemented"
)
def reset_snapshot_attribute(self):
if self.is_not_dryrun("ResetSnapshotAttribute"):
raise NotImplementedError(
"ElasticBlockStore.reset_snapshot_attribute is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"ElasticBlockStore.reset_snapshot_attribute is not yet implemented"
)
CREATE_VOLUME_RESPONSE = """<CreateVolumeResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">

132
moto/ec2/responses/elastic_ip_addresses.py
View File

@ -9,15 +9,16 @@ class ElasticIPAddresses(EC2BaseResponse):
tags = self._get_multi_param("TagSpecification")
tags = add_tag_specification(tags)
if self.is_not_dryrun("AllocateAddress"):
if reallocate_address:
address = self.ec2_backend.allocate_address(
domain, address=reallocate_address, tags=tags
)
else:
address = self.ec2_backend.allocate_address(domain, tags=tags)
template = self.response_template(ALLOCATE_ADDRESS_RESPONSE)
return template.render(address=address)
self.error_on_dryrun()
if reallocate_address:
address = self.ec2_backend.allocate_address(
domain, address=reallocate_address, tags=tags
)
else:
address = self.ec2_backend.allocate_address(domain, tags=tags)
template = self.response_template(ALLOCATE_ADDRESS_RESPONSE)
return template.render(address=address)
def associate_address(self):
instance = eni = None
@ -38,35 +39,36 @@ class ElasticIPAddresses(EC2BaseResponse):
if "AllowReassociation" in self.querystring:
reassociate = self._get_param("AllowReassociation") == "true"
if self.is_not_dryrun("AssociateAddress"):
if instance or eni:
if "PublicIp" in self.querystring:
eip = self.ec2_backend.associate_address(
instance=instance,
eni=eni,
address=self._get_param("PublicIp"),
reassociate=reassociate,
)
elif "AllocationId" in self.querystring:
eip = self.ec2_backend.associate_address(
instance=instance,
eni=eni,
allocation_id=self._get_param("AllocationId"),
reassociate=reassociate,
)
else:
self.ec2_backend.raise_error(
"MissingParameter",
"Invalid request, expect PublicIp/AllocationId parameter.",
)
self.error_on_dryrun()
if instance or eni:
if "PublicIp" in self.querystring:
eip = self.ec2_backend.associate_address(
instance=instance,
eni=eni,
address=self._get_param("PublicIp"),
reassociate=reassociate,
)
elif "AllocationId" in self.querystring:
eip = self.ec2_backend.associate_address(
instance=instance,
eni=eni,
allocation_id=self._get_param("AllocationId"),
reassociate=reassociate,
)
else:
self.ec2_backend.raise_error(
"MissingParameter",
"Invalid request, expect either instance or ENI.",
"Invalid request, expect PublicIp/AllocationId parameter.",
)
else:
self.ec2_backend.raise_error(
"MissingParameter",
"Invalid request, expect either instance or ENI.",
)
template = self.response_template(ASSOCIATE_ADDRESS_RESPONSE)
return template.render(address=eip)
template = self.response_template(ASSOCIATE_ADDRESS_RESPONSE)
return template.render(address=eip)
def describe_addresses(self):
self.error_on_dryrun()
@ -80,38 +82,46 @@ class ElasticIPAddresses(EC2BaseResponse):
return template.render(addresses=addresses)
def disassociate_address(self):
if self.is_not_dryrun("DisAssociateAddress"):
if "PublicIp" in self.querystring:
self.ec2_backend.disassociate_address(
address=self._get_param("PublicIp")
)
elif "AssociationId" in self.querystring:
self.ec2_backend.disassociate_address(
association_id=self._get_param("AssociationId")
)
else:
self.ec2_backend.raise_error(
"MissingParameter",
"Invalid request, expect PublicIp/AssociationId parameter.",
)
if (
"PublicIp" not in self.querystring
and "AssociationId" not in self.querystring
):
self.ec2_backend.raise_error(
"MissingParameter",
"Invalid request, expect PublicIp/AssociationId parameter.",
)
return self.response_template(DISASSOCIATE_ADDRESS_RESPONSE).render()
self.error_on_dryrun()
if "PublicIp" in self.querystring:
self.ec2_backend.disassociate_address(address=self._get_param("PublicIp"))
elif "AssociationId" in self.querystring:
self.ec2_backend.disassociate_address(
association_id=self._get_param("AssociationId")
)
return self.response_template(DISASSOCIATE_ADDRESS_RESPONSE).render()
def release_address(self):
if self.is_not_dryrun("ReleaseAddress"):
if "PublicIp" in self.querystring:
self.ec2_backend.release_address(address=self._get_param("PublicIp"))
elif "AllocationId" in self.querystring:
self.ec2_backend.release_address(
allocation_id=self._get_param("AllocationId")
)
else:
self.ec2_backend.raise_error(
"MissingParameter",
"Invalid request, expect PublicIp/AllocationId parameter.",
)
if (
"PublicIp" not in self.querystring
and "AllocationId" not in self.querystring
):
self.ec2_backend.raise_error(
"MissingParameter",
"Invalid request, expect PublicIp/AllocationId parameter.",
)
return self.response_template(RELEASE_ADDRESS_RESPONSE).render()
self.error_on_dryrun()
if "PublicIp" in self.querystring:
self.ec2_backend.release_address(address=self._get_param("PublicIp"))
elif "AllocationId" in self.querystring:
self.ec2_backend.release_address(
allocation_id=self._get_param("AllocationId")
)
return self.response_template(RELEASE_ADDRESS_RESPONSE).render()
ALLOCATE_ADDRESS_RESPONSE = """<AllocateAddressResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">

136
moto/ec2/responses/elastic_network_interfaces.py
View File

@ -17,96 +17,110 @@ class ElasticNetworkInterfaces(EC2BaseResponse):
tags = self._get_multi_param("TagSpecification")
tags = add_tag_specification(tags)
if self.is_not_dryrun("CreateNetworkInterface"):
eni = self.ec2_backend.create_network_interface(
subnet,
private_ip_address,
private_ip_addresses,
groups,
description,
tags,
secondary_ips_count=secondary_ips_count,
ipv6_addresses=ipv6_addresses,
ipv6_address_count=ipv6_address_count,
)
template = self.response_template(CREATE_NETWORK_INTERFACE_RESPONSE)
return template.render(eni=eni)
self.error_on_dryrun()
eni = self.ec2_backend.create_network_interface(
subnet,
private_ip_address,
private_ip_addresses,
groups,
description,
tags,
secondary_ips_count=secondary_ips_count,
ipv6_addresses=ipv6_addresses,
ipv6_address_count=ipv6_address_count,
)
template = self.response_template(CREATE_NETWORK_INTERFACE_RESPONSE)
return template.render(eni=eni)
def delete_network_interface(self):
eni_id = self._get_param("NetworkInterfaceId")
if self.is_not_dryrun("DeleteNetworkInterface"):
self.ec2_backend.delete_network_interface(eni_id)
template = self.response_template(DELETE_NETWORK_INTERFACE_RESPONSE)
return template.render()
self.error_on_dryrun()
self.ec2_backend.delete_network_interface(eni_id)
template = self.response_template(DELETE_NETWORK_INTERFACE_RESPONSE)
return template.render()
def describe_network_interface_attribute(self):
eni_id = self._get_param("NetworkInterfaceId")
attribute = self._get_param("Attribute")
if self.is_not_dryrun("DescribeNetworkInterfaceAttribute"):
eni = self.ec2_backend.get_all_network_interfaces([eni_id])[0]
if attribute == "description":
template = self.response_template(
DESCRIBE_NETWORK_INTERFACE_ATTRIBUTE_RESPONSE_DESCRIPTION
)
elif attribute == "groupSet":
template = self.response_template(
DESCRIBE_NETWORK_INTERFACE_ATTRIBUTE_RESPONSE_GROUPSET
)
elif attribute == "sourceDestCheck":
template = self.response_template(
DESCRIBE_NETWORK_INTERFACE_ATTRIBUTE_RESPONSE_SOURCEDESTCHECK
)
elif attribute == "attachment":
template = self.response_template(
DESCRIBE_NETWORK_INTERFACE_ATTRIBUTE_RESPONSE_ATTACHMENT
)
else:
raise InvalidParameterValueErrorUnknownAttribute(attribute)
self.error_on_dryrun()
eni = self.ec2_backend.get_all_network_interfaces([eni_id])[0]
if attribute == "description":
template = self.response_template(
DESCRIBE_NETWORK_INTERFACE_ATTRIBUTE_RESPONSE_DESCRIPTION
)
elif attribute == "groupSet":
template = self.response_template(
DESCRIBE_NETWORK_INTERFACE_ATTRIBUTE_RESPONSE_GROUPSET
)
elif attribute == "sourceDestCheck":
template = self.response_template(
DESCRIBE_NETWORK_INTERFACE_ATTRIBUTE_RESPONSE_SOURCEDESTCHECK
)
elif attribute == "attachment":
template = self.response_template(
DESCRIBE_NETWORK_INTERFACE_ATTRIBUTE_RESPONSE_ATTACHMENT
)
else:
raise InvalidParameterValueErrorUnknownAttribute(attribute)
return template.render(eni=eni)
def describe_network_interfaces(self):
eni_ids = self._get_multi_param("NetworkInterfaceId")
filters = self._filters_from_querystring()
if self.is_not_dryrun("DescribeNetworkInterfaces"):
enis = self.ec2_backend.get_all_network_interfaces(eni_ids, filters)
template = self.response_template(DESCRIBE_NETWORK_INTERFACES_RESPONSE)
return template.render(enis=enis)
self.error_on_dryrun()
enis = self.ec2_backend.get_all_network_interfaces(eni_ids, filters)
template = self.response_template(DESCRIBE_NETWORK_INTERFACES_RESPONSE)
return template.render(enis=enis)
def attach_network_interface(self):
eni_id = self._get_param("NetworkInterfaceId")
instance_id = self._get_param("InstanceId")
device_index = self._get_param("DeviceIndex")
if self.is_not_dryrun("AttachNetworkInterface"):
attachment_id = self.ec2_backend.attach_network_interface(
eni_id, instance_id, device_index
)
template = self.response_template(ATTACH_NETWORK_INTERFACE_RESPONSE)
return template.render(attachment_id=attachment_id)
self.error_on_dryrun()
attachment_id = self.ec2_backend.attach_network_interface(
eni_id, instance_id, device_index
)
template = self.response_template(ATTACH_NETWORK_INTERFACE_RESPONSE)
return template.render(attachment_id=attachment_id)
def detach_network_interface(self):
attachment_id = self._get_param("AttachmentId")
if self.is_not_dryrun("DetachNetworkInterface"):
self.ec2_backend.detach_network_interface(attachment_id)
template = self.response_template(DETACH_NETWORK_INTERFACE_RESPONSE)
return template.render()
self.error_on_dryrun()
self.ec2_backend.detach_network_interface(attachment_id)
template = self.response_template(DETACH_NETWORK_INTERFACE_RESPONSE)
return template.render()
def modify_network_interface_attribute(self):
eni_id = self._get_param("NetworkInterfaceId")
group_ids = self._get_multi_param("SecurityGroupId")
source_dest_check = get_attribute_value("SourceDestCheck", self.querystring)
description = get_attribute_value("Description", self.querystring)
if self.is_not_dryrun("ModifyNetworkInterface"):
self.ec2_backend.modify_network_interface_attribute(
eni_id, group_ids, source_dest_check, description
)
return MODIFY_NETWORK_INTERFACE_ATTRIBUTE_RESPONSE
self.error_on_dryrun()
self.ec2_backend.modify_network_interface_attribute(
eni_id, group_ids, source_dest_check, description
)
return MODIFY_NETWORK_INTERFACE_ATTRIBUTE_RESPONSE
def reset_network_interface_attribute(self):
if self.is_not_dryrun("ResetNetworkInterface"):
raise NotImplementedError(
"ElasticNetworkInterfaces(AmazonVPC).reset_network_interface_attribute is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"ElasticNetworkInterfaces(AmazonVPC).reset_network_interface_attribute is not yet implemented"
)
def assign_private_ip_addresses(self):
eni_id = self._get_param("NetworkInterfaceId")

49
moto/ec2/responses/flow_logs.py
View File

@ -17,37 +17,42 @@ class FlowLogs(EC2BaseResponse):
tags = self._parse_tag_specification()
tags = tags.get("vpc-flow-log", {})
if self.is_not_dryrun("CreateFlowLogs"):
flow_logs, errors = self.ec2_backend.create_flow_logs(
resource_type=resource_type,
resource_ids=resource_ids,
traffic_type=traffic_type,
deliver_logs_permission_arn=deliver_logs_permission_arn,
log_destination_type=log_destination_type,
log_destination=log_destination,
log_group_name=log_group_name,
log_format=log_format,
max_aggregation_interval=max_aggregation_interval,
)
for fl in flow_logs:
fl.add_tags(tags)
template = self.response_template(CREATE_FLOW_LOGS_RESPONSE)
return template.render(flow_logs=flow_logs, errors=errors)
self.error_on_dryrun()
flow_logs, errors = self.ec2_backend.create_flow_logs(
resource_type=resource_type,
resource_ids=resource_ids,
traffic_type=traffic_type,
deliver_logs_permission_arn=deliver_logs_permission_arn,
log_destination_type=log_destination_type,
log_destination=log_destination,
log_group_name=log_group_name,
log_format=log_format,
max_aggregation_interval=max_aggregation_interval,
)
for fl in flow_logs:
fl.add_tags(tags)
template = self.response_template(CREATE_FLOW_LOGS_RESPONSE)
return template.render(flow_logs=flow_logs, errors=errors)
def describe_flow_logs(self):
flow_log_ids = self._get_multi_param("FlowLogId")
filters = self._filters_from_querystring()
flow_logs = self.ec2_backend.describe_flow_logs(flow_log_ids, filters)
if self.is_not_dryrun("DescribeFlowLogs"):
template = self.response_template(DESCRIBE_FLOW_LOGS_RESPONSE)
return template.render(flow_logs=flow_logs)
self.error_on_dryrun()
template = self.response_template(DESCRIBE_FLOW_LOGS_RESPONSE)
return template.render(flow_logs=flow_logs)
def delete_flow_logs(self):
flow_log_ids = self._get_multi_param("FlowLogId")
self.error_on_dryrun()
self.ec2_backend.delete_flow_logs(flow_log_ids)
if self.is_not_dryrun("DeleteFlowLogs"):
template = self.response_template(DELETE_FLOW_LOGS_RESPONSE)
return template.render()
return self.response_template(DELETE_FLOW_LOGS_RESPONSE).render()
CREATE_FLOW_LOGS_RESPONSE = """

142
moto/ec2/responses/instances.py
View File

@ -87,62 +87,70 @@ class InstanceResponse(EC2BaseResponse):
if mappings:
kwargs["block_device_mappings"] = mappings
if self.is_not_dryrun("RunInstance"):
new_reservation = self.ec2_backend.add_instances(
image_id, min_count, user_data, security_group_names, **kwargs
)
if kwargs.get("iam_instance_profile_name"):
self.ec2_backend.associate_iam_instance_profile(
instance_id=new_reservation.instances[0].id,
iam_instance_profile_name=kwargs.get("iam_instance_profile_name"),
)
if kwargs.get("iam_instance_profile_arn"):
self.ec2_backend.associate_iam_instance_profile(
instance_id=new_reservation.instances[0].id,
iam_instance_profile_arn=kwargs.get("iam_instance_profile_arn"),
)
self.error_on_dryrun()
template = self.response_template(EC2_RUN_INSTANCES)
return template.render(
account_id=self.current_account, reservation=new_reservation
new_reservation = self.ec2_backend.add_instances(
image_id, min_count, user_data, security_group_names, **kwargs
)
if kwargs.get("iam_instance_profile_name"):
self.ec2_backend.associate_iam_instance_profile(
instance_id=new_reservation.instances[0].id,
iam_instance_profile_name=kwargs.get("iam_instance_profile_name"),
)
if kwargs.get("iam_instance_profile_arn"):
self.ec2_backend.associate_iam_instance_profile(
instance_id=new_reservation.instances[0].id,
iam_instance_profile_arn=kwargs.get("iam_instance_profile_arn"),
)
template = self.response_template(EC2_RUN_INSTANCES)
return template.render(
account_id=self.current_account, reservation=new_reservation
)
def terminate_instances(self):
instance_ids = self._get_multi_param("InstanceId")
if self.is_not_dryrun("TerminateInstance"):
instances = self.ec2_backend.terminate_instances(instance_ids)
from moto.autoscaling import autoscaling_backends
from moto.elbv2 import elbv2_backends
autoscaling_backends[self.current_account][
self.region
].notify_terminate_instances(instance_ids)
elbv2_backends[self.current_account][
self.region
].notify_terminate_instances(instance_ids)
template = self.response_template(EC2_TERMINATE_INSTANCES)
return template.render(instances=instances)
self.error_on_dryrun()
instances = self.ec2_backend.terminate_instances(instance_ids)
from moto.autoscaling import autoscaling_backends
from moto.elbv2 import elbv2_backends
autoscaling_backends[self.current_account][
self.region
].notify_terminate_instances(instance_ids)
elbv2_backends[self.current_account][self.region].notify_terminate_instances(
instance_ids
)
template = self.response_template(EC2_TERMINATE_INSTANCES)
return template.render(instances=instances)
def reboot_instances(self):
instance_ids = self._get_multi_param("InstanceId")
if self.is_not_dryrun("RebootInstance"):
instances = self.ec2_backend.reboot_instances(instance_ids)
template = self.response_template(EC2_REBOOT_INSTANCES)
return template.render(instances=instances)
self.error_on_dryrun()
instances = self.ec2_backend.reboot_instances(instance_ids)
template = self.response_template(EC2_REBOOT_INSTANCES)
return template.render(instances=instances)
def stop_instances(self):
instance_ids = self._get_multi_param("InstanceId")
if self.is_not_dryrun("StopInstance"):
instances = self.ec2_backend.stop_instances(instance_ids)
template = self.response_template(EC2_STOP_INSTANCES)
return template.render(instances=instances)
self.error_on_dryrun()
instances = self.ec2_backend.stop_instances(instance_ids)
template = self.response_template(EC2_STOP_INSTANCES)
return template.render(instances=instances)
def start_instances(self):
instance_ids = self._get_multi_param("InstanceId")
if self.is_not_dryrun("StartInstance"):
instances = self.ec2_backend.start_instances(instance_ids)
template = self.response_template(EC2_START_INSTANCES)
return template.render(instances=instances)
self.error_on_dryrun()
instances = self.ec2_backend.start_instances(instance_ids)
template = self.response_template(EC2_START_INSTANCES)
return template.render(instances=instances)
def _get_list_of_dict_params(self, param_prefix, _dct):
"""
@ -268,9 +276,10 @@ class InstanceResponse(EC2BaseResponse):
instance_id = self._get_param("InstanceId")
instance = self.ec2_backend.get_instance(instance_id)
if self.is_not_dryrun("ModifyInstanceAttribute"):
block_device_type = instance.block_device_mapping[device_name_value]
block_device_type.delete_on_termination = del_on_term_value
self.error_on_dryrun()
block_device_type = instance.block_device_mapping[device_name_value]
block_device_type.delete_on_termination = del_on_term_value
# +1 for the next device
mapping_counter += 1
@ -288,14 +297,15 @@ class InstanceResponse(EC2BaseResponse):
if not attribute_key:
return
if self.is_not_dryrun("Modify" + attribute_key.split(".")[0]):
value = self.querystring.get(attribute_key)[0]
normalized_attribute = camelcase_to_underscores(attribute_key.split(".")[0])
instance_id = self._get_param("InstanceId")
self.ec2_backend.modify_instance_attribute(
instance_id, normalized_attribute, value
)
return EC2_MODIFY_INSTANCE_ATTRIBUTE
self.error_on_dryrun()
value = self.querystring.get(attribute_key)[0]
normalized_attribute = camelcase_to_underscores(attribute_key.split(".")[0])
instance_id = self._get_param("InstanceId")
self.ec2_backend.modify_instance_attribute(
instance_id, normalized_attribute, value
)
return EC2_MODIFY_INSTANCE_ATTRIBUTE
def _attribute_value_handler(self):
attribute_key = self._get_param("Attribute")
@ -303,14 +313,15 @@ class InstanceResponse(EC2BaseResponse):
if attribute_key is None:
return
if self.is_not_dryrun("ModifyInstanceAttribute"):
value = self._get_param("Value")
normalized_attribute = camelcase_to_underscores(attribute_key)
instance_id = self._get_param("InstanceId")
self.ec2_backend.modify_instance_attribute(
instance_id, normalized_attribute, value
)
return EC2_MODIFY_INSTANCE_ATTRIBUTE
self.error_on_dryrun()
value = self._get_param("Value")
normalized_attribute = camelcase_to_underscores(attribute_key)
instance_id = self._get_param("InstanceId")
self.ec2_backend.modify_instance_attribute(
instance_id, normalized_attribute, value
)
return EC2_MODIFY_INSTANCE_ATTRIBUTE
def _security_grp_instance_attribute_handler(self):
new_security_grp_list = []
@ -319,11 +330,12 @@ class InstanceResponse(EC2BaseResponse):
new_security_grp_list.append(self.querystring.get(key)[0])
instance_id = self._get_param("InstanceId")
if self.is_not_dryrun("ModifyInstanceSecurityGroups"):
self.ec2_backend.modify_instance_security_groups(
instance_id, new_security_grp_list
)
return EC2_MODIFY_INSTANCE_ATTRIBUTE
self.error_on_dryrun()
self.ec2_backend.modify_instance_security_groups(
instance_id, new_security_grp_list
)
return EC2_MODIFY_INSTANCE_ATTRIBUTE
def _parse_block_device_mapping(self):
device_mappings = self._get_list_prefix("BlockDeviceMapping")

43
moto/ec2/responses/internet_gateways.py
View File

@ -5,28 +5,29 @@ class InternetGateways(EC2BaseResponse):
def attach_internet_gateway(self):
igw_id = self._get_param("InternetGatewayId")
vpc_id = self._get_param("VpcId")
if self.is_not_dryrun("AttachInternetGateway"):
self.ec2_backend.attach_internet_gateway(igw_id, vpc_id)
template = self.response_template(ATTACH_INTERNET_GATEWAY_RESPONSE)
return template.render()
self.error_on_dryrun()
self.ec2_backend.attach_internet_gateway(igw_id, vpc_id)
return self.response_template(ATTACH_INTERNET_GATEWAY_RESPONSE).render()
def create_internet_gateway(self):
if self.is_not_dryrun("CreateInternetGateway"):
tags = self._get_multi_param(
"TagSpecification", skip_result_conversion=True
)
if tags:
tags = tags[0].get("Tag") or []
igw = self.ec2_backend.create_internet_gateway(tags=tags)
template = self.response_template(CREATE_INTERNET_GATEWAY_RESPONSE)
return template.render(internet_gateway=igw)
self.error_on_dryrun()
tags = self._get_multi_param("TagSpecification", skip_result_conversion=True)
if tags:
tags = tags[0].get("Tag") or []
igw = self.ec2_backend.create_internet_gateway(tags=tags)
return self.response_template(CREATE_INTERNET_GATEWAY_RESPONSE).render(
internet_gateway=igw
)
def delete_internet_gateway(self):
igw_id = self._get_param("InternetGatewayId")
if self.is_not_dryrun("DeleteInternetGateway"):
self.ec2_backend.delete_internet_gateway(igw_id)
template = self.response_template(DELETE_INTERNET_GATEWAY_RESPONSE)
return template.render()
self.error_on_dryrun()
self.ec2_backend.delete_internet_gateway(igw_id)
return self.response_template(DELETE_INTERNET_GATEWAY_RESPONSE).render()
def describe_internet_gateways(self):
filter_dict = self._filters_from_querystring()
@ -46,10 +47,10 @@ class InternetGateways(EC2BaseResponse):
# raise else DependencyViolationError()
igw_id = self._get_param("InternetGatewayId")
vpc_id = self._get_param("VpcId")
if self.is_not_dryrun("DetachInternetGateway"):
self.ec2_backend.detach_internet_gateway(igw_id, vpc_id)
template = self.response_template(DETACH_INTERNET_GATEWAY_RESPONSE)
return template.render()
self.error_on_dryrun()
self.ec2_backend.detach_internet_gateway(igw_id, vpc_id)
return self.response_template(DETACH_INTERNET_GATEWAY_RESPONSE).render()
ATTACH_INTERNET_GATEWAY_RESPONSE = """<AttachInternetGatewayResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">

18
moto/ec2/responses/ip_addresses.py
View File

@ -3,13 +3,15 @@ from moto.core.responses import BaseResponse
class IPAddresses(BaseResponse):
def assign_private_ip_addresses(self):
if self.is_not_dryrun("AssignPrivateIPAddress"):
raise NotImplementedError(
"IPAddresses.assign_private_ip_addresses is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"IPAddresses.assign_private_ip_addresses is not yet implemented"
)
def unassign_private_ip_addresses(self):
if self.is_not_dryrun("UnAssignPrivateIPAddress"):
raise NotImplementedError(
"IPAddresses.unassign_private_ip_addresses is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"IPAddresses.unassign_private_ip_addresses is not yet implemented"
)

23
moto/ec2/responses/key_pairs.py
View File

@ -4,16 +4,17 @@ from ._base_response import EC2BaseResponse
class KeyPairs(EC2BaseResponse):
def create_key_pair(self):
name = self._get_param("KeyName")
if self.is_not_dryrun("CreateKeyPair"):
keypair = self.ec2_backend.create_key_pair(name)
template = self.response_template(CREATE_KEY_PAIR_RESPONSE)
return template.render(keypair=keypair)
self.error_on_dryrun()
keypair = self.ec2_backend.create_key_pair(name)
return self.response_template(CREATE_KEY_PAIR_RESPONSE).render(keypair=keypair)
def delete_key_pair(self):
name = self._get_param("KeyName")
if self.is_not_dryrun("DeleteKeyPair"):
self.ec2_backend.delete_key_pair(name)
return self.response_template(DELETE_KEY_PAIR_RESPONSE).render()
self.error_on_dryrun()
self.ec2_backend.delete_key_pair(name)
return self.response_template(DELETE_KEY_PAIR_RESPONSE).render()
def describe_key_pairs(self):
names = self._get_multi_param("KeyName")
@ -25,10 +26,10 @@ class KeyPairs(EC2BaseResponse):
def import_key_pair(self):
name = self._get_param("KeyName")
material = self._get_param("PublicKeyMaterial")
if self.is_not_dryrun("ImportKeyPair"):
keypair = self.ec2_backend.import_key_pair(name, material)
template = self.response_template(IMPORT_KEYPAIR_RESPONSE)
return template.render(keypair=keypair)
self.error_on_dryrun()
keypair = self.ec2_backend.import_key_pair(name, material)
return self.response_template(IMPORT_KEYPAIR_RESPONSE).render(keypair=keypair)
DESCRIBE_KEY_PAIRS_RESPONSE = """<DescribeKeyPairsResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">

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

@ -48,95 +48,53 @@ def pretty_xml(tree):
return parsed.toprettyxml(indent=" ")
def parse_object(raw_data):
out_data = {}
for key, value in raw_data.items():
key_fix_splits = key.split("_")
key_len = len(key_fix_splits)
new_key = ""
for i in range(0, key_len):
new_key += key_fix_splits[i][0].upper() + key_fix_splits[i][1:]
data = out_data
splits = new_key.split(".")
for split in splits[:-1]:
if split not in data:
data[split] = {}
data = data[split]
data[splits[-1]] = value
out_data = parse_lists(out_data)
return out_data
def parse_lists(data):
for key, value in data.items():
if isinstance(value, dict):
keys = data[key].keys()
is_list = all(map(lambda k: k.isnumeric(), keys))
if is_list:
new_value = []
keys = sorted(list(keys))
for k in keys:
lvalue = value[k]
if isinstance(lvalue, dict):
lvalue = parse_lists(lvalue)
new_value.append(lvalue)
data[key] = new_value
return data
class LaunchTemplates(EC2BaseResponse):
def create_launch_template(self):
name = self._get_param("LaunchTemplateName")
version_description = self._get_param("VersionDescription")
tag_spec = self._parse_tag_specification()
raw_template_data = self._get_dict_param("LaunchTemplateData.")
parsed_template_data = parse_object(raw_template_data)
parsed_template_data = self._get_multi_param_dict("LaunchTemplateData")
if self.is_not_dryrun("CreateLaunchTemplate"):
if tag_spec:
if "TagSpecifications" not in parsed_template_data:
parsed_template_data["TagSpecifications"] = []
converted_tag_spec = []
for resource_type, tags in tag_spec.items():
converted_tag_spec.append(
{
"ResourceType": resource_type,
"Tags": [
{"Key": key, "Value": value}
for key, value in tags.items()
],
}
)
self.error_on_dryrun()
parsed_template_data["TagSpecifications"].extend(converted_tag_spec)
if tag_spec:
if "TagSpecifications" not in parsed_template_data:
parsed_template_data["TagSpecifications"] = []
converted_tag_spec = []
for resource_type, tags in tag_spec.items():
converted_tag_spec.append(
{
"ResourceType": resource_type,
"Tags": [
{"Key": key, "Value": value} for key, value in tags.items()
],
}
)
template = self.ec2_backend.create_launch_template(
name, version_description, parsed_template_data, tag_spec
)
version = template.default_version()
parsed_template_data["TagSpecifications"].extend(converted_tag_spec)
tree = xml_root("CreateLaunchTemplateResponse")
xml_serialize(
tree,
"launchTemplate",
{
"createTime": version.create_time,
"createdBy": f"arn:aws:iam::{self.current_account}:root",
"defaultVersionNumber": template.default_version_number,
"latestVersionNumber": version.number,
"launchTemplateId": template.id,
"launchTemplateName": template.name,
"tags": template.tags,
},
)
template = self.ec2_backend.create_launch_template(
name, version_description, parsed_template_data, tag_spec
)
version = template.default_version()
return pretty_xml(tree)
tree = xml_root("CreateLaunchTemplateResponse")
xml_serialize(
tree,
"launchTemplate",
{
"createTime": version.create_time,
"createdBy": f"arn:aws:iam::{self.current_account}:root",
"defaultVersionNumber": template.default_version_number,
"latestVersionNumber": version.number,
"launchTemplateId": template.id,
"launchTemplateName": template.name,
"tags": template.tags,
},
)
return pretty_xml(tree)
def create_launch_template_version(self):
name = self._get_param("LaunchTemplateName")
@ -148,51 +106,49 @@ class LaunchTemplates(EC2BaseResponse):
version_description = self._get_param("VersionDescription")
raw_template_data = self._get_dict_param("LaunchTemplateData.")
template_data = parse_object(raw_template_data)
template_data = self._get_multi_param_dict("LaunchTemplateData")
if self.is_not_dryrun("CreateLaunchTemplate"):
version = template.create_version(template_data, version_description)
self.error_on_dryrun()
tree = xml_root("CreateLaunchTemplateVersionResponse")
xml_serialize(
tree,
"launchTemplateVersion",
{
"createTime": version.create_time,
"createdBy": f"arn:aws:iam::{self.current_account}:root",
"defaultVersion": template.is_default(version),
"launchTemplateData": version.data,
"launchTemplateId": template.id,
"launchTemplateName": template.name,
"versionDescription": version.description,
"versionNumber": version.number,
},
)
return pretty_xml(tree)
version = template.create_version(template_data, version_description)
tree = xml_root("CreateLaunchTemplateVersionResponse")
xml_serialize(
tree,
"launchTemplateVersion",
{
"createTime": version.create_time,
"createdBy": f"arn:aws:iam::{self.current_account}:root",
"defaultVersion": template.is_default(version),
"launchTemplateData": version.data,
"launchTemplateId": template.id,
"launchTemplateName": template.name,
"versionDescription": version.description,
"versionNumber": version.number,
},
)
return pretty_xml(tree)
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)
self.error_on_dryrun()
tree = xml_root("DeleteLaunchTemplatesResponse")
xml_serialize(
tree,
"launchTemplate",
{
"defaultVersionNumber": template.default_version_number,
"launchTemplateId": template.id,
"launchTemplateName": template.name,
},
)
template = self.ec2_backend.delete_launch_template(name, tid)
return pretty_xml(tree)
tree = xml_root("DeleteLaunchTemplatesResponse")
xml_serialize(
tree,
"launchTemplate",
{
"defaultVersionNumber": template.default_version_number,
"launchTemplateId": template.id,
"launchTemplateName": template.name,
},
)
# def delete_launch_template_versions(self):
# pass
return pretty_xml(tree)
def describe_launch_template_versions(self):
name = self._get_param("LaunchTemplateName")
@ -213,57 +169,58 @@ class LaunchTemplates(EC2BaseResponse):
"all filters", "DescribeLaunchTemplateVersions"
)
if self.is_not_dryrun("DescribeLaunchTemplateVersions"):
tree = ElementTree.Element(
"DescribeLaunchTemplateVersionsResponse",
{"xmlns": "http://ec2.amazonaws.com/doc/2016-11-15/"},
)
request_id = ElementTree.SubElement(tree, "requestId")
request_id.text = "65cadec1-b364-4354-8ca8-4176dexample"
self.error_on_dryrun()
versions_node = ElementTree.SubElement(tree, "launchTemplateVersionSet")
tree = ElementTree.Element(
"DescribeLaunchTemplateVersionsResponse",
{"xmlns": "http://ec2.amazonaws.com/doc/2016-11-15/"},
)
request_id = ElementTree.SubElement(tree, "requestId")
request_id.text = "65cadec1-b364-4354-8ca8-4176dexample"
ret_versions = []
if versions:
for v in versions:
if str(v).lower() == "$latest" or "$default":
tv = template.get_version(v)
else:
tv = template.get_version(int(v))
ret_versions.append(tv)
elif min_version:
if max_version:
vMax = max_version
versions_node = ElementTree.SubElement(tree, "launchTemplateVersionSet")
ret_versions = []
if versions:
for v in versions:
if str(v).lower() == "$latest" or "$default":
tv = template.get_version(v)
else:
vMax = min_version + max_results
vMin = min_version - 1
ret_versions = template.versions[vMin:vMax]
elif max_version:
tv = template.get_version(int(v))
ret_versions.append(tv)
elif min_version:
if max_version:
vMax = max_version
ret_versions = template.versions[:vMax]
else:
ret_versions = template.versions
vMax = min_version + max_results
ret_versions = ret_versions[:max_results]
vMin = min_version - 1
ret_versions = template.versions[vMin:vMax]
elif max_version:
vMax = max_version
ret_versions = template.versions[:vMax]
else:
ret_versions = template.versions
for version in ret_versions:
xml_serialize(
versions_node,
"item",
{
"createTime": version.create_time,
"createdBy": f"arn:aws:iam::{self.current_account}:root",
"defaultVersion": True,
"launchTemplateData": version.data,
"launchTemplateId": template.id,
"launchTemplateName": template.name,
"versionDescription": version.description,
"versionNumber": version.number,
},
)
ret_versions = ret_versions[:max_results]
return pretty_xml(tree)
for version in ret_versions:
xml_serialize(
versions_node,
"item",
{
"createTime": version.create_time,
"createdBy": f"arn:aws:iam::{self.current_account}:root",
"defaultVersion": True,
"launchTemplateData": version.data,
"launchTemplateId": template.id,
"launchTemplateName": template.name,
"versionDescription": version.description,
"versionNumber": version.number,
},
)
return pretty_xml(tree)
def describe_launch_templates(self):
max_results = self._get_int_param("MaxResults", 15)
@ -271,34 +228,32 @@ class LaunchTemplates(EC2BaseResponse):
template_ids = self._get_multi_param("LaunchTemplateId")
filters = self._filters_from_querystring()
if self.is_not_dryrun("DescribeLaunchTemplates"):
tree = ElementTree.Element("DescribeLaunchTemplatesResponse")
templates_node = ElementTree.SubElement(tree, "launchTemplates")
self.error_on_dryrun()
templates = self.ec2_backend.describe_launch_templates(
template_names=template_names,
template_ids=template_ids,
filters=filters,
tree = ElementTree.Element("DescribeLaunchTemplatesResponse")
templates_node = ElementTree.SubElement(tree, "launchTemplates")
templates = self.ec2_backend.describe_launch_templates(
template_names=template_names,
template_ids=template_ids,
filters=filters,
)
templates = templates[:max_results]
for template in templates:
xml_serialize(
templates_node,
"item",
{
"createTime": template.create_time,
"createdBy": f"arn:aws:iam::{self.current_account}:root",
"defaultVersionNumber": template.default_version_number,
"latestVersionNumber": template.latest_version_number,
"launchTemplateId": template.id,
"launchTemplateName": template.name,
"tags": template.tags,
},
)
templates = templates[:max_results]
for template in templates:
xml_serialize(
templates_node,
"item",
{
"createTime": template.create_time,
"createdBy": f"arn:aws:iam::{self.current_account}:root",
"defaultVersionNumber": template.default_version_number,
"latestVersionNumber": template.latest_version_number,
"launchTemplateId": template.id,
"launchTemplateName": template.name,
"tags": template.tags,
},
)
return pretty_xml(tree)
# def modify_launch_template(self):
# pass
return pretty_xml(tree)

16
moto/ec2/responses/monitoring.py
View File

@ -3,13 +3,13 @@ from moto.core.responses import BaseResponse
class Monitoring(BaseResponse):
def monitor_instances(self):
if self.is_not_dryrun("MonitorInstances"):
raise NotImplementedError(
"Monitoring.monitor_instances is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError("Monitoring.monitor_instances is not yet implemented")
def unmonitor_instances(self):
if self.is_not_dryrun("UnMonitorInstances"):
raise NotImplementedError(
"Monitoring.unmonitor_instances is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"Monitoring.unmonitor_instances is not yet implemented"
)

18
moto/ec2/responses/placement_groups.py
View File

@ -3,16 +3,18 @@ from moto.core.responses import BaseResponse
class PlacementGroups(BaseResponse):
def create_placement_group(self):
if self.is_not_dryrun("CreatePlacementGroup"):
raise NotImplementedError(
"PlacementGroups.create_placement_group is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"PlacementGroups.create_placement_group is not yet implemented"
)
def delete_placement_group(self):
if self.is_not_dryrun("DeletePlacementGroup"):
raise NotImplementedError(
"PlacementGroups.delete_placement_group is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"PlacementGroups.delete_placement_group is not yet implemented"
)
def describe_placement_groups(self):
raise NotImplementedError(

27
moto/ec2/responses/reserved_instances.py
View File

@ -3,16 +3,18 @@ from moto.core.responses import BaseResponse
class ReservedInstances(BaseResponse):
def cancel_reserved_instances_listing(self):
if self.is_not_dryrun("CancelReservedInstances"):
raise NotImplementedError(
"ReservedInstances.cancel_reserved_instances_listing is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"ReservedInstances.cancel_reserved_instances_listing is not yet implemented"
)
def create_reserved_instances_listing(self):
if self.is_not_dryrun("CreateReservedInstances"):
raise NotImplementedError(
"ReservedInstances.create_reserved_instances_listing is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"ReservedInstances.create_reserved_instances_listing is not yet implemented"
)
def describe_reserved_instances(self):
raise NotImplementedError(
@ -30,7 +32,8 @@ class ReservedInstances(BaseResponse):
)
def purchase_reserved_instances_offering(self):
if self.is_not_dryrun("PurchaseReservedInstances"):
raise NotImplementedError(
"ReservedInstances.purchase_reserved_instances_offering is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"ReservedInstances.purchase_reserved_instances_offering is not yet implemented"
)

90
moto/ec2/responses/security_groups.py
View File

@ -128,20 +128,22 @@ class SecurityGroups(EC2BaseResponse):
)
def authorize_security_group_egress(self):
if self.is_not_dryrun("GrantSecurityGroupEgress"):
for args in self._process_rules_from_querystring():
rule, group = self.ec2_backend.authorize_security_group_egress(*args)
self.ec2_backend.sg_old_egress_ruls[group.id] = group.egress_rules.copy()
template = self.response_template(AUTHORIZE_SECURITY_GROUP_EGRESS_RESPONSE)
return template.render(rule=rule, group=group)
self.error_on_dryrun()
for args in self._process_rules_from_querystring():
rule, group = self.ec2_backend.authorize_security_group_egress(*args)
self.ec2_backend.sg_old_egress_ruls[group.id] = group.egress_rules.copy()
template = self.response_template(AUTHORIZE_SECURITY_GROUP_EGRESS_RESPONSE)
return template.render(rule=rule, group=group)
def authorize_security_group_ingress(self):
if self.is_not_dryrun("GrantSecurityGroupIngress"):
for args in self._process_rules_from_querystring():
rule, group = self.ec2_backend.authorize_security_group_ingress(*args)
self.ec2_backend.sg_old_ingress_ruls[group.id] = group.ingress_rules.copy()
template = self.response_template(AUTHORIZE_SECURITY_GROUP_INGRESS_RESPONSE)
return template.render(rule=rule, group=group)
self.error_on_dryrun()
for args in self._process_rules_from_querystring():
rule, group = self.ec2_backend.authorize_security_group_ingress(*args)
self.ec2_backend.sg_old_ingress_ruls[group.id] = group.ingress_rules.copy()
template = self.response_template(AUTHORIZE_SECURITY_GROUP_INGRESS_RESPONSE)
return template.render(rule=rule, group=group)
def create_security_group(self):
name = self._get_param("GroupName")
@ -152,19 +154,16 @@ class SecurityGroups(EC2BaseResponse):
tags = (tags or {}).get("Tag", [])
tags = {t["Key"]: t["Value"] for t in tags}
if self.is_not_dryrun("CreateSecurityGroup"):
group = self.ec2_backend.create_security_group(
name, description, vpc_id=vpc_id, tags=tags
)
if group:
self.ec2_backend.sg_old_ingress_ruls[
group.id
] = group.ingress_rules.copy()
self.ec2_backend.sg_old_egress_ruls[
group.id
] = group.egress_rules.copy()
template = self.response_template(CREATE_SECURITY_GROUP_RESPONSE)
return template.render(group=group)
self.error_on_dryrun()
group = self.ec2_backend.create_security_group(
name, description, vpc_id=vpc_id, tags=tags
)
if group:
self.ec2_backend.sg_old_ingress_ruls[group.id] = group.ingress_rules.copy()
self.ec2_backend.sg_old_egress_ruls[group.id] = group.egress_rules.copy()
template = self.response_template(CREATE_SECURITY_GROUP_RESPONSE)
return template.render(group=group)
def delete_security_group(self):
# TODO this should raise an error if there are instances in the group.
@ -174,13 +173,14 @@ class SecurityGroups(EC2BaseResponse):
name = self._get_param("GroupName")
sg_id = self._get_param("GroupId")
if self.is_not_dryrun("DeleteSecurityGroup"):
if name:
self.ec2_backend.delete_security_group(name)
elif sg_id:
self.ec2_backend.delete_security_group(group_id=sg_id)
self.error_on_dryrun()
return DELETE_GROUP_RESPONSE
if name:
self.ec2_backend.delete_security_group(name)
elif sg_id:
self.ec2_backend.delete_security_group(group_id=sg_id)
return DELETE_GROUP_RESPONSE
def describe_security_groups(self):
groupnames = self._get_multi_param("GroupName")
@ -197,22 +197,26 @@ class SecurityGroups(EC2BaseResponse):
def describe_security_group_rules(self):
group_id = self._get_param("GroupId")
filters = self._get_param("Filter")
if self.is_not_dryrun("DescribeSecurityGroups"):
rules = self.ec2_backend.describe_security_group_rules(group_id, filters)
template = self.response_template(DESCRIBE_SECURITY_GROUP_RULES_RESPONSE)
return template.render(rules=rules)
self.error_on_dryrun()
rules = self.ec2_backend.describe_security_group_rules(group_id, filters)
template = self.response_template(DESCRIBE_SECURITY_GROUP_RULES_RESPONSE)
return template.render(rules=rules)
def revoke_security_group_egress(self):
if self.is_not_dryrun("RevokeSecurityGroupEgress"):
for args in self._process_rules_from_querystring():
self.ec2_backend.revoke_security_group_egress(*args)
return REVOKE_SECURITY_GROUP_EGRESS_RESPONSE
self.error_on_dryrun()
for args in self._process_rules_from_querystring():
self.ec2_backend.revoke_security_group_egress(*args)
return REVOKE_SECURITY_GROUP_EGRESS_RESPONSE
def revoke_security_group_ingress(self):
if self.is_not_dryrun("RevokeSecurityGroupIngress"):
for args in self._process_rules_from_querystring():
self.ec2_backend.revoke_security_group_ingress(*args)
return REVOKE_SECURITY_GROUP_INGRESS_RESPONSE
self.error_on_dryrun()
for args in self._process_rules_from_querystring():
self.ec2_backend.revoke_security_group_ingress(*args)
return REVOKE_SECURITY_GROUP_INGRESS_RESPONSE
def update_security_group_rule_descriptions_ingress(self):
for args in self._process_rules_from_querystring():

41
moto/ec2/responses/settings.py
View File

@ -3,32 +3,29 @@ from moto.core.responses import BaseResponse
class Settings(BaseResponse):
def disable_ebs_encryption_by_default(self):
if self.is_not_dryrun("DisableEbsEncryptionByDefault"):
self.ec2_backend.disable_ebs_encryption_by_default()
template = self.response_template(
DISABLE_EBS_ENCRYPTION_BY_DEFAULT_RESPONSE
)
return template.render(ebsEncryptionByDefault=False).replace(
"False", "false"
)
self.error_on_dryrun()
self.ec2_backend.disable_ebs_encryption_by_default()
template = self.response_template(DISABLE_EBS_ENCRYPTION_BY_DEFAULT_RESPONSE)
return template.render(ebsEncryptionByDefault=False).replace("False", "false")
def enable_ebs_encryption_by_default(self):
if self.is_not_dryrun("EnableEbsEncryptionByDefault"):
self.ec2_backend.enable_ebs_encryption_by_default()
template = self.response_template(
ENABLED_EBS_ENCRYPTION_BY_DEFAULT_RESPONSE
)
return template.render(ebsEncryptionByDefault=True).replace("True", "true")
self.error_on_dryrun()
self.ec2_backend.enable_ebs_encryption_by_default()
template = self.response_template(ENABLED_EBS_ENCRYPTION_BY_DEFAULT_RESPONSE)
return template.render(ebsEncryptionByDefault=True).replace("True", "true")
def get_ebs_encryption_by_default(self):
if self.is_not_dryrun("GetEbsEncryptionByDefault"):
result = self.ec2_backend.get_ebs_encryption_by_default()
template = self.response_template(GET_EBS_ENCRYPTION_BY_DEFAULT_RESPONSE)
return (
template.render(ebsEncryptionByDefault=result)
.replace("False", "false")
.replace("True", "true")
)
self.error_on_dryrun()
result = self.ec2_backend.get_ebs_encryption_by_default()
template = self.response_template(GET_EBS_ENCRYPTION_BY_DEFAULT_RESPONSE)
return (
template.render(ebsEncryptionByDefault=result)
.replace("False", "false")
.replace("True", "true")
)
DISABLE_EBS_ENCRYPTION_BY_DEFAULT_RESPONSE = """<DisableEbsEncryptionByDefaultResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">

77
moto/ec2/responses/spot_instances.py
View File

@ -4,22 +4,26 @@ from ._base_response import EC2BaseResponse
class SpotInstances(EC2BaseResponse):
def cancel_spot_instance_requests(self):
request_ids = self._get_multi_param("SpotInstanceRequestId")
if self.is_not_dryrun("CancelSpotInstance"):
requests = self.ec2_backend.cancel_spot_instance_requests(request_ids)
template = self.response_template(CANCEL_SPOT_INSTANCES_TEMPLATE)
return template.render(requests=requests)
self.error_on_dryrun()
requests = self.ec2_backend.cancel_spot_instance_requests(request_ids)
template = self.response_template(CANCEL_SPOT_INSTANCES_TEMPLATE)
return template.render(requests=requests)
def create_spot_datafeed_subscription(self):
if self.is_not_dryrun("CreateSpotDatafeedSubscription"):
raise NotImplementedError(
"SpotInstances.create_spot_datafeed_subscription is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"SpotInstances.create_spot_datafeed_subscription is not yet implemented"
)
def delete_spot_datafeed_subscription(self):
if self.is_not_dryrun("DeleteSpotDatafeedSubscription"):
raise NotImplementedError(
"SpotInstances.delete_spot_datafeed_subscription is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"SpotInstances.delete_spot_datafeed_subscription is not yet implemented"
)
def describe_spot_datafeed_subscription(self):
raise NotImplementedError(
@ -67,31 +71,32 @@ class SpotInstances(EC2BaseResponse):
)
tags = self._parse_tag_specification()
if self.is_not_dryrun("RequestSpotInstance"):
requests = self.ec2_backend.request_spot_instances(
price=price,
image_id=image_id,
count=count,
spot_instance_type=spot_instance_type,
valid_from=valid_from,
valid_until=valid_until,
launch_group=launch_group,
availability_zone_group=availability_zone_group,
key_name=key_name,
security_groups=security_groups,
user_data=user_data,
instance_type=instance_type,
placement=placement,
kernel_id=kernel_id,
ramdisk_id=ramdisk_id,
monitoring_enabled=monitoring_enabled,
subnet_id=subnet_id,
instance_interruption_behaviour=instance_interruption_behaviour,
tags=tags,
)
self.error_on_dryrun()
template = self.response_template(REQUEST_SPOT_INSTANCES_TEMPLATE)
return template.render(requests=requests)
requests = self.ec2_backend.request_spot_instances(
price=price,
image_id=image_id,
count=count,
spot_instance_type=spot_instance_type,
valid_from=valid_from,
valid_until=valid_until,
launch_group=launch_group,
availability_zone_group=availability_zone_group,
key_name=key_name,
security_groups=security_groups,
user_data=user_data,
instance_type=instance_type,
placement=placement,
kernel_id=kernel_id,
ramdisk_id=ramdisk_id,
monitoring_enabled=monitoring_enabled,
subnet_id=subnet_id,
instance_interruption_behaviour=instance_interruption_behaviour,
tags=tags,
)
template = self.response_template(REQUEST_SPOT_INSTANCES_TEMPLATE)
return template.render(requests=requests)
REQUEST_SPOT_INSTANCES_TEMPLATE = """<RequestSpotInstancesResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">

19
moto/ec2/responses/tags.py
View File

@ -9,20 +9,27 @@ class TagResponse(EC2BaseResponse):
validate_resource_ids(resource_ids)
self.ec2_backend.do_resources_exist(resource_ids)
tags = tags_from_query_string(self.querystring)
if self.is_not_dryrun("CreateTags"):
self.ec2_backend.create_tags(resource_ids, tags)
return CREATE_RESPONSE
self.error_on_dryrun()
self.ec2_backend.create_tags(resource_ids, tags)
return CREATE_RESPONSE
def delete_tags(self):
resource_ids = self._get_multi_param("ResourceId")
validate_resource_ids(resource_ids)
tags = tags_from_query_string(self.querystring)
if self.is_not_dryrun("DeleteTags"):
self.ec2_backend.delete_tags(resource_ids, tags)
return DELETE_RESPONSE
self.error_on_dryrun()
self.ec2_backend.delete_tags(resource_ids, tags)
return DELETE_RESPONSE
def describe_tags(self):
filters = self._filters_from_querystring()
self.error_on_dryrun()
tags = self.ec2_backend.describe_tags(filters=filters)
template = self.response_template(DESCRIBE_RESPONSE)
return template.render(tags=tags)

32
moto/ecr/responses.py
View File

@ -90,10 +90,10 @@ class ECRResponse(BaseResponse):
)
def batch_check_layer_availability(self):
if self.is_not_dryrun("BatchCheckLayerAvailability"):
raise NotImplementedError(
"ECR.batch_check_layer_availability is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"ECR.batch_check_layer_availability is not yet implemented"
)
def batch_delete_image(self):
repository_str = self._get_param("repositoryName")
@ -116,10 +116,8 @@ class ECRResponse(BaseResponse):
return json.dumps(response)
def complete_layer_upload(self):
if self.is_not_dryrun("CompleteLayerUpload"):
raise NotImplementedError(
"ECR.complete_layer_upload is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError("ECR.complete_layer_upload is not yet implemented")
def delete_repository_policy(self):
registry_id = self._get_param("registryId")
@ -149,10 +147,10 @@ class ECRResponse(BaseResponse):
return json.dumps({"authorizationData": auth_data})
def get_download_url_for_layer(self):
if self.is_not_dryrun("GetDownloadUrlForLayer"):
raise NotImplementedError(
"ECR.get_download_url_for_layer is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError(
"ECR.get_download_url_for_layer is not yet implemented"
)
def get_repository_policy(self):
registry_id = self._get_param("registryId")
@ -165,10 +163,8 @@ class ECRResponse(BaseResponse):
)
def initiate_layer_upload(self):
if self.is_not_dryrun("InitiateLayerUpload"):
raise NotImplementedError(
"ECR.initiate_layer_upload is not yet implemented"
)
self.error_on_dryrun()
raise NotImplementedError("ECR.initiate_layer_upload is not yet implemented")
def set_repository_policy(self):
registry_id = self._get_param("registryId")
@ -187,8 +183,8 @@ class ECRResponse(BaseResponse):
)
def upload_layer_part(self):
if self.is_not_dryrun("UploadLayerPart"):
raise NotImplementedError("ECR.upload_layer_part is not yet implemented")
self.error_on_dryrun()
raise NotImplementedError("ECR.upload_layer_part is not yet implemented")
def list_tags_for_resource(self):
arn = self._get_param("resourceArn")

2
tests/test_ec2/test_elastic_ip_addresses.py
View File

@ -155,7 +155,7 @@ def test_eip_associate_classic():
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the DisAssociateAddress operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the DisassociateAddress operation: Request would have succeeded, but DryRun flag is set"
)
client.disassociate_address(PublicIp=eip.public_ip)

2
tests/test_ec2/test_elastic_network_interfaces.py
View File

@ -182,7 +182,7 @@ def test_elastic_network_interfaces_modify_attribute():
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the ModifyNetworkInterface operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the ModifyNetworkInterfaceAttribute operation: Request would have succeeded, but DryRun flag is set"
)
client.modify_network_interface_attribute(

20
tests/test_ec2/test_instances.py
View File

@ -43,7 +43,7 @@ def test_instance_launch_and_terminate():
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the RunInstance operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the RunInstances operation: Request would have succeeded, but DryRun flag is set"
)
reservation = client.run_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1)
@ -85,7 +85,7 @@ def test_instance_launch_and_terminate():
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the TerminateInstance operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the TerminateInstances operation: Request would have succeeded, but DryRun flag is set"
)
response = client.terminate_instances(InstanceIds=[instance_id])
@ -920,7 +920,7 @@ def test_instance_start_and_stop():
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the StopInstance operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the StopInstances operation: Request would have succeeded, but DryRun flag is set"
)
stopped_instances = client.stop_instances(InstanceIds=instance_ids)[
@ -939,7 +939,7 @@ def test_instance_start_and_stop():
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the StartInstance operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the StartInstances operation: Request would have succeeded, but DryRun flag is set"
)
instance1.reload()
@ -967,7 +967,7 @@ def test_instance_reboot():
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the RebootInstance operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the RebootInstances operation: Request would have succeeded, but DryRun flag is set"
)
instance.state.should.equal({"Code": 16, "Name": "running"})
@ -988,7 +988,7 @@ def test_instance_attribute_instance_type():
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the ModifyInstanceType operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the ModifyInstanceAttribute operation: Request would have succeeded, but DryRun flag is set"
)
instance.modify_attribute(InstanceType={"Value": "m1.medium"})
@ -1019,7 +1019,7 @@ def test_modify_instance_attribute_security_groups():
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the ModifyInstanceSecurityGroups operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the ModifyInstanceAttribute operation: Request would have succeeded, but DryRun flag is set"
)
instance.modify_attribute(Groups=[sg_id, sg_id2])
@ -1043,7 +1043,7 @@ def test_instance_attribute_user_data():
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the ModifyUserData operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the ModifyInstanceAttribute operation: Request would have succeeded, but DryRun flag is set"
)
instance.modify_attribute(UserData={"Value": "this is my user data"})
@ -1068,7 +1068,7 @@ def test_instance_attribute_source_dest_check():
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the ModifySourceDestCheck operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the ModifyInstanceAttribute operation: Request would have succeeded, but DryRun flag is set"
)
instance.modify_attribute(SourceDestCheck={"Value": False})
@ -1971,7 +1971,7 @@ def test_get_instance_by_security_group():
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the ModifyInstanceSecurityGroups operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the ModifyInstanceAttribute operation: Request would have succeeded, but DryRun flag is set"
)
client.modify_instance_attribute(InstanceId=instance.id, Groups=[security_group.id])

48
tests/test_ec2/test_launch_templates.py
View File

@ -50,6 +50,28 @@ def test_launch_template_create():
)
@mock_ec2
def test_create_launch_template__dryrun():
cli = boto3.client("ec2", region_name="us-east-1")
template_name = str(uuid4())
with pytest.raises(ClientError) as exc:
cli.create_launch_template(
DryRun=True,
LaunchTemplateName=template_name,
LaunchTemplateData={"ImageId": "ami-abc123"},
TagSpecifications=[
{"ResourceType": "instance", "Tags": [{"Key": "key", "Value": "value"}]}
],
)
err = exc.value.response["Error"]
err.should.have.key("Code").equals("DryRunOperation")
err["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the CreateLaunchTemplate operation: Request would have succeeded, but DryRun flag is set"
)
@mock_ec2
def test_describe_launch_template_versions():
template_data = {
@ -142,6 +164,29 @@ def test_create_launch_template_version():
version["VersionNumber"].should.equal(2)
@mock_ec2
def test_create_launch_template_version__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"}
)
with pytest.raises(ClientError) as exc:
cli.create_launch_template_version(
DryRun=True,
LaunchTemplateName=template_name,
LaunchTemplateData={"ImageId": "ami-def456"},
VersionDescription="new ami",
)
err = exc.value.response["Error"]
err.should.have.key("Code").equals("DryRunOperation")
err["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the CreateLaunchTemplateVersion operation: Request would have succeeded, but DryRun flag is set"
)
@mock_ec2
def test_create_launch_template_version_by_id():
cli = boto3.client("ec2", region_name="us-east-1")
@ -476,6 +521,9 @@ def test_delete_launch_template__dryrun():
cli.delete_launch_template(DryRun=True, LaunchTemplateName=template_name)
err = exc.value.response["Error"]
err.should.have.key("Code").equals("DryRunOperation")
err["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the DeleteLaunchTemplate operation: Request would have succeeded, but DryRun flag is set"
)
# Template still exists
cli.describe_launch_templates(LaunchTemplateNames=[template_name])[

4
tests/test_ec2/test_security_groups.py
View File

@ -224,7 +224,7 @@ def test_authorize_ip_range_and_revoke():
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the GrantSecurityGroupIngress operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the AuthorizeSecurityGroupIngress operation: Request would have succeeded, but DryRun flag is set"
)
ingress_permissions = [
@ -287,7 +287,7 @@ def test_authorize_ip_range_and_revoke():
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the GrantSecurityGroupEgress operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the AuthorizeSecurityGroupEgress operation: Request would have succeeded, but DryRun flag is set"
)
egress_security_group.authorize_egress(IpPermissions=egress_permissions)

4
tests/test_ec2/test_spot_instances.py
View File

@ -57,7 +57,7 @@ def test_request_spot_instances():
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the RequestSpotInstance operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the RequestSpotInstances operation: Request would have succeeded, but DryRun flag is set"
)
request = conn.request_spot_instances(
@ -178,7 +178,7 @@ def test_cancel_spot_instance_request():
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the CancelSpotInstance operation: Request would have succeeded, but DryRun flag is set"
"An error occurred (DryRunOperation) when calling the CancelSpotInstanceRequests operation: Request would have succeeded, but DryRun flag is set"
)
client.cancel_spot_instance_requests(