diff --git a/moto/ec2/exceptions.py b/moto/ec2/exceptions.py index aad3b6738..fce22c1bb 100644 --- a/moto/ec2/exceptions.py +++ b/moto/ec2/exceptions.py @@ -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__( diff --git a/moto/ec2/models/instances.py b/moto/ec2/models/instances.py index fee0c6348..02c5587b2 100644 --- a/moto/ec2/models/instances.py +++ b/moto/ec2/models/instances.py @@ -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) diff --git a/moto/ec2/responses/__init__.py b/moto/ec2/responses/__init__.py index fcbb99ba6..7c6a9eeac 100644 --- a/moto/ec2/responses/__init__.py +++ b/moto/ec2/responses/__init__.py @@ -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 diff --git a/moto/ec2/responses/_base_response.py b/moto/ec2/responses/_base_response.py index 885286368..5529baea3 100644 --- a/moto/ec2/responses/_base_response.py +++ b/moto/ec2/responses/_base_response.py @@ -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] diff --git a/moto/ec2/responses/account_attributes.py b/moto/ec2/responses/account_attributes.py index 5ecc8ab5f..e5715143f 100644 --- a/moto/ec2/responses/account_attributes.py +++ b/moto/ec2/responses/account_attributes.py @@ -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() diff --git a/moto/ec2/responses/amazon_dev_pay.py b/moto/ec2/responses/amazon_dev_pay.py deleted file mode 100644 index c5a188a35..000000000 --- a/moto/ec2/responses/amazon_dev_pay.py +++ /dev/null @@ -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" - ) diff --git a/moto/ec2/responses/amis.py b/moto/ec2/responses/amis.py index 689cf7f59..a950131ef 100644 --- a/moto/ec2/responses/amis.py +++ b/moto/ec2/responses/amis.py @@ -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") diff --git a/moto/ec2/responses/availability_zones_and_regions.py b/moto/ec2/responses/availability_zones_and_regions.py index c8a03bb9f..0ef8238d0 100644 --- a/moto/ec2/responses/availability_zones_and_regions.py +++ b/moto/ec2/responses/availability_zones_and_regions.py @@ -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) diff --git a/moto/ec2/responses/carrier_gateways.py b/moto/ec2/responses/carrier_gateways.py index 4464cb9e7..7e9a7618f 100644 --- a/moto/ec2/responses/carrier_gateways.py +++ b/moto/ec2/responses/carrier_gateways.py @@ -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() diff --git a/moto/ec2/responses/customer_gateways.py b/moto/ec2/responses/customer_gateways.py index f78fcf6ec..0b51c9858 100644 --- a/moto/ec2/responses/customer_gateways.py +++ b/moto/ec2/responses/customer_gateways.py @@ -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") diff --git a/moto/ec2/responses/dhcp_options.py b/moto/ec2/responses/dhcp_options.py index 76b7b1a60..6c3160004 100644 --- a/moto/ec2/responses/dhcp_options.py +++ b/moto/ec2/responses/dhcp_options.py @@ -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) diff --git a/moto/ec2/responses/egress_only_internet_gateways.py b/moto/ec2/responses/egress_only_internet_gateways.py index aec0dc1fe..317ead69d 100644 --- a/moto/ec2/responses/egress_only_internet_gateways.py +++ b/moto/ec2/responses/egress_only_internet_gateways.py @@ -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 diff --git a/moto/ec2/responses/elastic_block_store.py b/moto/ec2/responses/elastic_block_store.py index c234eeef3..a2b2e3feb 100644 --- a/moto/ec2/responses/elastic_block_store.py +++ b/moto/ec2/responses/elastic_block_store.py @@ -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( diff --git a/moto/ec2/responses/elastic_ip_addresses.py b/moto/ec2/responses/elastic_ip_addresses.py index 6d03ec6ce..dc96c22b9 100644 --- a/moto/ec2/responses/elastic_ip_addresses.py +++ b/moto/ec2/responses/elastic_ip_addresses.py @@ -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 diff --git a/moto/ec2/responses/elastic_network_interfaces.py b/moto/ec2/responses/elastic_network_interfaces.py index 567dd1cfd..af193d4f6 100644 --- a/moto/ec2/responses/elastic_network_interfaces.py +++ b/moto/ec2/responses/elastic_network_interfaces.py @@ -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) diff --git a/moto/ec2/responses/fleets.py b/moto/ec2/responses/fleets.py index a22f61620..9bf247672 100644 --- a/moto/ec2/responses/fleets.py +++ b/moto/ec2/responses/fleets.py @@ -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( diff --git a/moto/ec2/responses/flow_logs.py b/moto/ec2/responses/flow_logs.py index c827d4faa..72c9ee461 100644 --- a/moto/ec2/responses/flow_logs.py +++ b/moto/ec2/responses/flow_logs.py @@ -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() diff --git a/moto/ec2/responses/general.py b/moto/ec2/responses/general.py index 018784a77..2fd98481e 100644 --- a/moto/ec2/responses/general.py +++ b/moto/ec2/responses/general.py @@ -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. diff --git a/moto/ec2/responses/hosts.py b/moto/ec2/responses/hosts.py index daf0d7c29..cb1825c5b 100644 --- a/moto/ec2/responses/hosts.py +++ b/moto/ec2/responses/hosts.py @@ -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) diff --git a/moto/ec2/responses/iam_instance_profiles.py b/moto/ec2/responses/iam_instance_profiles.py index 0d4f2f50c..2695be856 100644 --- a/moto/ec2/responses/iam_instance_profiles.py +++ b/moto/ec2/responses/iam_instance_profiles.py @@ -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") diff --git a/moto/ec2/responses/instances.py b/moto/ec2/responses/instances.py index 380be566b..ed53545fb 100644 --- a/moto/ec2/responses/instances.py +++ b/moto/ec2/responses/instances.py @@ -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 diff --git a/moto/ec2/responses/internet_gateways.py b/moto/ec2/responses/internet_gateways.py index 7428f938a..c8fd1eff2 100644 --- a/moto/ec2/responses/internet_gateways.py +++ b/moto/ec2/responses/internet_gateways.py @@ -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") diff --git a/moto/ec2/responses/ip_addresses.py b/moto/ec2/responses/ip_addresses.py index bd1a6b066..3f106ee08 100644 --- a/moto/ec2/responses/ip_addresses.py +++ b/moto/ec2/responses/ip_addresses.py @@ -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( diff --git a/moto/ec2/responses/key_pairs.py b/moto/ec2/responses/key_pairs.py index c922b18f7..343f20bde 100644 --- a/moto/ec2/responses/key_pairs.py +++ b/moto/ec2/responses/key_pairs.py @@ -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() diff --git a/moto/ec2/responses/launch_templates.py b/moto/ec2/responses/launch_templates.py index bb539c106..efb3ceda5 100644 --- a/moto/ec2/responses/launch_templates.py +++ b/moto/ec2/responses/launch_templates.py @@ -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") diff --git a/moto/ec2/responses/monitoring.py b/moto/ec2/responses/monitoring.py index b755650ea..0a90f4013 100644 --- a/moto/ec2/responses/monitoring.py +++ b/moto/ec2/responses/monitoring.py @@ -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( diff --git a/moto/ec2/responses/nat_gateways.py b/moto/ec2/responses/nat_gateways.py index 01c01ad56..2c0c13365 100644 --- a/moto/ec2/responses/nat_gateways.py +++ b/moto/ec2/responses/nat_gateways.py @@ -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) diff --git a/moto/ec2/responses/network_acls.py b/moto/ec2/responses/network_acls.py index ec254599b..b99b87c35 100644 --- a/moto/ec2/responses/network_acls.py +++ b/moto/ec2/responses/network_acls.py @@ -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") diff --git a/moto/ec2/responses/placement_groups.py b/moto/ec2/responses/placement_groups.py deleted file mode 100644 index 4a72cc487..000000000 --- a/moto/ec2/responses/placement_groups.py +++ /dev/null @@ -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" - ) diff --git a/moto/ec2/responses/reserved_instances.py b/moto/ec2/responses/reserved_instances.py index e32eb82ac..96de88421 100644 --- a/moto/ec2/responses/reserved_instances.py +++ b/moto/ec2/responses/reserved_instances.py @@ -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( diff --git a/moto/ec2/responses/route_tables.py b/moto/ec2/responses/route_tables.py index 1703310e6..bab437680 100644 --- a/moto/ec2/responses/route_tables.py +++ b/moto/ec2/responses/route_tables.py @@ -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( diff --git a/moto/ec2/responses/security_groups.py b/moto/ec2/responses/security_groups.py index 30bb039dd..034a0152f 100644 --- a/moto/ec2/responses/security_groups.py +++ b/moto/ec2/responses/security_groups.py @@ -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 diff --git a/moto/ec2/responses/settings.py b/moto/ec2/responses/settings.py index b3fdeba33..ebdd872dc 100644 --- a/moto/ec2/responses/settings.py +++ b/moto/ec2/responses/settings.py @@ -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() diff --git a/moto/ec2/responses/spot_fleets.py b/moto/ec2/responses/spot_fleets.py index e4306c330..254525fbc 100644 --- a/moto/ec2/responses/spot_fleets.py +++ b/moto/ec2/responses/spot_fleets.py @@ -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"] diff --git a/moto/ec2/responses/spot_instances.py b/moto/ec2/responses/spot_instances.py index da88de9f9..c7e89c84a 100644 --- a/moto/ec2/responses/spot_instances.py +++ b/moto/ec2/responses/spot_instances.py @@ -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) diff --git a/moto/ec2/responses/subnets.py b/moto/ec2/responses/subnets.py index 48b45c0db..139882cca 100644 --- a/moto/ec2/responses/subnets.py +++ b/moto/ec2/responses/subnets.py @@ -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( diff --git a/moto/ec2/responses/tags.py b/moto/ec2/responses/tags.py index 215739318..a57945d92 100644 --- a/moto/ec2/responses/tags.py +++ b/moto/ec2/responses/tags.py @@ -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() diff --git a/moto/ec2/responses/transit_gateway_attachments.py b/moto/ec2/responses/transit_gateway_attachments.py index 03582c629..751750d92 100644 --- a/moto/ec2/responses/transit_gateway_attachments.py +++ b/moto/ec2/responses/transit_gateway_attachments.py @@ -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( diff --git a/moto/ec2/responses/transit_gateway_route_tables.py b/moto/ec2/responses/transit_gateway_route_tables.py index 22a6d89cc..06039fc61 100644 --- a/moto/ec2/responses/transit_gateway_route_tables.py +++ b/moto/ec2/responses/transit_gateway_route_tables.py @@ -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 = ( diff --git a/moto/ec2/responses/transit_gateways.py b/moto/ec2/responses/transit_gateways.py index 4817751c7..f86afaab3 100644 --- a/moto/ec2/responses/transit_gateways.py +++ b/moto/ec2/responses/transit_gateways.py @@ -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") diff --git a/moto/ec2/responses/virtual_private_gateways.py b/moto/ec2/responses/virtual_private_gateways.py index d3f92f0c2..5f2f8e425 100644 --- a/moto/ec2/responses/virtual_private_gateways.py +++ b/moto/ec2/responses/virtual_private_gateways.py @@ -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) diff --git a/moto/ec2/responses/vm_export.py b/moto/ec2/responses/vm_export.py deleted file mode 100644 index 8a3fa0d76..000000000 --- a/moto/ec2/responses/vm_export.py +++ /dev/null @@ -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" - ) diff --git a/moto/ec2/responses/vm_import.py b/moto/ec2/responses/vm_import.py deleted file mode 100644 index 9f43de092..000000000 --- a/moto/ec2/responses/vm_import.py +++ /dev/null @@ -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") diff --git a/moto/ec2/responses/vpc_peering_connections.py b/moto/ec2/responses/vpc_peering_connections.py index 965cc6548..a85b5fffc 100644 --- a/moto/ec2/responses/vpc_peering_connections.py +++ b/moto/ec2/responses/vpc_peering_connections.py @@ -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" diff --git a/moto/ec2/responses/vpc_service_configuration.py b/moto/ec2/responses/vpc_service_configuration.py index 25f19aadf..86de3fa28 100644 --- a/moto/ec2/responses/vpc_service_configuration.py +++ b/moto/ec2/responses/vpc_service_configuration.py @@ -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") diff --git a/moto/ec2/responses/vpcs.py b/moto/ec2/responses/vpcs.py index 5e4bc5d55..2f0f2e91f 100644 --- a/moto/ec2/responses/vpcs.py +++ b/moto/ec2/responses/vpcs.py @@ -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") diff --git a/moto/ec2/responses/vpn_connections.py b/moto/ec2/responses/vpn_connections.py index e2857b02b..5db579adb 100644 --- a/moto/ec2/responses/vpn_connections.py +++ b/moto/ec2/responses/vpn_connections.py @@ -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( diff --git a/moto/ec2/responses/windows.py b/moto/ec2/responses/windows.py index 5e37f155d..0169adb11 100644 --- a/moto/ec2/responses/windows.py +++ b/moto/ec2/responses/windows.py @@ -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) diff --git a/moto/ec2/utils.py b/moto/ec2/utils.py index dc4c95eb7..b633bf9c9 100644 --- a/moto/ec2/utils.py +++ b/moto/ec2/utils.py @@ -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"]] = {} diff --git a/moto/utilities/utils.py b/moto/utilities/utils.py index 8b3d7cc5f..d415b57aa 100644 --- a/moto/utilities/utils.py +++ b/moto/utilities/utils.py @@ -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]: diff --git a/setup.cfg b/setup.cfg index 80aaa293b..1b31e3e43 100644 --- a/setup.cfg +++ b/setup.cfg @@ -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 diff --git a/tests/test_ec2/test_prefix_lists.py b/tests/test_ec2/test_prefix_lists.py index 9611ebcb7..447ff3449 100644 --- a/tests/test_ec2/test_prefix_lists.py +++ b/tests/test_ec2/test_prefix_lists.py @@ -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]+")