Techdebt: MyPy EC2 (r-models) (#5909)

This commit is contained in:
Bert Blommers 2023-02-07 22:04:11 -01:00 committed by GitHub
parent 749a8572ba
commit 2ee5dcea63
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 217 additions and 188 deletions

View File

@ -220,7 +220,7 @@ class InvalidRouteTableIdError(EC2ClientError):
class InvalidRouteError(EC2ClientError): class InvalidRouteError(EC2ClientError):
def __init__(self, route_table_id, cidr): def __init__(self, route_table_id: str, cidr: str):
super().__init__( super().__init__(
"InvalidRoute.NotFound", "InvalidRoute.NotFound",
f"no route with destination-cidr-block {cidr} in route table {route_table_id}", f"no route with destination-cidr-block {cidr} in route table {route_table_id}",
@ -228,7 +228,7 @@ class InvalidRouteError(EC2ClientError):
class RouteAlreadyExistsError(EC2ClientError): class RouteAlreadyExistsError(EC2ClientError):
def __init__(self, cidr): def __init__(self, cidr: str):
super().__init__( super().__init__(
"RouteAlreadyExists", f"The route identified by {cidr} already exists" "RouteAlreadyExists", f"The route identified by {cidr} already exists"
) )
@ -593,7 +593,7 @@ class InvalidCIDRBlockParameterError(EC2ClientError):
class InvalidDestinationCIDRBlockParameterError(EC2ClientError): class InvalidDestinationCIDRBlockParameterError(EC2ClientError):
def __init__(self, cidr_block): def __init__(self, cidr_block: str):
super().__init__( super().__init__(
"InvalidParameterValue", "InvalidParameterValue",
f"Value ({cidr_block}) for parameter destinationCidrBlock is invalid. This is not a valid CIDR block.", f"Value ({cidr_block}) for parameter destinationCidrBlock is invalid. This is not a valid CIDR block.",

View File

@ -28,7 +28,7 @@ from .instance_types import InstanceTypeBackend, InstanceTypeOfferingBackend
from .nat_gateways import NatGatewayBackend from .nat_gateways import NatGatewayBackend
from .network_acls import NetworkAclBackend from .network_acls import NetworkAclBackend
from .availability_zones_and_regions import RegionsAndZonesBackend from .availability_zones_and_regions import RegionsAndZonesBackend
from .route_tables import RouteBackend, RouteTableBackend from .route_tables import RouteBackend
from .security_groups import SecurityGroupBackend from .security_groups import SecurityGroupBackend
from .spot_requests import ( from .spot_requests import (
SpotRequestBackend, SpotRequestBackend,
@ -101,7 +101,6 @@ class EC2Backend(
VPNConnectionBackend, VPNConnectionBackend,
VPCServiceConfigurationBackend, VPCServiceConfigurationBackend,
VPCPeeringConnectionBackend, VPCPeeringConnectionBackend,
RouteTableBackend,
RouteBackend, RouteBackend,
InternetGatewayBackend, InternetGatewayBackend,
EgressOnlyInternetGatewayBackend, EgressOnlyInternetGatewayBackend,

View File

@ -1,6 +1,16 @@
import ipaddress import ipaddress
from typing import Any, Dict, List, Optional, Set
from moto.core import CloudFormationModel from moto.core import CloudFormationModel
from moto.ec2.models.carrier_gateways import CarrierGateway
from moto.ec2.models.elastic_network_interfaces import NetworkInterface
from moto.ec2.models.internet_gateways import EgressOnlyInternetGateway
from moto.ec2.models.instances import Instance
from moto.ec2.models.managed_prefixes import ManagedPrefixList
from moto.ec2.models.nat_gateways import NatGateway
from moto.ec2.models.transit_gateway import TransitGateway
from moto.ec2.models.vpc_peering_connections import VPCPeeringConnection
from moto.ec2.models.vpn_gateway import VpnGateway
from .core import TaggedEC2Resource from .core import TaggedEC2Resource
from ..exceptions import ( from ..exceptions import (
DependencyViolationError, DependencyViolationError,
@ -16,39 +26,42 @@ from ..utils import (
generic_filter, generic_filter,
random_subnet_association_id, random_subnet_association_id,
random_route_table_id, random_route_table_id,
split_route_id,
) )
class RouteTable(TaggedEC2Resource, CloudFormationModel): class RouteTable(TaggedEC2Resource, CloudFormationModel):
def __init__(self, ec2_backend, route_table_id, vpc_id, main=False): def __init__(
self, ec2_backend: Any, route_table_id: str, vpc_id: str, main: bool = False
):
self.ec2_backend = ec2_backend self.ec2_backend = ec2_backend
self.id = route_table_id self.id = route_table_id
self.vpc_id = vpc_id self.vpc_id = vpc_id
if main: self.main_association_id = random_subnet_association_id() if main else None
self.main_association_id = random_subnet_association_id() self.associations: Dict[str, str] = {}
else: self.routes: Dict[str, Route] = {}
self.main_association_id = None
self.associations = {}
self.routes = {}
@property @property
def owner_id(self): def owner_id(self) -> str:
return self.ec2_backend.account_id return self.ec2_backend.account_id
@staticmethod @staticmethod
def cloudformation_name_type(): def cloudformation_name_type() -> str:
return None return ""
@staticmethod @staticmethod
def cloudformation_type(): def cloudformation_type() -> str:
# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-ec2-routetable.html # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-ec2-routetable.html
return "AWS::EC2::RouteTable" return "AWS::EC2::RouteTable"
@classmethod @classmethod
def create_from_cloudformation_json( def create_from_cloudformation_json( # type: ignore[misc]
cls, resource_name, cloudformation_json, account_id, region_name, **kwargs cls,
): resource_name: str,
cloudformation_json: Any,
account_id: str,
region_name: str,
**kwargs: Any,
) -> "RouteTable":
from ..models import ec2_backends from ..models import ec2_backends
properties = cloudformation_json["Properties"] properties = cloudformation_json["Properties"]
@ -59,10 +72,12 @@ class RouteTable(TaggedEC2Resource, CloudFormationModel):
return route_table return route_table
@property @property
def physical_resource_id(self): def physical_resource_id(self) -> str:
return self.id return self.id
def get_filter_value(self, filter_name): def get_filter_value(
self, filter_name: str, method_name: Optional[str] = None
) -> Any:
if filter_name == "association.main": if filter_name == "association.main":
# Note: Boto only supports 'true'. # Note: Boto only supports 'true'.
# https://github.com/boto/boto/issues/1742 # https://github.com/boto/boto/issues/1742
@ -96,7 +111,7 @@ class RouteTable(TaggedEC2Resource, CloudFormationModel):
return super().get_filter_value(filter_name, "DescribeRouteTables") return super().get_filter_value(filter_name, "DescribeRouteTables")
@property @property
def all_associations_ids(self): def all_associations_ids(self) -> Set[str]:
# NOTE(yoctozepto): Doing an explicit copy to not touch the original. # NOTE(yoctozepto): Doing an explicit copy to not touch the original.
all_associations = set(self.associations) all_associations = set(self.associations)
if self.main_association_id is not None: if self.main_association_id is not None:
@ -104,16 +119,108 @@ class RouteTable(TaggedEC2Resource, CloudFormationModel):
return all_associations return all_associations
class RouteTableBackend: class Route(CloudFormationModel):
def __init__(self): def __init__(
self.route_tables = {} self,
route_table: RouteTable,
destination_cidr_block: Optional[str],
destination_ipv6_cidr_block: Optional[str],
destination_prefix_list: Optional[ManagedPrefixList] = None,
local: bool = False,
gateway: Optional[VpnGateway] = None,
instance: Optional[Instance] = None,
nat_gateway: Optional[NatGateway] = None,
egress_only_igw: Optional[EgressOnlyInternetGateway] = None,
transit_gateway: Optional[TransitGateway] = None,
interface: Optional[NetworkInterface] = None,
vpc_pcx: Optional[VPCPeeringConnection] = None,
carrier_gateway: Optional[CarrierGateway] = None,
):
self.id = generate_route_id(
route_table.id,
destination_cidr_block,
destination_ipv6_cidr_block,
destination_prefix_list.id if destination_prefix_list else None,
)
self.route_table = route_table
self.destination_cidr_block = destination_cidr_block
self.destination_ipv6_cidr_block = destination_ipv6_cidr_block
self.destination_prefix_list = destination_prefix_list
self.local = local
self.gateway = gateway
self.instance = instance
self.nat_gateway = nat_gateway
self.egress_only_igw = egress_only_igw
self.transit_gateway = transit_gateway
self.interface = interface
self.vpc_pcx = vpc_pcx
self.carrier_gateway = carrier_gateway
def create_route_table(self, vpc_id, tags=None, main=False): @property
def physical_resource_id(self) -> str:
return self.id
@staticmethod
def cloudformation_name_type() -> str:
return ""
@staticmethod
def cloudformation_type() -> str:
# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-ec2-route.html
return "AWS::EC2::Route"
@classmethod
def create_from_cloudformation_json( # type: ignore[misc]
cls,
resource_name: str,
cloudformation_json: Any,
account_id: str,
region_name: str,
**kwargs: str,
) -> "Route":
from ..models import ec2_backends
properties = cloudformation_json["Properties"]
gateway_id = properties.get("GatewayId")
instance_id = properties.get("InstanceId")
interface_id = properties.get("NetworkInterfaceId")
nat_gateway_id = properties.get("NatGatewayId")
egress_only_igw_id = properties.get("EgressOnlyInternetGatewayId")
transit_gateway_id = properties.get("TransitGatewayId")
pcx_id = properties.get("VpcPeeringConnectionId")
route_table_id = properties["RouteTableId"]
ec2_backend = ec2_backends[account_id][region_name]
route = ec2_backend.create_route(
route_table_id=route_table_id,
destination_cidr_block=properties.get("DestinationCidrBlock"),
gateway_id=gateway_id,
instance_id=instance_id,
nat_gateway_id=nat_gateway_id,
egress_only_igw_id=egress_only_igw_id,
transit_gateway_id=transit_gateway_id,
interface_id=interface_id,
vpc_peering_connection_id=pcx_id,
)
return route
class RouteBackend:
def __init__(self) -> None:
self.route_tables: Dict[str, RouteTable] = {}
def create_route_table(
self,
vpc_id: str,
tags: Optional[List[Dict[str, str]]] = None,
main: bool = False,
) -> RouteTable:
route_table_id = random_route_table_id() route_table_id = random_route_table_id()
vpc = self.get_vpc(vpc_id) # Validate VPC exists vpc = self.get_vpc(vpc_id) # type: ignore[attr-defined] # Validate VPC exists
route_table = RouteTable(self, route_table_id, vpc_id, main=main) route_table = RouteTable(self, route_table_id, vpc_id, main=main)
for tag in tags or []: for tag in tags or []:
route_table.add_tag(tag.get("Key"), tag.get("Value")) route_table.add_tag(tag["Key"], tag["Value"])
self.route_tables[route_table_id] = route_table self.route_tables[route_table_id] = route_table
# creating default routes for ipv4 cirds # creating default routes for ipv4 cirds
@ -133,14 +240,16 @@ class RouteTableBackend:
return route_table return route_table
def get_route_table(self, route_table_id): def get_route_table(self, route_table_id: str) -> RouteTable:
route_table = self.route_tables.get(route_table_id, None) route_table = self.route_tables.get(route_table_id, None)
if not route_table: if not route_table:
raise InvalidRouteTableIdError(route_table_id) raise InvalidRouteTableIdError(route_table_id)
return route_table return route_table
def describe_route_tables(self, route_table_ids=None, filters=None): def describe_route_tables(
route_tables = self.route_tables.copy().values() self, route_table_ids: Optional[List[str]] = None, filters: Any = None
) -> List[RouteTable]:
route_tables = list(self.route_tables.values())
if route_table_ids: if route_table_ids:
route_tables = [ route_tables = [
@ -158,16 +267,20 @@ class RouteTableBackend:
return generic_filter(filters, route_tables) return generic_filter(filters, route_tables)
def delete_route_table(self, route_table_id): def delete_route_table(self, route_table_id: str) -> None:
route_table = self.get_route_table(route_table_id) route_table = self.get_route_table(route_table_id)
if route_table.associations: if route_table.associations:
raise DependencyViolationError( raise DependencyViolationError(
f"The routeTable '{route_table_id}' has dependencies and cannot be deleted." f"The routeTable '{route_table_id}' has dependencies and cannot be deleted."
) )
self.route_tables.pop(route_table_id) self.route_tables.pop(route_table_id)
return True
def associate_route_table(self, route_table_id, gateway_id=None, subnet_id=None): def associate_route_table(
self,
route_table_id: str,
gateway_id: Optional[str] = None,
subnet_id: Optional[str] = None,
) -> str:
# Idempotent if association already exists. # Idempotent if association already exists.
route_tables_by_subnet = self.describe_route_tables( route_tables_by_subnet = self.describe_route_tables(
filters={"association.subnet-id": [subnet_id]} filters={"association.subnet-id": [subnet_id]}
@ -182,22 +295,24 @@ class RouteTableBackend:
# Association does not yet exist, so create it. # Association does not yet exist, so create it.
route_table = self.get_route_table(route_table_id) route_table = self.get_route_table(route_table_id)
if gateway_id is None: if gateway_id is None:
self.get_subnet(subnet_id) # Validate subnet exists self.get_subnet(subnet_id) # type: ignore[attr-defined] # Validate subnet exists
association_id = random_subnet_association_id() association_id = random_subnet_association_id()
route_table.associations[association_id] = subnet_id route_table.associations[association_id] = subnet_id # type: ignore[assignment]
return association_id return association_id
if subnet_id is None: else:
association_id = random_subnet_association_id() association_id = random_subnet_association_id()
route_table.associations[association_id] = gateway_id route_table.associations[association_id] = gateway_id
return association_id return association_id
def disassociate_route_table(self, association_id): def disassociate_route_table(self, association_id: str) -> Optional[str]:
for route_table in self.route_tables.values(): for route_table in self.route_tables.values():
if association_id in route_table.associations: if association_id in route_table.associations:
return route_table.associations.pop(association_id, None) return route_table.associations.pop(association_id, None)
raise InvalidAssociationIdError(association_id) raise InvalidAssociationIdError(association_id)
def replace_route_table_association(self, association_id, route_table_id): def replace_route_table_association(
self, association_id: str, route_table_id: str
) -> str:
# Idempotent if association already exists. # Idempotent if association already exists.
new_route_table = self.get_route_table(route_table_id) new_route_table = self.get_route_table(route_table_id)
if association_id in new_route_table.all_associations_ids: if association_id in new_route_table.all_associations_ids:
@ -223,107 +338,22 @@ class RouteTableBackend:
new_route_table.associations[new_association_id] = association_target_id new_route_table.associations[new_association_id] = association_target_id
return new_association_id return new_association_id
# TODO: refractor to isloate class methods from backend logic
class Route(CloudFormationModel):
def __init__(
self,
route_table,
destination_cidr_block,
destination_ipv6_cidr_block,
destination_prefix_list=None,
local=False,
gateway=None,
instance=None,
nat_gateway=None,
egress_only_igw=None,
transit_gateway=None,
interface=None,
vpc_pcx=None,
carrier_gateway=None,
):
self.id = generate_route_id(
route_table.id,
destination_cidr_block,
destination_ipv6_cidr_block,
destination_prefix_list.id if destination_prefix_list else None,
)
self.route_table = route_table
self.destination_cidr_block = destination_cidr_block
self.destination_ipv6_cidr_block = destination_ipv6_cidr_block
self.destination_prefix_list = destination_prefix_list
self.local = local
self.gateway = gateway
self.instance = instance
self.nat_gateway = nat_gateway
self.egress_only_igw = egress_only_igw
self.transit_gateway = transit_gateway
self.interface = interface
self.vpc_pcx = vpc_pcx
self.carrier_gateway = carrier_gateway
@property
def physical_resource_id(self):
return self.id
@staticmethod
def cloudformation_name_type():
return None
@staticmethod
def cloudformation_type():
# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-ec2-route.html
return "AWS::EC2::Route"
@classmethod
def create_from_cloudformation_json(
cls, resource_name, cloudformation_json, account_id, region_name, **kwargs
):
from ..models import ec2_backends
properties = cloudformation_json["Properties"]
gateway_id = properties.get("GatewayId")
instance_id = properties.get("InstanceId")
interface_id = properties.get("NetworkInterfaceId")
nat_gateway_id = properties.get("NatGatewayId")
egress_only_igw_id = properties.get("EgressOnlyInternetGatewayId")
transit_gateway_id = properties.get("TransitGatewayId")
pcx_id = properties.get("VpcPeeringConnectionId")
route_table_id = properties["RouteTableId"]
ec2_backend = ec2_backends[account_id][region_name]
route_table = ec2_backend.create_route(
route_table_id=route_table_id,
destination_cidr_block=properties.get("DestinationCidrBlock"),
gateway_id=gateway_id,
instance_id=instance_id,
nat_gateway_id=nat_gateway_id,
egress_only_igw_id=egress_only_igw_id,
transit_gateway_id=transit_gateway_id,
interface_id=interface_id,
vpc_peering_connection_id=pcx_id,
)
return route_table
class RouteBackend:
def create_route( def create_route(
self, self,
route_table_id, route_table_id: str,
destination_cidr_block, destination_cidr_block: Optional[str],
destination_ipv6_cidr_block=None, destination_ipv6_cidr_block: Optional[str] = None,
destination_prefix_list_id=None, destination_prefix_list_id: Optional[str] = None,
local=False, local: bool = False,
gateway_id=None, gateway_id: Optional[str] = None,
instance_id=None, instance_id: Optional[str] = None,
nat_gateway_id=None, nat_gateway_id: Optional[str] = None,
egress_only_igw_id=None, egress_only_igw_id: Optional[str] = None,
transit_gateway_id=None, transit_gateway_id: Optional[str] = None,
interface_id=None, interface_id: Optional[str] = None,
vpc_peering_connection_id=None, vpc_peering_connection_id: Optional[str] = None,
carrier_gateway_id=None, carrier_gateway_id: Optional[str] = None,
): ) -> Route:
gateway = None gateway = None
nat_gateway = None nat_gateway = None
transit_gateway = None transit_gateway = None
@ -336,16 +366,16 @@ class RouteBackend:
if interface_id: if interface_id:
# for validating interface Id whether it is valid or not. # for validating interface Id whether it is valid or not.
interface = self.get_network_interface(interface_id) interface = self.get_network_interface(interface_id) # type: ignore[attr-defined]
else: else:
if gateway_id: if gateway_id:
if EC2_RESOURCE_TO_PREFIX["vpn-gateway"] in gateway_id: if EC2_RESOURCE_TO_PREFIX["vpn-gateway"] in gateway_id:
gateway = self.get_vpn_gateway(gateway_id) gateway = self.get_vpn_gateway(gateway_id) # type: ignore[attr-defined]
elif EC2_RESOURCE_TO_PREFIX["internet-gateway"] in gateway_id: elif EC2_RESOURCE_TO_PREFIX["internet-gateway"] in gateway_id:
gateway = self.get_internet_gateway(gateway_id) gateway = self.get_internet_gateway(gateway_id) # type: ignore[attr-defined]
elif EC2_RESOURCE_TO_PREFIX["vpc-endpoint"] in gateway_id: elif EC2_RESOURCE_TO_PREFIX["vpc-endpoint"] in gateway_id:
gateway = self.get_vpc_end_point(gateway_id) gateway = self.get_vpc_end_point(gateway_id) # type: ignore[attr-defined]
if destination_cidr_block: if destination_cidr_block:
self.__validate_destination_cidr_block( self.__validate_destination_cidr_block(
@ -353,17 +383,17 @@ class RouteBackend:
) )
if nat_gateway_id is not None: if nat_gateway_id is not None:
nat_gateway = self.nat_gateways.get(nat_gateway_id) nat_gateway = self.nat_gateways.get(nat_gateway_id) # type: ignore[attr-defined]
if egress_only_igw_id is not None: if egress_only_igw_id is not None:
egress_only_igw = self.get_egress_only_igw(egress_only_igw_id) egress_only_igw = self.get_egress_only_igw(egress_only_igw_id) # type: ignore[attr-defined]
if transit_gateway_id is not None: if transit_gateway_id is not None:
transit_gateway = self.transit_gateways.get(transit_gateway_id) transit_gateway = self.transit_gateways.get(transit_gateway_id) # type: ignore[attr-defined]
if destination_prefix_list_id is not None: if destination_prefix_list_id is not None:
destination_prefix_list = self.managed_prefix_lists.get( destination_prefix_list = self.managed_prefix_lists.get( # type: ignore[attr-defined]
destination_prefix_list_id destination_prefix_list_id
) )
if carrier_gateway_id is not None: if carrier_gateway_id is not None:
carrier_gateway = self.carrier_gateways.get(carrier_gateway_id) carrier_gateway = self.carrier_gateways.get(carrier_gateway_id) # type: ignore[attr-defined]
route = Route( route = Route(
route_table, route_table,
@ -372,13 +402,13 @@ class RouteBackend:
destination_prefix_list, destination_prefix_list,
local=local, local=local,
gateway=gateway, gateway=gateway,
instance=self.get_instance(instance_id) if instance_id else None, instance=self.get_instance(instance_id) if instance_id else None, # type: ignore[attr-defined]
nat_gateway=nat_gateway, nat_gateway=nat_gateway,
egress_only_igw=egress_only_igw, egress_only_igw=egress_only_igw,
transit_gateway=transit_gateway, transit_gateway=transit_gateway,
interface=interface, interface=interface,
carrier_gateway=carrier_gateway, carrier_gateway=carrier_gateway,
vpc_pcx=self.get_vpc_peering_connection(vpc_peering_connection_id) vpc_pcx=self.get_vpc_peering_connection(vpc_peering_connection_id) # type: ignore[attr-defined]
if vpc_peering_connection_id if vpc_peering_connection_id
else None, else None,
) )
@ -387,18 +417,18 @@ class RouteBackend:
def replace_route( def replace_route(
self, self,
route_table_id, route_table_id: str,
destination_cidr_block, destination_cidr_block: str,
destination_ipv6_cidr_block=None, destination_ipv6_cidr_block: Optional[str] = None,
destination_prefix_list_id=None, destination_prefix_list_id: Optional[str] = None,
nat_gateway_id=None, nat_gateway_id: Optional[str] = None,
egress_only_igw_id=None, egress_only_igw_id: Optional[str] = None,
transit_gateway_id=None, transit_gateway_id: Optional[str] = None,
gateway_id=None, gateway_id: Optional[str] = None,
instance_id=None, instance_id: Optional[str] = None,
interface_id=None, interface_id: Optional[str] = None,
vpc_peering_connection_id=None, vpc_peering_connection_id: Optional[str] = None,
): ) -> Route:
route_table = self.get_route_table(route_table_id) route_table = self.get_route_table(route_table_id)
route_id = generate_route_id( route_id = generate_route_id(
route_table.id, destination_cidr_block, destination_ipv6_cidr_block route_table.id, destination_cidr_block, destination_ipv6_cidr_block
@ -406,7 +436,7 @@ class RouteBackend:
route = route_table.routes[route_id] route = route_table.routes[route_id]
if interface_id: if interface_id:
self.raise_not_implemented_error("ReplaceRoute to NetworkInterfaceId") self.raise_not_implemented_error("ReplaceRoute to NetworkInterfaceId") # type: ignore[attr-defined]
route.gateway = None route.gateway = None
route.nat_gateway = None route.nat_gateway = None
@ -414,25 +444,25 @@ class RouteBackend:
route.transit_gateway = None route.transit_gateway = None
if gateway_id: if gateway_id:
if EC2_RESOURCE_TO_PREFIX["vpn-gateway"] in gateway_id: if EC2_RESOURCE_TO_PREFIX["vpn-gateway"] in gateway_id:
route.gateway = self.get_vpn_gateway(gateway_id) route.gateway = self.get_vpn_gateway(gateway_id) # type: ignore[attr-defined]
elif EC2_RESOURCE_TO_PREFIX["internet-gateway"] in gateway_id: elif EC2_RESOURCE_TO_PREFIX["internet-gateway"] in gateway_id:
route.gateway = self.get_internet_gateway(gateway_id) route.gateway = self.get_internet_gateway(gateway_id) # type: ignore[attr-defined]
if nat_gateway_id is not None: if nat_gateway_id is not None:
route.nat_gateway = self.nat_gateways.get(nat_gateway_id) route.nat_gateway = self.nat_gateways.get(nat_gateway_id) # type: ignore[attr-defined]
if egress_only_igw_id is not None: if egress_only_igw_id is not None:
route.egress_only_igw = self.get_egress_only_igw(egress_only_igw_id) route.egress_only_igw = self.get_egress_only_igw(egress_only_igw_id) # type: ignore[attr-defined]
if transit_gateway_id is not None: if transit_gateway_id is not None:
route.transit_gateway = self.transit_gateways.get(transit_gateway_id) route.transit_gateway = self.transit_gateways.get(transit_gateway_id) # type: ignore[attr-defined]
if destination_prefix_list_id is not None: if destination_prefix_list_id is not None:
route.prefix_list = self.managed_prefix_lists.get( route.prefix_list = self.managed_prefix_lists.get( # type: ignore[attr-defined]
destination_prefix_list_id destination_prefix_list_id
) )
route.instance = self.get_instance(instance_id) if instance_id else None route.instance = self.get_instance(instance_id) if instance_id else None # type: ignore[attr-defined]
route.interface = None route.interface = None
route.vpc_pcx = ( route.vpc_pcx = (
self.get_vpc_peering_connection(vpc_peering_connection_id) self.get_vpc_peering_connection(vpc_peering_connection_id) # type: ignore[attr-defined]
if vpc_peering_connection_id if vpc_peering_connection_id
else None else None
) )
@ -440,18 +470,13 @@ class RouteBackend:
route_table.routes[route.id] = route route_table.routes[route.id] = route
return route return route
def get_route(self, route_id):
route_table_id, _ = split_route_id(route_id)
route_table = self.get_route_table(route_table_id)
return route_table.get(route_id)
def delete_route( def delete_route(
self, self,
route_table_id, route_table_id: str,
destination_cidr_block, destination_cidr_block: str,
destination_ipv6_cidr_block=None, destination_ipv6_cidr_block: Optional[str] = None,
destination_prefix_list_id=None, destination_prefix_list_id: Optional[str] = None,
): ) -> Route:
cidr = destination_cidr_block cidr = destination_cidr_block
route_table = self.get_route_table(route_table_id) route_table = self.get_route_table(route_table_id)
if destination_ipv6_cidr_block: if destination_ipv6_cidr_block:
@ -464,7 +489,9 @@ class RouteBackend:
raise InvalidRouteError(route_table_id, cidr) raise InvalidRouteError(route_table_id, cidr)
return deleted return deleted
def __validate_destination_cidr_block(self, destination_cidr_block, route_table): def __validate_destination_cidr_block(
self, destination_cidr_block: str, route_table: RouteTable
) -> None:
""" """
Utility function to check the destination CIDR block Utility function to check the destination CIDR block
Will validate the format and check for overlap with existing routes Will validate the format and check for overlap with existing routes

View File

@ -7,7 +7,7 @@ from datetime import datetime
from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.primitives.asymmetric import rsa
from typing import Any, Dict, List, TypeVar, Optional from typing import Any, Dict, List, TypeVar, Tuple, Optional
from moto.iam import iam_backends from moto.iam import iam_backends
from moto.moto_api._internal import mock_random as random from moto.moto_api._internal import mock_random as random
@ -122,7 +122,7 @@ def random_subnet_ipv6_cidr_block_association_id():
) )
def random_subnet_association_id(): def random_subnet_association_id() -> str:
return random_id(prefix=EC2_RESOURCE_TO_PREFIX["route-table-association"]) return random_id(prefix=EC2_RESOURCE_TO_PREFIX["route-table-association"])
@ -184,7 +184,7 @@ def random_egress_only_internet_gateway_id() -> str:
) )
def random_route_table_id(): def random_route_table_id() -> str:
return random_id(prefix=EC2_RESOURCE_TO_PREFIX["route-table"]) return random_id(prefix=EC2_RESOURCE_TO_PREFIX["route-table"])
@ -281,7 +281,10 @@ def random_ipv6_cidr():
def generate_route_id( def generate_route_id(
route_table_id, cidr_block, ipv6_cidr_block=None, prefix_list=None route_table_id: str,
cidr_block: Optional[str],
ipv6_cidr_block: Optional[str] = None,
prefix_list: Optional[str] = None,
): ):
if ipv6_cidr_block and not cidr_block: if ipv6_cidr_block and not cidr_block:
cidr_block = ipv6_cidr_block cidr_block = ipv6_cidr_block
@ -309,7 +312,7 @@ def utc_date_and_time() -> str:
return f"{x.year}-{x.month:02d}-{x.day:02d}T{x.hour:02d}:{x.minute:02d}:{x.second:02d}.000Z" return f"{x.year}-{x.month:02d}-{x.day:02d}T{x.hour:02d}:{x.minute:02d}:{x.second:02d}.000Z"
def split_route_id(route_id): def split_route_id(route_id: str) -> Tuple[str, str]:
values = route_id.split("~") values = route_id.split("~")
return values[0], values[1] return values[0], values[1]

View File

@ -229,7 +229,7 @@ disable = W,C,R,E
enable = anomalous-backslash-in-string, arguments-renamed, dangerous-default-value, deprecated-module, function-redefined, import-self, redefined-builtin, redefined-outer-name, reimported, pointless-statement, super-with-arguments, unused-argument, unused-import, unused-variable, useless-else-on-loop, wildcard-import 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] [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/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/moto_api
show_column_numbers=True show_column_numbers=True
show_error_codes = True show_error_codes = True
disable_error_code=abstract disable_error_code=abstract