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

Techdebt: MyPy EC2 (#5936)

Browse Source
This commit is contained in:
Bert Blommers 2023-02-16 20:43:43 -01:00 committed by GitHub
parent e019473dee
commit 418c69f477
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
52 changed files with 380 additions and 474 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
moto/ec2/exceptions.py
View File

@ -64,7 +64,7 @@ class InvalidRequest(EC2ClientError):
class InvalidParameterCombination(EC2ClientError):
def __init__(self, msg):
def __init__(self, msg: str):
super().__init__("InvalidParameterCombination", msg)
@ -340,14 +340,6 @@ class InvalidAddressError(EC2ClientError):
super().__init__("InvalidAddress.NotFound", f"Address '{ip}' not found.")
class LogDestinationNotFoundError(EC2ClientError):
def __init__(self, bucket_name):
super().__init__(
"LogDestinationNotFoundException",
f"LogDestination: '{bucket_name}' does not exist.",
)
class InvalidAllocationIdError(EC2ClientError):
def __init__(self, allocation_id: Any):
super().__init__(

6
moto/ec2/models/instances.py
View File

@ -22,9 +22,9 @@ from moto.packages.boto.ec2.instance import Reservation
from ..exceptions import (
AvailabilityZoneNotFromRegionError,
EC2ClientError,
InvalidInstanceIdError,
InvalidInstanceTypeError,
InvalidParameterCombination,
InvalidParameterValueErrorUnknownAttribute,
InvalidSecurityGroupNotFoundError,
OperationNotPermitted4,
@ -779,9 +779,7 @@ class InstanceBackend:
) -> List[Tuple[Instance, InstanceState]]:
terminated_instances = []
if not instance_ids:
raise EC2ClientError(
"InvalidParameterCombination", "No instances specified"
)
raise InvalidParameterCombination("No instances specified")
for instance in self.get_multi_instances_by_id(instance_ids):
if instance.disable_api_termination == "true":
raise OperationNotPermitted4(instance.id)

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

@ -1,5 +1,4 @@
from .account_attributes import AccountAttributes
from .amazon_dev_pay import AmazonDevPay
from .amis import AmisResponse
from .availability_zones_and_regions import AvailabilityZonesAndRegions
from .customer_gateways import CustomerGateways
@ -18,7 +17,6 @@ from .key_pairs import KeyPairs
from .launch_templates import LaunchTemplates
from .monitoring import Monitoring
from .network_acls import NetworkACLs
from .placement_groups import PlacementGroups
from .reserved_instances import ReservedInstances
from .route_tables import RouteTables
from .security_groups import SecurityGroups
@ -29,8 +27,6 @@ from .subnets import Subnets
from .flow_logs import FlowLogs
from .tags import TagResponse
from .virtual_private_gateways import VirtualPrivateGateways
from .vm_export import VMExport
from .vm_import import VMImport
from .vpcs import VPCs
from .vpc_service_configuration import VPCEndpointServiceConfiguration
from .vpc_peering_connections import VPCPeeringConnections
@ -46,7 +42,6 @@ from .carrier_gateways import CarrierGateway
class EC2Response(
AccountAttributes,
AmazonDevPay,
AmisResponse,
AvailabilityZonesAndRegions,
CustomerGateways,
@ -65,7 +60,6 @@ class EC2Response(
LaunchTemplates,
Monitoring,
NetworkACLs,
PlacementGroups,
ReservedInstances,
RouteTables,
SecurityGroups,
@ -76,8 +70,6 @@ class EC2Response(
FlowLogs,
TagResponse,
VirtualPrivateGateways,
VMExport,
VMImport,
VPCs,
VPCEndpointServiceConfiguration,
VPCPeeringConnections,
@ -90,15 +82,9 @@ class EC2Response(
IamInstanceProfiles,
CarrierGateway,
):
def __init__(self):
def __init__(self) -> None:
super().__init__(service_name="ec2")
@property
def ec2_backend(self):
from moto.ec2.models import ec2_backends
return ec2_backends[self.current_account][self.region]
@property
def should_autoescape(self):
def should_autoescape(self) -> bool:
return True

13
moto/ec2/responses/_base_response.py
View File

@ -1,18 +1,27 @@
from typing import Any, Dict
from moto.core.responses import BaseResponse
from ..exceptions import EmptyTagSpecError
from ..utils import convert_tag_spec
class EC2BaseResponse(BaseResponse):
def _filters_from_querystring(self):
@property
def ec2_backend(self) -> Any: # type: ignore[misc]
from moto.ec2.models import ec2_backends
return ec2_backends[self.current_account][self.region]
def _filters_from_querystring(self) -> Dict[str, str]:
# [{"Name": x1, "Value": y1}, ..]
_filters = self._get_multi_param("Filter.")
# return {x1: y1, ...}
return {f["Name"]: f["Value"] for f in _filters}
def _parse_tag_specification(self):
def _parse_tag_specification(self) -> Dict[str, Dict[str, str]]:
# [{"ResourceType": _type, "Tag": [{"Key": k, "Value": v}, ..]}]
tag_spec_set = self._get_multi_param("TagSpecification")
if not tag_spec_set:
tag_spec_set = self._get_multi_param("TagSpecifications")
# If we do not pass any Tags, this method will convert this to [_type] instead
if isinstance(tag_spec_set, list) and any(
[isinstance(spec, str) for spec in tag_spec_set]

2
moto/ec2/responses/account_attributes.py
View File

@ -2,7 +2,7 @@ from moto.core.responses import BaseResponse
class AccountAttributes(BaseResponse):
def describe_account_attributes(self):
def describe_account_attributes(self) -> str:
template = self.response_template(DESCRIBE_ACCOUNT_ATTRIBUTES_RESULT)
return template.render()

8
moto/ec2/responses/amazon_dev_pay.py
View File

@ -1,8 +0,0 @@
from moto.core.responses import BaseResponse
class AmazonDevPay(BaseResponse):
def confirm_product_instance(self):
raise NotImplementedError(
"AmazonDevPay.confirm_product_instance is not yet implemented"
)

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

@ -2,15 +2,15 @@ from ._base_response import EC2BaseResponse
class AmisResponse(EC2BaseResponse):
def create_image(self):
name = self.querystring.get("Name")[0]
def create_image(self) -> str:
name = self.querystring.get("Name")[0] # type: ignore[index]
description = self._get_param("Description", if_none="")
instance_id = self._get_param("InstanceId")
tag_specifications = self._get_multi_param("TagSpecification")
self.error_on_dryrun()
image = self.ec2_backend.create_image(
image = self.ec2_backend.create_image( # type: ignore[attr-defined]
instance_id,
name,
description,
@ -19,7 +19,7 @@ class AmisResponse(EC2BaseResponse):
template = self.response_template(CREATE_IMAGE_RESPONSE)
return template.render(image=image)
def copy_image(self):
def copy_image(self) -> str:
source_image_id = self._get_param("SourceImageId")
source_region = self._get_param("SourceRegion")
name = self._get_param("Name")
@ -33,7 +33,7 @@ class AmisResponse(EC2BaseResponse):
template = self.response_template(COPY_IMAGE_RESPONSE)
return template.render(image=image)
def deregister_image(self):
def deregister_image(self) -> str:
ami_id = self._get_param("ImageId")
self.error_on_dryrun()
@ -42,7 +42,7 @@ class AmisResponse(EC2BaseResponse):
template = self.response_template(DEREGISTER_IMAGE_RESPONSE)
return template.render(success="true")
def describe_images(self):
def describe_images(self) -> str:
self.error_on_dryrun()
ami_ids = self._get_multi_param("ImageId")
filters = self._filters_from_querystring()
@ -54,14 +54,14 @@ class AmisResponse(EC2BaseResponse):
template = self.response_template(DESCRIBE_IMAGES_RESPONSE)
return template.render(images=images)
def describe_image_attribute(self):
def describe_image_attribute(self) -> str:
ami_id = self._get_param("ImageId")
groups = self.ec2_backend.get_launch_permission_groups(ami_id)
users = self.ec2_backend.get_launch_permission_users(ami_id)
template = self.response_template(DESCRIBE_IMAGE_ATTRIBUTES_RESPONSE)
return template.render(ami_id=ami_id, groups=groups, users=users)
def modify_image_attribute(self):
def modify_image_attribute(self) -> str:
ami_id = self._get_param("ImageId")
operation_type = self._get_param("OperationType")
group = self._get_param("UserGroup.1")
@ -79,8 +79,8 @@ class AmisResponse(EC2BaseResponse):
)
return MODIFY_IMAGE_ATTRIBUTE_RESPONSE
def register_image(self):
name = self.querystring.get("Name")[0]
def register_image(self) -> str:
name = self.querystring.get("Name")[0] # type: ignore[index]
description = self._get_param("Description", if_none="")
self.error_on_dryrun()
@ -89,7 +89,7 @@ class AmisResponse(EC2BaseResponse):
template = self.response_template(REGISTER_IMAGE_RESPONSE)
return template.render(image=image)
def reset_image_attribute(self):
def reset_image_attribute(self) -> str:
self.error_on_dryrun()
raise NotImplementedError("AMIs.reset_image_attribute is not yet implemented")

8
moto/ec2/responses/availability_zones_and_regions.py
View File

@ -1,15 +1,15 @@
from moto.core.responses import BaseResponse
from ._base_response import EC2BaseResponse
class AvailabilityZonesAndRegions(BaseResponse):
def describe_availability_zones(self):
class AvailabilityZonesAndRegions(EC2BaseResponse):
def describe_availability_zones(self) -> str:
self.error_on_dryrun()
filters = self._filters_from_querystring()
zones = self.ec2_backend.describe_availability_zones(filters)
template = self.response_template(DESCRIBE_ZONES_RESPONSE)
return template.render(zones=zones)
def describe_regions(self):
def describe_regions(self) -> str:
self.error_on_dryrun()
region_names = self._get_multi_param("RegionName")
regions = self.ec2_backend.describe_regions(region_names)

10
moto/ec2/responses/carrier_gateways.py
View File

@ -3,10 +3,10 @@ from ._base_response import EC2BaseResponse
class CarrierGateway(EC2BaseResponse):
def create_carrier_gateway(self):
def create_carrier_gateway(self) -> str:
vpc_id = self._get_param("VpcId")
tags = self._get_multi_param("TagSpecification")
tags = add_tag_specification(tags)
tag_param = self._get_multi_param("TagSpecification")
tags = add_tag_specification(tag_param)
carrier_gateway = self.ec2_backend.create_carrier_gateway(
vpc_id=vpc_id, tags=tags
@ -14,14 +14,14 @@ class CarrierGateway(EC2BaseResponse):
template = self.response_template(CREATE_CARRIER_GATEWAY_RESPONSE)
return template.render(carrier_gateway=carrier_gateway)
def delete_carrier_gateway(self):
def delete_carrier_gateway(self) -> str:
carrier_gateway_id = self._get_param("CarrierGatewayId")
carrier_gateway = self.ec2_backend.delete_carrier_gateway(carrier_gateway_id)
template = self.response_template(DELETE_CARRIER_GATEWAY_RESPONSE)
return template.render(carrier_gateway=carrier_gateway)
def describe_carrier_gateways(self):
def describe_carrier_gateways(self) -> str:
carrier_gateway_ids = self._get_multi_param("CarrierGatewayId")
filters = self._filters_from_querystring()

11
moto/ec2/responses/customer_gateways.py
View File

@ -2,27 +2,24 @@ from ._base_response import EC2BaseResponse
class CustomerGateways(EC2BaseResponse):
def create_customer_gateway(self):
def create_customer_gateway(self) -> str:
gateway_type = self._get_param("Type")
ip_address = self._get_param("IpAddress") or self._get_param("PublicIp")
bgp_asn = self._get_param("BgpAsn")
tags = self._get_multi_param("TagSpecification")
tags = tags[0] if isinstance(tags, list) and len(tags) == 1 else tags
tags = (tags or {}).get("Tag", [])
tags = {t["Key"]: t["Value"] for t in tags}
tags = self._parse_tag_specification().get("customer-gateway", {})
customer_gateway = self.ec2_backend.create_customer_gateway(
gateway_type, ip_address=ip_address, bgp_asn=bgp_asn, tags=tags
)
template = self.response_template(CREATE_CUSTOMER_GATEWAY_RESPONSE)
return template.render(customer_gateway=customer_gateway)
def delete_customer_gateway(self):
def delete_customer_gateway(self) -> str:
customer_gateway_id = self._get_param("CustomerGatewayId")
self.ec2_backend.delete_customer_gateway(customer_gateway_id)
template = self.response_template(DELETE_CUSTOMER_GATEWAY_RESPONSE)
return template.render(delete_status="true")
def describe_customer_gateways(self):
def describe_customer_gateways(self) -> str:
self.error_on_dryrun()
filters = self._filters_from_querystring()
customer_gateway_ids = self._get_multi_param("CustomerGatewayId")

22
moto/ec2/responses/dhcp_options.py
View File

@ -2,7 +2,7 @@ from ._base_response import EC2BaseResponse
class DHCPOptions(EC2BaseResponse):
def associate_dhcp_options(self):
def associate_dhcp_options(self) -> str:
dhcp_opt_id = self._get_param("DhcpOptionsId")
vpc_id = self._get_param("VpcId")
@ -14,17 +14,17 @@ class DHCPOptions(EC2BaseResponse):
template = self.response_template(ASSOCIATE_DHCP_OPTIONS_RESPONSE)
return template.render()
def create_dhcp_options(self):
dhcp_config = self._get_multi_param("DhcpConfiguration")
dhcp_config = {f["Key"]: f["Value"] for f in dhcp_config}
def create_dhcp_options(self) -> str:
provided_config = self._get_multi_param("DhcpConfiguration")
flat_config = {f["Key"]: f["Value"] for f in provided_config}
# TODO validate we only got the options we know about
domain_name_servers = dhcp_config.get("domain-name-servers", None)
domain_name = dhcp_config.get("domain-name", None)
ntp_servers = dhcp_config.get("ntp-servers", None)
netbios_name_servers = dhcp_config.get("netbios-name-servers", None)
netbios_node_type = dhcp_config.get("netbios-node-type", None)
domain_name_servers = flat_config.get("domain-name-servers", None)
domain_name = flat_config.get("domain-name", None)
ntp_servers = flat_config.get("ntp-servers", None)
netbios_name_servers = flat_config.get("netbios-name-servers", None)
netbios_node_type = flat_config.get("netbios-node-type", None)
dhcp_options_set = self.ec2_backend.create_dhcp_options(
domain_name_servers=domain_name_servers,
@ -37,13 +37,13 @@ class DHCPOptions(EC2BaseResponse):
template = self.response_template(CREATE_DHCP_OPTIONS_RESPONSE)
return template.render(dhcp_options_set=dhcp_options_set)
def delete_dhcp_options(self):
def delete_dhcp_options(self) -> str:
dhcp_opt_id = self._get_param("DhcpOptionsId")
self.ec2_backend.delete_dhcp_options_set(dhcp_opt_id)
template = self.response_template(DELETE_DHCP_OPTIONS_RESPONSE)
return template.render(delete_status="true")
def describe_dhcp_options(self):
def describe_dhcp_options(self) -> str:
dhcp_opt_ids = self._get_multi_param("DhcpOptionsId")
filters = self._filters_from_querystring()
dhcp_opts = self.ec2_backend.describe_dhcp_options(dhcp_opt_ids, filters)

14
moto/ec2/responses/egress_only_internet_gateways.py
View File

@ -1,12 +1,12 @@
from moto.core.responses import BaseResponse
from ._base_response import EC2BaseResponse
from moto.ec2.utils import add_tag_specification
class EgressOnlyInternetGateway(BaseResponse):
def create_egress_only_internet_gateway(self):
class EgressOnlyInternetGateway(EC2BaseResponse):
def create_egress_only_internet_gateway(self) -> str:
vpc_id = self._get_param("VpcId")
tags = self._get_multi_param("TagSpecification")
tags = add_tag_specification(tags)
tag_param = self._get_multi_param("TagSpecification")
tags = add_tag_specification(tag_param)
egress_only_igw = self.ec2_backend.create_egress_only_internet_gateway(
vpc_id=vpc_id, tags=tags
@ -14,7 +14,7 @@ class EgressOnlyInternetGateway(BaseResponse):
template = self.response_template(CREATE_EGRESS_ONLY_IGW_RESPONSE)
return template.render(egress_only_igw=egress_only_igw)
def describe_egress_only_internet_gateways(self):
def describe_egress_only_internet_gateways(self) -> str:
egress_only_igw_ids = self._get_multi_param("EgressOnlyInternetGatewayId")
egress_only_igws = self.ec2_backend.describe_egress_only_internet_gateways(
egress_only_igw_ids
@ -22,7 +22,7 @@ class EgressOnlyInternetGateway(BaseResponse):
template = self.response_template(DESCRIBE_EGRESS_ONLY_IGW_RESPONSE)
return template.render(egress_only_igws=egress_only_igws)
def delete_egress_only_internet_gateway(self):
def delete_egress_only_internet_gateway(self) -> str:
egress_only_igw_id = self._get_param("EgressOnlyInternetGatewayId")
self.ec2_backend.delete_egress_only_internet_gateway(
gateway_id=egress_only_igw_id

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

@ -2,7 +2,7 @@ from ._base_response import EC2BaseResponse
class ElasticBlockStore(EC2BaseResponse):
def attach_volume(self):
def attach_volume(self) -> str:
volume_id = self._get_param("VolumeId")
instance_id = self._get_param("InstanceId")
device_path = self._get_param("Device")
@ -13,7 +13,7 @@ class ElasticBlockStore(EC2BaseResponse):
template = self.response_template(ATTACHED_VOLUME_RESPONSE)
return template.render(attachment=attachment)
def copy_snapshot(self):
def copy_snapshot(self) -> str:
source_snapshot_id = self._get_param("SourceSnapshotId")
source_region = self._get_param("SourceRegion")
description = self._get_param("Description")
@ -29,7 +29,7 @@ class ElasticBlockStore(EC2BaseResponse):
template = self.response_template(COPY_SNAPSHOT_RESPONSE)
return template.render(snapshot=snapshot)
def create_snapshot(self):
def create_snapshot(self) -> str:
volume_id = self._get_param("VolumeId")
description = self._get_param("Description")
tags = self._parse_tag_specification()
@ -42,7 +42,7 @@ class ElasticBlockStore(EC2BaseResponse):
template = self.response_template(CREATE_SNAPSHOT_RESPONSE)
return template.render(snapshot=snapshot)
def create_snapshots(self):
def create_snapshots(self) -> str:
params = self._get_params()
instance_spec = params.get("InstanceSpecification")
description = params.get("Description", "")
@ -57,7 +57,7 @@ class ElasticBlockStore(EC2BaseResponse):
template = self.response_template(CREATE_SNAPSHOTS_RESPONSE)
return template.render(snapshots=snapshots)
def create_volume(self):
def create_volume(self) -> str:
size = self._get_param("Size")
zone = self._get_param("AvailabilityZone")
snapshot_id = self._get_param("SnapshotId")
@ -83,7 +83,7 @@ class ElasticBlockStore(EC2BaseResponse):
template = self.response_template(CREATE_VOLUME_RESPONSE)
return template.render(volume=volume)
def modify_volume(self):
def modify_volume(self) -> str:
volume_id = self._get_param("VolumeId")
target_size = self._get_param("Size")
target_volume_type = self._get_param("VolumeType")
@ -96,7 +96,7 @@ class ElasticBlockStore(EC2BaseResponse):
template = self.response_template(MODIFY_VOLUME_RESPONSE)
return template.render(volume=volume)
def describe_volumes_modifications(self):
def describe_volumes_modifications(self) -> str:
filters = self._filters_from_querystring()
volume_ids = self._get_multi_param("VolumeId")
modifications = self.ec2_backend.describe_volumes_modifications(
@ -105,7 +105,7 @@ class ElasticBlockStore(EC2BaseResponse):
template = self.response_template(DESCRIBE_VOLUMES_MODIFICATIONS_RESPONSE)
return template.render(modifications=modifications)
def delete_snapshot(self):
def delete_snapshot(self) -> str:
snapshot_id = self._get_param("SnapshotId")
self.error_on_dryrun()
@ -113,7 +113,7 @@ class ElasticBlockStore(EC2BaseResponse):
self.ec2_backend.delete_snapshot(snapshot_id)
return DELETE_SNAPSHOT_RESPONSE
def delete_volume(self):
def delete_volume(self) -> str:
volume_id = self._get_param("VolumeId")
self.error_on_dryrun()
@ -121,7 +121,7 @@ class ElasticBlockStore(EC2BaseResponse):
self.ec2_backend.delete_volume(volume_id)
return DELETE_VOLUME_RESPONSE
def describe_snapshots(self):
def describe_snapshots(self) -> str:
filters = self._filters_from_querystring()
snapshot_ids = self._get_multi_param("SnapshotId")
snapshots = self.ec2_backend.describe_snapshots(
@ -130,7 +130,7 @@ class ElasticBlockStore(EC2BaseResponse):
template = self.response_template(DESCRIBE_SNAPSHOTS_RESPONSE)
return template.render(snapshots=snapshots)
def describe_volumes(self):
def describe_volumes(self) -> str:
filters = self._filters_from_querystring()
volume_ids = self._get_multi_param("VolumeId")
volumes = self.ec2_backend.describe_volumes(
@ -139,17 +139,17 @@ class ElasticBlockStore(EC2BaseResponse):
template = self.response_template(DESCRIBE_VOLUMES_RESPONSE)
return template.render(volumes=volumes)
def describe_volume_attribute(self):
def describe_volume_attribute(self) -> str:
raise NotImplementedError(
"ElasticBlockStore.describe_volume_attribute is not yet implemented"
)
def describe_volume_status(self):
def describe_volume_status(self) -> str:
raise NotImplementedError(
"ElasticBlockStore.describe_volume_status is not yet implemented"
)
def detach_volume(self):
def detach_volume(self) -> str:
volume_id = self._get_param("VolumeId")
instance_id = self._get_param("InstanceId")
device_path = self._get_param("Device")
@ -160,28 +160,28 @@ class ElasticBlockStore(EC2BaseResponse):
template = self.response_template(DETATCH_VOLUME_RESPONSE)
return template.render(attachment=attachment)
def enable_volume_io(self):
def enable_volume_io(self) -> str:
self.error_on_dryrun()
raise NotImplementedError(
"ElasticBlockStore.enable_volume_io is not yet implemented"
)
def import_volume(self):
def import_volume(self) -> str:
self.error_on_dryrun()
raise NotImplementedError(
"ElasticBlockStore.import_volume is not yet implemented"
)
def describe_snapshot_attribute(self):
def describe_snapshot_attribute(self) -> str:
snapshot_id = self._get_param("SnapshotId")
groups = self.ec2_backend.get_create_volume_permission_groups(snapshot_id)
user_ids = self.ec2_backend.get_create_volume_permission_userids(snapshot_id)
template = self.response_template(DESCRIBE_SNAPSHOT_ATTRIBUTES_RESPONSE)
return template.render(snapshot_id=snapshot_id, groups=groups, userIds=user_ids)
def modify_snapshot_attribute(self):
def modify_snapshot_attribute(self) -> str:
snapshot_id = self._get_param("SnapshotId")
operation_type = self._get_param("OperationType")
groups = self._get_multi_param("UserGroup")
@ -199,14 +199,14 @@ class ElasticBlockStore(EC2BaseResponse):
)
return MODIFY_SNAPSHOT_ATTRIBUTE_RESPONSE
def modify_volume_attribute(self):
def modify_volume_attribute(self) -> str:
self.error_on_dryrun()
raise NotImplementedError(
"ElasticBlockStore.modify_volume_attribute is not yet implemented"
)
def reset_snapshot_attribute(self):
def reset_snapshot_attribute(self) -> str:
self.error_on_dryrun()
raise NotImplementedError(

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

@ -3,11 +3,11 @@ from ._base_response import EC2BaseResponse
class ElasticIPAddresses(EC2BaseResponse):
def allocate_address(self):
def allocate_address(self) -> str:
domain = self._get_param("Domain", if_none=None)
reallocate_address = self._get_param("Address", if_none=None)
tags = self._get_multi_param("TagSpecification")
tags = add_tag_specification(tags)
tag_param = self._get_multi_param("TagSpecification")
tags = add_tag_specification(tag_param)
self.error_on_dryrun()
@ -20,7 +20,7 @@ class ElasticIPAddresses(EC2BaseResponse):
template = self.response_template(ALLOCATE_ADDRESS_RESPONSE)
return template.render(address=address)
def associate_address(self):
def associate_address(self) -> str:
instance = eni = None
if "InstanceId" in self.querystring:
@ -70,7 +70,7 @@ class ElasticIPAddresses(EC2BaseResponse):
template = self.response_template(ASSOCIATE_ADDRESS_RESPONSE)
return template.render(address=eip)
def describe_addresses(self):
def describe_addresses(self) -> str:
self.error_on_dryrun()
allocation_ids = self._get_multi_param("AllocationId")
public_ips = self._get_multi_param("PublicIp")
@ -81,7 +81,7 @@ class ElasticIPAddresses(EC2BaseResponse):
template = self.response_template(DESCRIBE_ADDRESS_RESPONSE)
return template.render(addresses=addresses)
def disassociate_address(self):
def disassociate_address(self) -> str:
if (
"PublicIp" not in self.querystring
and "AssociationId" not in self.querystring
@ -102,7 +102,7 @@ class ElasticIPAddresses(EC2BaseResponse):
return self.response_template(DISASSOCIATE_ADDRESS_RESPONSE).render()
def release_address(self):
def release_address(self) -> str:
if (
"PublicIp" not in self.querystring
and "AllocationId" not in self.querystring

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

@ -4,7 +4,7 @@ from ._base_response import EC2BaseResponse
class ElasticNetworkInterfaces(EC2BaseResponse):
def create_network_interface(self):
def create_network_interface(self) -> str:
subnet_id = self._get_param("SubnetId")
private_ip_address = self._get_param("PrivateIpAddress")
private_ip_addresses = self._get_multi_param("PrivateIpAddresses")
@ -14,8 +14,7 @@ class ElasticNetworkInterfaces(EC2BaseResponse):
groups = self._get_multi_param("SecurityGroupId")
subnet = self.ec2_backend.get_subnet(subnet_id)
description = self._get_param("Description")
tags = self._get_multi_param("TagSpecification")
tags = add_tag_specification(tags)
tags = add_tag_specification(self._get_multi_param("TagSpecification"))
self.error_on_dryrun()
@ -33,7 +32,7 @@ class ElasticNetworkInterfaces(EC2BaseResponse):
template = self.response_template(CREATE_NETWORK_INTERFACE_RESPONSE)
return template.render(eni=eni)
def delete_network_interface(self):
def delete_network_interface(self) -> str:
eni_id = self._get_param("NetworkInterfaceId")
self.error_on_dryrun()
@ -42,7 +41,7 @@ class ElasticNetworkInterfaces(EC2BaseResponse):
template = self.response_template(DELETE_NETWORK_INTERFACE_RESPONSE)
return template.render()
def describe_network_interface_attribute(self):
def describe_network_interface_attribute(self) -> str:
eni_id = self._get_param("NetworkInterfaceId")
attribute = self._get_param("Attribute")
@ -70,7 +69,7 @@ class ElasticNetworkInterfaces(EC2BaseResponse):
raise InvalidParameterValueErrorUnknownAttribute(attribute)
return template.render(eni=eni)
def describe_network_interfaces(self):
def describe_network_interfaces(self) -> str:
eni_ids = self._get_multi_param("NetworkInterfaceId")
filters = self._filters_from_querystring()
@ -80,7 +79,7 @@ class ElasticNetworkInterfaces(EC2BaseResponse):
template = self.response_template(DESCRIBE_NETWORK_INTERFACES_RESPONSE)
return template.render(enis=enis)
def attach_network_interface(self):
def attach_network_interface(self) -> str:
eni_id = self._get_param("NetworkInterfaceId")
instance_id = self._get_param("InstanceId")
device_index = self._get_param("DeviceIndex")
@ -93,7 +92,7 @@ class ElasticNetworkInterfaces(EC2BaseResponse):
template = self.response_template(ATTACH_NETWORK_INTERFACE_RESPONSE)
return template.render(attachment_id=attachment_id)
def detach_network_interface(self):
def detach_network_interface(self) -> str:
attachment_id = self._get_param("AttachmentId")
self.error_on_dryrun()
@ -102,7 +101,7 @@ class ElasticNetworkInterfaces(EC2BaseResponse):
template = self.response_template(DETACH_NETWORK_INTERFACE_RESPONSE)
return template.render()
def modify_network_interface_attribute(self):
def modify_network_interface_attribute(self) -> str:
eni_id = self._get_param("NetworkInterfaceId")
group_ids = self._get_multi_param("SecurityGroupId")
source_dest_check = get_attribute_value("SourceDestCheck", self.querystring)
@ -115,28 +114,28 @@ class ElasticNetworkInterfaces(EC2BaseResponse):
)
return MODIFY_NETWORK_INTERFACE_ATTRIBUTE_RESPONSE
def reset_network_interface_attribute(self):
def reset_network_interface_attribute(self) -> str:
self.error_on_dryrun()
raise NotImplementedError(
"ElasticNetworkInterfaces(AmazonVPC).reset_network_interface_attribute is not yet implemented"
)
def assign_private_ip_addresses(self):
def assign_private_ip_addresses(self) -> str:
eni_id = self._get_param("NetworkInterfaceId")
secondary_ips_count = self._get_int_param("SecondaryPrivateIpAddressCount", 0)
eni = self.ec2_backend.assign_private_ip_addresses(eni_id, secondary_ips_count)
template = self.response_template(ASSIGN_PRIVATE_IP_ADDRESSES)
return template.render(eni=eni)
def unassign_private_ip_addresses(self):
def unassign_private_ip_addresses(self) -> str:
eni_id = self._get_param("NetworkInterfaceId")
private_ip_address = self._get_multi_param("PrivateIpAddress")
eni = self.ec2_backend.unassign_private_ip_addresses(eni_id, private_ip_address)
template = self.response_template(UNASSIGN_PRIVATE_IP_ADDRESSES)
return template.render(eni=eni)
def assign_ipv6_addresses(self):
def assign_ipv6_addresses(self) -> str:
eni_id = self._get_param("NetworkInterfaceId")
ipv6_count = self._get_int_param("Ipv6AddressCount", 0)
ipv6_addresses = self._get_multi_param("Ipv6Addresses")
@ -144,7 +143,7 @@ class ElasticNetworkInterfaces(EC2BaseResponse):
template = self.response_template(ASSIGN_IPV6_ADDRESSES)
return template.render(eni=eni)
def unassign_ipv6_addresses(self):
def unassign_ipv6_addresses(self) -> str:
eni_id = self._get_param("NetworkInterfaceId")
ips = self._get_multi_param("Ipv6Addresses")
eni = self.ec2_backend.unassign_ipv6_addresses(eni_id, ips)

12
moto/ec2/responses/fleets.py
View File

@ -1,22 +1,22 @@
from moto.core.responses import BaseResponse
from ._base_response import EC2BaseResponse
class Fleets(BaseResponse):
def delete_fleets(self):
class Fleets(EC2BaseResponse):
def delete_fleets(self) -> str:
fleet_ids = self._get_multi_param("FleetId.")
terminate_instances = self._get_param("TerminateInstances")
fleets = self.ec2_backend.delete_fleets(fleet_ids, terminate_instances)
template = self.response_template(DELETE_FLEETS_TEMPLATE)
return template.render(fleets=fleets)
def describe_fleet_instances(self):
def describe_fleet_instances(self) -> str:
fleet_id = self._get_param("FleetId")
instances = self.ec2_backend.describe_fleet_instances(fleet_id)
template = self.response_template(DESCRIBE_FLEET_INSTANCES_TEMPLATE)
return template.render(fleet_id=fleet_id, instances=instances)
def describe_fleets(self):
def describe_fleets(self) -> str:
fleet_ids = self._get_multi_param("FleetId.")
requests = self.ec2_backend.describe_fleets(fleet_ids)
@ -24,7 +24,7 @@ class Fleets(BaseResponse):
rend = template.render(requests=requests)
return rend
def create_fleet(self):
def create_fleet(self) -> str:
on_demand_options = self._get_multi_param_dict("OnDemandOptions")
spot_options = self._get_multi_param_dict("SpotOptions")
target_capacity_specification = self._get_multi_param_dict(

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

@ -3,7 +3,7 @@ from ._base_response import EC2BaseResponse
class FlowLogs(EC2BaseResponse):
def create_flow_logs(self):
def create_flow_logs(self) -> str:
resource_type = self._get_param("ResourceType")
resource_ids = self._get_multi_param("ResourceId")
traffic_type = self._get_param("TrafficType")
@ -15,8 +15,7 @@ class FlowLogs(EC2BaseResponse):
max_aggregation_interval = self._get_param("MaxAggregationInterval")
validate_resource_ids(resource_ids)
tags = self._parse_tag_specification()
tags = tags.get("vpc-flow-log", {})
tags = self._parse_tag_specification().get("vpc-flow-log", {})
self.error_on_dryrun()
@ -36,7 +35,7 @@ class FlowLogs(EC2BaseResponse):
template = self.response_template(CREATE_FLOW_LOGS_RESPONSE)
return template.render(flow_logs=flow_logs, errors=errors)
def describe_flow_logs(self):
def describe_flow_logs(self) -> str:
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)
@ -46,7 +45,7 @@ class FlowLogs(EC2BaseResponse):
template = self.response_template(DESCRIBE_FLOW_LOGS_RESPONSE)
return template.render(flow_logs=flow_logs)
def delete_flow_logs(self):
def delete_flow_logs(self) -> str:
flow_log_ids = self._get_multi_param("FlowLogId")
self.error_on_dryrun()

6
moto/ec2/responses/general.py
View File

@ -1,8 +1,8 @@
from moto.core.responses import BaseResponse
from ._base_response import EC2BaseResponse
class General(BaseResponse):
def get_console_output(self):
class General(EC2BaseResponse):
def get_console_output(self) -> str:
instance_id = self._get_param("InstanceId")
if not instance_id:
# For compatibility with boto.

10
moto/ec2/responses/hosts.py
View File

@ -2,9 +2,9 @@ from ._base_response import EC2BaseResponse
class HostsResponse(EC2BaseResponse):
def allocate_hosts(self):
def allocate_hosts(self) -> str:
params = self._get_params()
quantity = int(params.get("Quantity"))
quantity = int(params.get("Quantity")) # type: ignore
host_recovery = params.get("HostRecovery")
zone = params.get("AvailabilityZone")
instance_type = params.get("InstanceType")
@ -24,14 +24,14 @@ class HostsResponse(EC2BaseResponse):
template = self.response_template(EC2_ALLOCATE_HOSTS)
return template.render(host_ids=host_ids)
def describe_hosts(self):
def describe_hosts(self) -> str:
host_ids = list(self._get_params().get("HostId", {}).values())
filters = self._filters_from_querystring()
hosts = self.ec2_backend.describe_hosts(host_ids, filters)
template = self.response_template(EC2_DESCRIBE_HOSTS)
return template.render(account_id=self.current_account, hosts=hosts)
def modify_hosts(self):
def modify_hosts(self) -> str:
params = self._get_params()
host_ids = list(self._get_params().get("HostId", {}).values())
auto_placement = params.get("AutoPlacement")
@ -44,7 +44,7 @@ class HostsResponse(EC2BaseResponse):
template = self.response_template(EC2_MODIFY_HOSTS)
return template.render(host_ids=host_ids)
def release_hosts(self):
def release_hosts(self) -> str:
host_ids = list(self._get_params().get("HostId", {}).values())
self.ec2_backend.release_hosts(host_ids)
template = self.response_template(EC2_RELEASE_HOSTS)

12
moto/ec2/responses/iam_instance_profiles.py
View File

@ -1,8 +1,8 @@
from moto.core.responses import BaseResponse
from ._base_response import EC2BaseResponse
class IamInstanceProfiles(BaseResponse):
def associate_iam_instance_profile(self):
class IamInstanceProfiles(EC2BaseResponse):
def associate_iam_instance_profile(self) -> str:
instance_id = self._get_param("InstanceId")
iam_instance_profile_name = self._get_param("IamInstanceProfile.Name")
iam_instance_profile_arn = self._get_param("IamInstanceProfile.Arn")
@ -12,7 +12,7 @@ class IamInstanceProfiles(BaseResponse):
template = self.response_template(IAM_INSTANCE_PROFILE_RESPONSE)
return template.render(iam_association=iam_association, state="associating")
def describe_iam_instance_profile_associations(self):
def describe_iam_instance_profile_associations(self) -> str:
association_ids = self._get_multi_param("AssociationId")
filters = self._get_object_map("Filter")
max_items = self._get_param("MaxItems")
@ -26,7 +26,7 @@ class IamInstanceProfiles(BaseResponse):
template = self.response_template(DESCRIBE_IAM_INSTANCE_PROFILE_RESPONSE)
return template.render(iam_associations=iam_associations, next_token=next_token)
def disassociate_iam_instance_profile(self):
def disassociate_iam_instance_profile(self) -> str:
association_id = self._get_param("AssociationId")
iam_association = self.ec2_backend.disassociate_iam_instance_profile(
association_id
@ -34,7 +34,7 @@ class IamInstanceProfiles(BaseResponse):
template = self.response_template(IAM_INSTANCE_PROFILE_RESPONSE)
return template.render(iam_association=iam_association, state="disassociating")
def replace_iam_instance_profile_association(self):
def replace_iam_instance_profile_association(self) -> str:
association_id = self._get_param("AssociationId")
iam_instance_profile_name = self._get_param("IamInstanceProfile.Name")
iam_instance_profile_arn = self._get_param("IamInstanceProfile.Arn")

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

@ -1,3 +1,4 @@
from typing import Any, Dict, List, Optional
from moto.core.utils import camelcase_to_underscores
from moto.ec2.exceptions import (
MissingParameterError,
@ -11,7 +12,7 @@ from ._base_response import EC2BaseResponse
class InstanceResponse(EC2BaseResponse):
def describe_instances(self):
def describe_instances(self) -> str:
self.error_on_dryrun()
filter_dict = self._filters_from_querystring()
instance_ids = self._get_multi_param("InstanceId")
@ -44,7 +45,7 @@ class InstanceResponse(EC2BaseResponse):
.replace("False", "false")
)
def run_instances(self):
def run_instances(self) -> str:
min_count = int(self._get_param("MinCount", if_none="1"))
image_id = self._get_param("ImageId")
owner_id = self._get_param("OwnerId")
@ -108,7 +109,7 @@ class InstanceResponse(EC2BaseResponse):
account_id=self.current_account, reservation=new_reservation
)
def terminate_instances(self):
def terminate_instances(self) -> str:
instance_ids = self._get_multi_param("InstanceId")
self.error_on_dryrun()
@ -126,7 +127,7 @@ class InstanceResponse(EC2BaseResponse):
template = self.response_template(EC2_TERMINATE_INSTANCES)
return template.render(instances=instances)
def reboot_instances(self):
def reboot_instances(self) -> str:
instance_ids = self._get_multi_param("InstanceId")
self.error_on_dryrun()
@ -135,7 +136,7 @@ class InstanceResponse(EC2BaseResponse):
template = self.response_template(EC2_REBOOT_INSTANCES)
return template.render(instances=instances)
def stop_instances(self):
def stop_instances(self) -> str:
instance_ids = self._get_multi_param("InstanceId")
self.error_on_dryrun()
@ -144,7 +145,7 @@ class InstanceResponse(EC2BaseResponse):
template = self.response_template(EC2_STOP_INSTANCES)
return template.render(instances=instances)
def start_instances(self):
def start_instances(self) -> str:
instance_ids = self._get_multi_param("InstanceId")
self.error_on_dryrun()
@ -152,7 +153,9 @@ class InstanceResponse(EC2BaseResponse):
template = self.response_template(EC2_START_INSTANCES)
return template.render(instances=instances)
def _get_list_of_dict_params(self, param_prefix, _dct):
def _get_list_of_dict_params(
self, param_prefix: str, _dct: Dict[str, Any]
) -> List[Any]:
"""
Simplified version of _get_dict_param
Allows you to pass in a custom dict instead of using self.querystring by default
@ -163,7 +166,7 @@ class InstanceResponse(EC2BaseResponse):
params.append(value)
return params
def describe_instance_status(self):
def describe_instance_status(self) -> str:
instance_ids = self._get_multi_param("InstanceId")
include_all_instances = self._get_param("IncludeAllInstances") == "true"
filters = self._get_list_prefix("Filter")
@ -179,7 +182,7 @@ class InstanceResponse(EC2BaseResponse):
template = self.response_template(EC2_INSTANCE_STATUS)
return template.render(instances=instances)
def describe_instance_types(self):
def describe_instance_types(self) -> str:
instance_type_filters = self._get_multi_param("InstanceType")
filter_dict = self._filters_from_querystring()
instance_types = self.ec2_backend.describe_instance_types(
@ -188,7 +191,7 @@ class InstanceResponse(EC2BaseResponse):
template = self.response_template(EC2_DESCRIBE_INSTANCE_TYPES)
return template.render(instance_types=instance_types)
def describe_instance_type_offerings(self):
def describe_instance_type_offerings(self) -> str:
location_type_filters = self._get_param("LocationType")
filter_dict = self._filters_from_querystring()
offerings = self.ec2_backend.describe_instance_type_offerings(
@ -197,7 +200,7 @@ class InstanceResponse(EC2BaseResponse):
template = self.response_template(EC2_DESCRIBE_INSTANCE_TYPE_OFFERINGS)
return template.render(instance_type_offerings=offerings)
def describe_instance_attribute(self):
def describe_instance_attribute(self) -> str:
# TODO this and modify below should raise IncorrectInstanceState if
# instance not in stopped state
attribute = self._get_param("Attribute")
@ -213,7 +216,7 @@ class InstanceResponse(EC2BaseResponse):
return template.render(instance=instance, attribute=attribute, value=value)
def describe_instance_credit_specifications(self):
def describe_instance_credit_specifications(self) -> str:
instance_ids = self._get_multi_param("InstanceId")
instance = self.ec2_backend.describe_instance_credit_specifications(
instance_ids
@ -221,7 +224,7 @@ class InstanceResponse(EC2BaseResponse):
template = self.response_template(EC2_DESCRIBE_INSTANCE_CREDIT_SPECIFICATIONS)
return template.render(instances=instance)
def modify_instance_attribute(self):
def modify_instance_attribute(self) -> str:
handlers = [
self._attribute_value_handler,
self._dot_value_instance_attribute_handler,
@ -241,7 +244,7 @@ class InstanceResponse(EC2BaseResponse):
)
raise NotImplementedError(msg)
def _block_device_mapping_handler(self):
def _block_device_mapping_handler(self) -> Optional[str]:
"""
Handles requests which are generated by code similar to:
@ -286,8 +289,9 @@ class InstanceResponse(EC2BaseResponse):
if mapping_counter > 1:
return EC2_MODIFY_INSTANCE_ATTRIBUTE
return None
def _dot_value_instance_attribute_handler(self):
def _dot_value_instance_attribute_handler(self) -> Optional[str]:
attribute_key = None
for key, value in self.querystring.items():
if ".Value" in key:
@ -295,11 +299,11 @@ class InstanceResponse(EC2BaseResponse):
break
if not attribute_key:
return
return None
self.error_on_dryrun()
value = self.querystring.get(attribute_key)[0]
value = self.querystring.get(attribute_key)[0] # type: ignore
normalized_attribute = camelcase_to_underscores(attribute_key.split(".")[0])
instance_id = self._get_param("InstanceId")
self.ec2_backend.modify_instance_attribute(
@ -307,11 +311,11 @@ class InstanceResponse(EC2BaseResponse):
)
return EC2_MODIFY_INSTANCE_ATTRIBUTE
def _attribute_value_handler(self):
def _attribute_value_handler(self) -> Optional[str]:
attribute_key = self._get_param("Attribute")
if attribute_key is None:
return
return None
self.error_on_dryrun()
@ -323,11 +327,11 @@ class InstanceResponse(EC2BaseResponse):
)
return EC2_MODIFY_INSTANCE_ATTRIBUTE
def _security_grp_instance_attribute_handler(self):
def _security_grp_instance_attribute_handler(self) -> str:
new_security_grp_list = []
for key in self.querystring:
if "GroupId." in key:
new_security_grp_list.append(self.querystring.get(key)[0])
new_security_grp_list.append(self.querystring.get(key)[0]) # type: ignore
instance_id = self._get_param("InstanceId")
self.error_on_dryrun()
@ -337,12 +341,12 @@ class InstanceResponse(EC2BaseResponse):
)
return EC2_MODIFY_INSTANCE_ATTRIBUTE
def _parse_block_device_mapping(self):
def _parse_block_device_mapping(self) -> List[Dict[str, Any]]:
device_mappings = self._get_list_prefix("BlockDeviceMapping")
mappings = []
for device_mapping in device_mappings:
self._validate_block_device_mapping(device_mapping)
device_template = deepcopy(BLOCK_DEVICE_MAPPING_TEMPLATE)
device_template: Dict[str, Any] = deepcopy(BLOCK_DEVICE_MAPPING_TEMPLATE)
device_template["VirtualName"] = device_mapping.get("virtual_name")
device_template["DeviceName"] = device_mapping.get("device_name")
device_template["Ebs"]["SnapshotId"] = device_mapping.get(
@ -368,7 +372,7 @@ class InstanceResponse(EC2BaseResponse):
return mappings
@staticmethod
def _validate_block_device_mapping(device_mapping):
def _validate_block_device_mapping(device_mapping: Dict[str, Any]) -> None: # type: ignore[misc]
from botocore import __version__ as botocore_version
@ -393,7 +397,7 @@ class InstanceResponse(EC2BaseResponse):
raise MissingParameterError("size or snapshotId")
@staticmethod
def _convert_to_bool(bool_str):
def _convert_to_bool(bool_str: Any) -> bool: # type: ignore[misc]
if isinstance(bool_str, bool):
return bool_str

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

@ -2,7 +2,7 @@ from ._base_response import EC2BaseResponse
class InternetGateways(EC2BaseResponse):
def attach_internet_gateway(self):
def attach_internet_gateway(self) -> str:
igw_id = self._get_param("InternetGatewayId")
vpc_id = self._get_param("VpcId")
@ -11,7 +11,7 @@ class InternetGateways(EC2BaseResponse):
self.ec2_backend.attach_internet_gateway(igw_id, vpc_id)
return self.response_template(ATTACH_INTERNET_GATEWAY_RESPONSE).render()
def create_internet_gateway(self):
def create_internet_gateway(self) -> str:
self.error_on_dryrun()
tags = self._get_multi_param("TagSpecification", skip_result_conversion=True)
@ -22,14 +22,14 @@ class InternetGateways(EC2BaseResponse):
internet_gateway=igw
)
def delete_internet_gateway(self):
def delete_internet_gateway(self) -> str:
igw_id = self._get_param("InternetGatewayId")
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):
def describe_internet_gateways(self) -> str:
filter_dict = self._filters_from_querystring()
if "InternetGatewayId.1" in self.querystring:
igw_ids = self._get_multi_param("InternetGatewayId")
@ -42,7 +42,7 @@ class InternetGateways(EC2BaseResponse):
template = self.response_template(DESCRIBE_INTERNET_GATEWAYS_RESPONSE)
return template.render(internet_gateways=igws)
def detach_internet_gateway(self):
def detach_internet_gateway(self) -> str:
# TODO validate no instances with EIPs in VPC before detaching
# raise else DependencyViolationError()
igw_id = self._get_param("InternetGatewayId")

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

@ -2,14 +2,14 @@ from moto.core.responses import BaseResponse
class IPAddresses(BaseResponse):
def assign_private_ip_addresses(self):
def assign_private_ip_addresses(self) -> str:
self.error_on_dryrun()
raise NotImplementedError(
"IPAddresses.assign_private_ip_addresses is not yet implemented"
)
def unassign_private_ip_addresses(self):
def unassign_private_ip_addresses(self) -> str:
self.error_on_dryrun()
raise NotImplementedError(

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

@ -2,28 +2,28 @@ from ._base_response import EC2BaseResponse
class KeyPairs(EC2BaseResponse):
def create_key_pair(self):
def create_key_pair(self) -> str:
name = self._get_param("KeyName")
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):
def delete_key_pair(self) -> str:
name = self._get_param("KeyName")
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):
def describe_key_pairs(self) -> str:
names = self._get_multi_param("KeyName")
filters = self._filters_from_querystring()
keypairs = self.ec2_backend.describe_key_pairs(names, filters)
template = self.response_template(DESCRIBE_KEY_PAIRS_RESPONSE)
return template.render(keypairs=keypairs)
def import_key_pair(self):
def import_key_pair(self) -> str:
name = self._get_param("KeyName")
material = self._get_param("PublicKeyMaterial")
self.error_on_dryrun()

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

@ -1,3 +1,4 @@
from typing import Any
from moto.ec2.exceptions import FilterNotImplementedError
from moto.moto_api._internal import mock_random
from ._base_response import EC2BaseResponse
@ -6,7 +7,7 @@ from xml.etree import ElementTree
from xml.dom import minidom
def xml_root(name):
def xml_root(name: str) -> ElementTree.Element:
root = ElementTree.Element(
name, {"xmlns": "http://ec2.amazonaws.com/doc/2016-11-15/"}
)
@ -16,7 +17,7 @@ def xml_root(name):
return root
def xml_serialize(tree, key, value):
def xml_serialize(tree: ElementTree.Element, key: str, value: Any) -> None:
name = key[0].lower() + key[1:]
if isinstance(value, list):
if name[-1] == "s":
@ -42,14 +43,14 @@ def xml_serialize(tree, key, value):
)
def pretty_xml(tree):
def pretty_xml(tree: ElementTree.Element) -> str:
rough = ElementTree.tostring(tree, "utf-8")
parsed = minidom.parseString(rough)
return parsed.toprettyxml(indent=" ")
class LaunchTemplates(EC2BaseResponse):
def create_launch_template(self):
def create_launch_template(self) -> str:
name = self._get_param("LaunchTemplateName")
version_description = self._get_param("VersionDescription")
tag_spec = self._parse_tag_specification()
@ -96,7 +97,7 @@ class LaunchTemplates(EC2BaseResponse):
return pretty_xml(tree)
def create_launch_template_version(self):
def create_launch_template_version(self) -> str:
name = self._get_param("LaunchTemplateName")
tmpl_id = self._get_param("LaunchTemplateId")
if name:
@ -129,7 +130,7 @@ class LaunchTemplates(EC2BaseResponse):
)
return pretty_xml(tree)
def delete_launch_template(self):
def delete_launch_template(self) -> str:
name = self._get_param("LaunchTemplateName")
tid = self._get_param("LaunchTemplateId")
@ -150,7 +151,7 @@ class LaunchTemplates(EC2BaseResponse):
return pretty_xml(tree)
def describe_launch_template_versions(self):
def describe_launch_template_versions(self) -> str:
name = self._get_param("LaunchTemplateName")
template_id = self._get_param("LaunchTemplateId")
if name:
@ -222,7 +223,7 @@ class LaunchTemplates(EC2BaseResponse):
return pretty_xml(tree)
def describe_launch_templates(self):
def describe_launch_templates(self) -> str:
max_results = self._get_int_param("MaxResults", 15)
template_names = self._get_multi_param("LaunchTemplateName")
template_ids = self._get_multi_param("LaunchTemplateId")

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

@ -2,12 +2,12 @@ from moto.core.responses import BaseResponse
class Monitoring(BaseResponse):
def monitor_instances(self):
def monitor_instances(self) -> str:
self.error_on_dryrun()
raise NotImplementedError("Monitoring.monitor_instances is not yet implemented")
def unmonitor_instances(self):
def unmonitor_instances(self) -> str:
self.error_on_dryrun()
raise NotImplementedError(

9
moto/ec2/responses/nat_gateways.py
View File

@ -3,12 +3,11 @@ from ._base_response import EC2BaseResponse
class NatGateways(EC2BaseResponse):
def create_nat_gateway(self):
def create_nat_gateway(self) -> str:
subnet_id = self._get_param("SubnetId")
allocation_id = self._get_param("AllocationId")
connectivity_type = self._get_param("ConnectivityType")
tags = self._get_multi_param("TagSpecification")
tags = add_tag_specification(tags)
tags = add_tag_specification(self._get_multi_param("TagSpecification"))
nat_gateway = self.ec2_backend.create_nat_gateway(
subnet_id=subnet_id,
@ -19,13 +18,13 @@ class NatGateways(EC2BaseResponse):
template = self.response_template(CREATE_NAT_GATEWAY)
return template.render(nat_gateway=nat_gateway)
def delete_nat_gateway(self):
def delete_nat_gateway(self) -> str:
nat_gateway_id = self._get_param("NatGatewayId")
nat_gateway = self.ec2_backend.delete_nat_gateway(nat_gateway_id)
template = self.response_template(DELETE_NAT_GATEWAY_RESPONSE)
return template.render(nat_gateway=nat_gateway)
def describe_nat_gateways(self):
def describe_nat_gateways(self) -> str:
filters = self._filters_from_querystring()
nat_gateway_ids = self._get_multi_param("NatGatewayId")
nat_gateways = self.ec2_backend.describe_nat_gateways(filters, nat_gateway_ids)

14
moto/ec2/responses/network_acls.py
View File

@ -2,7 +2,7 @@ from ._base_response import EC2BaseResponse
class NetworkACLs(EC2BaseResponse):
def create_network_acl(self):
def create_network_acl(self) -> str:
vpc_id = self._get_param("VpcId")
tags = self._get_multi_param("TagSpecification")
if tags:
@ -11,7 +11,7 @@ class NetworkACLs(EC2BaseResponse):
template = self.response_template(CREATE_NETWORK_ACL_RESPONSE)
return template.render(network_acl=network_acl)
def create_network_acl_entry(self):
def create_network_acl_entry(self) -> str:
network_acl_id = self._get_param("NetworkAclId")
rule_number = self._get_param("RuleNumber")
protocol = self._get_param("Protocol")
@ -39,13 +39,13 @@ class NetworkACLs(EC2BaseResponse):
template = self.response_template(CREATE_NETWORK_ACL_ENTRY_RESPONSE)
return template.render(network_acl_entry=network_acl_entry)
def delete_network_acl(self):
def delete_network_acl(self) -> str:
network_acl_id = self._get_param("NetworkAclId")
self.ec2_backend.delete_network_acl(network_acl_id)
template = self.response_template(DELETE_NETWORK_ACL_ASSOCIATION)
return template.render()
def delete_network_acl_entry(self):
def delete_network_acl_entry(self) -> str:
network_acl_id = self._get_param("NetworkAclId")
rule_number = self._get_param("RuleNumber")
egress = self._get_param("Egress")
@ -53,7 +53,7 @@ class NetworkACLs(EC2BaseResponse):
template = self.response_template(DELETE_NETWORK_ACL_ENTRY_RESPONSE)
return template.render()
def replace_network_acl_entry(self):
def replace_network_acl_entry(self) -> str:
network_acl_id = self._get_param("NetworkAclId")
rule_number = self._get_param("RuleNumber")
protocol = self._get_param("Protocol")
@ -81,14 +81,14 @@ class NetworkACLs(EC2BaseResponse):
template = self.response_template(REPLACE_NETWORK_ACL_ENTRY_RESPONSE)
return template.render()
def describe_network_acls(self):
def describe_network_acls(self) -> str:
network_acl_ids = self._get_multi_param("NetworkAclId")
filters = self._filters_from_querystring()
network_acls = self.ec2_backend.describe_network_acls(network_acl_ids, filters)
template = self.response_template(DESCRIBE_NETWORK_ACL_RESPONSE)
return template.render(network_acls=network_acls)
def replace_network_acl_association(self):
def replace_network_acl_association(self) -> str:
association_id = self._get_param("AssociationId")
network_acl_id = self._get_param("NetworkAclId")

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

@ -1,22 +0,0 @@
from moto.core.responses import BaseResponse
class PlacementGroups(BaseResponse):
def create_placement_group(self):
self.error_on_dryrun()
raise NotImplementedError(
"PlacementGroups.create_placement_group is not yet implemented"
)
def delete_placement_group(self):
self.error_on_dryrun()
raise NotImplementedError(
"PlacementGroups.delete_placement_group is not yet implemented"
)
def describe_placement_groups(self):
raise NotImplementedError(
"PlacementGroups.describe_placement_groups is not yet implemented"
)

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

@ -2,36 +2,36 @@ from moto.core.responses import BaseResponse
class ReservedInstances(BaseResponse):
def cancel_reserved_instances_listing(self):
def cancel_reserved_instances_listing(self) -> None:
self.error_on_dryrun()
raise NotImplementedError(
"ReservedInstances.cancel_reserved_instances_listing is not yet implemented"
)
def create_reserved_instances_listing(self):
def create_reserved_instances_listing(self) -> None:
self.error_on_dryrun()
raise NotImplementedError(
"ReservedInstances.create_reserved_instances_listing is not yet implemented"
)
def describe_reserved_instances(self):
def describe_reserved_instances(self) -> None:
raise NotImplementedError(
"ReservedInstances.describe_reserved_instances is not yet implemented"
)
def describe_reserved_instances_listings(self):
def describe_reserved_instances_listings(self) -> None:
raise NotImplementedError(
"ReservedInstances.describe_reserved_instances_listings is not yet implemented"
)
def describe_reserved_instances_offerings(self):
def describe_reserved_instances_offerings(self) -> None:
raise NotImplementedError(
"ReservedInstances.describe_reserved_instances_offerings is not yet implemented"
)
def purchase_reserved_instances_offering(self):
def purchase_reserved_instances_offering(self) -> None:
self.error_on_dryrun()
raise NotImplementedError(

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

@ -2,7 +2,7 @@ from ._base_response import EC2BaseResponse
class RouteTables(EC2BaseResponse):
def associate_route_table(self):
def associate_route_table(self) -> str:
route_table_id = self._get_param("RouteTableId")
gateway_id = self._get_param("GatewayId")
subnet_id = self._get_param("SubnetId")
@ -12,7 +12,7 @@ class RouteTables(EC2BaseResponse):
template = self.response_template(ASSOCIATE_ROUTE_TABLE_RESPONSE)
return template.render(association_id=association_id)
def create_route(self):
def create_route(self) -> str:
route_table_id = self._get_param("RouteTableId")
destination_cidr_block = self._get_param("DestinationCidrBlock")
destination_ipv6_cidr_block = self._get_param("DestinationIpv6CidrBlock")
@ -44,7 +44,7 @@ class RouteTables(EC2BaseResponse):
template = self.response_template(CREATE_ROUTE_RESPONSE)
return template.render()
def create_route_table(self):
def create_route_table(self) -> str:
vpc_id = self._get_param("VpcId")
tags = self._get_multi_param("TagSpecification", skip_result_conversion=True)
if tags:
@ -53,7 +53,7 @@ class RouteTables(EC2BaseResponse):
template = self.response_template(CREATE_ROUTE_TABLE_RESPONSE)
return template.render(route_table=route_table)
def delete_route(self):
def delete_route(self) -> str:
route_table_id = self._get_param("RouteTableId")
destination_cidr_block = self._get_param("DestinationCidrBlock")
destination_ipv6_cidr_block = self._get_param("DestinationIpv6CidrBlock")
@ -67,26 +67,26 @@ class RouteTables(EC2BaseResponse):
template = self.response_template(DELETE_ROUTE_RESPONSE)
return template.render()
def delete_route_table(self):
def delete_route_table(self) -> str:
route_table_id = self._get_param("RouteTableId")
self.ec2_backend.delete_route_table(route_table_id)
template = self.response_template(DELETE_ROUTE_TABLE_RESPONSE)
return template.render()
def describe_route_tables(self):
def describe_route_tables(self) -> str:
route_table_ids = self._get_multi_param("RouteTableId")
filters = self._filters_from_querystring()
route_tables = self.ec2_backend.describe_route_tables(route_table_ids, filters)
template = self.response_template(DESCRIBE_ROUTE_TABLES_RESPONSE)
return template.render(route_tables=route_tables)
def disassociate_route_table(self):
def disassociate_route_table(self) -> str:
association_id = self._get_param("AssociationId")
self.ec2_backend.disassociate_route_table(association_id)
template = self.response_template(DISASSOCIATE_ROUTE_TABLE_RESPONSE)
return template.render()
def replace_route(self):
def replace_route(self) -> str:
route_table_id = self._get_param("RouteTableId")
destination_cidr_block = self._get_param("DestinationCidrBlock")
destination_ipv6_cidr_block = self._get_param("DestinationIpv6CidrBlock")
@ -116,7 +116,7 @@ class RouteTables(EC2BaseResponse):
template = self.response_template(REPLACE_ROUTE_RESPONSE)
return template.render()
def replace_route_table_association(self):
def replace_route_table_association(self) -> str:
route_table_id = self._get_param("RouteTableId")
association_id = self._get_param("AssociationId")
new_association_id = self.ec2_backend.replace_route_table_association(

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

@ -1,19 +1,20 @@
from typing import Any, Dict, List, Tuple
from ._base_response import EC2BaseResponse
def try_parse_int(value, default=None):
def try_parse_int(value: Any, default: Any = None) -> Any:
try:
return int(value)
except (TypeError, ValueError):
return default
def parse_sg_attributes_from_dict(sg_attributes):
def parse_sg_attributes_from_dict(sg_attributes: Dict[str, Any]) -> Tuple[Any, ...]:
ip_protocol = sg_attributes.get("IpProtocol", [None])[0]
from_port = sg_attributes.get("FromPort", [None])[0]
to_port = sg_attributes.get("ToPort", [None])[0]
ip_ranges = []
ip_ranges: List[Dict[str, Any]] = []
ip_ranges_tree = sg_attributes.get("IpRanges") or {}
for ip_range_idx in sorted(ip_ranges_tree.keys()):
ip_range = {"CidrIp": ip_ranges_tree[ip_range_idx]["CidrIp"][0]}
@ -22,7 +23,7 @@ def parse_sg_attributes_from_dict(sg_attributes):
ip_ranges.append(ip_range)
ip_ranges_tree = sg_attributes.get("Ipv6Ranges") or {}
ip_ranges_tree: Dict[str, Any] = sg_attributes.get("Ipv6Ranges") or {} # type: ignore[no-redef]
for ip_range_idx in sorted(ip_ranges_tree.keys()):
ip_range = {"CidrIpv6": ip_ranges_tree[ip_range_idx]["CidrIpv6"][0]}
if ip_ranges_tree[ip_range_idx].get("Description"):
@ -31,15 +32,15 @@ def parse_sg_attributes_from_dict(sg_attributes):
ip_ranges.append(ip_range)
if "CidrIp" in sg_attributes:
cidr_ip = sg_attributes.get("CidrIp")[0]
cidr_ip = sg_attributes.get("CidrIp")[0] # type: ignore
ip_ranges.append({"CidrIp": cidr_ip})
if "CidrIpv6" in sg_attributes:
cidr_ipv6 = sg_attributes.get("CidrIpv6")[0]
cidr_ipv6 = sg_attributes.get("CidrIpv6")[0] # type: ignore
ip_ranges.append({"CidrIpv6": cidr_ipv6})
source_groups = []
groups_tree = sg_attributes.get("Groups") or {}
source_groups: List[Dict[str, Any]] = []
groups_tree: Dict[str, Any] = sg_attributes.get("Groups") or {}
for group_idx in sorted(groups_tree.keys()):
group_dict = groups_tree[group_idx]
source_group = {}
@ -53,8 +54,8 @@ def parse_sg_attributes_from_dict(sg_attributes):
source_group["OwnerId"] = group_dict["OwnerId"][0]
source_groups.append(source_group)
prefix_list_ids = []
pl_tree = sg_attributes.get("PrefixListIds") or {}
prefix_list_ids: List[Dict[str, Any]] = []
pl_tree: Dict[str, Any] = sg_attributes.get("PrefixListIds") or {}
for pl_index in sorted(pl_tree):
pl_dict = pl_tree.get(pl_index, {})
pl_item = {}
@ -68,10 +69,10 @@ def parse_sg_attributes_from_dict(sg_attributes):
class SecurityGroups(EC2BaseResponse):
def _process_rules_from_querystring(self):
def _process_rules_from_querystring(self) -> Any:
group_name_or_id = self._get_param("GroupName") or self._get_param("GroupId")
querytree = {}
querytree: Dict[str, Any] = {}
for key, value in self.querystring.items():
key_splitted = key.split(".")
key_splitted = [try_parse_int(e, e) for e in key_splitted]
@ -127,7 +128,7 @@ class SecurityGroups(EC2BaseResponse):
prefix_list_ids,
)
def authorize_security_group_egress(self):
def authorize_security_group_egress(self) -> str:
self.error_on_dryrun()
for args in self._process_rules_from_querystring():
@ -136,7 +137,7 @@ class SecurityGroups(EC2BaseResponse):
template = self.response_template(AUTHORIZE_SECURITY_GROUP_EGRESS_RESPONSE)
return template.render(rule=rule, group=group)
def authorize_security_group_ingress(self):
def authorize_security_group_ingress(self) -> str:
self.error_on_dryrun()
for args in self._process_rules_from_querystring():
@ -145,14 +146,11 @@ class SecurityGroups(EC2BaseResponse):
template = self.response_template(AUTHORIZE_SECURITY_GROUP_INGRESS_RESPONSE)
return template.render(rule=rule, group=group)
def create_security_group(self):
def create_security_group(self) -> str:
name = self._get_param("GroupName")
description = self._get_param("GroupDescription")
vpc_id = self._get_param("VpcId")
tags = self._get_multi_param("TagSpecification")
tags = tags[0] if isinstance(tags, list) and len(tags) == 1 else tags
tags = (tags or {}).get("Tag", [])
tags = {t["Key"]: t["Value"] for t in tags}
tags = self._parse_tag_specification().get("security-group", {})
self.error_on_dryrun()
@ -165,7 +163,7 @@ class SecurityGroups(EC2BaseResponse):
template = self.response_template(CREATE_SECURITY_GROUP_RESPONSE)
return template.render(group=group)
def delete_security_group(self):
def delete_security_group(self) -> str:
# TODO this should raise an error if there are instances in the group.
# See
# http://docs.aws.amazon.com/AWSEC2/latest/APIReference/ApiReference-query-DeleteSecurityGroup.html
@ -182,7 +180,7 @@ class SecurityGroups(EC2BaseResponse):
return DELETE_GROUP_RESPONSE
def describe_security_groups(self):
def describe_security_groups(self) -> str:
groupnames = self._get_multi_param("GroupName")
group_ids = self._get_multi_param("GroupId")
filters = self._filters_from_querystring()
@ -194,7 +192,7 @@ class SecurityGroups(EC2BaseResponse):
template = self.response_template(DESCRIBE_SECURITY_GROUPS_RESPONSE)
return template.render(groups=groups)
def describe_security_group_rules(self):
def describe_security_group_rules(self) -> str:
group_id = self._get_param("GroupId")
filters = self._get_param("Filter")
@ -204,21 +202,21 @@ class SecurityGroups(EC2BaseResponse):
template = self.response_template(DESCRIBE_SECURITY_GROUP_RULES_RESPONSE)
return template.render(rules=rules)
def revoke_security_group_egress(self):
def revoke_security_group_egress(self) -> str:
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):
def revoke_security_group_ingress(self) -> str:
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):
def update_security_group_rule_descriptions_ingress(self) -> str:
for args in self._process_rules_from_querystring():
group = self.ec2_backend.update_security_group_rule_descriptions_ingress(
*args
@ -226,7 +224,7 @@ class SecurityGroups(EC2BaseResponse):
self.ec2_backend.sg_old_ingress_ruls[group.id] = group.ingress_rules.copy()
return UPDATE_SECURITY_GROUP_RULE_DESCRIPTIONS_INGRESS
def update_security_group_rule_descriptions_egress(self):
def update_security_group_rule_descriptions_egress(self) -> str:
for args in self._process_rules_from_querystring():
group = self.ec2_backend.update_security_group_rule_descriptions_egress(
*args

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

@ -1,22 +1,22 @@
from moto.core.responses import BaseResponse
from ._base_response import EC2BaseResponse
class Settings(BaseResponse):
def disable_ebs_encryption_by_default(self):
class Settings(EC2BaseResponse):
def disable_ebs_encryption_by_default(self) -> str:
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):
def enable_ebs_encryption_by_default(self) -> str:
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):
def get_ebs_encryption_by_default(self) -> str:
self.error_on_dryrun()
result = self.ec2_backend.get_ebs_encryption_by_default()

14
moto/ec2/responses/spot_fleets.py
View File

@ -1,8 +1,8 @@
from moto.core.responses import BaseResponse
from ._base_response import EC2BaseResponse
class SpotFleets(BaseResponse):
def cancel_spot_fleet_requests(self):
class SpotFleets(EC2BaseResponse):
def cancel_spot_fleet_requests(self) -> str:
spot_fleet_request_ids = self._get_multi_param("SpotFleetRequestId.")
terminate_instances = self._get_bool_param("TerminateInstances")
spot_fleets = self.ec2_backend.cancel_spot_fleet_requests(
@ -11,7 +11,7 @@ class SpotFleets(BaseResponse):
template = self.response_template(CANCEL_SPOT_FLEETS_TEMPLATE)
return template.render(spot_fleets=spot_fleets)
def describe_spot_fleet_instances(self):
def describe_spot_fleet_instances(self) -> str:
spot_fleet_request_id = self._get_param("SpotFleetRequestId")
spot_requests = self.ec2_backend.describe_spot_fleet_instances(
@ -22,14 +22,14 @@ class SpotFleets(BaseResponse):
spot_request_id=spot_fleet_request_id, spot_requests=spot_requests
)
def describe_spot_fleet_requests(self):
def describe_spot_fleet_requests(self) -> str:
spot_fleet_request_ids = self._get_multi_param("SpotFleetRequestId.")
requests = self.ec2_backend.describe_spot_fleet_requests(spot_fleet_request_ids)
template = self.response_template(DESCRIBE_SPOT_FLEET_TEMPLATE)
return template.render(requests=requests)
def modify_spot_fleet_request(self):
def modify_spot_fleet_request(self) -> str:
spot_fleet_request_id = self._get_param("SpotFleetRequestId")
target_capacity = self._get_int_param("TargetCapacity")
terminate_instances = self._get_param(
@ -40,7 +40,7 @@ class SpotFleets(BaseResponse):
)
return self.response_template(MODIFY_SPOT_FLEET_REQUEST_TEMPLATE).render()
def request_spot_fleet(self):
def request_spot_fleet(self) -> str:
spot_config = self._get_multi_param_dict("SpotFleetRequestConfig")
spot_price = spot_config.get("SpotPrice")
target_capacity = spot_config["TargetCapacity"]

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

@ -2,7 +2,7 @@ from ._base_response import EC2BaseResponse
class SpotInstances(EC2BaseResponse):
def cancel_spot_instance_requests(self):
def cancel_spot_instance_requests(self) -> str:
request_ids = self._get_multi_param("SpotInstanceRequestId")
self.error_on_dryrun()
@ -11,26 +11,26 @@ class SpotInstances(EC2BaseResponse):
template = self.response_template(CANCEL_SPOT_INSTANCES_TEMPLATE)
return template.render(requests=requests)
def create_spot_datafeed_subscription(self):
def create_spot_datafeed_subscription(self) -> None:
self.error_on_dryrun()
raise NotImplementedError(
"SpotInstances.create_spot_datafeed_subscription is not yet implemented"
)
def delete_spot_datafeed_subscription(self):
def delete_spot_datafeed_subscription(self) -> None:
self.error_on_dryrun()
raise NotImplementedError(
"SpotInstances.delete_spot_datafeed_subscription is not yet implemented"
)
def describe_spot_datafeed_subscription(self):
def describe_spot_datafeed_subscription(self) -> None:
raise NotImplementedError(
"SpotInstances.describe_spot_datafeed_subscription is not yet implemented"
)
def describe_spot_instance_requests(self):
def describe_spot_instance_requests(self) -> str:
spot_instance_ids = self._get_multi_param("SpotInstanceRequestId")
filters = self._filters_from_querystring()
requests = self.ec2_backend.describe_spot_instance_requests(
@ -39,7 +39,7 @@ class SpotInstances(EC2BaseResponse):
template = self.response_template(DESCRIBE_SPOT_INSTANCES_TEMPLATE)
return template.render(requests=requests)
def describe_spot_price_history(self):
def describe_spot_price_history(self) -> str:
instance_types_filters = self._get_multi_param("InstanceType")
filter_dict = self._filters_from_querystring()
prices = self.ec2_backend.describe_spot_price_history(
@ -48,7 +48,7 @@ class SpotInstances(EC2BaseResponse):
template = self.response_template(DESCRIBE_SPOT_PRICE_HISTORY_TEMPLATE)
return template.render(prices=prices)
def request_spot_instances(self):
def request_spot_instances(self) -> str:
price = self._get_param("SpotPrice")
image_id = self._get_param("LaunchSpecification.ImageId")
count = self._get_int_param("InstanceCount", 1)

15
moto/ec2/responses/subnets.py
View File

@ -5,7 +5,7 @@ from ._base_response import EC2BaseResponse
class Subnets(EC2BaseResponse):
def create_subnet(self):
def create_subnet(self) -> str:
vpc_id = self._get_param("VpcId")
cidr_block = self._get_param("CidrBlock")
ipv6_cidr_block = self._get_param("Ipv6CidrBlock")
@ -30,13 +30,13 @@ class Subnets(EC2BaseResponse):
template = self.response_template(CREATE_SUBNET_RESPONSE)
return template.render(subnet=subnet)
def delete_subnet(self):
def delete_subnet(self) -> str:
subnet_id = self._get_param("SubnetId")
subnet = self.ec2_backend.delete_subnet(subnet_id)
template = self.response_template(DELETE_SUBNET_RESPONSE)
return template.render(subnet=subnet)
def describe_subnets(self):
def describe_subnets(self) -> str:
self.error_on_dryrun()
subnet_ids = self._get_multi_param("SubnetId")
filters = self._filters_from_querystring()
@ -44,19 +44,20 @@ class Subnets(EC2BaseResponse):
template = self.response_template(DESCRIBE_SUBNETS_RESPONSE)
return template.render(subnets=subnets)
def modify_subnet_attribute(self):
def modify_subnet_attribute(self) -> str:
subnet_id = self._get_param("SubnetId")
for attribute in ("MapPublicIpOnLaunch", "AssignIpv6AddressOnCreation"):
if self.querystring.get(f"{attribute}.Value"):
attr_name = camelcase_to_underscores(attribute)
attr_value = self.querystring.get(f"{attribute}.Value")[0]
attr_value = self.querystring[f"{attribute}.Value"][0]
self.ec2_backend.modify_subnet_attribute(
subnet_id, attr_name, attr_value
)
return MODIFY_SUBNET_ATTRIBUTE_RESPONSE
return ""
def associate_subnet_cidr_block(self):
def associate_subnet_cidr_block(self) -> str:
ipv6_cidr_block = self._get_param("Ipv6CidrBlock")
subnet_id = self._get_param("SubnetId")
@ -66,7 +67,7 @@ class Subnets(EC2BaseResponse):
template = self.response_template(ASSOCIATE_SUBNET_CIDR_BLOCK_RESPONSE)
return template.render(subnet_id=subnet_id, association=association)
def disassociate_subnet_cidr_block(self):
def disassociate_subnet_cidr_block(self) -> str:
association_id = self._get_param("AssociationId")
subnet_id, association = self.ec2_backend.disassociate_subnet_cidr_block(

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

@ -4,7 +4,7 @@ from ._base_response import EC2BaseResponse
class TagResponse(EC2BaseResponse):
def create_tags(self):
def create_tags(self) -> str:
resource_ids = self._get_multi_param("ResourceId")
validate_resource_ids(resource_ids)
self.ec2_backend.do_resources_exist(resource_ids)
@ -15,7 +15,7 @@ class TagResponse(EC2BaseResponse):
self.ec2_backend.create_tags(resource_ids, tags)
return CREATE_RESPONSE
def delete_tags(self):
def delete_tags(self) -> str:
resource_ids = self._get_multi_param("ResourceId")
validate_resource_ids(resource_ids)
tags = tags_from_query_string(self.querystring)
@ -25,7 +25,7 @@ class TagResponse(EC2BaseResponse):
self.ec2_backend.delete_tags(resource_ids, tags)
return DELETE_RESPONSE
def describe_tags(self):
def describe_tags(self) -> str:
filters = self._filters_from_querystring()
self.error_on_dryrun()

33
moto/ec2/responses/transit_gateway_attachments.py
View File

@ -3,16 +3,13 @@ from ._base_response import EC2BaseResponse
class TransitGatewayAttachment(EC2BaseResponse):
def create_transit_gateway_vpc_attachment(self):
def create_transit_gateway_vpc_attachment(self) -> str:
options = self._get_multi_param_dict("Options")
subnet_ids = self._get_multi_param("SubnetIds")
transit_gateway_id = self._get_param("TransitGatewayId")
vpc_id = self._get_param("VpcId")
tags = self._get_multi_param("TagSpecifications")
tags = tags[0] if isinstance(tags, list) and len(tags) == 1 else tags
tags = (tags or {}).get("Tag", [])
tags = {t["Key"]: t["Value"] for t in tags}
tags = self._parse_tag_specification().get("transit-gateway-route-table", {})
transit_gateway_attachment = (
self.ec2_backend.create_transit_gateway_vpc_attachment(
@ -26,7 +23,7 @@ class TransitGatewayAttachment(EC2BaseResponse):
template = self.response_template(CREATE_TRANSIT_GATEWAY_VPC_ATTACHMENT)
return template.render(transit_gateway_attachment=transit_gateway_attachment)
def describe_transit_gateway_vpc_attachments(self):
def describe_transit_gateway_vpc_attachments(self) -> str:
transit_gateways_attachment_ids = self._get_multi_param(
"TransitGatewayAttachmentIds"
)
@ -42,7 +39,7 @@ class TransitGatewayAttachment(EC2BaseResponse):
transit_gateway_vpc_attachments=transit_gateway_vpc_attachments
)
def modify_transit_gateway_vpc_attachment(self):
def modify_transit_gateway_vpc_attachment(self) -> str:
add_subnet_ids = self._get_multi_param("AddSubnetIds")
options = self._get_multi_param_dict("Options")
remove_subnet_ids = self._get_multi_param("RemoveSubnetIds")
@ -59,7 +56,7 @@ class TransitGatewayAttachment(EC2BaseResponse):
template = self.response_template(MODIFY_TRANSIT_GATEWAY_VPC_ATTACHMENTS)
return template.render(transit_gateway_attachment=transit_gateway_attachment)
def describe_transit_gateway_attachments(self):
def describe_transit_gateway_attachments(self) -> str:
transit_gateways_attachment_ids = self._get_multi_param(
"TransitGatewayAttachmentIds"
)
@ -73,7 +70,7 @@ class TransitGatewayAttachment(EC2BaseResponse):
template = self.response_template(DESCRIBE_TRANSIT_GATEWAY_ATTACHMENTS)
return template.render(transit_gateway_attachments=transit_gateway_attachments)
def delete_transit_gateway_vpc_attachment(self):
def delete_transit_gateway_vpc_attachment(self) -> str:
transit_gateway_attachment_id = self._get_param("TransitGatewayAttachmentId")
transit_gateway_attachment = (
self.ec2_backend.delete_transit_gateway_vpc_attachment(
@ -83,7 +80,7 @@ class TransitGatewayAttachment(EC2BaseResponse):
template = self.response_template(DELETE_TRANSIT_GATEWAY_VPC_ATTACHMENTS)
return template.render(transit_gateway_attachment=transit_gateway_attachment)
def associate_transit_gateway_route_table(self):
def associate_transit_gateway_route_table(self) -> str:
transit_gateway_attachment_id = self._get_param("TransitGatewayAttachmentId")
transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId")
transit_gateway_association = (
@ -95,7 +92,7 @@ class TransitGatewayAttachment(EC2BaseResponse):
template = self.response_template(TRANSIT_GATEWAY_ASSOCIATION)
return template.render(transit_gateway_association=transit_gateway_association)
def disassociate_transit_gateway_route_table(self):
def disassociate_transit_gateway_route_table(self) -> str:
tgw_attach_id = self._get_param("TransitGatewayAttachmentId")
tgw_rt_id = self._get_param("TransitGatewayRouteTableId")
@ -105,7 +102,7 @@ class TransitGatewayAttachment(EC2BaseResponse):
template = self.response_template(TRANSIT_GATEWAY_DISASSOCIATION)
return template.render(tgw_association=tgw_association)
def enable_transit_gateway_route_table_propagation(self):
def enable_transit_gateway_route_table_propagation(self) -> str:
transit_gateway_attachment_id = self._get_param("TransitGatewayAttachmentId")
transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId")
transit_gateway_propagation = (
@ -117,7 +114,7 @@ class TransitGatewayAttachment(EC2BaseResponse):
template = self.response_template(TRANSIT_GATEWAY_PROPAGATION)
return template.render(transit_gateway_propagation=transit_gateway_propagation)
def disable_transit_gateway_route_table_propagation(self):
def disable_transit_gateway_route_table_propagation(self) -> str:
transit_gateway_attachment_id = self._get_param("TransitGatewayAttachmentId")
transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId")
transit_gateway_propagation = (
@ -129,7 +126,7 @@ class TransitGatewayAttachment(EC2BaseResponse):
template = self.response_template(TRANSIT_GATEWAY_PROPAGATION)
return template.render(transit_gateway_propagation=transit_gateway_propagation)
def create_transit_gateway_peering_attachment(self):
def create_transit_gateway_peering_attachment(self) -> str:
peer_account_id = self._get_param("PeerAccountId")
peer_region = self._get_param("PeerRegion")
peer_transit_gateway_id = self._get_param("PeerTransitGatewayId")
@ -150,7 +147,7 @@ class TransitGatewayAttachment(EC2BaseResponse):
transit_gateway_peering_attachment=transit_gateway_peering_attachment,
)
def describe_transit_gateway_peering_attachments(self):
def describe_transit_gateway_peering_attachments(self) -> str:
transit_gateways_attachment_ids = self._get_multi_param(
"TransitGatewayAttachmentIds"
)
@ -166,7 +163,7 @@ class TransitGatewayAttachment(EC2BaseResponse):
transit_gateway_peering_attachments=transit_gateway_peering_attachments
)
def accept_transit_gateway_peering_attachment(self):
def accept_transit_gateway_peering_attachment(self) -> str:
transit_gateway_attachment_id = self._get_param("TransitGatewayAttachmentId")
transit_gateway_peering_attachment = (
self.ec2_backend.accept_transit_gateway_peering_attachment(
@ -179,7 +176,7 @@ class TransitGatewayAttachment(EC2BaseResponse):
transit_gateway_peering_attachment=transit_gateway_peering_attachment,
)
def delete_transit_gateway_peering_attachment(self):
def delete_transit_gateway_peering_attachment(self) -> str:
transit_gateway_attachment_id = self._get_param("TransitGatewayAttachmentId")
transit_gateway_peering_attachment = (
self.ec2_backend.delete_transit_gateway_peering_attachment(
@ -192,7 +189,7 @@ class TransitGatewayAttachment(EC2BaseResponse):
transit_gateway_peering_attachment=transit_gateway_peering_attachment,
)
def reject_transit_gateway_peering_attachment(self):
def reject_transit_gateway_peering_attachment(self) -> str:
transit_gateway_attachment_id = self._get_param("TransitGatewayAttachmentId")
transit_gateway_peering_attachment = (
self.ec2_backend.reject_transit_gateway_peering_attachment(

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

@ -3,12 +3,9 @@ from moto.utilities.utils import str2bool
class TransitGatewayRouteTable(EC2BaseResponse):
def create_transit_gateway_route_table(self):
def create_transit_gateway_route_table(self) -> str:
transit_gateway_id = self._get_param("TransitGatewayId")
tags = self._get_multi_param("TagSpecifications")
tags = tags[0] if isinstance(tags, list) and len(tags) == 1 else tags
tags = (tags or {}).get("Tag", [])
tags = {t["Key"]: t["Value"] for t in tags}
tags = self._parse_tag_specification().get("transit-gateway-route-table", {})
transit_gateway_route_table = (
self.ec2_backend.create_transit_gateway_route_table(
@ -18,7 +15,7 @@ class TransitGatewayRouteTable(EC2BaseResponse):
template = self.response_template(CREATE_TRANSIT_GATEWAY_ROUTE_TABLE_RESPONSE)
return template.render(transit_gateway_route_table=transit_gateway_route_table)
def describe_transit_gateway_route_tables(self):
def describe_transit_gateway_route_tables(self) -> str:
filters = self._filters_from_querystring()
transit_gateway_route_table_ids = (
self._get_multi_param("TransitGatewayRouteTableIds") or None
@ -33,7 +30,7 @@ class TransitGatewayRouteTable(EC2BaseResponse):
transit_gateway_route_tables=transit_gateway_route_tables
)
def delete_transit_gateway_route_table(self):
def delete_transit_gateway_route_table(self) -> str:
transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId")
transit_gateway_route_table = (
self.ec2_backend.delete_transit_gateway_route_table(
@ -43,7 +40,7 @@ class TransitGatewayRouteTable(EC2BaseResponse):
template = self.response_template(DELETE_TRANSIT_GATEWAY_ROUTE_TABLE_RESPONSE)
return template.render(transit_gateway_route_table=transit_gateway_route_table)
def create_transit_gateway_route(self):
def create_transit_gateway_route(self) -> str:
transit_gateway_attachment_id = self._get_param("TransitGatewayAttachmentId")
destination_cidr_block = self._get_param("DestinationCidrBlock")
transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId")
@ -60,7 +57,7 @@ class TransitGatewayRouteTable(EC2BaseResponse):
destination_cidr_block=destination_cidr_block,
)
def delete_transit_gateway_route(self):
def delete_transit_gateway_route(self) -> str:
destination_cidr_block = self._get_param("DestinationCidrBlock")
transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId")
transit_gateway_route_table = self.ec2_backend.delete_transit_gateway_route(
@ -75,7 +72,7 @@ class TransitGatewayRouteTable(EC2BaseResponse):
del transit_gateway_route_table.routes[destination_cidr_block]
return rendered_template
def search_transit_gateway_routes(self):
def search_transit_gateway_routes(self) -> str:
transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId")
filters = self._filters_from_querystring()
max_results = self._get_param("MaxResults")
@ -87,7 +84,7 @@ class TransitGatewayRouteTable(EC2BaseResponse):
template = self.response_template(SEARCH_TRANSIT_GATEWAY_ROUTES_RESPONSE)
return template.render(transit_gateway_routes=transit_gateway_routes)
def get_transit_gateway_route_table_associations(self):
def get_transit_gateway_route_table_associations(self) -> str:
transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId")
filters = self._filters_from_querystring()
transit_gateway_route_table_associations = (
@ -102,7 +99,7 @@ class TransitGatewayRouteTable(EC2BaseResponse):
transit_gateway_route_table_associations=transit_gateway_route_table_associations
)
def get_transit_gateway_route_table_propagations(self):
def get_transit_gateway_route_table_propagations(self) -> str:
transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId")
filters = self._filters_from_querystring()
transit_gateway_route_table_propagations = (

8
moto/ec2/responses/transit_gateways.py
View File

@ -2,7 +2,7 @@ from ._base_response import EC2BaseResponse
class TransitGateways(EC2BaseResponse):
def create_transit_gateway(self):
def create_transit_gateway(self) -> str:
description = self._get_param("Description") or None
options = self._get_multi_param_dict("Options")
tags = self._get_multi_param("TagSpecification")
@ -32,13 +32,13 @@ class TransitGateways(EC2BaseResponse):
template = self.response_template(CREATE_TRANSIT_GATEWAY_RESPONSE)
return template.render(transit_gateway=transit_gateway)
def delete_transit_gateway(self):
def delete_transit_gateway(self) -> str:
transit_gateway_id = self._get_param("TransitGatewayId")
transit_gateway = self.ec2_backend.delete_transit_gateway(transit_gateway_id)
template = self.response_template(DELETE_TRANSIT_GATEWAY_RESPONSE)
return template.render(transit_gateway=transit_gateway)
def describe_transit_gateways(self):
def describe_transit_gateways(self) -> str:
transit_gateway_ids = self._get_multi_param("TransitGatewayIds")
filters = self._filters_from_querystring()
transit_gateways = self.ec2_backend.describe_transit_gateways(
@ -47,7 +47,7 @@ class TransitGateways(EC2BaseResponse):
template = self.response_template(DESCRIBE_TRANSIT_GATEWAY_RESPONSE)
return template.render(transit_gateways=transit_gateways)
def modify_transit_gateway(self):
def modify_transit_gateway(self) -> str:
transit_gateway_id = self._get_param("TransitGatewayId")
description = self._get_param("Description") or None
options = self._get_multi_param_dict("Options")

15
moto/ec2/responses/virtual_private_gateways.py
View File

@ -2,21 +2,18 @@ from ._base_response import EC2BaseResponse
class VirtualPrivateGateways(EC2BaseResponse):
def attach_vpn_gateway(self):
def attach_vpn_gateway(self) -> str:
vpn_gateway_id = self._get_param("VpnGatewayId")
vpc_id = self._get_param("VpcId")
attachment = self.ec2_backend.attach_vpn_gateway(vpn_gateway_id, vpc_id)
template = self.response_template(ATTACH_VPN_GATEWAY_RESPONSE)
return template.render(attachment=attachment)
def create_vpn_gateway(self):
def create_vpn_gateway(self) -> str:
gateway_type = self._get_param("Type")
amazon_side_asn = self._get_param("AmazonSideAsn")
availability_zone = self._get_param("AvailabilityZone")
tags = self._get_multi_param("TagSpecification")
tags = tags[0] if isinstance(tags, list) and len(tags) == 1 else tags
tags = (tags or {}).get("Tag", [])
tags = {t["Key"]: t["Value"] for t in tags}
tags = self._parse_tag_specification().get("virtual-private-gateway", {})
vpn_gateway = self.ec2_backend.create_vpn_gateway(
gateway_type=gateway_type,
amazon_side_asn=amazon_side_asn,
@ -26,20 +23,20 @@ class VirtualPrivateGateways(EC2BaseResponse):
template = self.response_template(CREATE_VPN_GATEWAY_RESPONSE)
return template.render(vpn_gateway=vpn_gateway)
def delete_vpn_gateway(self):
def delete_vpn_gateway(self) -> str:
vpn_gateway_id = self._get_param("VpnGatewayId")
vpn_gateway = self.ec2_backend.delete_vpn_gateway(vpn_gateway_id)
template = self.response_template(DELETE_VPN_GATEWAY_RESPONSE)
return template.render(vpn_gateway=vpn_gateway)
def describe_vpn_gateways(self):
def describe_vpn_gateways(self) -> str:
filters = self._filters_from_querystring()
vpn_gw_ids = self._get_multi_param("VpnGatewayId")
vpn_gateways = self.ec2_backend.describe_vpn_gateways(filters, vpn_gw_ids)
template = self.response_template(DESCRIBE_VPN_GATEWAYS_RESPONSE)
return template.render(vpn_gateways=vpn_gateways)
def detach_vpn_gateway(self):
def detach_vpn_gateway(self) -> str:
vpn_gateway_id = self._get_param("VpnGatewayId")
vpc_id = self._get_param("VpcId")
attachment = self.ec2_backend.detach_vpn_gateway(vpn_gateway_id, vpc_id)

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

@ -1,16 +0,0 @@
from moto.core.responses import BaseResponse
class VMExport(BaseResponse):
def cancel_export_task(self):
raise NotImplementedError("VMExport.cancel_export_task is not yet implemented")
def create_instance_export_task(self):
raise NotImplementedError(
"VMExport.create_instance_export_task is not yet implemented"
)
def describe_export_tasks(self):
raise NotImplementedError(
"VMExport.describe_export_tasks is not yet implemented"
)

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

@ -1,19 +0,0 @@
from moto.core.responses import BaseResponse
class VMImport(BaseResponse):
def cancel_conversion_task(self):
raise NotImplementedError(
"VMImport.cancel_conversion_task is not yet implemented"
)
def describe_conversion_tasks(self):
raise NotImplementedError(
"VMImport.describe_conversion_tasks is not yet implemented"
)
def import_instance(self):
raise NotImplementedError("VMImport.import_instance is not yet implemented")
def import_volume(self):
raise NotImplementedError("VMImport.import_volume is not yet implemented")

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

@ -1,13 +1,10 @@
from moto.core.responses import BaseResponse
from ._base_response import EC2BaseResponse
class VPCPeeringConnections(BaseResponse):
def create_vpc_peering_connection(self):
class VPCPeeringConnections(EC2BaseResponse):
def create_vpc_peering_connection(self) -> str:
peer_region = self._get_param("PeerRegion")
tags = self._get_multi_param("TagSpecification")
tags = tags[0] if isinstance(tags, list) and len(tags) == 1 else tags
tags = (tags or {}).get("Tag", [])
tags = {t["Key"]: t["Value"] for t in tags}
tags = self._parse_tag_specification().get("vpc-peering-connection", {})
if peer_region == self.region or peer_region is None:
peer_vpc = self.ec2_backend.get_vpc(self._get_param("PeerVpcId"))
@ -22,13 +19,13 @@ class VPCPeeringConnections(BaseResponse):
template = self.response_template(CREATE_VPC_PEERING_CONNECTION_RESPONSE)
return template.render(account_id=self.current_account, vpc_pcx=vpc_pcx)
def delete_vpc_peering_connection(self):
def delete_vpc_peering_connection(self) -> str:
vpc_pcx_id = self._get_param("VpcPeeringConnectionId")
vpc_pcx = self.ec2_backend.delete_vpc_peering_connection(vpc_pcx_id)
template = self.response_template(DELETE_VPC_PEERING_CONNECTION_RESPONSE)
return template.render(vpc_pcx=vpc_pcx)
def describe_vpc_peering_connections(self):
def describe_vpc_peering_connections(self) -> str:
ids = self._get_multi_param("VpcPeeringConnectionId")
vpc_pcxs = self.ec2_backend.describe_vpc_peering_connections(
vpc_peering_ids=ids
@ -36,19 +33,19 @@ class VPCPeeringConnections(BaseResponse):
template = self.response_template(DESCRIBE_VPC_PEERING_CONNECTIONS_RESPONSE)
return template.render(account_id=self.current_account, vpc_pcxs=vpc_pcxs)
def accept_vpc_peering_connection(self):
def accept_vpc_peering_connection(self) -> str:
vpc_pcx_id = self._get_param("VpcPeeringConnectionId")
vpc_pcx = self.ec2_backend.accept_vpc_peering_connection(vpc_pcx_id)
template = self.response_template(ACCEPT_VPC_PEERING_CONNECTION_RESPONSE)
return template.render(account_id=self.current_account, vpc_pcx=vpc_pcx)
def reject_vpc_peering_connection(self):
def reject_vpc_peering_connection(self) -> str:
vpc_pcx_id = self._get_param("VpcPeeringConnectionId")
self.ec2_backend.reject_vpc_peering_connection(vpc_pcx_id)
template = self.response_template(REJECT_VPC_PEERING_CONNECTION_RESPONSE)
return template.render()
def modify_vpc_peering_connection_options(self):
def modify_vpc_peering_connection_options(self) -> str:
vpc_pcx_id = self._get_param("VpcPeeringConnectionId")
accepter_options = self._get_multi_param_dict(
"AccepterPeeringConnectionOptions"

17
moto/ec2/responses/vpc_service_configuration.py
View File

@ -1,10 +1,9 @@
from moto.core.responses import BaseResponse
from ._base_response import EC2BaseResponse
from ..exceptions import NoLoadBalancersProvided
class VPCEndpointServiceConfiguration(BaseResponse):
def create_vpc_endpoint_service_configuration(self):
class VPCEndpointServiceConfiguration(EC2BaseResponse):
def create_vpc_endpoint_service_configuration(self) -> str:
gateway_lbs = self._get_multi_param("GatewayLoadBalancerArn")
network_lbs = self._get_multi_param("NetworkLoadBalancerArn")
if not gateway_lbs and not network_lbs:
@ -27,7 +26,7 @@ class VPCEndpointServiceConfiguration(BaseResponse):
template = self.response_template(CREATE_VPC_ENDPOINT_SERVICE_CONFIGURATION)
return template.render(config=config)
def describe_vpc_endpoint_service_configurations(self):
def describe_vpc_endpoint_service_configurations(self) -> str:
service_ids = self._get_multi_param("ServiceId")
configs = self.ec2_backend.describe_vpc_endpoint_service_configurations(
@ -37,7 +36,7 @@ class VPCEndpointServiceConfiguration(BaseResponse):
template = self.response_template(DESCRIBE_VPC_ENDPOINT_SERVICE_CONFIGURATION)
return template.render(configs=configs)
def delete_vpc_endpoint_service_configurations(self):
def delete_vpc_endpoint_service_configurations(self) -> str:
service_ids = self._get_multi_param("ServiceId")
missing_configs = self.ec2_backend.delete_vpc_endpoint_service_configurations(
service_ids
@ -46,7 +45,7 @@ class VPCEndpointServiceConfiguration(BaseResponse):
template = self.response_template(DELETE_VPC_ENDPOINT_SERVICE_CONFIGURATION)
return template.render(missing=missing_configs)
def describe_vpc_endpoint_service_permissions(self):
def describe_vpc_endpoint_service_permissions(self) -> str:
service_id = self._get_param("ServiceId")
principals = self.ec2_backend.describe_vpc_endpoint_service_permissions(
@ -56,7 +55,7 @@ class VPCEndpointServiceConfiguration(BaseResponse):
template = self.response_template(DESCRIBE_VPC_ENDPOINT_SERVICE_PERMISSIONS)
return template.render(principals=principals)
def modify_vpc_endpoint_service_configuration(self):
def modify_vpc_endpoint_service_configuration(self) -> str:
service_id = self._get_param("ServiceId")
private_dns_name = self._get_param("PrivateDnsName")
acceptance_required = self._get_param("AcceptanceRequired")
@ -77,7 +76,7 @@ class VPCEndpointServiceConfiguration(BaseResponse):
return MODIFY_VPC_ENDPOINT_SERVICE_CONFIGURATION
def modify_vpc_endpoint_service_permissions(self):
def modify_vpc_endpoint_service_permissions(self) -> str:
service_id = self._get_param("ServiceId")
add_principals = self._get_multi_param("AddAllowedPrincipals")
remove_principals = self._get_multi_param("RemoveAllowedPrincipals")

66
moto/ec2/responses/vpcs.py
View File

@ -4,20 +4,20 @@ from ._base_response import EC2BaseResponse
class VPCs(EC2BaseResponse):
def _get_doc_date(self):
def _get_doc_date(self) -> str:
return (
"2013-10-15"
if "Boto/" in self.headers.get("user-agent", "")
else "2016-11-15"
)
def create_default_vpc(self):
def create_default_vpc(self) -> str:
vpc = self.ec2_backend.create_default_vpc()
doc_date = self._get_doc_date()
template = self.response_template(CREATE_VPC_RESPONSE)
return template.render(vpc=vpc, doc_date=doc_date)
def create_vpc(self):
def create_vpc(self) -> str:
cidr_block = self._get_param("CidrBlock")
tags = self._get_multi_param("TagSpecification")
instance_tenancy = self._get_param("InstanceTenancy", if_none="default")
@ -43,13 +43,13 @@ class VPCs(EC2BaseResponse):
template = self.response_template(CREATE_VPC_RESPONSE)
return template.render(vpc=vpc, doc_date=doc_date)
def delete_vpc(self):
def delete_vpc(self) -> str:
vpc_id = self._get_param("VpcId")
vpc = self.ec2_backend.delete_vpc(vpc_id)
template = self.response_template(DELETE_VPC_RESPONSE)
return template.render(vpc=vpc)
def describe_vpcs(self):
def describe_vpcs(self) -> str:
self.error_on_dryrun()
vpc_ids = self._get_multi_param("VpcId")
filters = self._filters_from_querystring()
@ -62,13 +62,13 @@ class VPCs(EC2BaseResponse):
template = self.response_template(DESCRIBE_VPCS_RESPONSE)
return template.render(vpcs=vpcs, doc_date=doc_date, region=self.region)
def modify_vpc_tenancy(self):
def modify_vpc_tenancy(self) -> str:
vpc_id = self._get_param("VpcId")
tenancy = self._get_param("InstanceTenancy")
self.ec2_backend.modify_vpc_tenancy(vpc_id, tenancy)
return self.response_template(MODIFY_VPC_TENANCY_RESPONSE).render()
def describe_vpc_attribute(self):
def describe_vpc_attribute(self) -> str:
vpc_id = self._get_param("VpcId")
attribute = self._get_param("Attribute")
attr_name = camelcase_to_underscores(attribute)
@ -76,7 +76,7 @@ class VPCs(EC2BaseResponse):
template = self.response_template(DESCRIBE_VPC_ATTRIBUTE_RESPONSE)
return template.render(vpc_id=vpc_id, attribute=attribute, value=value)
def describe_vpc_classic_link_dns_support(self):
def describe_vpc_classic_link_dns_support(self) -> str:
vpc_ids = self._get_multi_param("VpcIds")
filters = self._filters_from_querystring()
vpcs = self.ec2_backend.describe_vpcs(vpc_ids=vpc_ids, filters=filters)
@ -86,7 +86,7 @@ class VPCs(EC2BaseResponse):
)
return template.render(vpcs=vpcs, doc_date=doc_date)
def enable_vpc_classic_link_dns_support(self):
def enable_vpc_classic_link_dns_support(self) -> str:
vpc_id = self._get_param("VpcId")
classic_link_dns_supported = (
self.ec2_backend.enable_vpc_classic_link_dns_support(vpc_id=vpc_id)
@ -97,7 +97,7 @@ class VPCs(EC2BaseResponse):
classic_link_dns_supported=classic_link_dns_supported, doc_date=doc_date
)
def disable_vpc_classic_link_dns_support(self):
def disable_vpc_classic_link_dns_support(self) -> str:
vpc_id = self._get_param("VpcId")
classic_link_dns_supported = (
self.ec2_backend.disable_vpc_classic_link_dns_support(vpc_id=vpc_id)
@ -108,7 +108,7 @@ class VPCs(EC2BaseResponse):
classic_link_dns_supported=classic_link_dns_supported, doc_date=doc_date
)
def describe_vpc_classic_link(self):
def describe_vpc_classic_link(self) -> str:
vpc_ids = self._get_multi_param("VpcId")
filters = self._filters_from_querystring()
vpcs = self.ec2_backend.describe_vpcs(vpc_ids=vpc_ids, filters=filters)
@ -116,7 +116,7 @@ class VPCs(EC2BaseResponse):
template = self.response_template(DESCRIBE_VPC_CLASSIC_LINK_RESPONSE)
return template.render(vpcs=vpcs, doc_date=doc_date)
def enable_vpc_classic_link(self):
def enable_vpc_classic_link(self) -> str:
vpc_id = self._get_param("VpcId")
classic_link_enabled = self.ec2_backend.enable_vpc_classic_link(vpc_id=vpc_id)
doc_date = self._get_doc_date()
@ -125,7 +125,7 @@ class VPCs(EC2BaseResponse):
classic_link_enabled=classic_link_enabled, doc_date=doc_date
)
def disable_vpc_classic_link(self):
def disable_vpc_classic_link(self) -> str:
vpc_id = self._get_param("VpcId")
classic_link_enabled = self.ec2_backend.disable_vpc_classic_link(vpc_id=vpc_id)
doc_date = self._get_doc_date()
@ -134,7 +134,7 @@ class VPCs(EC2BaseResponse):
classic_link_enabled=classic_link_enabled, doc_date=doc_date
)
def modify_vpc_attribute(self):
def modify_vpc_attribute(self) -> str:
vpc_id = self._get_param("VpcId")
for attribute in (
"EnableDnsSupport",
@ -143,12 +143,12 @@ class VPCs(EC2BaseResponse):
):
if self.querystring.get(f"{attribute}.Value"):
attr_name = camelcase_to_underscores(attribute)
attr_value = self.querystring.get(f"{attribute}.Value")[0]
attr_value = self.querystring[f"{attribute}.Value"][0]
self.ec2_backend.modify_vpc_attribute(vpc_id, attr_name, attr_value)
return MODIFY_VPC_ATTRIBUTE_RESPONSE
return None
return ""
def associate_vpc_cidr_block(self):
def associate_vpc_cidr_block(self) -> str:
vpc_id = self._get_param("VpcId")
amazon_provided_ipv6_cidr_blocks = self._get_param(
"AmazonProvidedIpv6CidrBlock"
@ -175,7 +175,7 @@ class VPCs(EC2BaseResponse):
cidr_block_state="associating",
)
def disassociate_vpc_cidr_block(self):
def disassociate_vpc_cidr_block(self) -> str:
association_id = self._get_param("AssociationId")
value = self.ec2_backend.disassociate_vpc_cidr_block(association_id)
if "::" in value.get("cidr_block", ""):
@ -190,7 +190,7 @@ class VPCs(EC2BaseResponse):
cidr_block_state="disassociating",
)
def create_vpc_endpoint(self):
def create_vpc_endpoint(self) -> str:
vpc_id = self._get_param("VpcId")
service_name = self._get_param("ServiceName")
route_table_ids = self._get_multi_param("RouteTableId")
@ -198,11 +198,10 @@ class VPCs(EC2BaseResponse):
endpoint_type = self._get_param("VpcEndpointType")
policy_document = self._get_param("PolicyDocument")
client_token = self._get_param("ClientToken")
tags = self._get_multi_param("TagSpecification")
private_dns_enabled = self._get_bool_param("PrivateDnsEnabled", if_none=True)
security_group_ids = self._get_multi_param("SecurityGroupId")
tags = add_tag_specification(tags)
tags = add_tag_specification(self._get_multi_param("TagSpecification"))
vpc_end_point = self.ec2_backend.create_vpc_endpoint(
vpc_id=vpc_id,
service_name=service_name,
@ -218,7 +217,7 @@ class VPCs(EC2BaseResponse):
template = self.response_template(CREATE_VPC_END_POINT)
return template.render(vpc_end_point=vpc_end_point)
def modify_vpc_endpoint(self):
def modify_vpc_endpoint(self) -> str:
vpc_id = self._get_param("VpcEndpointId")
add_subnets = self._get_multi_param("AddSubnetId")
add_route_tables = self._get_multi_param("AddRouteTableId")
@ -234,7 +233,7 @@ class VPCs(EC2BaseResponse):
template = self.response_template(MODIFY_VPC_END_POINT)
return template.render()
def describe_vpc_endpoint_services(self):
def describe_vpc_endpoint_services(self) -> str:
vpc_end_point_services = self.ec2_backend.describe_vpc_endpoint_services(
service_names=self._get_multi_param("ServiceName"),
filters=self._get_multi_param("Filter"),
@ -245,7 +244,7 @@ class VPCs(EC2BaseResponse):
template = self.response_template(DESCRIBE_VPC_ENDPOINT_SERVICES_RESPONSE)
return template.render(vpc_end_points=vpc_end_point_services)
def describe_vpc_endpoints(self):
def describe_vpc_endpoints(self) -> str:
vpc_end_points_ids = self._get_multi_param("VpcEndpointId")
filters = self._filters_from_querystring()
vpc_end_points = self.ec2_backend.describe_vpc_endpoints(
@ -256,21 +255,18 @@ class VPCs(EC2BaseResponse):
vpc_end_points=vpc_end_points, account_id=self.current_account
)
def delete_vpc_endpoints(self):
def delete_vpc_endpoints(self) -> str:
vpc_end_points_ids = self._get_multi_param("VpcEndpointId")
self.ec2_backend.delete_vpc_endpoints(vpce_ids=vpc_end_points_ids)
return self.response_template(DELETE_VPC_ENDPOINT_RESPONSE).render()
def create_managed_prefix_list(self):
def create_managed_prefix_list(self) -> str:
address_family = self._get_param("AddressFamily")
max_entries = self._get_param("MaxEntries")
prefix_list_name = self._get_param("PrefixListName")
entry = self._get_multi_param("Entry")
tags = self._get_multi_param("TagSpecification")
tags = tags[0] if isinstance(tags, list) and len(tags) == 1 else tags
tags = (tags or {}).get("Tag", [])
tags = {t["Key"]: t["Value"] for t in tags}
tags = self._parse_tag_specification().get("prefix-list", {})
managed_prefix_list = self.ec2_backend.create_managed_prefix_list(
address_family=address_family,
@ -282,7 +278,7 @@ class VPCs(EC2BaseResponse):
template = self.response_template(CREATE_MANAGED_PREFIX_LIST)
return template.render(managed_prefix_list=managed_prefix_list)
def describe_managed_prefix_lists(self):
def describe_managed_prefix_lists(self) -> str:
prefix_list_ids = self._get_multi_param("PrefixListId")
filters = self._filters_from_querystring()
managed_prefix_lists = self.ec2_backend.describe_managed_prefix_lists(
@ -291,7 +287,7 @@ class VPCs(EC2BaseResponse):
template = self.response_template(DESCRIBE_MANAGED_PREFIX_LIST)
return template.render(managed_prefix_lists=managed_prefix_lists)
def get_managed_prefix_list_entries(self):
def get_managed_prefix_list_entries(self) -> str:
prefix_list_id = self._get_param("PrefixListId")
target_version = self._get_param("TargetVersion")
managed_prefix_list = self.ec2_backend.get_managed_prefix_list_entries(
@ -310,7 +306,7 @@ class VPCs(EC2BaseResponse):
template = self.response_template(GET_MANAGED_PREFIX_LIST_ENTRIES)
return template.render(entries=entries)
def delete_managed_prefix_list(self):
def delete_managed_prefix_list(self) -> str:
prefix_list_id = self._get_param("PrefixListId")
managed_prefix_list = self.ec2_backend.delete_managed_prefix_list(
prefix_list_id
@ -318,7 +314,7 @@ class VPCs(EC2BaseResponse):
template = self.response_template(DELETE_MANAGED_PREFIX_LIST)
return template.render(managed_prefix_list=managed_prefix_list)
def describe_prefix_lists(self):
def describe_prefix_lists(self) -> str:
prefix_list_ids = self._get_multi_param("PrefixListId")
filters = self._filters_from_querystring()
managed_pls = self.ec2_backend.describe_managed_prefix_lists(
@ -327,7 +323,7 @@ class VPCs(EC2BaseResponse):
template = self.response_template(DESCRIBE_PREFIX_LIST)
return template.render(managed_pls=managed_pls)
def modify_managed_prefix_list(self):
def modify_managed_prefix_list(self) -> str:
add_entry = self._get_multi_param("AddEntry")
prefix_list_id = self._get_param("PrefixListId")
current_version = self._get_param("CurrentVersion")

6
moto/ec2/responses/vpn_connections.py
View File

@ -4,7 +4,7 @@ from xml.sax.saxutils import escape
class VPNConnections(EC2BaseResponse):
def create_vpn_connection(self):
def create_vpn_connection(self) -> str:
vpn_conn_type = self._get_param("Type")
cgw_id = self._get_param("CustomerGatewayId")
vgw_id = self._get_param("VpnGatewayId")
@ -26,7 +26,7 @@ class VPNConnections(EC2BaseResponse):
template = self.response_template(CREATE_VPN_CONNECTION_RESPONSE)
return template.render(vpn_connection=vpn_connection)
def delete_vpn_connection(self):
def delete_vpn_connection(self) -> str:
vpn_connection_id = self._get_param("VpnConnectionId")
vpn_connection = self.ec2_backend.delete_vpn_connection(vpn_connection_id)
if vpn_connection.transit_gateway_id:
@ -39,7 +39,7 @@ class VPNConnections(EC2BaseResponse):
template = self.response_template(DELETE_VPN_CONNECTION_RESPONSE)
return template.render(vpn_connection=vpn_connection)
def describe_vpn_connections(self):
def describe_vpn_connections(self) -> str:
vpn_connection_ids = self._get_multi_param("VpnConnectionId")
filters = self._filters_from_querystring()
vpn_connections = self.ec2_backend.describe_vpn_connections(

12
moto/ec2/responses/windows.py
View File

@ -1,20 +1,20 @@
from moto.core.responses import BaseResponse
from ._base_response import EC2BaseResponse
from moto.ec2.utils import utc_date_and_time
class Windows(BaseResponse):
def bundle_instance(self):
class Windows(EC2BaseResponse):
def bundle_instance(self) -> str:
raise NotImplementedError("Windows.bundle_instance is not yet implemented")
def cancel_bundle_task(self):
def cancel_bundle_task(self) -> str:
raise NotImplementedError("Windows.cancel_bundle_task is not yet implemented")
def describe_bundle_tasks(self):
def describe_bundle_tasks(self) -> str:
raise NotImplementedError(
"Windows.describe_bundle_tasks is not yet implemented"
)
def get_password_data(self):
def get_password_data(self) -> str:
instance_id = self._get_param("InstanceId")
password_data = self.ec2_backend.get_password_data(instance_id)
template = self.response_template(GET_PASSWORD_DATA_RESPONSE)

56
moto/ec2/utils.py
View File

@ -7,7 +7,7 @@ from datetime import datetime
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
from typing import Any, Dict, List, TypeVar, Tuple, Optional
from typing import Any, Dict, List, Set, TypeVar, Tuple, Optional, Union
from moto.iam import iam_backends
from moto.moto_api._internal import mock_random as random
@ -64,11 +64,11 @@ EC2_PREFIX_TO_RESOURCE = dict((v, k) for (k, v) in EC2_RESOURCE_TO_PREFIX.items(
HEX_CHARS = list(str(x) for x in range(10)) + ["a", "b", "c", "d", "e", "f"]
def random_resource_id(size=8):
def random_resource_id(size: int = 8) -> str:
return "".join(random.choice(HEX_CHARS) for _ in range(size))
def random_id(prefix="", size=8):
def random_id(prefix: str = "", size: int = 8) -> str:
return f"{prefix}-{random_resource_id(size)}"
@ -244,7 +244,7 @@ def random_dedicated_host_id() -> str:
return random_id(prefix=EC2_RESOURCE_TO_PREFIX["dedicated_host"], size=17)
def random_private_ip(cidr: str = None, ipv6: bool = False) -> str:
def random_private_ip(cidr: Optional[str] = None, ipv6: bool = False) -> str:
# prefix - ula.prefixlen : get number of remaing length for the IP.
# prefix will be 32 for IPv4 and 128 for IPv6.
# random.getrandbits() will generate remaining bits for IPv6 or Ipv4 in decimal format
@ -252,7 +252,7 @@ def random_private_ip(cidr: str = None, ipv6: bool = False) -> str:
if ipv6:
ula = ipaddress.IPv6Network(cidr)
return str(ula.network_address + (random.getrandbits(128 - ula.prefixlen)))
ula = ipaddress.IPv4Network(cidr)
ula = ipaddress.IPv4Network(cidr) # type: ignore[assignment]
return str(ula.network_address + (random.getrandbits(32 - ula.prefixlen)))
if ipv6:
return f"2001::cafe:{random.getrandbits(16)}x/64"
@ -285,7 +285,7 @@ def generate_route_id(
cidr_block: Optional[str],
ipv6_cidr_block: Optional[str] = None,
prefix_list: Optional[str] = None,
):
) -> str:
if ipv6_cidr_block and not cidr_block:
cidr_block = ipv6_cidr_block
if prefix_list and not cidr_block:
@ -315,7 +315,9 @@ def split_route_id(route_id: str) -> Tuple[str, str]:
return values[0], values[1]
def get_attribute_value(parameter, querystring_dict):
def get_attribute_value(
parameter: str, querystring_dict: Dict[str, List[str]]
) -> Union[None, bool, str]:
for key, value in querystring_dict.items():
match = re.search(rf"{parameter}.Value", key)
if match:
@ -325,7 +327,7 @@ def get_attribute_value(parameter, querystring_dict):
return None
def get_object_value(obj, attr):
def get_object_value(obj: Any, attr: str) -> Any:
keys = attr.split(".")
val = obj
for key in keys:
@ -353,25 +355,25 @@ def is_tag_filter(filter_name: str) -> bool:
)
def get_obj_tag(obj, filter_name):
def get_obj_tag(obj: Any, filter_name: str) -> Optional[str]:
tag_name = filter_name.replace("tag:", "", 1)
tags = dict((tag["key"], tag["value"]) for tag in obj.get_tags())
return tags.get(tag_name)
def get_obj_tag_names(obj):
def get_obj_tag_names(obj: Any) -> Set[str]:
tags = set((tag["key"] for tag in obj.get_tags()))
return tags
def get_obj_tag_values(obj, key=None):
def get_obj_tag_values(obj: Any, key: Optional[str] = None) -> Set[str]:
tags = set(
(tag["value"] for tag in obj.get_tags() if tag["key"] == key or key is None)
)
return tags
def add_tag_specification(tags):
def add_tag_specification(tags: Any) -> Dict[str, str]:
tags = tags[0] if isinstance(tags, list) and len(tags) == 1 else tags
tags = (tags or {}).get("Tag", [])
tags = {t["Key"]: t["Value"] for t in tags}
@ -388,7 +390,7 @@ def tag_filter_matches(obj: Any, filter_name: str, filter_values: List[str]) ->
key = filter_name[4:]
tag_values = get_obj_tag_values(obj, key=key)
else:
tag_values = [get_obj_tag(obj, filter_name) or ""]
tag_values = [get_obj_tag(obj, filter_name) or ""] # type: ignore[assignment]
for tag_value in tag_values:
if any(regex.match(tag_value) for regex in regex_filters):
@ -421,7 +423,7 @@ filter_dict_attribute_mapping = {
}
def passes_filter_dict(instance, filter_dict):
def passes_filter_dict(instance: Any, filter_dict: Dict[str, Any]) -> bool:
for filter_name, filter_values in filter_dict.items():
if filter_name in filter_dict_attribute_mapping:
instance_attr = filter_dict_attribute_mapping[filter_name]
@ -440,7 +442,7 @@ def passes_filter_dict(instance, filter_dict):
return True
def instance_value_in_filter_values(instance_value, filter_values):
def instance_value_in_filter_values(instance_value: Any, filter_values: Any) -> bool:
if isinstance(instance_value, list):
if not set(filter_values).intersection(set(instance_value)):
return False
@ -458,11 +460,11 @@ def filter_reservations(
result = []
for reservation in reservations:
new_instances = []
for instance in reservation.instances:
for instance in reservation.instances: # type: ignore[attr-defined]
if passes_filter_dict(instance, filter_dict):
new_instances.append(instance)
if new_instances:
reservation.instances = new_instances
reservation.instances = new_instances # type: ignore[attr-defined]
result.append(reservation)
return result
@ -474,7 +476,7 @@ filter_dict_igw_mapping = {
}
def passes_igw_filter_dict(igw, filter_dict):
def passes_igw_filter_dict(igw: Any, filter_dict: Dict[str, Any]) -> bool:
for filter_name, filter_values in filter_dict.items():
if filter_name in filter_dict_igw_mapping:
igw_attr = filter_dict_igw_mapping[filter_name]
@ -501,7 +503,7 @@ def filter_internet_gateways(
return result
def is_filter_matching(obj, _filter, filter_value):
def is_filter_matching(obj: Any, _filter: str, filter_value: Any) -> bool:
value = obj.get_filter_value(_filter)
if filter_value is None:
@ -583,7 +585,7 @@ def get_prefix(resource_id: str) -> str:
resource_id_prefix = EC2_RESOURCE_TO_PREFIX["reserved-instance"]
else:
# We should probably raise an error here, to make it more obvious this is not yet supported
return None
return None # type: ignore[return-value]
return resource_id_prefix
@ -609,7 +611,7 @@ def is_valid_ipv6_cidr(cird: str) -> bool:
return cidr_pattern_re.match(cird) is not None
def generate_instance_identity_document(instance):
def generate_instance_identity_document(instance: Any) -> Dict[str, Any]:
"""
http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/instance-identity-documents.html
@ -681,12 +683,12 @@ def filter_iam_instance_profile_associations(
filter_passed = True
if filter_dict.get("instance-id"):
if (
iam_instance_association.instance.id
iam_instance_association.instance.id # type: ignore[attr-defined]
not in filter_dict.get("instance-id").values()
):
filter_passed = False
if filter_dict.get("state"):
if iam_instance_association.state not in filter_dict.get("state").values():
if iam_instance_association.state not in filter_dict.get("state").values(): # type: ignore[attr-defined]
filter_passed = False
if filter_passed:
result.append(iam_instance_association)
@ -732,7 +734,7 @@ def describe_tag_filter(
if match:
tag_key_name = match.group(1)
need_delete = True
for tag in instance.get_tags():
for tag in instance.get_tags(): # type: ignore[attr-defined]
if tag.get("key") == tag_key_name and tag.get(
"value"
) in filters.get(key):
@ -746,7 +748,9 @@ def describe_tag_filter(
return result
def gen_moto_amis(described_images, drop_images_missing_keys=True):
def gen_moto_amis(
described_images: List[Dict[str, Any]], drop_images_missing_keys: bool = True
) -> List[Dict[str, Any]]:
"""Convert `boto3.EC2.Client.describe_images` output to form acceptable to `MOTO_AMIS_PATH`
Parameters
@ -804,7 +808,7 @@ def convert_tag_spec(
# IN: [{"ResourceType": _type, "Tag": [{"Key": k, "Value": v}, ..]}]
# (or) [{"ResourceType": _type, "Tags": [{"Key": k, "Value": v}, ..]}] <-- special cfn case
# OUT: {_type: {k: v, ..}}
tags = {}
tags: Dict[str, Dict[str, str]] = {}
for tag_spec in tag_spec_set:
if tag_spec["ResourceType"] not in tags:
tags[tag_spec["ResourceType"]] = {}

5
moto/utilities/utils.py
View File

@ -3,14 +3,15 @@ import hashlib
import pkgutil
from collections.abc import MutableMapping
from typing import Any, Dict, List, TypeVar, Tuple
from typing import Any, Dict, List, TypeVar, Tuple, Optional
def str2bool(v):
def str2bool(v: Any) -> Optional[bool]:
if v in ("yes", True, "true", "True", "TRUE", "t", "1"):
return True
elif v in ("no", False, "false", "False", "FALSE", "f", "0"):
return False
return None
def load_resource(package: str, resource: str) -> Dict[str, Any]:

2
setup.cfg
View File

@ -229,7 +229,7 @@ disable = W,C,R,E
enable = anomalous-backslash-in-string, arguments-renamed, dangerous-default-value, deprecated-module, function-redefined, import-self, redefined-builtin, redefined-outer-name, reimported, pointless-statement, super-with-arguments, unused-argument, unused-import, unused-variable, useless-else-on-loop, wildcard-import
[mypy]
files= moto/a*,moto/b*,moto/c*,moto/d*,moto/ebs/,moto/ec2/models/**/*.py,moto/moto_api
files= moto/a*,moto/b*,moto/c*,moto/d*,moto/ebs/,moto/ec2/,moto/moto_api
show_column_numbers=True
show_error_codes = True
disable_error_code=abstract

2
tests/test_ec2/test_prefix_lists.py
View File

@ -32,7 +32,7 @@ def test_create_with_tags():
MaxEntries=2,
AddressFamily="?",
TagSpecifications=[
{"ResourceType": "", "Tags": [{"Key": "key1", "Value": "val1"}]}
{"ResourceType": "prefix-list", "Tags": [{"Key": "key1", "Value": "val1"}]}
],
)["PrefixList"]
prefix_list.should.have.key("PrefixListId").match("pl-[a-z0-9]+")