Techdebt: MyPy ELB (#5996)

This commit is contained in:
Bert Blommers 2023-02-27 22:08:24 -01:00 committed by GitHub
parent c4f99f3102
commit 8eaa9fa370
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 227 additions and 176 deletions

View File

@ -1,29 +1,30 @@
from typing import Any
from moto.core.exceptions import RESTError from moto.core.exceptions import RESTError
class ELBClientError(RESTError): class ELBClientError(RESTError):
code = 400 code = 400
def __init__(self, error_type, message): def __init__(self, error_type: str, message: str):
super().__init__(error_type, message, template="wrapped_single_error") super().__init__(error_type, message, template="wrapped_single_error")
class DuplicateTagKeysError(ELBClientError): class DuplicateTagKeysError(ELBClientError):
def __init__(self, cidr): def __init__(self, cidr: Any):
super().__init__( super().__init__(
"DuplicateTagKeys", f"Tag key was specified more than once: {cidr}" "DuplicateTagKeys", f"Tag key was specified more than once: {cidr}"
) )
class CertificateNotFoundException(ELBClientError): class CertificateNotFoundException(ELBClientError):
def __init__(self): def __init__(self) -> None:
super().__init__( super().__init__(
"CertificateNotFoundException", "Supplied certificate was not found" "CertificateNotFoundException", "Supplied certificate was not found"
) )
class LoadBalancerNotFoundError(ELBClientError): class LoadBalancerNotFoundError(ELBClientError):
def __init__(self, name): def __init__(self, name: str):
super().__init__( super().__init__(
"LoadBalancerNotFound", "LoadBalancerNotFound",
f"The specified load balancer does not exist: {name}", f"The specified load balancer does not exist: {name}",
@ -31,21 +32,21 @@ class LoadBalancerNotFoundError(ELBClientError):
class NoActiveLoadBalancerFoundError(ELBClientError): class NoActiveLoadBalancerFoundError(ELBClientError):
def __init__(self, name): def __init__(self, name: str):
super().__init__( super().__init__(
"LoadBalancerNotFound", f"There is no ACTIVE Load Balancer named '{name}'" "LoadBalancerNotFound", f"There is no ACTIVE Load Balancer named '{name}'"
) )
class PolicyNotFoundError(ELBClientError): class PolicyNotFoundError(ELBClientError):
def __init__(self): def __init__(self) -> None:
super().__init__( super().__init__(
"PolicyNotFound", "There is no policy with name . for load balancer ." "PolicyNotFound", "There is no policy with name . for load balancer ."
) )
class TooManyTagsError(ELBClientError): class TooManyTagsError(ELBClientError):
def __init__(self): def __init__(self) -> None:
super().__init__( super().__init__(
"LoadBalancerNotFound", "LoadBalancerNotFound",
"The quota for the number of tags that can be assigned to a load balancer has been reached", "The quota for the number of tags that can be assigned to a load balancer has been reached",
@ -53,7 +54,7 @@ class TooManyTagsError(ELBClientError):
class BadHealthCheckDefinition(ELBClientError): class BadHealthCheckDefinition(ELBClientError):
def __init__(self): def __init__(self) -> None:
super().__init__( super().__init__(
"ValidationError", "ValidationError",
"HealthCheck Target must begin with one of HTTP, TCP, HTTPS, SSL", "HealthCheck Target must begin with one of HTTP, TCP, HTTPS, SSL",
@ -61,7 +62,7 @@ class BadHealthCheckDefinition(ELBClientError):
class DuplicateListenerError(ELBClientError): class DuplicateListenerError(ELBClientError):
def __init__(self, name, port): def __init__(self, name: str, port: str):
super().__init__( super().__init__(
"DuplicateListener", "DuplicateListener",
f"A listener already exists for {name} with LoadBalancerPort {port}, but with a different InstancePort, Protocol, or SSLCertificateId", f"A listener already exists for {name} with LoadBalancerPort {port}, but with a different InstancePort, Protocol, or SSLCertificateId",
@ -69,7 +70,7 @@ class DuplicateListenerError(ELBClientError):
class DuplicateLoadBalancerName(ELBClientError): class DuplicateLoadBalancerName(ELBClientError):
def __init__(self, name): def __init__(self, name: str):
super().__init__( super().__init__(
"DuplicateLoadBalancerName", "DuplicateLoadBalancerName",
f"The specified load balancer name already exists for this account: {name}", f"The specified load balancer name already exists for this account: {name}",
@ -77,12 +78,12 @@ class DuplicateLoadBalancerName(ELBClientError):
class EmptyListenersError(ELBClientError): class EmptyListenersError(ELBClientError):
def __init__(self): def __init__(self) -> None:
super().__init__("ValidationError", "Listeners cannot be empty") super().__init__("ValidationError", "Listeners cannot be empty")
class InvalidSecurityGroupError(ELBClientError): class InvalidSecurityGroupError(ELBClientError):
def __init__(self): def __init__(self) -> None:
super().__init__( super().__init__(
"ValidationError", "ValidationError",
"One or more of the specified security groups do not exist.", "One or more of the specified security groups do not exist.",

View File

@ -1,6 +1,6 @@
import datetime import datetime
from collections import OrderedDict from collections import OrderedDict
from typing import List, Iterable from typing import Any, Dict, List, Iterable, Optional
from moto.core import BaseBackend, BackendDict, BaseModel, CloudFormationModel from moto.core import BaseBackend, BackendDict, BaseModel, CloudFormationModel
from moto.ec2.models import ec2_backends from moto.ec2.models import ec2_backends
@ -18,12 +18,22 @@ from .exceptions import (
TooManyTagsError, TooManyTagsError,
CertificateNotFoundException, CertificateNotFoundException,
) )
from .policies import AppCookieStickinessPolicy, LbCookieStickinessPolicy, OtherPolicy from .policies import (
AppCookieStickinessPolicy,
LbCookieStickinessPolicy,
OtherPolicy,
Policy,
)
class FakeHealthCheck(BaseModel): class FakeHealthCheck(BaseModel):
def __init__( def __init__(
self, timeout, healthy_threshold, unhealthy_threshold, interval, target self,
timeout: str,
healthy_threshold: str,
unhealthy_threshold: str,
interval: str,
target: str,
): ):
self.timeout = timeout self.timeout = timeout
self.healthy_threshold = healthy_threshold self.healthy_threshold = healthy_threshold
@ -35,53 +45,59 @@ class FakeHealthCheck(BaseModel):
class FakeListener(BaseModel): class FakeListener(BaseModel):
def __init__(self, load_balancer_port, instance_port, protocol, ssl_certificate_id): def __init__(
self,
load_balancer_port: str,
instance_port: str,
protocol: str,
ssl_certificate_id: Optional[str],
):
self.load_balancer_port = load_balancer_port self.load_balancer_port = load_balancer_port
self.instance_port = instance_port self.instance_port = instance_port
self.protocol = protocol.upper() self.protocol = protocol.upper()
self.ssl_certificate_id = ssl_certificate_id self.ssl_certificate_id = ssl_certificate_id
self.policy_names = [] self.policy_names: List[str] = []
def __repr__(self): def __repr__(self) -> str:
return f"FakeListener(lbp: {self.load_balancer_port}, inp: {self.instance_port}, pro: {self.protocol}, cid: {self.ssl_certificate_id}, policies: {self.policy_names})" return f"FakeListener(lbp: {self.load_balancer_port}, inp: {self.instance_port}, pro: {self.protocol}, cid: {self.ssl_certificate_id}, policies: {self.policy_names})"
class FakeBackend(BaseModel): class FakeBackend(BaseModel):
def __init__(self, instance_port): def __init__(self, instance_port: str):
self.instance_port = instance_port self.instance_port = instance_port
self.policy_names = [] self.policy_names: List[str] = []
def __repr__(self): def __repr__(self) -> str:
return f"FakeBackend(inp: {self.instance_port}, policies: {self.policy_names})" return f"FakeBackend(inp: {self.instance_port}, policies: {self.policy_names})"
class FakeLoadBalancer(CloudFormationModel): class FakeLoadBalancer(CloudFormationModel):
def __init__( def __init__(
self, self,
name, name: str,
zones, zones: List[str],
ports, ports: List[Dict[str, Any]],
scheme=None, scheme: Optional[str],
vpc_id=None, vpc_id: Optional[str],
subnets=None, subnets: Optional[List[str]],
security_groups=None, security_groups: Optional[List[str]],
elb_backend=None, elb_backend: "ELBBackend",
): ):
self.name = name self.name = name
self.health_check = None self.health_check: Optional[FakeHealthCheck] = None
self.instance_sparse_ids = [] self.instance_sparse_ids: List[str] = []
self.instance_autoscaling_ids = [] self.instance_autoscaling_ids: List[str] = []
self.zones = zones self.zones = zones
self.listeners = [] self.listeners = []
self.backends = [] self.backends = []
self.created_time = datetime.datetime.now(datetime.timezone.utc) self.created_time = datetime.datetime.now(datetime.timezone.utc)
self.scheme = scheme or "internet-facing" self.scheme = scheme or "internet-facing"
self.attributes = FakeLoadBalancer.get_default_attributes() self.attributes = FakeLoadBalancer.get_default_attributes()
self.policies = [] self.policies: List[Policy] = []
self.security_groups = security_groups or [] self.security_groups = security_groups or []
self.subnets = subnets or [] self.subnets = subnets or []
self.vpc_id = vpc_id self.vpc_id = vpc_id
self.tags = {} self.tags: Dict[str, str] = {}
self.dns_name = f"{name}.us-east-1.elb.amazonaws.com" self.dns_name = f"{name}.us-east-1.elb.amazonaws.com"
for port in ports: for port in ports:
@ -113,23 +129,28 @@ class FakeLoadBalancer(CloudFormationModel):
self.backends.append(backend) self.backends.append(backend)
@property @property
def instance_ids(self): def instance_ids(self) -> List[str]:
"""Return all the instances attached to the ELB""" """Return all the instances attached to the ELB"""
return self.instance_sparse_ids + self.instance_autoscaling_ids return self.instance_sparse_ids + self.instance_autoscaling_ids
@staticmethod @staticmethod
def cloudformation_name_type(): def cloudformation_name_type() -> str:
return "LoadBalancerName" return "LoadBalancerName"
@staticmethod @staticmethod
def cloudformation_type(): def cloudformation_type() -> str:
# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-elasticloadbalancing-loadbalancer.html # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-elasticloadbalancing-loadbalancer.html
return "AWS::ElasticLoadBalancing::LoadBalancer" return "AWS::ElasticLoadBalancing::LoadBalancer"
@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,
) -> "FakeLoadBalancer":
properties = cloudformation_json["Properties"] properties = cloudformation_json["Properties"]
elb_backend = elb_backends[account_id][region_name] elb_backend = elb_backends[account_id][region_name]
@ -145,14 +166,13 @@ class FakeLoadBalancer(CloudFormationModel):
elb_backend.register_instances(new_elb.name, [instance_id]) elb_backend.register_instances(new_elb.name, [instance_id])
policies = properties.get("Policies", []) policies = properties.get("Policies", [])
port_policies = {} port_policies: Dict[str, Any] = {}
for policy in policies: for policy in policies:
policy_name = policy["PolicyName"] policy_name = policy["PolicyName"]
other_policy = OtherPolicy() other_policy = OtherPolicy(policy_name, "", [])
other_policy.policy_name = policy_name
elb_backend.create_lb_other_policy(new_elb.name, other_policy) elb_backend.create_lb_other_policy(new_elb.name, other_policy)
for port in policy.get("InstancePorts", []): for port in policy.get("InstancePorts", []):
policies_for_port = port_policies.get(port, set()) policies_for_port: Any = port_policies.get(port, set())
policies_for_port.add(policy_name) policies_for_port.add(policy_name)
port_policies[port] = policies_for_port port_policies[port] = policies_for_port
@ -175,14 +195,14 @@ class FakeLoadBalancer(CloudFormationModel):
return new_elb return new_elb
@classmethod @classmethod
def update_from_cloudformation_json( def update_from_cloudformation_json( # type: ignore[misc]
cls, cls,
original_resource, original_resource: Any,
new_resource_name, new_resource_name: str,
cloudformation_json, cloudformation_json: Any,
account_id, account_id: str,
region_name, region_name: str,
): ) -> "FakeLoadBalancer":
cls.delete_from_cloudformation_json( cls.delete_from_cloudformation_json(
original_resource.name, cloudformation_json, account_id, region_name original_resource.name, cloudformation_json, account_id, region_name
) )
@ -191,9 +211,13 @@ class FakeLoadBalancer(CloudFormationModel):
) )
@classmethod @classmethod
def delete_from_cloudformation_json( def delete_from_cloudformation_json( # type: ignore[misc]
cls, resource_name, cloudformation_json, account_id, region_name cls,
): resource_name: str,
cloudformation_json: Any,
account_id: str,
region_name: str,
) -> None:
elb_backend = elb_backends[account_id][region_name] elb_backend = elb_backends[account_id][region_name]
try: try:
elb_backend.delete_load_balancer(resource_name) elb_backend.delete_load_balancer(resource_name)
@ -201,11 +225,11 @@ class FakeLoadBalancer(CloudFormationModel):
pass pass
@property @property
def physical_resource_id(self): def physical_resource_id(self) -> str:
return self.name return self.name
@classmethod @classmethod
def has_cfn_attr(cls, attr): def has_cfn_attr(cls, attr: str) -> bool:
return attr in [ return attr in [
"CanonicalHostedZoneName", "CanonicalHostedZoneName",
"CanonicalHostedZoneNameID", "CanonicalHostedZoneNameID",
@ -214,7 +238,7 @@ class FakeLoadBalancer(CloudFormationModel):
"SourceSecurityGroup.OwnerAlias", "SourceSecurityGroup.OwnerAlias",
] ]
def get_cfn_attribute(self, attribute_name): def get_cfn_attribute(self, attribute_name: str) -> Any:
from moto.cloudformation.exceptions import UnformattedGetAttTemplateException from moto.cloudformation.exceptions import UnformattedGetAttTemplateException
if attribute_name == "CanonicalHostedZoneName": if attribute_name == "CanonicalHostedZoneName":
@ -238,8 +262,8 @@ class FakeLoadBalancer(CloudFormationModel):
raise UnformattedGetAttTemplateException() raise UnformattedGetAttTemplateException()
@classmethod @classmethod
def get_default_attributes(cls): def get_default_attributes(cls) -> Dict[str, Any]: # type: ignore[misc]
attributes = dict() attributes: Dict[str, Any] = dict()
attributes["cross_zone_load_balancing"] = {"enabled": False} attributes["cross_zone_load_balancing"] = {"enabled": False}
attributes["connection_draining"] = {"enabled": False} attributes["connection_draining"] = {"enabled": False}
attributes["access_log"] = {"enabled": False} attributes["access_log"] = {"enabled": False}
@ -247,37 +271,37 @@ class FakeLoadBalancer(CloudFormationModel):
return attributes return attributes
def add_tag(self, key, value): def add_tag(self, key: str, value: str) -> None:
if len(self.tags) >= 10 and key not in self.tags: if len(self.tags) >= 10 and key not in self.tags:
raise TooManyTagsError() raise TooManyTagsError()
self.tags[key] = value self.tags[key] = value
def list_tags(self): def list_tags(self) -> Dict[str, str]:
return self.tags return self.tags
def remove_tag(self, key): def remove_tag(self, key: str) -> None:
if key in self.tags: if key in self.tags:
del self.tags[key] del self.tags[key]
def delete(self, account_id, region): def delete(self, account_id: str, region: str) -> None:
"""Not exposed as part of the ELB API - used for CloudFormation.""" """Not exposed as part of the ELB API - used for CloudFormation."""
elb_backends[account_id][region].delete_load_balancer(self.name) elb_backends[account_id][region].delete_load_balancer(self.name)
class ELBBackend(BaseBackend): class ELBBackend(BaseBackend):
def __init__(self, region_name, account_id): def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id) super().__init__(region_name, account_id)
self.load_balancers = OrderedDict() self.load_balancers: Dict[str, FakeLoadBalancer] = OrderedDict()
def create_load_balancer( def create_load_balancer(
self, self,
name, name: str,
zones, zones: List[str],
ports, ports: List[Dict[str, Any]],
scheme="internet-facing", scheme: str = "internet-facing",
subnets=None, subnets: Optional[List[str]] = None,
security_groups=None, security_groups: Optional[List[str]] = None,
): ) -> FakeLoadBalancer:
vpc_id = None vpc_id = None
ec2_backend = ec2_backends[self.account_id][self.region_name] ec2_backend = ec2_backends[self.account_id][self.region_name]
if subnets: if subnets:
@ -317,7 +341,9 @@ class ELBBackend(BaseBackend):
self.load_balancers[name] = new_load_balancer self.load_balancers[name] = new_load_balancer
return new_load_balancer return new_load_balancer
def create_load_balancer_listeners(self, name, ports): def create_load_balancer_listeners(
self, name: str, ports: List[Dict[str, Any]]
) -> Optional[FakeLoadBalancer]:
balancer = self.load_balancers.get(name, None) balancer = self.load_balancers.get(name, None)
if balancer: if balancer:
for port in ports: for port in ports:
@ -350,19 +376,21 @@ class ELBBackend(BaseBackend):
return balancer return balancer
def describe_load_balancers(self, names: List[str]) -> List[FakeLoadBalancer]: def describe_load_balancers(self, names: List[str]) -> List[FakeLoadBalancer]:
balancers = self.load_balancers.values() balancers = list(self.load_balancers.values())
if names: if names:
matched_balancers = [ matched_balancers = [
balancer for balancer in balancers if balancer.name in names balancer for balancer in balancers if balancer.name in names
] ]
if len(names) != len(matched_balancers): if len(names) != len(matched_balancers):
missing_elb = list(set(names) - set(matched_balancers))[0] missing_elb = list(set(names) - set(matched_balancers))[0] # type: ignore[arg-type]
raise LoadBalancerNotFoundError(missing_elb) raise LoadBalancerNotFoundError(missing_elb)
return matched_balancers return matched_balancers
else: else:
return balancers return balancers
def describe_load_balancer_policies(self, lb_name, policy_names): def describe_load_balancer_policies(
self, lb_name: str, policy_names: List[str]
) -> List[Policy]:
lb = self.describe_load_balancers([lb_name])[0] lb = self.describe_load_balancers([lb_name])[0]
policies = lb.policies policies = lb.policies
if policy_names: if policy_names:
@ -371,7 +399,9 @@ class ELBBackend(BaseBackend):
raise PolicyNotFoundError() raise PolicyNotFoundError()
return policies return policies
def describe_instance_health(self, lb_name, instances): def describe_instance_health(
self, lb_name: str, instances: List[Dict[str, str]]
) -> List[Dict[str, Any]]:
elb = self.get_load_balancer(lb_name) elb = self.get_load_balancer(lb_name)
if elb is None: if elb is None:
raise NoActiveLoadBalancerFoundError(name=lb_name) raise NoActiveLoadBalancerFoundError(name=lb_name)
@ -394,8 +424,10 @@ class ELBBackend(BaseBackend):
return instances return instances
def delete_load_balancer_listeners(self, name, ports): def delete_load_balancer_listeners(
balancer = self.load_balancers.get(name, None) self, name: str, ports: List[str]
) -> Optional[FakeLoadBalancer]:
balancer = self.get_load_balancer(name)
listeners = [] listeners = []
if balancer: if balancer:
for lb_port in ports: for lb_port in ports:
@ -407,20 +439,20 @@ class ELBBackend(BaseBackend):
balancer.listeners = listeners balancer.listeners = listeners
return balancer return balancer
def delete_load_balancer(self, load_balancer_name): def delete_load_balancer(self, load_balancer_name: str) -> None:
self.load_balancers.pop(load_balancer_name, None) self.load_balancers.pop(load_balancer_name, None)
def delete_load_balancer_policy(self, lb_name, policy_name): def delete_load_balancer_policy(self, lb_name: str, policy_name: str) -> None:
lb = self.get_load_balancer(lb_name) lb = self.get_load_balancer(lb_name)
lb.policies = [p for p in lb.policies if p.policy_name != policy_name] lb.policies = [p for p in lb.policies if p.policy_name != policy_name]
def get_load_balancer(self, load_balancer_name): def get_load_balancer(self, load_balancer_name: str) -> FakeLoadBalancer:
return self.load_balancers.get(load_balancer_name) return self.load_balancers.get(load_balancer_name) # type: ignore[return-value]
def apply_security_groups_to_load_balancer( def apply_security_groups_to_load_balancer(
self, load_balancer_name, security_group_ids self, load_balancer_name: str, security_group_ids: List[str]
): ) -> None:
load_balancer = self.load_balancers.get(load_balancer_name) load_balancer = self.get_load_balancer(load_balancer_name)
ec2_backend = ec2_backends[self.account_id][self.region_name] ec2_backend = ec2_backends[self.account_id][self.region_name]
for security_group_id in security_group_ids: for security_group_id in security_group_ids:
if ec2_backend.get_security_group_from_id(security_group_id) is None: if ec2_backend.get_security_group_from_id(security_group_id) is None:
@ -429,13 +461,13 @@ class ELBBackend(BaseBackend):
def configure_health_check( def configure_health_check(
self, self,
load_balancer_name, load_balancer_name: str,
timeout, timeout: str,
healthy_threshold, healthy_threshold: str,
unhealthy_threshold, unhealthy_threshold: str,
interval, interval: str,
target, target: str,
): ) -> FakeHealthCheck:
check = FakeHealthCheck( check = FakeHealthCheck(
timeout, healthy_threshold, unhealthy_threshold, interval, target timeout, healthy_threshold, unhealthy_threshold, interval, target
) )
@ -444,8 +476,8 @@ class ELBBackend(BaseBackend):
return check return check
def set_load_balancer_listener_ssl_certificate( def set_load_balancer_listener_ssl_certificate(
self, name, lb_port, ssl_certificate_id self, name: str, lb_port: str, ssl_certificate_id: str
): ) -> Optional[FakeLoadBalancer]:
balancer = self.load_balancers.get(name, None) balancer = self.load_balancers.get(name, None)
if balancer: if balancer:
for idx, listener in enumerate(balancer.listeners): for idx, listener in enumerate(balancer.listeners):
@ -500,12 +532,12 @@ class ELBBackend(BaseBackend):
def modify_load_balancer_attributes( def modify_load_balancer_attributes(
self, self,
load_balancer_name, load_balancer_name: str,
cross_zone=None, cross_zone: Optional[Dict[str, Any]] = None,
connection_settings=None, connection_settings: Optional[Dict[str, Any]] = None,
connection_draining=None, connection_draining: Optional[Dict[str, Any]] = None,
access_log=None, access_log: Optional[Dict[str, Any]] = None,
): ) -> None:
load_balancer = self.get_load_balancer(load_balancer_name) load_balancer = self.get_load_balancer(load_balancer_name)
if cross_zone: if cross_zone:
load_balancer.attributes["cross_zone_load_balancing"] = cross_zone load_balancer.attributes["cross_zone_load_balancing"] = cross_zone
@ -521,8 +553,12 @@ class ELBBackend(BaseBackend):
load_balancer.attributes["access_log"] = access_log load_balancer.attributes["access_log"] = access_log
def create_lb_other_policy( def create_lb_other_policy(
self, load_balancer_name, policy_name, policy_type_name, policy_attrs self,
): load_balancer_name: str,
policy_name: str,
policy_type_name: str,
policy_attrs: List[Dict[str, str]],
) -> FakeLoadBalancer:
load_balancer = self.get_load_balancer(load_balancer_name) load_balancer = self.get_load_balancer(load_balancer_name)
if policy_name not in [p.policy_name for p in load_balancer.policies]: if policy_name not in [p.policy_name for p in load_balancer.policies]:
load_balancer.policies.append( load_balancer.policies.append(
@ -532,24 +568,27 @@ class ELBBackend(BaseBackend):
return load_balancer return load_balancer
def create_app_cookie_stickiness_policy( def create_app_cookie_stickiness_policy(
self, load_balancer_name, policy_name, cookie_name self, load_balancer_name: str, policy_name: str, cookie_name: str
): ) -> FakeLoadBalancer:
load_balancer = self.get_load_balancer(load_balancer_name) load_balancer = self.get_load_balancer(load_balancer_name)
policy = AppCookieStickinessPolicy(policy_name, cookie_name) policy = AppCookieStickinessPolicy(policy_name, cookie_name)
load_balancer.policies.append(policy) load_balancer.policies.append(policy)
return load_balancer return load_balancer
def create_lb_cookie_stickiness_policy( def create_lb_cookie_stickiness_policy(
self, load_balancer_name, policy_name, cookie_expiration_period self,
): load_balancer_name: str,
policy_name: str,
cookie_expiration_period: Optional[int],
) -> FakeLoadBalancer:
load_balancer = self.get_load_balancer(load_balancer_name) load_balancer = self.get_load_balancer(load_balancer_name)
policy = LbCookieStickinessPolicy(policy_name, cookie_expiration_period) policy = LbCookieStickinessPolicy(policy_name, cookie_expiration_period)
load_balancer.policies.append(policy) load_balancer.policies.append(policy)
return load_balancer return load_balancer
def set_load_balancer_policies_of_backend_server( def set_load_balancer_policies_of_backend_server(
self, load_balancer_name, instance_port, policies self, load_balancer_name: str, instance_port: int, policies: List[str]
): ) -> FakeLoadBalancer:
load_balancer = self.get_load_balancer(load_balancer_name) load_balancer = self.get_load_balancer(load_balancer_name)
backend = [ backend = [
b for b in load_balancer.backends if int(b.instance_port) == instance_port b for b in load_balancer.backends if int(b.instance_port) == instance_port
@ -560,8 +599,8 @@ class ELBBackend(BaseBackend):
return load_balancer return load_balancer
def set_load_balancer_policies_of_listener( def set_load_balancer_policies_of_listener(
self, load_balancer_name, load_balancer_port, policies self, load_balancer_name: str, load_balancer_port: int, policies: List[str]
): ) -> FakeLoadBalancer:
load_balancer = self.get_load_balancer(load_balancer_name) load_balancer = self.get_load_balancer(load_balancer_name)
listener = [ listener = [
l_listener l_listener
@ -573,7 +612,7 @@ class ELBBackend(BaseBackend):
load_balancer.listeners[listener_idx] = listener load_balancer.listeners[listener_idx] = listener
return load_balancer return load_balancer
def _register_certificate(self, ssl_certificate_id, dns_name): def _register_certificate(self, ssl_certificate_id: str, dns_name: str) -> None:
from moto.acm.models import acm_backends, CertificateNotFound from moto.acm.models import acm_backends, CertificateNotFound
acm_backend = acm_backends[self.account_id][self.region_name] acm_backend = acm_backends[self.account_id][self.region_name]
@ -583,8 +622,8 @@ class ELBBackend(BaseBackend):
raise CertificateNotFoundException() raise CertificateNotFoundException()
def enable_availability_zones_for_load_balancer( def enable_availability_zones_for_load_balancer(
self, load_balancer_name, availability_zones self, load_balancer_name: str, availability_zones: List[str]
): ) -> List[str]:
load_balancer = self.get_load_balancer(load_balancer_name) load_balancer = self.get_load_balancer(load_balancer_name)
load_balancer.zones = sorted( load_balancer.zones = sorted(
list(set(load_balancer.zones + availability_zones)) list(set(load_balancer.zones + availability_zones))
@ -592,8 +631,8 @@ class ELBBackend(BaseBackend):
return load_balancer.zones return load_balancer.zones
def disable_availability_zones_for_load_balancer( def disable_availability_zones_for_load_balancer(
self, load_balancer_name, availability_zones self, load_balancer_name: str, availability_zones: List[str]
): ) -> List[str]:
load_balancer = self.get_load_balancer(load_balancer_name) load_balancer = self.get_load_balancer(load_balancer_name)
load_balancer.zones = sorted( load_balancer.zones = sorted(
list( list(
@ -602,12 +641,16 @@ class ELBBackend(BaseBackend):
) )
return load_balancer.zones return load_balancer.zones
def attach_load_balancer_to_subnets(self, load_balancer_name, subnets): def attach_load_balancer_to_subnets(
self, load_balancer_name: str, subnets: List[str]
) -> List[str]:
load_balancer = self.get_load_balancer(load_balancer_name) load_balancer = self.get_load_balancer(load_balancer_name)
load_balancer.subnets = list(set(load_balancer.subnets + subnets)) load_balancer.subnets = list(set(load_balancer.subnets + subnets))
return load_balancer.subnets return load_balancer.subnets
def detach_load_balancer_from_subnets(self, load_balancer_name, subnets): def detach_load_balancer_from_subnets(
self, load_balancer_name: str, subnets: List[str]
) -> List[str]:
load_balancer = self.get_load_balancer(load_balancer_name) load_balancer = self.get_load_balancer(load_balancer_name)
load_balancer.subnets = [s for s in load_balancer.subnets if s not in subnets] load_balancer.subnets = [s for s in load_balancer.subnets if s not in subnets]
return load_balancer.subnets return load_balancer.subnets

View File

@ -1,24 +1,30 @@
from typing import Any, Dict, List, Optional
class Policy(object): class Policy(object):
def __init__(self, policy_type_name): def __init__(self, policy_name: str, policy_type_name: str):
self.policy_name = policy_name
self.policy_type_name = policy_type_name self.policy_type_name = policy_type_name
class AppCookieStickinessPolicy(Policy): class AppCookieStickinessPolicy(Policy):
def __init__(self, policy_name, cookie_name): def __init__(self, policy_name: str, cookie_name: str):
super().__init__(policy_type_name="AppCookieStickinessPolicy") super().__init__(policy_name, policy_type_name="AppCookieStickinessPolicy")
self.policy_name = policy_name
self.cookie_name = cookie_name self.cookie_name = cookie_name
class LbCookieStickinessPolicy(Policy): class LbCookieStickinessPolicy(Policy):
def __init__(self, policy_name, cookie_expiration_period): def __init__(self, policy_name: str, cookie_expiration_period: Optional[int]):
super().__init__(policy_type_name="LbCookieStickinessPolicy") super().__init__(policy_name, policy_type_name="LbCookieStickinessPolicy")
self.policy_name = policy_name
self.cookie_expiration_period = cookie_expiration_period self.cookie_expiration_period = cookie_expiration_period
class OtherPolicy(Policy): class OtherPolicy(Policy):
def __init__(self, policy_name, policy_type_name, policy_attrs): def __init__(
super().__init__(policy_type_name=policy_type_name) self,
self.policy_name = policy_name policy_name: str,
policy_type_name: str,
policy_attrs: List[Dict[str, Any]],
):
super().__init__(policy_name, policy_type_name=policy_type_name)
self.attributes = policy_attrs or [] self.attributes = policy_attrs or []

View File

@ -1,17 +1,17 @@
from moto.core.responses import BaseResponse from moto.core.responses import BaseResponse
from .models import elb_backends from .models import elb_backends, ELBBackend, FakeLoadBalancer
from .exceptions import DuplicateTagKeysError, LoadBalancerNotFoundError from .exceptions import DuplicateTagKeysError, LoadBalancerNotFoundError
class ELBResponse(BaseResponse): class ELBResponse(BaseResponse):
def __init__(self): def __init__(self) -> None:
super().__init__(service_name="elb") super().__init__(service_name="elb")
@property @property
def elb_backend(self): def elb_backend(self) -> ELBBackend:
return elb_backends[self.current_account][self.region] return elb_backends[self.current_account][self.region]
def create_load_balancer(self): def create_load_balancer(self) -> str:
load_balancer_name = self._get_param("LoadBalancerName") load_balancer_name = self._get_param("LoadBalancerName")
availability_zones = self._get_multi_param("AvailabilityZones.member") availability_zones = self._get_multi_param("AvailabilityZones.member")
ports = self._get_list_prefix("Listeners.member") ports = self._get_list_prefix("Listeners.member")
@ -31,7 +31,7 @@ class ELBResponse(BaseResponse):
template = self.response_template(CREATE_LOAD_BALANCER_TEMPLATE) template = self.response_template(CREATE_LOAD_BALANCER_TEMPLATE)
return template.render(load_balancer=load_balancer) return template.render(load_balancer=load_balancer)
def create_load_balancer_listeners(self): def create_load_balancer_listeners(self) -> str:
load_balancer_name = self._get_param("LoadBalancerName") load_balancer_name = self._get_param("LoadBalancerName")
ports = self._get_list_prefix("Listeners.member") ports = self._get_list_prefix("Listeners.member")
@ -42,7 +42,7 @@ class ELBResponse(BaseResponse):
template = self.response_template(CREATE_LOAD_BALANCER_LISTENERS_TEMPLATE) template = self.response_template(CREATE_LOAD_BALANCER_LISTENERS_TEMPLATE)
return template.render() return template.render()
def describe_load_balancers(self): def describe_load_balancers(self) -> str:
names = self._get_multi_param("LoadBalancerNames.member") names = self._get_multi_param("LoadBalancerNames.member")
all_load_balancers = list(self.elb_backend.describe_load_balancers(names)) all_load_balancers = list(self.elb_backend.describe_load_balancers(names))
marker = self._get_param("Marker") marker = self._get_param("Marker")
@ -66,7 +66,7 @@ class ELBResponse(BaseResponse):
marker=next_marker, marker=next_marker,
) )
def delete_load_balancer_listeners(self): def delete_load_balancer_listeners(self) -> str:
load_balancer_name = self._get_param("LoadBalancerName") load_balancer_name = self._get_param("LoadBalancerName")
ports = self._get_multi_param("LoadBalancerPorts.member") ports = self._get_multi_param("LoadBalancerPorts.member")
ports = [int(port) for port in ports] ports = [int(port) for port in ports]
@ -75,14 +75,14 @@ class ELBResponse(BaseResponse):
template = self.response_template(DELETE_LOAD_BALANCER_LISTENERS) template = self.response_template(DELETE_LOAD_BALANCER_LISTENERS)
return template.render() return template.render()
def delete_load_balancer(self): def delete_load_balancer(self) -> str:
load_balancer_name = self._get_param("LoadBalancerName") load_balancer_name = self._get_param("LoadBalancerName")
self.elb_backend.delete_load_balancer(load_balancer_name) self.elb_backend.delete_load_balancer(load_balancer_name)
template = self.response_template(DELETE_LOAD_BALANCER_TEMPLATE) template = self.response_template(DELETE_LOAD_BALANCER_TEMPLATE)
return template.render() return template.render()
def delete_load_balancer_policy(self): def delete_load_balancer_policy(self) -> str:
load_balancer_name = self.querystring.get("LoadBalancerName")[0] load_balancer_name = self.querystring.get("LoadBalancerName")[0] # type: ignore
names = self._get_param("PolicyName") names = self._get_param("PolicyName")
self.elb_backend.delete_load_balancer_policy( self.elb_backend.delete_load_balancer_policy(
lb_name=load_balancer_name, policy_name=names lb_name=load_balancer_name, policy_name=names
@ -91,7 +91,7 @@ class ELBResponse(BaseResponse):
template = self.response_template(DELETE_LOAD_BALANCER_POLICY) template = self.response_template(DELETE_LOAD_BALANCER_POLICY)
return template.render() return template.render()
def apply_security_groups_to_load_balancer(self): def apply_security_groups_to_load_balancer(self) -> str:
load_balancer_name = self._get_param("LoadBalancerName") load_balancer_name = self._get_param("LoadBalancerName")
security_group_ids = self._get_multi_param("SecurityGroups.member") security_group_ids = self._get_multi_param("SecurityGroups.member")
self.elb_backend.apply_security_groups_to_load_balancer( self.elb_backend.apply_security_groups_to_load_balancer(
@ -100,7 +100,7 @@ class ELBResponse(BaseResponse):
template = self.response_template(APPLY_SECURITY_GROUPS_TEMPLATE) template = self.response_template(APPLY_SECURITY_GROUPS_TEMPLATE)
return template.render(security_group_ids=security_group_ids) return template.render(security_group_ids=security_group_ids)
def configure_health_check(self): def configure_health_check(self) -> str:
check = self.elb_backend.configure_health_check( check = self.elb_backend.configure_health_check(
load_balancer_name=self._get_param("LoadBalancerName"), load_balancer_name=self._get_param("LoadBalancerName"),
timeout=self._get_param("HealthCheck.Timeout"), timeout=self._get_param("HealthCheck.Timeout"),
@ -112,7 +112,7 @@ class ELBResponse(BaseResponse):
template = self.response_template(CONFIGURE_HEALTH_CHECK_TEMPLATE) template = self.response_template(CONFIGURE_HEALTH_CHECK_TEMPLATE)
return template.render(check=check) return template.render(check=check)
def register_instances_with_load_balancer(self): def register_instances_with_load_balancer(self) -> str:
load_balancer_name = self._get_param("LoadBalancerName") load_balancer_name = self._get_param("LoadBalancerName")
instance_ids = [ instance_ids = [
list(param.values())[0] list(param.values())[0]
@ -124,7 +124,7 @@ class ELBResponse(BaseResponse):
) )
return template.render(load_balancer=load_balancer) return template.render(load_balancer=load_balancer)
def set_load_balancer_listener_ssl_certificate(self): def set_load_balancer_listener_ssl_certificate(self) -> str:
load_balancer_name = self._get_param("LoadBalancerName") load_balancer_name = self._get_param("LoadBalancerName")
ssl_certificate_id = self.querystring["SSLCertificateId"][0] ssl_certificate_id = self.querystring["SSLCertificateId"][0]
lb_port = self.querystring["LoadBalancerPort"][0] lb_port = self.querystring["LoadBalancerPort"][0]
@ -136,7 +136,7 @@ class ELBResponse(BaseResponse):
template = self.response_template(SET_LOAD_BALANCER_SSL_CERTIFICATE) template = self.response_template(SET_LOAD_BALANCER_SSL_CERTIFICATE)
return template.render() return template.render()
def deregister_instances_from_load_balancer(self): def deregister_instances_from_load_balancer(self) -> str:
load_balancer_name = self._get_param("LoadBalancerName") load_balancer_name = self._get_param("LoadBalancerName")
instance_ids = [ instance_ids = [
list(param.values())[0] list(param.values())[0]
@ -148,13 +148,13 @@ class ELBResponse(BaseResponse):
) )
return template.render(load_balancer=load_balancer) return template.render(load_balancer=load_balancer)
def describe_load_balancer_attributes(self): def describe_load_balancer_attributes(self) -> str:
load_balancer_name = self._get_param("LoadBalancerName") load_balancer_name = self._get_param("LoadBalancerName")
load_balancer = self.elb_backend.get_load_balancer(load_balancer_name) load_balancer = self.elb_backend.get_load_balancer(load_balancer_name)
template = self.response_template(DESCRIBE_ATTRIBUTES_TEMPLATE) template = self.response_template(DESCRIBE_ATTRIBUTES_TEMPLATE)
return template.render(attributes=load_balancer.attributes) return template.render(attributes=load_balancer.attributes)
def modify_load_balancer_attributes(self): def modify_load_balancer_attributes(self) -> str:
load_balancer_name = self._get_param("LoadBalancerName") load_balancer_name = self._get_param("LoadBalancerName")
load_balancer = self.elb_backend.get_load_balancer(load_balancer_name) load_balancer = self.elb_backend.get_load_balancer(load_balancer_name)
@ -193,7 +193,7 @@ class ELBResponse(BaseResponse):
load_balancer=load_balancer, attributes=load_balancer.attributes load_balancer=load_balancer, attributes=load_balancer.attributes
) )
def create_load_balancer_policy(self): def create_load_balancer_policy(self) -> str:
load_balancer_name = self._get_param("LoadBalancerName") load_balancer_name = self._get_param("LoadBalancerName")
policy_name = self._get_param("PolicyName") policy_name = self._get_param("PolicyName")
@ -207,7 +207,7 @@ class ELBResponse(BaseResponse):
template = self.response_template(CREATE_LOAD_BALANCER_POLICY_TEMPLATE) template = self.response_template(CREATE_LOAD_BALANCER_POLICY_TEMPLATE)
return template.render() return template.render()
def create_app_cookie_stickiness_policy(self): def create_app_cookie_stickiness_policy(self) -> str:
load_balancer_name = self._get_param("LoadBalancerName") load_balancer_name = self._get_param("LoadBalancerName")
policy_name = self._get_param("PolicyName") policy_name = self._get_param("PolicyName")
@ -220,7 +220,7 @@ class ELBResponse(BaseResponse):
template = self.response_template(CREATE_APP_COOKIE_STICKINESS_POLICY_TEMPLATE) template = self.response_template(CREATE_APP_COOKIE_STICKINESS_POLICY_TEMPLATE)
return template.render() return template.render()
def create_lb_cookie_stickiness_policy(self): def create_lb_cookie_stickiness_policy(self) -> str:
load_balancer_name = self._get_param("LoadBalancerName") load_balancer_name = self._get_param("LoadBalancerName")
policy_name = self._get_param("PolicyName") policy_name = self._get_param("PolicyName")
@ -237,7 +237,7 @@ class ELBResponse(BaseResponse):
template = self.response_template(CREATE_LB_COOKIE_STICKINESS_POLICY_TEMPLATE) template = self.response_template(CREATE_LB_COOKIE_STICKINESS_POLICY_TEMPLATE)
return template.render() return template.render()
def set_load_balancer_policies_of_listener(self): def set_load_balancer_policies_of_listener(self) -> str:
load_balancer_name = self._get_param("LoadBalancerName") load_balancer_name = self._get_param("LoadBalancerName")
load_balancer = self.elb_backend.get_load_balancer(load_balancer_name) load_balancer = self.elb_backend.get_load_balancer(load_balancer_name)
load_balancer_port = int(self._get_param("LoadBalancerPort")) load_balancer_port = int(self._get_param("LoadBalancerPort"))
@ -259,10 +259,10 @@ class ELBResponse(BaseResponse):
) )
return template.render() return template.render()
def set_load_balancer_policies_for_backend_server(self): def set_load_balancer_policies_for_backend_server(self) -> str:
load_balancer_name = self.querystring.get("LoadBalancerName")[0] load_balancer_name = self.querystring.get("LoadBalancerName")[0] # type: ignore
load_balancer = self.elb_backend.get_load_balancer(load_balancer_name) load_balancer = self.elb_backend.get_load_balancer(load_balancer_name)
instance_port = int(self.querystring.get("InstancePort")[0]) instance_port = int(self.querystring.get("InstancePort")[0]) # type: ignore
mb_backend = [ mb_backend = [
b for b in load_balancer.backends if int(b.instance_port) == instance_port b for b in load_balancer.backends if int(b.instance_port) == instance_port
@ -270,7 +270,7 @@ class ELBResponse(BaseResponse):
if mb_backend: if mb_backend:
policies = self._get_multi_param("PolicyNames.member") policies = self._get_multi_param("PolicyNames.member")
self.elb_backend.set_load_balancer_policies_of_backend_server( self.elb_backend.set_load_balancer_policies_of_backend_server(
load_balancer_name, instance_port, policies load_balancer_name, instance_port, policies # type: ignore[arg-type]
) )
# else: explode? # else: explode?
@ -279,8 +279,8 @@ class ELBResponse(BaseResponse):
) )
return template.render() return template.render()
def describe_load_balancer_policies(self): def describe_load_balancer_policies(self) -> str:
load_balancer_name = self.querystring.get("LoadBalancerName")[0] load_balancer_name = self.querystring.get("LoadBalancerName")[0] # type: ignore
names = self._get_multi_param("PolicyNames.member.") names = self._get_multi_param("PolicyNames.member.")
policies = self.elb_backend.describe_load_balancer_policies( policies = self.elb_backend.describe_load_balancer_policies(
lb_name=load_balancer_name, policy_names=names lb_name=load_balancer_name, policy_names=names
@ -289,14 +289,14 @@ class ELBResponse(BaseResponse):
template = self.response_template(DESCRIBE_LOAD_BALANCER_POLICIES_TEMPLATE) template = self.response_template(DESCRIBE_LOAD_BALANCER_POLICIES_TEMPLATE)
return template.render(policies=policies) return template.render(policies=policies)
def describe_instance_health(self): def describe_instance_health(self) -> str:
lb_name = self._get_param("LoadBalancerName") lb_name = self._get_param("LoadBalancerName")
instances = self._get_params().get("Instances", []) instances = self._get_params().get("Instances", [])
instances = self.elb_backend.describe_instance_health(lb_name, instances) instances = self.elb_backend.describe_instance_health(lb_name, instances)
template = self.response_template(DESCRIBE_INSTANCE_HEALTH_TEMPLATE) template = self.response_template(DESCRIBE_INSTANCE_HEALTH_TEMPLATE)
return template.render(instances=instances) return template.render(instances=instances)
def add_tags(self): def add_tags(self) -> str:
for key, value in self.querystring.items(): for key, value in self.querystring.items():
if "LoadBalancerNames.member" in key: if "LoadBalancerNames.member" in key:
@ -310,7 +310,7 @@ class ELBResponse(BaseResponse):
template = self.response_template(ADD_TAGS_TEMPLATE) template = self.response_template(ADD_TAGS_TEMPLATE)
return template.render() return template.render()
def remove_tags(self): def remove_tags(self) -> str:
for key in self.querystring: for key in self.querystring:
if "LoadBalancerNames.member" in key: if "LoadBalancerNames.member" in key:
number = key.split(".")[2] number = key.split(".")[2]
@ -329,7 +329,7 @@ class ELBResponse(BaseResponse):
template = self.response_template(REMOVE_TAGS_TEMPLATE) template = self.response_template(REMOVE_TAGS_TEMPLATE)
return template.render() return template.render()
def describe_tags(self): def describe_tags(self) -> str:
elbs = [] elbs = []
for key in self.querystring: for key in self.querystring:
if "LoadBalancerNames.member" in key: if "LoadBalancerNames.member" in key:
@ -345,7 +345,7 @@ class ELBResponse(BaseResponse):
template = self.response_template(DESCRIBE_TAGS_TEMPLATE) template = self.response_template(DESCRIBE_TAGS_TEMPLATE)
return template.render(load_balancers=elbs) return template.render(load_balancers=elbs)
def _add_tags(self, elb): def _add_tags(self, elb: FakeLoadBalancer) -> None:
tag_values = [] tag_values = []
tag_keys = [] tag_keys = []
@ -356,11 +356,11 @@ class ELBResponse(BaseResponse):
elif t_key.split(".")[3] == "Value": elif t_key.split(".")[3] == "Value":
tag_values.extend(t_val) tag_values.extend(t_val)
counts = {} count_dict = {}
for i in tag_keys: for i in tag_keys:
counts[i] = tag_keys.count(i) count_dict[i] = tag_keys.count(i)
counts = sorted(counts.items(), key=lambda i: i[1], reverse=True) counts = sorted(count_dict.items(), key=lambda i: i[1], reverse=True)
if counts and counts[0][1] > 1: if counts and counts[0][1] > 1:
# We have dupes... # We have dupes...
@ -369,14 +369,14 @@ class ELBResponse(BaseResponse):
for tag_key, tag_value in zip(tag_keys, tag_values): for tag_key, tag_value in zip(tag_keys, tag_values):
elb.add_tag(tag_key, tag_value) elb.add_tag(tag_key, tag_value)
def enable_availability_zones_for_load_balancer(self): def enable_availability_zones_for_load_balancer(self) -> str:
params = self._get_params() params = self._get_params()
load_balancer_name = params.get("LoadBalancerName") load_balancer_name = params.get("LoadBalancerName")
availability_zones = params.get("AvailabilityZones") availability_zones = params.get("AvailabilityZones")
availability_zones = ( availability_zones = (
self.elb_backend.enable_availability_zones_for_load_balancer( self.elb_backend.enable_availability_zones_for_load_balancer(
load_balancer_name=load_balancer_name, load_balancer_name=load_balancer_name, # type: ignore[arg-type]
availability_zones=availability_zones, availability_zones=availability_zones, # type: ignore[arg-type]
) )
) )
template = self.response_template( template = self.response_template(
@ -384,14 +384,14 @@ class ELBResponse(BaseResponse):
) )
return template.render(availability_zones=availability_zones) return template.render(availability_zones=availability_zones)
def disable_availability_zones_for_load_balancer(self): def disable_availability_zones_for_load_balancer(self) -> str:
params = self._get_params() params = self._get_params()
load_balancer_name = params.get("LoadBalancerName") load_balancer_name = params.get("LoadBalancerName")
availability_zones = params.get("AvailabilityZones") availability_zones = params.get("AvailabilityZones")
availability_zones = ( availability_zones = (
self.elb_backend.disable_availability_zones_for_load_balancer( self.elb_backend.disable_availability_zones_for_load_balancer(
load_balancer_name=load_balancer_name, load_balancer_name=load_balancer_name, # type: ignore[arg-type]
availability_zones=availability_zones, availability_zones=availability_zones, # type: ignore[arg-type]
) )
) )
template = self.response_template( template = self.response_template(
@ -399,24 +399,24 @@ class ELBResponse(BaseResponse):
) )
return template.render(availability_zones=availability_zones) return template.render(availability_zones=availability_zones)
def attach_load_balancer_to_subnets(self): def attach_load_balancer_to_subnets(self) -> str:
params = self._get_params() params = self._get_params()
load_balancer_name = params.get("LoadBalancerName") load_balancer_name = params.get("LoadBalancerName")
subnets = params.get("Subnets") subnets = params.get("Subnets")
all_subnets = self.elb_backend.attach_load_balancer_to_subnets( all_subnets = self.elb_backend.attach_load_balancer_to_subnets(
load_balancer_name, subnets load_balancer_name, subnets # type: ignore[arg-type]
) )
template = self.response_template(ATTACH_LB_TO_SUBNETS_TEMPLATE) template = self.response_template(ATTACH_LB_TO_SUBNETS_TEMPLATE)
return template.render(subnets=all_subnets) return template.render(subnets=all_subnets)
def detach_load_balancer_from_subnets(self): def detach_load_balancer_from_subnets(self) -> str:
params = self._get_params() params = self._get_params()
load_balancer_name = params.get("LoadBalancerName") load_balancer_name = params.get("LoadBalancerName")
subnets = params.get("Subnets") subnets = params.get("Subnets")
all_subnets = self.elb_backend.detach_load_balancer_from_subnets( all_subnets = self.elb_backend.detach_load_balancer_from_subnets(
load_balancer_name, subnets load_balancer_name, subnets # type: ignore[arg-type]
) )
template = self.response_template(DETACH_LB_FROM_SUBNETS_TEMPLATE) template = self.response_template(DETACH_LB_FROM_SUBNETS_TEMPLATE)
return template.render(subnets=all_subnets) return template.render(subnets=all_subnets)

View File

@ -1,3 +1,4 @@
from typing import Any
from urllib.parse import parse_qs from urllib.parse import parse_qs
from botocore.awsrequest import AWSPreparedRequest from botocore.awsrequest import AWSPreparedRequest
@ -5,7 +6,7 @@ from moto.elb.responses import ELBResponse
from moto.elbv2.responses import ELBV2Response from moto.elbv2.responses import ELBV2Response
def api_version_elb_backend(*args, **kwargs): def api_version_elb_backend(*args: Any, **kwargs: Any) -> Any:
""" """
ELB and ELBV2 (Classic and Application load balancers) use the same ELB and ELBV2 (Classic and Application load balancers) use the same
hostname and url space. To differentiate them we must read the hostname and url space. To differentiate them we must read the
@ -20,7 +21,7 @@ def api_version_elb_backend(*args, **kwargs):
version = request.values.get("Version") version = request.values.get("Version")
elif isinstance(request, AWSPreparedRequest): elif isinstance(request, AWSPreparedRequest):
# boto in-memory # boto in-memory
version = parse_qs(request.body).get("Version")[0] version = parse_qs(request.body).get("Version")[0] # type: ignore
else: else:
# boto in server mode # boto in server mode
request.parse_request() request.parse_request()

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,moto/ec2instanceconnect,moto/ecr,moto/ecs,moto/efs,moto/eks,moto/elasticache,moto/elasticbeanstalk,moto/elastictranscoder,moto/es,moto/moto_api,moto/neptune files= moto/a*,moto/b*,moto/c*,moto/d*,moto/ebs/,moto/ec2,moto/ec2instanceconnect,moto/ecr,moto/ecs,moto/efs,moto/eks,moto/elasticache,moto/elasticbeanstalk,moto/elastictranscoder,moto/elb,moto/es,moto/moto_api,moto/neptune
show_column_numbers=True show_column_numbers=True
show_error_codes = True show_error_codes = True
disable_error_code=abstract disable_error_code=abstract