From b8350f2801e1505b089ae7fc1b3aeb2123864668 Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Sun, 12 Feb 2023 16:53:36 -0100 Subject: [PATCH] Techdebt: MyPy EC2 (t-models) (#5922) --- IMPLEMENTATION_COVERAGE.md | 6 +- docs/docs/services/ec2.rst | 6 +- moto/autoscaling/models.py | 2 +- moto/ds/models.py | 2 +- moto/ec2/exceptions.py | 4 +- moto/ec2/models/__init__.py | 6 +- moto/ec2/models/subnets.py | 8 +- moto/ec2/models/tags.py | 14 +- moto/ec2/models/transit_gateway.py | 63 ++++-- .../ec2/models/transit_gateway_attachments.py | 150 ++++++++------ .../models/transit_gateway_route_tables.py | 188 +++++++++--------- moto/ec2/models/vpc_peering_connections.py | 14 +- moto/ec2/responses/subnets.py | 2 +- .../responses/transit_gateway_route_tables.py | 4 +- moto/ec2/utils.py | 11 +- moto/redshift/models.py | 2 +- moto/route53resolver/models.py | 8 +- moto/utilities/utils.py | 2 +- setup.cfg | 2 +- 19 files changed, 278 insertions(+), 216 deletions(-) diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md index e759d2e82..2900adbca 100644 --- a/IMPLEMENTATION_COVERAGE.md +++ b/IMPLEMENTATION_COVERAGE.md @@ -2063,7 +2063,7 @@ - [X] describe_spot_price_history - [ ] describe_stale_security_groups - [ ] describe_store_image_tasks -- [ ] describe_subnets +- [X] describe_subnets - [X] describe_tags - [ ] describe_traffic_mirror_filters - [ ] describe_traffic_mirror_sessions @@ -2187,8 +2187,8 @@ - [ ] get_transit_gateway_policy_table_associations - [ ] get_transit_gateway_policy_table_entries - [ ] get_transit_gateway_prefix_list_references -- [ ] get_transit_gateway_route_table_associations -- [ ] get_transit_gateway_route_table_propagations +- [X] get_transit_gateway_route_table_associations +- [X] get_transit_gateway_route_table_propagations - [ ] get_verified_access_endpoint_policy - [ ] get_verified_access_group_policy - [ ] get_vpn_connection_device_sample_configuration diff --git a/docs/docs/services/ec2.rst b/docs/docs/services/ec2.rst index 58a95f519..5cd8dc981 100644 --- a/docs/docs/services/ec2.rst +++ b/docs/docs/services/ec2.rst @@ -360,7 +360,7 @@ ec2 - [X] describe_spot_price_history - [ ] describe_stale_security_groups - [ ] describe_store_image_tasks -- [ ] describe_subnets +- [X] describe_subnets - [X] describe_tags - [ ] describe_traffic_mirror_filters - [ ] describe_traffic_mirror_sessions @@ -503,8 +503,8 @@ ec2 - [ ] get_transit_gateway_policy_table_associations - [ ] get_transit_gateway_policy_table_entries - [ ] get_transit_gateway_prefix_list_references -- [ ] get_transit_gateway_route_table_associations -- [ ] get_transit_gateway_route_table_propagations +- [X] get_transit_gateway_route_table_associations +- [X] get_transit_gateway_route_table_propagations - [ ] get_verified_access_endpoint_policy - [ ] get_verified_access_group_policy - [ ] get_vpn_connection_device_sample_configuration diff --git a/moto/autoscaling/models.py b/moto/autoscaling/models.py index a9bde31a7..56fe72b65 100644 --- a/moto/autoscaling/models.py +++ b/moto/autoscaling/models.py @@ -484,7 +484,7 @@ class FakeAutoScalingGroup(CloudFormationModel): if vpc_zone_identifier: # extract azs for vpcs subnet_ids = vpc_zone_identifier.split(",") - subnets = self.autoscaling_backend.ec2_backend.get_all_subnets( + subnets = self.autoscaling_backend.ec2_backend.describe_subnets( subnet_ids=subnet_ids ) vpc_zones = [subnet.availability_zone for subnet in subnets] diff --git a/moto/ds/models.py b/moto/ds/models.py index e2e987115..1d003f498 100644 --- a/moto/ds/models.py +++ b/moto/ds/models.py @@ -219,7 +219,7 @@ class DirectoryServiceBackend(BaseBackend): # Subnet IDs are checked before the VPC ID. The Subnet IDs must # be valid and in different availability zones. try: - subnets = ec2_backends[self.account_id][region].get_all_subnets( + subnets = ec2_backends[self.account_id][region].describe_subnets( subnet_ids=vpc_settings["SubnetIds"] ) except InvalidSubnetIdError as exc: diff --git a/moto/ec2/exceptions.py b/moto/ec2/exceptions.py index 7dd6f5e23..4a441a085 100644 --- a/moto/ec2/exceptions.py +++ b/moto/ec2/exceptions.py @@ -443,7 +443,7 @@ class EmptyTagSpecError(EC2ClientError): class InvalidParameterValueErrorTagNull(EC2ClientError): - def __init__(self): + def __init__(self) -> None: super().__init__( "InvalidParameterValue", "Tag value cannot be null. Use empty string instead.", @@ -490,7 +490,7 @@ class ResourceAlreadyAssociatedError(EC2ClientError): class TagLimitExceeded(EC2ClientError): - def __init__(self): + def __init__(self) -> None: super().__init__( "TagLimitExceeded", "The maximum number of Tags for a resource has been reached.", diff --git a/moto/ec2/models/__init__.py b/moto/ec2/models/__init__.py index ade8411f4..c57182bc8 100644 --- a/moto/ec2/models/__init__.py +++ b/moto/ec2/models/__init__.py @@ -34,10 +34,7 @@ from .spot_requests import SpotRequestBackend from .subnets import SubnetBackend from .tags import TagBackend from .transit_gateway import TransitGatewayBackend -from .transit_gateway_route_tables import ( - TransitGatewayRelationsBackend, - TransitGatewayRouteTableBackend, -) +from .transit_gateway_route_tables import TransitGatewayRouteTableBackend from .transit_gateway_attachments import TransitGatewayAttachmentBackend from .vpn_gateway import VpnGatewayBackend from .vpn_connections import VPNConnectionBackend @@ -111,7 +108,6 @@ class EC2Backend( TransitGatewayBackend, TransitGatewayRouteTableBackend, TransitGatewayAttachmentBackend, - TransitGatewayRelationsBackend, LaunchTemplateBackend, IamInstanceProfileAssociationBackend, CarrierGatewayBackend, diff --git a/moto/ec2/models/subnets.py b/moto/ec2/models/subnets.py index 1921eca80..96de49faa 100644 --- a/moto/ec2/models/subnets.py +++ b/moto/ec2/models/subnets.py @@ -290,7 +290,7 @@ class SubnetBackend: def get_default_subnet(self, availability_zone: str) -> Subnet: return [ subnet - for subnet in self.get_all_subnets( + for subnet in self.describe_subnets( filters={"availabilityZone": availability_zone} ) if subnet.default_for_az @@ -336,7 +336,7 @@ class SubnetBackend: if ipv6_cidr_block and "::/64" not in ipv6_cidr_block: raise GenericInvalidParameterValueError("ipv6-cidr-block", ipv6_cidr_block) - for subnet in self.get_all_subnets(filters={"vpc-id": vpc_id}): + for subnet in self.describe_subnets(filters={"vpc-id": vpc_id}): if subnet.cidr.overlaps(subnet_cidr_block): raise InvalidSubnetConflictError(cidr_block) @@ -394,7 +394,7 @@ class SubnetBackend: self.subnets[availability_zone][subnet_id] = subnet # type: ignore[index] return subnet - def get_all_subnets( + def describe_subnets( self, subnet_ids: Optional[List[str]] = None, filters: Optional[Any] = None ) -> Iterable[Subnet]: # Extract a list of all subnets @@ -428,7 +428,7 @@ class SubnetBackend: def get_subnet_from_ipv6_association(self, association_id: str) -> Optional[Subnet]: subnet = None - for s in self.get_all_subnets(): + for s in self.describe_subnets(): if association_id in s.ipv6_cidr_block_associations: subnet = s return subnet diff --git a/moto/ec2/models/tags.py b/moto/ec2/models/tags.py index 6f8c9843d..fdf9a1205 100644 --- a/moto/ec2/models/tags.py +++ b/moto/ec2/models/tags.py @@ -1,6 +1,6 @@ import re from collections import defaultdict -from typing import Dict, List +from typing import Dict, List, Optional from ..exceptions import ( InvalidParameterValueErrorTagNull, @@ -16,10 +16,10 @@ from ..utils import ( class TagBackend: VALID_TAG_FILTERS = ["key", "resource-id", "resource-type", "value"] - def __init__(self): - self.tags = defaultdict(dict) + def __init__(self) -> None: + self.tags: Dict[str, Dict[str, str]] = defaultdict(dict) - def create_tags(self, resource_ids: List[str], tags: Dict[str, str]): + def create_tags(self, resource_ids: List[str], tags: Dict[str, str]) -> bool: if None in set([tags[tag] for tag in tags]): raise InvalidParameterValueErrorTagNull() for resource_id in resource_ids: @@ -37,7 +37,7 @@ class TagBackend: self.tags[resource_id][tag] = tags[tag] return True - def delete_tags(self, resource_ids: List[str], tags: Dict[str, str]): + def delete_tags(self, resource_ids: List[str], tags: Dict[str, str]) -> bool: for resource_id in resource_ids: for tag in tags: if tag in self.tags[resource_id]: @@ -47,7 +47,9 @@ class TagBackend: self.tags[resource_id].pop(tag) return True - def describe_tags(self, filters=None): + def describe_tags( + self, filters: Optional[Dict[str, List[str]]] = None + ) -> List[Dict[str, str]]: results = [] key_filters = [] resource_id_filters = [] diff --git a/moto/ec2/models/transit_gateway.py b/moto/ec2/models/transit_gateway.py index e4d301f5a..b6e6bfd2d 100644 --- a/moto/ec2/models/transit_gateway.py +++ b/moto/ec2/models/transit_gateway.py @@ -1,4 +1,5 @@ from datetime import datetime +from typing import Any, Dict, List, Optional from moto.core import CloudFormationModel from moto.core.utils import iso_8601_datetime_with_milliseconds from moto.utilities.utils import filter_resources, merge_multiple_dicts @@ -20,7 +21,12 @@ class TransitGateway(TaggedEC2Resource, CloudFormationModel): "VpnEcmpSupport": "enable", } - def __init__(self, backend, description=None, options=None): + def __init__( + self, + backend: Any, + description: Optional[str], + options: Optional[Dict[str, str]] = None, + ): self.ec2_backend = backend self.id = random_transit_gateway_id() self.description = description @@ -29,30 +35,35 @@ class TransitGateway(TaggedEC2Resource, CloudFormationModel): self._created_at = datetime.utcnow() @property - def physical_resource_id(self): + def physical_resource_id(self) -> str: return self.id @property - def create_time(self): + def create_time(self) -> str: return iso_8601_datetime_with_milliseconds(self._created_at) @property - def owner_id(self): + def owner_id(self) -> str: return self.ec2_backend.account_id @staticmethod - def cloudformation_name_type(): - return None + def cloudformation_name_type() -> str: + return "" @staticmethod - def cloudformation_type(): + def cloudformation_type() -> str: # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-ec2-natgateway.html return "AWS::EC2::TransitGateway" @classmethod - def create_from_cloudformation_json( - cls, resource_name, cloudformation_json, account_id, region_name, **kwargs - ): + def create_from_cloudformation_json( # type: ignore[misc] + cls, + resource_name: str, + cloudformation_json: Any, + account_id: str, + region_name: str, + **kwargs: Any + ) -> "TransitGateway": from ..models import ec2_backends ec2_backend = ec2_backends[account_id][region_name] @@ -73,21 +84,26 @@ class TransitGateway(TaggedEC2Resource, CloudFormationModel): class TransitGatewayBackend: - def __init__(self): - self.transit_gateways = {} + def __init__(self) -> None: + self.transit_gateways: Dict[str, TransitGateway] = {} - def create_transit_gateway(self, description=None, options=None, tags=None): + def create_transit_gateway( + self, + description: Optional[str], + options: Optional[Dict[str, str]] = None, + tags: Optional[List[Dict[str, str]]] = None, + ) -> TransitGateway: transit_gateway = TransitGateway(self, description, options) for tag in tags or []: - tag_key = tag.get("Key") - tag_value = tag.get("Value") - transit_gateway.add_tag(tag_key, tag_value) + transit_gateway.add_tag(tag["Key"], tag["Value"]) self.transit_gateways[transit_gateway.id] = transit_gateway return transit_gateway - def describe_transit_gateways(self, filters, transit_gateway_ids): - transit_gateways = list(self.transit_gateways.copy().values()) + def describe_transit_gateways( + self, filters: Any, transit_gateway_ids: Optional[List[str]] + ) -> List[TransitGateway]: + transit_gateways = list(self.transit_gateways.values()) if transit_gateway_ids: transit_gateways = [ @@ -105,13 +121,16 @@ class TransitGatewayBackend: result = filter_resources(transit_gateways, filters, attr_pairs) return result - def delete_transit_gateway(self, transit_gateway_id): + def delete_transit_gateway(self, transit_gateway_id: str) -> TransitGateway: return self.transit_gateways.pop(transit_gateway_id) def modify_transit_gateway( - self, transit_gateway_id, description=None, options=None - ): - transit_gateway = self.transit_gateways.get(transit_gateway_id) + self, + transit_gateway_id: str, + description: Optional[str] = None, + options: Optional[Dict[str, str]] = None, + ) -> TransitGateway: + transit_gateway = self.transit_gateways[transit_gateway_id] if description: transit_gateway.description = description if options: diff --git a/moto/ec2/models/transit_gateway_attachments.py b/moto/ec2/models/transit_gateway_attachments.py index 4b58d1725..60c3b1317 100644 --- a/moto/ec2/models/transit_gateway_attachments.py +++ b/moto/ec2/models/transit_gateway_attachments.py @@ -1,4 +1,5 @@ from datetime import datetime +from typing import Any, Dict, List, Optional from moto.core.utils import iso_8601_datetime_with_milliseconds from moto.utilities.utils import merge_multiple_dicts, filter_resources from .core import TaggedEC2Resource @@ -8,11 +9,16 @@ from ..utils import random_transit_gateway_attachment_id, describe_tag_filter class TransitGatewayAttachment(TaggedEC2Resource): def __init__( - self, backend, resource_id, resource_type, transit_gateway_id, tags=None + self, + backend: Any, + resource_id: str, + resource_type: str, + transit_gateway_id: str, + tags: Optional[Dict[str, str]] = None, ): self.ec2_backend = backend - self.association = {} - self.propagation = {} + self.association: Dict[str, str] = {} + self.propagation: Dict[str, str] = {} self.resource_id = resource_id self.resource_type = resource_type @@ -28,7 +34,7 @@ class TransitGatewayAttachment(TaggedEC2Resource): self.owner_id = backend.account_id @property - def create_time(self): + def create_time(self) -> str: return iso_8601_datetime_with_milliseconds(self._created_at) @@ -40,7 +46,13 @@ class TransitGatewayVpcAttachment(TransitGatewayAttachment): } def __init__( - self, backend, transit_gateway_id, vpc_id, subnet_ids, tags=None, options=None + self, + backend: Any, + transit_gateway_id: str, + vpc_id: str, + subnet_ids: List[str], + tags: Optional[Dict[str, str]] = None, + options: Optional[Dict[str, str]] = None, ): super().__init__( backend=backend, @@ -58,13 +70,13 @@ class TransitGatewayVpcAttachment(TransitGatewayAttachment): class TransitGatewayPeeringAttachment(TransitGatewayAttachment): def __init__( self, - backend, - transit_gateway_id=None, - peer_transit_gateway_id=None, - peer_region=None, - peer_account_id=None, - tags=None, - region_name=None, + backend: Any, + transit_gateway_id: str, + peer_transit_gateway_id: str, + peer_region: str, + peer_account_id: str, + tags: Dict[str, str], + region_name: str, ): super().__init__( backend=backend, @@ -88,12 +100,15 @@ class TransitGatewayPeeringAttachment(TransitGatewayAttachment): class TransitGatewayAttachmentBackend: - def __init__(self): - self.transit_gateway_attachments = {} + def __init__(self) -> None: + self.transit_gateway_attachments: Dict[str, TransitGatewayAttachment] = {} def create_transit_gateway_vpn_attachment( - self, vpn_id, transit_gateway_id, tags=None - ): + self, + vpn_id: str, + transit_gateway_id: str, + tags: Optional[Dict[str, str]] = None, + ) -> TransitGatewayAttachment: transit_gateway_vpn_attachment = TransitGatewayAttachment( self, resource_id=vpn_id, @@ -107,8 +122,13 @@ class TransitGatewayAttachmentBackend: return transit_gateway_vpn_attachment def create_transit_gateway_vpc_attachment( - self, transit_gateway_id, vpc_id, subnet_ids, tags=None, options=None - ): + self, + transit_gateway_id: str, + vpc_id: str, + subnet_ids: List[str], + tags: Optional[Dict[str, str]] = None, + options: Optional[Dict[str, str]] = None, + ) -> TransitGatewayVpcAttachment: transit_gateway_vpc_attachment = TransitGatewayVpcAttachment( self, transit_gateway_id=transit_gateway_id, @@ -123,8 +143,10 @@ class TransitGatewayAttachmentBackend: return transit_gateway_vpc_attachment def describe_transit_gateway_attachments( - self, transit_gateways_attachment_ids=None, filters=None - ): + self, + transit_gateways_attachment_ids: Optional[List[str]] = None, + filters: Any = None, + ) -> List[TransitGatewayAttachment]: transit_gateway_attachments = list(self.transit_gateway_attachments.values()) attr_pairs = ( @@ -149,8 +171,10 @@ class TransitGatewayAttachmentBackend: return result def describe_transit_gateway_vpc_attachments( - self, transit_gateways_attachment_ids=None, filters=None - ): + self, + transit_gateways_attachment_ids: Optional[List[str]] = None, + filters: Any = None, + ) -> List[TransitGatewayAttachment]: transit_gateway_attachments = list(self.transit_gateway_attachments.values()) attr_pairs = ( @@ -175,7 +199,9 @@ class TransitGatewayAttachmentBackend: result = filter_resources(transit_gateway_attachments, filters, attr_pairs) return result - def delete_transit_gateway_vpc_attachment(self, transit_gateway_attachment_id=None): + def delete_transit_gateway_vpc_attachment( + self, transit_gateway_attachment_id: str + ) -> TransitGatewayAttachment: transit_gateway_attachment = self.transit_gateway_attachments.pop( transit_gateway_attachment_id ) @@ -184,62 +210,64 @@ class TransitGatewayAttachmentBackend: def modify_transit_gateway_vpc_attachment( self, - add_subnet_ids=None, - options=None, - remove_subnet_ids=None, - transit_gateway_attachment_id=None, - ): + transit_gateway_attachment_id: str, + add_subnet_ids: Optional[List[str]] = None, + options: Optional[Dict[str, str]] = None, + remove_subnet_ids: Optional[List[str]] = None, + ) -> TransitGatewayAttachment: tgw_attachment = self.transit_gateway_attachments[transit_gateway_attachment_id] if remove_subnet_ids: - tgw_attachment.subnet_ids = [ - id for id in tgw_attachment.subnet_ids if id not in remove_subnet_ids + tgw_attachment.subnet_ids = [ # type: ignore[attr-defined] + id for id in tgw_attachment.subnet_ids if id not in remove_subnet_ids # type: ignore[attr-defined] ] if options: - tgw_attachment.options.update(options) + tgw_attachment.options.update(options) # type: ignore[attr-defined] if add_subnet_ids: for subnet_id in add_subnet_ids: - tgw_attachment.subnet_ids.append(subnet_id) + tgw_attachment.subnet_ids.append(subnet_id) # type: ignore[attr-defined] return tgw_attachment def set_attachment_association( - self, transit_gateway_attachment_id=None, transit_gateway_route_table_id=None - ): + self, transit_gateway_attachment_id: str, transit_gateway_route_table_id: str + ) -> None: self.transit_gateway_attachments[transit_gateway_attachment_id].association = { "state": "associated", "transitGatewayRouteTableId": transit_gateway_route_table_id, } - def unset_attachment_association(self, tgw_attach_id): - self.transit_gateway_attachments.get(tgw_attach_id).association = {} + def unset_attachment_association(self, tgw_attach_id: str) -> None: + self.transit_gateway_attachments[tgw_attach_id].association = {} def set_attachment_propagation( - self, transit_gateway_attachment_id=None, transit_gateway_route_table_id=None - ): + self, transit_gateway_attachment_id: str, transit_gateway_route_table_id: str + ) -> None: self.transit_gateway_attachments[transit_gateway_attachment_id].propagation = { "state": "enabled", "transitGatewayRouteTableId": transit_gateway_route_table_id, } - def unset_attachment_propagation(self, tgw_attach_id): - self.transit_gateway_attachments.get(tgw_attach_id).propagation = {} + def unset_attachment_propagation(self, tgw_attach_id: str) -> None: + self.transit_gateway_attachments[tgw_attach_id].propagation = {} - def disable_attachment_propagation(self, transit_gateway_attachment_id=None): + def disable_attachment_propagation( + self, transit_gateway_attachment_id: str + ) -> None: self.transit_gateway_attachments[transit_gateway_attachment_id].propagation[ "state" ] = "disabled" def create_transit_gateway_peering_attachment( self, - transit_gateway_id, - peer_transit_gateway_id, - peer_region, - peer_account_id, - tags, - ): + transit_gateway_id: str, + peer_transit_gateway_id: str, + peer_region: str, + peer_account_id: str, + tags: Dict[str, str], + ) -> TransitGatewayPeeringAttachment: transit_gateway_peering_attachment = TransitGatewayPeeringAttachment( self, transit_gateway_id=transit_gateway_id, @@ -247,7 +275,7 @@ class TransitGatewayAttachmentBackend: peer_region=peer_region, peer_account_id=peer_account_id, tags=tags, - region_name=self.region_name, + region_name=self.region_name, # type: ignore[attr-defined] ) transit_gateway_peering_attachment.status.accept() transit_gateway_peering_attachment.state = "available" @@ -257,8 +285,10 @@ class TransitGatewayAttachmentBackend: return transit_gateway_peering_attachment def describe_transit_gateway_peering_attachments( - self, transit_gateways_attachment_ids=None, filters=None - ): + self, + transit_gateways_attachment_ids: Optional[List[str]] = None, + filters: Any = None, + ) -> List[TransitGatewayAttachment]: transit_gateway_attachments = list(self.transit_gateway_attachments.values()) attr_pairs = ( @@ -284,26 +314,34 @@ class TransitGatewayAttachmentBackend: ) return transit_gateway_attachments - def accept_transit_gateway_peering_attachment(self, transit_gateway_attachment_id): + def accept_transit_gateway_peering_attachment( + self, transit_gateway_attachment_id: str + ) -> TransitGatewayAttachment: transit_gateway_attachment = self.transit_gateway_attachments[ transit_gateway_attachment_id ] transit_gateway_attachment.state = "available" - transit_gateway_attachment.status.accept() + # Bit dodgy - we just assume that we act on a TransitGatewayPeeringAttachment + # We could just as easily have another sub-class of TransitGatewayAttachment on our hands, which does not have a status-attribute + transit_gateway_attachment.status.accept() # type: ignore[attr-defined] return transit_gateway_attachment - def reject_transit_gateway_peering_attachment(self, transit_gateway_attachment_id): + def reject_transit_gateway_peering_attachment( + self, transit_gateway_attachment_id: str + ) -> TransitGatewayAttachment: transit_gateway_attachment = self.transit_gateway_attachments[ transit_gateway_attachment_id ] transit_gateway_attachment.state = "rejected" - transit_gateway_attachment.status.reject() + transit_gateway_attachment.status.reject() # type: ignore[attr-defined] return transit_gateway_attachment - def delete_transit_gateway_peering_attachment(self, transit_gateway_attachment_id): + def delete_transit_gateway_peering_attachment( + self, transit_gateway_attachment_id: str + ) -> TransitGatewayAttachment: transit_gateway_attachment = self.transit_gateway_attachments[ transit_gateway_attachment_id ] transit_gateway_attachment.state = "deleted" - transit_gateway_attachment.status.deleted() + transit_gateway_attachment.status.deleted() # type: ignore[attr-defined] return transit_gateway_attachment diff --git a/moto/ec2/models/transit_gateway_route_tables.py b/moto/ec2/models/transit_gateway_route_tables.py index 14a9a6c02..824200311 100644 --- a/moto/ec2/models/transit_gateway_route_tables.py +++ b/moto/ec2/models/transit_gateway_route_tables.py @@ -1,4 +1,5 @@ from datetime import datetime +from typing import Any, Dict, List, Optional from moto.core.utils import iso_8601_datetime_with_milliseconds from moto.utilities.utils import filter_resources from .core import TaggedEC2Resource @@ -8,11 +9,11 @@ from ..utils import random_transit_gateway_route_table_id class TransitGatewayRouteTable(TaggedEC2Resource): def __init__( self, - backend, - transit_gateway_id, - tags=None, - default_association_route_table=False, - default_propagation_route_table=False, + backend: Any, + transit_gateway_id: str, + tags: Optional[Dict[str, str]] = None, + default_association_route_table: bool = False, + default_propagation_route_table: bool = False, ): self.ec2_backend = backend self.id = random_transit_gateway_route_table_id() @@ -23,31 +24,54 @@ class TransitGatewayRouteTable(TaggedEC2Resource): self.default_association_route_table = default_association_route_table self.default_propagation_route_table = default_propagation_route_table self.state = "available" - self.routes = {} + self.routes: Dict[str, Dict[str, Optional[str]]] = {} self.add_tags(tags or {}) - self.route_table_association = {} - self.route_table_propagation = {} + self.route_table_association: Dict[str, str] = {} + self.route_table_propagation: Dict[str, str] = {} @property - def physical_resource_id(self): + def physical_resource_id(self) -> str: return self.id @property - def create_time(self): + def create_time(self) -> str: return iso_8601_datetime_with_milliseconds(self._created_at) +class TransitGatewayRelations: + # this class is for TransitGatewayAssociation and TransitGatewayPropagation + def __init__( + self, + backend: Any, + transit_gateway_attachment_id: str, + transit_gateway_route_table_id: str, + state: str, + ): + self.ec2_backend = backend + self.transit_gateway_attachment_id = transit_gateway_attachment_id + self.transit_gateway_route_table_id = transit_gateway_route_table_id + self.resource_id = backend.transit_gateway_attachments[ + transit_gateway_attachment_id + ].resource_id + self.resource_type = backend.transit_gateway_attachments[ + transit_gateway_attachment_id + ].resource_type + self.state = state + + class TransitGatewayRouteTableBackend: - def __init__(self): - self.transit_gateways_route_tables = {} + def __init__(self) -> None: + self.transit_gateways_route_tables: Dict[str, TransitGatewayRouteTable] = {} + self.transit_gateway_associations: Dict[str, TransitGatewayRelations] = {} + self.transit_gateway_propagations: Dict[str, TransitGatewayRelations] = {} def create_transit_gateway_route_table( self, - transit_gateway_id, - tags=None, - default_association_route_table=False, - default_propagation_route_table=False, - ): + transit_gateway_id: str, + tags: Optional[Dict[str, str]] = None, + default_association_route_table: bool = False, + default_propagation_route_table: bool = False, + ) -> TransitGatewayRouteTable: transit_gateways_route_table = TransitGatewayRouteTable( self, transit_gateway_id=transit_gateway_id, @@ -61,8 +85,8 @@ class TransitGatewayRouteTableBackend: return transit_gateways_route_table def get_all_transit_gateway_route_tables( - self, transit_gateway_route_table_ids=None, filters=None - ): + self, transit_gateway_route_table_ids: Optional[str] = None, filters: Any = None + ) -> List[TransitGatewayRouteTable]: transit_gateway_route_tables = list(self.transit_gateways_route_tables.values()) attr_pairs = ( @@ -85,7 +109,9 @@ class TransitGatewayRouteTableBackend: result = filter_resources(transit_gateway_route_tables, filters, attr_pairs) return result - def delete_transit_gateway_route_table(self, transit_gateway_route_table_id): + def delete_transit_gateway_route_table( + self, transit_gateway_route_table_id: str + ) -> TransitGatewayRouteTable: transit_gateways_route_table = self.transit_gateways_route_tables[ transit_gateway_route_table_id ] @@ -94,15 +120,15 @@ class TransitGatewayRouteTableBackend: def create_transit_gateway_route( self, - transit_gateway_route_table_id, - destination_cidr_block, - transit_gateway_attachment_id=None, - blackhole=False, - ): - transit_gateways_route_table = self.transit_gateways_route_tables.get( + transit_gateway_route_table_id: str, + destination_cidr_block: str, + transit_gateway_attachment_id: Optional[str] = None, + blackhole: bool = False, + ) -> Dict[str, Optional[str]]: + transit_gateways_route_table = self.transit_gateways_route_tables[ transit_gateway_route_table_id - ) - transit_gateway_attachment = self.transit_gateway_attachments.get( + ] + transit_gateway_attachment = self.transit_gateway_attachments.get( # type: ignore[attr-defined] transit_gateway_attachment_id ) transit_gateways_route_table.routes[destination_cidr_block] = { @@ -121,13 +147,13 @@ class TransitGatewayRouteTableBackend: } } transit_gateways_route_table.routes[destination_cidr_block].update( - transit_gateway_attachment_dict + transit_gateway_attachment_dict # type: ignore ) return transit_gateways_route_table.routes[destination_cidr_block] def delete_transit_gateway_route( - self, transit_gateway_route_table_id, destination_cidr_block - ): + self, transit_gateway_route_table_id: str, destination_cidr_block: str + ) -> TransitGatewayRouteTable: transit_gateways_route_table = self.transit_gateways_route_tables[ transit_gateway_route_table_id ] @@ -135,8 +161,11 @@ class TransitGatewayRouteTableBackend: return transit_gateways_route_table def search_transit_gateway_routes( - self, transit_gateway_route_table_id, filters, max_results=None - ): + self, + transit_gateway_route_table_id: str, + filters: Any, + max_results: Optional[int] = None, + ) -> Dict[str, Dict[str, Optional[str]]]: """ The following filters are currently supported: type, state, route-search.exact-match """ @@ -144,7 +173,7 @@ class TransitGatewayRouteTableBackend: transit_gateway_route_table_id ) if not transit_gateway_route_table: - return [] + return {} attr_pairs = ( ("type", "type"), @@ -157,61 +186,63 @@ class TransitGatewayRouteTableBackend: for attrs in attr_pairs: values = filters.get(attrs[0]) or None if values: - if routes.get(key).get(attrs[1]) not in values: + if routes[key].get(attrs[1]) not in values: routes.pop(key) break if max_results: - routes = routes[: int(max_results)] + routes = routes[: int(max_results)] # type: ignore return routes def set_route_table_association( - self, transit_gateway_attachment_id, transit_gateway_route_table_id - ): + self, transit_gateway_attachment_id: str, transit_gateway_route_table_id: str + ) -> None: self.transit_gateways_route_tables[ transit_gateway_route_table_id ].route_table_association = { - "resourceId": self.transit_gateway_attachments[ + "resourceId": self.transit_gateway_attachments[ # type: ignore[attr-defined] transit_gateway_attachment_id ].resource_id, - "resourceType": self.transit_gateway_attachments[ + "resourceType": self.transit_gateway_attachments[ # type: ignore[attr-defined] transit_gateway_attachment_id ].resource_type, "state": "associated", "transitGatewayAttachmentId": transit_gateway_attachment_id, } - def unset_route_table_association(self, tgw_rt_id): + def unset_route_table_association(self, tgw_rt_id: str) -> None: tgw_rt = self.transit_gateways_route_tables[tgw_rt_id] tgw_rt.route_table_association = {} def set_route_table_propagation( - self, transit_gateway_attachment_id, transit_gateway_route_table_id - ): + self, transit_gateway_attachment_id: str, transit_gateway_route_table_id: str + ) -> None: self.transit_gateways_route_tables[ transit_gateway_route_table_id ].route_table_propagation = { - "resourceId": self.transit_gateway_attachments[ + "resourceId": self.transit_gateway_attachments[ # type: ignore[attr-defined] transit_gateway_attachment_id ].resource_id, - "resourceType": self.transit_gateway_attachments[ + "resourceType": self.transit_gateway_attachments[ # type: ignore[attr-defined] transit_gateway_attachment_id ].resource_type, "state": "enabled", "transitGatewayAttachmentId": transit_gateway_attachment_id, } - def unset_route_table_propagation(self, tgw_rt_id): + def unset_route_table_propagation(self, tgw_rt_id: str) -> None: tgw_rt = self.transit_gateways_route_tables[tgw_rt_id] tgw_rt.route_table_propagation = {} - def disable_route_table_propagation(self, transit_gateway_route_table_id): + def disable_route_table_propagation( + self, transit_gateway_route_table_id: str + ) -> None: self.transit_gateways_route_tables[ transit_gateway_route_table_id ].route_table_propagation = {} - def get_all_transit_gateway_route_table_associations( - self, transit_gateway_route_table_id=None, filters=None - ): + def get_transit_gateway_route_table_associations( + self, transit_gateway_route_table_id: List[str], filters: Any = None + ) -> List[TransitGatewayRouteTable]: transit_gateway_route_tables = list(self.transit_gateways_route_tables.values()) if transit_gateway_route_tables: @@ -236,9 +267,9 @@ class TransitGatewayRouteTableBackend: result = filter_resources(transit_gateway_route_tables, filters, attr_pairs) return result - def get_all_transit_gateway_route_table_propagations( - self, transit_gateway_route_table_id=None, filters=None - ): + def get_transit_gateway_route_table_propagations( + self, transit_gateway_route_table_id: str, filters: Any = None + ) -> List[TransitGatewayRouteTable]: transit_gateway_route_tables = list(self.transit_gateways_route_tables.values()) if transit_gateway_route_tables: @@ -263,36 +294,9 @@ class TransitGatewayRouteTableBackend: result = filter_resources(transit_gateway_route_tables, filters, attr_pairs) return result - -class TransitGatewayRelations(object): - # this class is for TransitGatewayAssociation and TransitGatewayPropagation - def __init__( - self, - backend, - transit_gateway_attachment_id=None, - transit_gateway_route_table_id=None, - state=None, - ): - self.ec2_backend = backend - self.transit_gateway_attachment_id = transit_gateway_attachment_id - self.transit_gateway_route_table_id = transit_gateway_route_table_id - self.resource_id = backend.transit_gateway_attachments[ - transit_gateway_attachment_id - ].resource_id - self.resource_type = backend.transit_gateway_attachments[ - transit_gateway_attachment_id - ].resource_type - self.state = state - - -class TransitGatewayRelationsBackend: - def __init__(self): - self.transit_gateway_associations = {} - self.transit_gateway_propagations = {} - def associate_transit_gateway_route_table( - self, transit_gateway_attachment_id=None, transit_gateway_route_table_id=None - ): + self, transit_gateway_attachment_id: str, transit_gateway_route_table_id: str + ) -> TransitGatewayRelations: transit_gateway_association = TransitGatewayRelations( self, transit_gateway_attachment_id, @@ -302,7 +306,7 @@ class TransitGatewayRelationsBackend: self.set_route_table_association( transit_gateway_attachment_id, transit_gateway_route_table_id ) - self.set_attachment_association( + self.set_attachment_association( # type: ignore[attr-defined] transit_gateway_attachment_id, transit_gateway_route_table_id ) self.transit_gateway_associations[ @@ -312,8 +316,8 @@ class TransitGatewayRelationsBackend: return transit_gateway_association def enable_transit_gateway_route_table_propagation( - self, transit_gateway_attachment_id=None, transit_gateway_route_table_id=None - ): + self, transit_gateway_attachment_id: str, transit_gateway_route_table_id: str + ) -> TransitGatewayRelations: transit_gateway_propagation = TransitGatewayRelations( self, transit_gateway_attachment_id, @@ -323,7 +327,7 @@ class TransitGatewayRelationsBackend: self.set_route_table_propagation( transit_gateway_attachment_id, transit_gateway_route_table_id ) - self.set_attachment_propagation( + self.set_attachment_propagation( # type: ignore[attr-defined] transit_gateway_attachment_id, transit_gateway_route_table_id ) self.transit_gateway_propagations[ @@ -333,12 +337,12 @@ class TransitGatewayRelationsBackend: return transit_gateway_propagation def disable_transit_gateway_route_table_propagation( - self, transit_gateway_attachment_id=None, transit_gateway_route_table_id=None - ): + self, transit_gateway_attachment_id: str, transit_gateway_route_table_id: str + ) -> TransitGatewayRelations: self.disable_route_table_propagation( transit_gateway_route_table_id=transit_gateway_route_table_id ) - self.disable_attachment_propagation( + self.disable_attachment_propagation( # type: ignore[attr-defined] transit_gateway_attachment_id=transit_gateway_attachment_id ) self.transit_gateway_propagations[ @@ -350,11 +354,13 @@ class TransitGatewayRelationsBackend: return transit_gateway_propagation - def disassociate_transit_gateway_route_table(self, tgw_attach_id, tgw_rt_id): + def disassociate_transit_gateway_route_table( + self, tgw_attach_id: str, tgw_rt_id: str + ) -> TransitGatewayRelations: tgw_association = self.transit_gateway_associations.pop(tgw_attach_id) tgw_association.state = "disassociated" self.unset_route_table_association(tgw_rt_id) - self.unset_attachment_association(tgw_attach_id) + self.unset_attachment_association(tgw_attach_id) # type: ignore[attr-defined] return tgw_association diff --git a/moto/ec2/models/vpc_peering_connections.py b/moto/ec2/models/vpc_peering_connections.py index fbbc6a9a1..dfb026328 100644 --- a/moto/ec2/models/vpc_peering_connections.py +++ b/moto/ec2/models/vpc_peering_connections.py @@ -11,28 +11,28 @@ from .core import TaggedEC2Resource from ..utils import random_vpc_peering_connection_id -class PeeringConnectionStatus(object): - def __init__(self, code="initiating-request", message=""): +class PeeringConnectionStatus: + def __init__(self, code: str = "initiating-request", message: str = ""): self.code = code self.message = message - def deleted(self): + def deleted(self) -> None: self.code = "deleted" self.message = "Deleted by {deleter ID}" - def initiating(self): + def initiating(self) -> None: self.code = "initiating-request" self.message = "Initiating Request to {accepter ID}" - def pending(self): + def pending(self) -> None: self.code = "pending-acceptance" self.message = "Pending Acceptance by {accepter ID}" - def accept(self): + def accept(self) -> None: self.code = "active" self.message = "Active" - def reject(self): + def reject(self) -> None: self.code = "rejected" self.message = "Inactive" diff --git a/moto/ec2/responses/subnets.py b/moto/ec2/responses/subnets.py index 8700989bb..48b45c0db 100644 --- a/moto/ec2/responses/subnets.py +++ b/moto/ec2/responses/subnets.py @@ -40,7 +40,7 @@ class Subnets(EC2BaseResponse): self.error_on_dryrun() subnet_ids = self._get_multi_param("SubnetId") filters = self._filters_from_querystring() - subnets = self.ec2_backend.get_all_subnets(subnet_ids, filters) + subnets = self.ec2_backend.describe_subnets(subnet_ids, filters) template = self.response_template(DESCRIBE_SUBNETS_RESPONSE) return template.render(subnets=subnets) diff --git a/moto/ec2/responses/transit_gateway_route_tables.py b/moto/ec2/responses/transit_gateway_route_tables.py index f0dc4f260..22a6d89cc 100644 --- a/moto/ec2/responses/transit_gateway_route_tables.py +++ b/moto/ec2/responses/transit_gateway_route_tables.py @@ -91,7 +91,7 @@ class TransitGatewayRouteTable(EC2BaseResponse): transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId") filters = self._filters_from_querystring() transit_gateway_route_table_associations = ( - self.ec2_backend.get_all_transit_gateway_route_table_associations( + self.ec2_backend.get_transit_gateway_route_table_associations( transit_gateway_route_table_id, filters ) ) @@ -106,7 +106,7 @@ class TransitGatewayRouteTable(EC2BaseResponse): transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId") filters = self._filters_from_querystring() transit_gateway_route_table_propagations = ( - self.ec2_backend.get_all_transit_gateway_route_table_propagations( + self.ec2_backend.get_transit_gateway_route_table_propagations( transit_gateway_route_table_id, filters ) ) diff --git a/moto/ec2/utils.py b/moto/ec2/utils.py index 22eb15015..3c724732a 100644 --- a/moto/ec2/utils.py +++ b/moto/ec2/utils.py @@ -208,17 +208,17 @@ def random_nat_gateway_id() -> str: return random_id(prefix=EC2_RESOURCE_TO_PREFIX["nat-gateway"], size=17) -def random_transit_gateway_id(): +def random_transit_gateway_id() -> str: return random_id(prefix=EC2_RESOURCE_TO_PREFIX["transit-gateway"], size=17) -def random_transit_gateway_route_table_id(): +def random_transit_gateway_route_table_id() -> str: return random_id( prefix=EC2_RESOURCE_TO_PREFIX["transit-gateway-route-table"], size=17 ) -def random_transit_gateway_attachment_id(): +def random_transit_gateway_attachment_id() -> str: return random_id( prefix=EC2_RESOURCE_TO_PREFIX["transit-gateway-attachment"], size=17 ) @@ -543,7 +543,7 @@ def generic_filter( return objects -def simple_aws_filter_to_re(filter_string): +def simple_aws_filter_to_re(filter_string: str) -> str: tmp_filter = filter_string.replace(r"\?", "[?]") tmp_filter = tmp_filter.replace(r"\*", "[*]") tmp_filter = fnmatch.translate(tmp_filter) @@ -567,7 +567,7 @@ def random_key_pair() -> Dict[str, str]: } -def get_prefix(resource_id): +def get_prefix(resource_id: str) -> str: resource_id_prefix, _, after = resource_id.partition("-") if resource_id_prefix == EC2_RESOURCE_TO_PREFIX["transit-gateway"]: if after.startswith("rtb"): @@ -584,6 +584,7 @@ def get_prefix(resource_id): if uuid4hex.match(resource_id) is not None: 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 resource_id_prefix diff --git a/moto/redshift/models.py b/moto/redshift/models.py index 02e91aff0..0be0216da 100644 --- a/moto/redshift/models.py +++ b/moto/redshift/models.py @@ -384,7 +384,7 @@ class SubnetGroup(TaggableResourceMixin, CloudFormationModel): @property def subnets(self): - return self.ec2_backend.get_all_subnets(filters={"subnet-id": self.subnet_ids}) + return self.ec2_backend.describe_subnets(filters={"subnet-id": self.subnet_ids}) @property def vpc_id(self): diff --git a/moto/route53resolver/models.py b/moto/route53resolver/models.py index 237aeb827..81b8de7b3 100644 --- a/moto/route53resolver/models.py +++ b/moto/route53resolver/models.py @@ -218,7 +218,7 @@ class ResolverEndpoint(BaseModel): # pylint: disable=too-many-instance-attribut of the subnets has already been checked. """ first_subnet_id = self.ip_addresses[0]["SubnetId"] - subnet_info = self.ec2_backend.get_all_subnets(subnet_ids=[first_subnet_id])[0] + subnet_info = self.ec2_backend.describe_subnets(subnet_ids=[first_subnet_id])[0] return subnet_info.vpc_id def _build_subnet_info(self): @@ -415,7 +415,7 @@ class Route53ResolverBackend(BaseBackend): subnets = defaultdict(set) for subnet_id, ip_addr in [(x["SubnetId"], x["Ip"]) for x in ip_addresses]: try: - subnet_info = self.ec2_backend.get_all_subnets(subnet_ids=[subnet_id])[ + subnet_info = self.ec2_backend.describe_subnets(subnet_ids=[subnet_id])[ 0 ] except InvalidSubnetIdError as exc: @@ -496,7 +496,7 @@ class Route53ResolverBackend(BaseBackend): for x in ip_addresses: if not x.get("Ip"): - subnet_info = self.ec2_backend.get_all_subnets( + subnet_info = self.ec2_backend.describe_subnets( subnet_ids=[x["SubnetId"]] )[0] x["Ip"] = subnet_info.get_available_subnet_ip(self) @@ -877,7 +877,7 @@ class Route53ResolverBackend(BaseBackend): resolver_endpoint = self.resolver_endpoints[resolver_endpoint_id] if not value.get("Ip"): - subnet_info = self.ec2_backend.get_all_subnets( + subnet_info = self.ec2_backend.describe_subnets( subnet_ids=[value.get("SubnetId")] )[0] value["Ip"] = subnet_info.get_available_subnet_ip(self) diff --git a/moto/utilities/utils.py b/moto/utilities/utils.py index 4cf578127..8b3d7cc5f 100644 --- a/moto/utilities/utils.py +++ b/moto/utilities/utils.py @@ -41,7 +41,7 @@ RESOURCE_TYPE = TypeVar("RESOURCE_TYPE") def filter_resources( resources: List[RESOURCE_TYPE], filters: Any, - attr_pairs: Tuple[Tuple[str, str], ...], + attr_pairs: Tuple[Tuple[str, ...], ...], ) -> List[RESOURCE_TYPE]: """ Used to filter resources. Usually in get and describe apis. diff --git a/setup.cfg b/setup.cfg index 289e84d00..ebedb8289 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/ec2/models/c*,moto/ec2/models/d*,moto/ec2/models/e*,moto/ec2/models/f*,moto/ec2/models/h*,moto/ec2/models/i*,moto/ec2/models/k*,moto/ec2/models/l*,moto/ec2/models/m*,moto/ec2/models/n*,moto/ec2/models/r*,moto/ec2/models/s*,moto/moto_api +files= moto/a*,moto/b*,moto/c*,moto/d*,moto/ebs/,moto/ec2/models/a*,moto/ec2/models/c*,moto/ec2/models/d*,moto/ec2/models/e*,moto/ec2/models/f*,moto/ec2/models/h*,moto/ec2/models/i*,moto/ec2/models/k*,moto/ec2/models/l*,moto/ec2/models/m*,moto/ec2/models/n*,moto/ec2/models/r*,moto/ec2/models/s*,moto/ec2/models/t*,moto/moto_api show_column_numbers=True show_error_codes = True disable_error_code=abstract