diff --git a/moto/ec2/exceptions.py b/moto/ec2/exceptions.py index 41c0459a8..1763b5a0e 100644 --- a/moto/ec2/exceptions.py +++ b/moto/ec2/exceptions.py @@ -1,5 +1,5 @@ from moto.core.exceptions import RESTError -from typing import List, Union +from typing import List, Optional, Union # EC2 has a custom root-tag - vs @@ -98,8 +98,7 @@ class InvalidKeyPairFormatError(EC2ClientError): class InvalidVPCIdError(EC2ClientError): - def __init__(self, vpc_id): - + def __init__(self, vpc_id: str): super().__init__("InvalidVpcID.NotFound", f"VpcID {vpc_id} does not exist.") @@ -159,7 +158,7 @@ class InvalidVpnConnectionIdError(EC2ClientError): class InvalidCustomerGatewayIdError(EC2ClientError): - def __init__(self, customer_gateway_id): + def __init__(self, customer_gateway_id: str): super().__init__( "InvalidCustomerGatewayID.NotFound", f"The customer gateway ID '{customer_gateway_id}' does not exist", @@ -526,7 +525,7 @@ class MotoNotImplementedError(NotImplementedError): class FilterNotImplementedError(MotoNotImplementedError): - def __init__(self, filter_name, method_name): + def __init__(self, filter_name: str, method_name: Optional[str]): super().__init__(f"The filter '{filter_name}' for {method_name}") @@ -719,7 +718,7 @@ class InvalidSubnetCidrBlockAssociationID(EC2ClientError): class InvalidCarrierGatewayID(EC2ClientError): - def __init__(self, carrier_gateway_id): + def __init__(self, carrier_gateway_id: str): super().__init__( "InvalidCarrierGatewayID.NotFound", f"The CarrierGateway ID '{carrier_gateway_id}' does not exist", diff --git a/moto/ec2/models/carrier_gateways.py b/moto/ec2/models/carrier_gateways.py index 51ef97a6b..9c9e5b6bb 100644 --- a/moto/ec2/models/carrier_gateways.py +++ b/moto/ec2/models/carrier_gateways.py @@ -1,3 +1,4 @@ +from typing import Any, Dict, List, Optional from moto.utilities.utils import filter_resources from .core import TaggedEC2Resource @@ -6,7 +7,9 @@ from ..utils import random_carrier_gateway_id class CarrierGateway(TaggedEC2Resource): - def __init__(self, ec2_backend, vpc_id, tags=None): + def __init__( + self, ec2_backend: Any, vpc_id: str, tags: Optional[Dict[str, str]] = None + ): self.id = random_carrier_gateway_id() self.ec2_backend = ec2_backend self.vpc_id = vpc_id @@ -14,34 +17,38 @@ class CarrierGateway(TaggedEC2Resource): self.add_tags(tags or {}) @property - def physical_resource_id(self): + def physical_resource_id(self) -> str: return self.id @property - def owner_id(self): + def owner_id(self) -> str: return self.ec2_backend.account_id class CarrierGatewayBackend: - def __init__(self): - self.carrier_gateways = {} + def __init__(self) -> None: + self.carrier_gateways: Dict[str, CarrierGateway] = {} - def create_carrier_gateway(self, vpc_id, tags=None): - vpc = self.get_vpc(vpc_id) + def create_carrier_gateway( + self, vpc_id: str, tags: Optional[Dict[str, str]] = None + ) -> CarrierGateway: + vpc = self.get_vpc(vpc_id) # type: ignore[attr-defined] if not vpc: raise InvalidVPCIdError(vpc_id) carrier_gateway = CarrierGateway(self, vpc_id, tags) self.carrier_gateways[carrier_gateway.id] = carrier_gateway return carrier_gateway - def delete_carrier_gateway(self, gateway_id): + def delete_carrier_gateway(self, gateway_id: str) -> CarrierGateway: if not self.carrier_gateways.get(gateway_id): raise InvalidCarrierGatewayID(gateway_id) carrier_gateway = self.carrier_gateways.pop(gateway_id) carrier_gateway.state = "deleted" return carrier_gateway - def describe_carrier_gateways(self, ids=None, filters=None): + def describe_carrier_gateways( + self, ids: Optional[List[str]] = None, filters: Any = None + ) -> List[CarrierGateway]: carrier_gateways = list(self.carrier_gateways.values()) if ids: diff --git a/moto/ec2/models/core.py b/moto/ec2/models/core.py index bd60d7ccb..f07e7c31e 100644 --- a/moto/ec2/models/core.py +++ b/moto/ec2/models/core.py @@ -7,16 +7,16 @@ from ..exceptions import FilterNotImplementedError class TaggedEC2Resource(BaseModel): def get_tags(self) -> List[Dict[str, str]]: tags = [] - if self.id: - tags = self.ec2_backend.describe_tags(filters={"resource-id": [self.id]}) + if self.id: # type: ignore[attr-defined] + tags = self.ec2_backend.describe_tags(filters={"resource-id": [self.id]}) # type: ignore[attr-defined] return tags def add_tag(self, key: str, value: str) -> None: - self.ec2_backend.create_tags([self.id], {key: value}) + self.ec2_backend.create_tags([self.id], {key: value}) # type: ignore[attr-defined] def add_tags(self, tag_map: Dict[str, str]) -> None: for key, value in tag_map.items(): - self.ec2_backend.create_tags([self.id], {key: value}) + self.ec2_backend.create_tags([self.id], {key: value}) # type: ignore[attr-defined] def get_filter_value( self, filter_name: str, method_name: Optional[str] = None diff --git a/moto/ec2/models/customer_gateways.py b/moto/ec2/models/customer_gateways.py index 3ab4e5262..4d712d227 100644 --- a/moto/ec2/models/customer_gateways.py +++ b/moto/ec2/models/customer_gateways.py @@ -1,3 +1,5 @@ +from typing import Any, Dict, List, Optional + from .core import TaggedEC2Resource from ..exceptions import InvalidCustomerGatewayIdError from ..utils import random_customer_gateway_id @@ -6,35 +8,40 @@ from ..utils import random_customer_gateway_id class CustomerGateway(TaggedEC2Resource): def __init__( self, - ec2_backend, - gateway_id, - gateway_type, - ip_address, - bgp_asn, - state="available", - tags=None, + ec2_backend: Any, + gateway_id: str, + gateway_type: str, + ip_address: str, + bgp_asn: str, + state: str = "available", + tags: Optional[Dict[str, str]] = None, ): self.ec2_backend = ec2_backend self.id = gateway_id - self.type = gateway_type + self.type = gateway_type or "ipsec.1" self.ip_address = ip_address self.bgp_asn = bgp_asn - self.attachments = {} self.state = state self.add_tags(tags or {}) super().__init__() - def get_filter_value(self, filter_name): + def get_filter_value( + self, filter_name: str, method_name: Optional[str] = None + ) -> Any: return super().get_filter_value(filter_name, "DescribeCustomerGateways") class CustomerGatewayBackend: - def __init__(self): - self.customer_gateways = {} + def __init__(self) -> None: + self.customer_gateways: Dict[str, CustomerGateway] = {} def create_customer_gateway( - self, gateway_type="ipsec.1", ip_address=None, bgp_asn=None, tags=None - ): + self, + gateway_type: str, + ip_address: str, + bgp_asn: str, + tags: Optional[Dict[str, str]] = None, + ) -> CustomerGateway: customer_gateway_id = random_customer_gateway_id() customer_gateway = CustomerGateway( self, customer_gateway_id, gateway_type, ip_address, bgp_asn, tags=tags @@ -42,8 +49,10 @@ class CustomerGatewayBackend: self.customer_gateways[customer_gateway_id] = customer_gateway return customer_gateway - def get_all_customer_gateways(self, filters=None, customer_gateway_ids=None): - customer_gateways = self.customer_gateways.copy().values() + def describe_customer_gateways( + self, filters: Any = None, customer_gateway_ids: Optional[List[str]] = None + ) -> List[CustomerGateway]: + customer_gateways = list(self.customer_gateways.copy().values()) if customer_gateway_ids: customer_gateways = [ cg for cg in customer_gateways if cg.id in customer_gateway_ids @@ -76,17 +85,12 @@ class CustomerGatewayBackend: ] return customer_gateways - def get_customer_gateway(self, customer_gateway_id): - customer_gateway = self.customer_gateways.get(customer_gateway_id, None) + def get_customer_gateway(self, customer_gateway_id: str) -> CustomerGateway: + customer_gateway = self.customer_gateways.get(customer_gateway_id) if not customer_gateway: raise InvalidCustomerGatewayIdError(customer_gateway_id) return customer_gateway - def delete_customer_gateway(self, customer_gateway_id): + def delete_customer_gateway(self, customer_gateway_id: str) -> None: customer_gateway = self.get_customer_gateway(customer_gateway_id) customer_gateway.state = "deleted" - # deleted = self.customer_gateways.pop(customer_gateway_id, None) - deleted = True - if not deleted: - raise InvalidCustomerGatewayIdError(customer_gateway_id) - return deleted diff --git a/moto/ec2/responses/customer_gateways.py b/moto/ec2/responses/customer_gateways.py index 51afb7037..f78fcf6ec 100644 --- a/moto/ec2/responses/customer_gateways.py +++ b/moto/ec2/responses/customer_gateways.py @@ -18,15 +18,15 @@ class CustomerGateways(EC2BaseResponse): def delete_customer_gateway(self): customer_gateway_id = self._get_param("CustomerGatewayId") - delete_status = self.ec2_backend.delete_customer_gateway(customer_gateway_id) + self.ec2_backend.delete_customer_gateway(customer_gateway_id) template = self.response_template(DELETE_CUSTOMER_GATEWAY_RESPONSE) - return template.render(delete_status=delete_status) + return template.render(delete_status="true") def describe_customer_gateways(self): self.error_on_dryrun() filters = self._filters_from_querystring() customer_gateway_ids = self._get_multi_param("CustomerGatewayId") - customer_gateways = self.ec2_backend.get_all_customer_gateways( + customer_gateways = self.ec2_backend.describe_customer_gateways( filters, customer_gateway_ids ) template = self.response_template(DESCRIBE_CUSTOMER_GATEWAYS_RESPONSE) diff --git a/moto/ec2/utils.py b/moto/ec2/utils.py index 32f44418b..77d2e76cc 100644 --- a/moto/ec2/utils.py +++ b/moto/ec2/utils.py @@ -142,7 +142,7 @@ def random_vpn_connection_id(): return random_id(prefix=EC2_RESOURCE_TO_PREFIX["vpn-connection"]) -def random_customer_gateway_id(): +def random_customer_gateway_id() -> str: return random_id(prefix=EC2_RESOURCE_TO_PREFIX["customer-gateway"]) @@ -232,7 +232,7 @@ def random_iam_instance_profile_association_id(): return random_id(prefix=EC2_RESOURCE_TO_PREFIX["iam-instance-profile-association"]) -def random_carrier_gateway_id(): +def random_carrier_gateway_id() -> str: return random_id(prefix=EC2_RESOURCE_TO_PREFIX["carrier-gateway"], size=17) diff --git a/moto/utilities/utils.py b/moto/utilities/utils.py index 01e98499f..4cf578127 100644 --- a/moto/utilities/utils.py +++ b/moto/utilities/utils.py @@ -3,7 +3,7 @@ import hashlib import pkgutil from collections.abc import MutableMapping -from typing import Any, Dict, List, TypeVar +from typing import Any, Dict, List, TypeVar, Tuple def str2bool(v): @@ -39,7 +39,9 @@ RESOURCE_TYPE = TypeVar("RESOURCE_TYPE") def filter_resources( - resources: List[RESOURCE_TYPE], filters: Any, attr_pairs: Any + resources: List[RESOURCE_TYPE], + filters: Any, + attr_pairs: Tuple[Tuple[str, str], ...], ) -> List[RESOURCE_TYPE]: """ Used to filter resources. Usually in get and describe apis. diff --git a/setup.cfg b/setup.cfg index d8cc6bae6..fda491699 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/a*,moto/moto_api +files= moto/a*,moto/b*,moto/c*,moto/d*,moto/ebs/,moto/ec2/models/a*,moto/ec2/models/c*,moto/moto_api show_column_numbers=True show_error_codes = True disable_error_code=abstract