From 8eaa9fa3700345ceb16084178c78247fc1512dfc Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Mon, 27 Feb 2023 22:08:24 -0100 Subject: [PATCH] Techdebt: MyPy ELB (#5996) --- moto/elb/exceptions.py | 25 ++-- moto/elb/models.py | 255 ++++++++++++++++++++++++----------------- moto/elb/policies.py | 26 +++-- moto/elb/responses.py | 90 +++++++-------- moto/elb/urls.py | 5 +- setup.cfg | 2 +- 6 files changed, 227 insertions(+), 176 deletions(-) diff --git a/moto/elb/exceptions.py b/moto/elb/exceptions.py index bc49b9648..63e806869 100644 --- a/moto/elb/exceptions.py +++ b/moto/elb/exceptions.py @@ -1,29 +1,30 @@ +from typing import Any from moto.core.exceptions import RESTError class ELBClientError(RESTError): 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") class DuplicateTagKeysError(ELBClientError): - def __init__(self, cidr): + def __init__(self, cidr: Any): super().__init__( "DuplicateTagKeys", f"Tag key was specified more than once: {cidr}" ) class CertificateNotFoundException(ELBClientError): - def __init__(self): + def __init__(self) -> None: super().__init__( "CertificateNotFoundException", "Supplied certificate was not found" ) class LoadBalancerNotFoundError(ELBClientError): - def __init__(self, name): + def __init__(self, name: str): super().__init__( "LoadBalancerNotFound", f"The specified load balancer does not exist: {name}", @@ -31,21 +32,21 @@ class LoadBalancerNotFoundError(ELBClientError): class NoActiveLoadBalancerFoundError(ELBClientError): - def __init__(self, name): + def __init__(self, name: str): super().__init__( "LoadBalancerNotFound", f"There is no ACTIVE Load Balancer named '{name}'" ) class PolicyNotFoundError(ELBClientError): - def __init__(self): + def __init__(self) -> None: super().__init__( "PolicyNotFound", "There is no policy with name . for load balancer ." ) class TooManyTagsError(ELBClientError): - def __init__(self): + def __init__(self) -> None: super().__init__( "LoadBalancerNotFound", "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): - def __init__(self): + def __init__(self) -> None: super().__init__( "ValidationError", "HealthCheck Target must begin with one of HTTP, TCP, HTTPS, SSL", @@ -61,7 +62,7 @@ class BadHealthCheckDefinition(ELBClientError): class DuplicateListenerError(ELBClientError): - def __init__(self, name, port): + def __init__(self, name: str, port: str): super().__init__( "DuplicateListener", 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): - def __init__(self, name): + def __init__(self, name: str): super().__init__( "DuplicateLoadBalancerName", f"The specified load balancer name already exists for this account: {name}", @@ -77,12 +78,12 @@ class DuplicateLoadBalancerName(ELBClientError): class EmptyListenersError(ELBClientError): - def __init__(self): + def __init__(self) -> None: super().__init__("ValidationError", "Listeners cannot be empty") class InvalidSecurityGroupError(ELBClientError): - def __init__(self): + def __init__(self) -> None: super().__init__( "ValidationError", "One or more of the specified security groups do not exist.", diff --git a/moto/elb/models.py b/moto/elb/models.py index a085e3db8..e70faeafc 100644 --- a/moto/elb/models.py +++ b/moto/elb/models.py @@ -1,6 +1,6 @@ import datetime 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.ec2.models import ec2_backends @@ -18,12 +18,22 @@ from .exceptions import ( TooManyTagsError, CertificateNotFoundException, ) -from .policies import AppCookieStickinessPolicy, LbCookieStickinessPolicy, OtherPolicy +from .policies import ( + AppCookieStickinessPolicy, + LbCookieStickinessPolicy, + OtherPolicy, + Policy, +) class FakeHealthCheck(BaseModel): 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.healthy_threshold = healthy_threshold @@ -35,53 +45,59 @@ class FakeHealthCheck(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.instance_port = instance_port self.protocol = protocol.upper() 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})" class FakeBackend(BaseModel): - def __init__(self, instance_port): + def __init__(self, instance_port: str): 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})" class FakeLoadBalancer(CloudFormationModel): def __init__( self, - name, - zones, - ports, - scheme=None, - vpc_id=None, - subnets=None, - security_groups=None, - elb_backend=None, + name: str, + zones: List[str], + ports: List[Dict[str, Any]], + scheme: Optional[str], + vpc_id: Optional[str], + subnets: Optional[List[str]], + security_groups: Optional[List[str]], + elb_backend: "ELBBackend", ): self.name = name - self.health_check = None - self.instance_sparse_ids = [] - self.instance_autoscaling_ids = [] + self.health_check: Optional[FakeHealthCheck] = None + self.instance_sparse_ids: List[str] = [] + self.instance_autoscaling_ids: List[str] = [] self.zones = zones self.listeners = [] self.backends = [] self.created_time = datetime.datetime.now(datetime.timezone.utc) self.scheme = scheme or "internet-facing" self.attributes = FakeLoadBalancer.get_default_attributes() - self.policies = [] + self.policies: List[Policy] = [] self.security_groups = security_groups or [] self.subnets = subnets or [] self.vpc_id = vpc_id - self.tags = {} + self.tags: Dict[str, str] = {} self.dns_name = f"{name}.us-east-1.elb.amazonaws.com" for port in ports: @@ -113,23 +129,28 @@ class FakeLoadBalancer(CloudFormationModel): self.backends.append(backend) @property - def instance_ids(self): + def instance_ids(self) -> List[str]: """Return all the instances attached to the ELB""" return self.instance_sparse_ids + self.instance_autoscaling_ids @staticmethod - def cloudformation_name_type(): + def cloudformation_name_type() -> str: return "LoadBalancerName" @staticmethod - def cloudformation_type(): + def cloudformation_type() -> str: # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-elasticloadbalancing-loadbalancer.html return "AWS::ElasticLoadBalancing::LoadBalancer" @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, + ) -> "FakeLoadBalancer": properties = cloudformation_json["Properties"] elb_backend = elb_backends[account_id][region_name] @@ -145,14 +166,13 @@ class FakeLoadBalancer(CloudFormationModel): elb_backend.register_instances(new_elb.name, [instance_id]) policies = properties.get("Policies", []) - port_policies = {} + port_policies: Dict[str, Any] = {} for policy in policies: policy_name = policy["PolicyName"] - other_policy = OtherPolicy() - other_policy.policy_name = policy_name + other_policy = OtherPolicy(policy_name, "", []) elb_backend.create_lb_other_policy(new_elb.name, other_policy) 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) port_policies[port] = policies_for_port @@ -175,14 +195,14 @@ class FakeLoadBalancer(CloudFormationModel): return new_elb @classmethod - def update_from_cloudformation_json( + def update_from_cloudformation_json( # type: ignore[misc] cls, - original_resource, - new_resource_name, - cloudformation_json, - account_id, - region_name, - ): + original_resource: Any, + new_resource_name: str, + cloudformation_json: Any, + account_id: str, + region_name: str, + ) -> "FakeLoadBalancer": cls.delete_from_cloudformation_json( original_resource.name, cloudformation_json, account_id, region_name ) @@ -191,9 +211,13 @@ class FakeLoadBalancer(CloudFormationModel): ) @classmethod - def delete_from_cloudformation_json( - cls, resource_name, cloudformation_json, account_id, region_name - ): + def delete_from_cloudformation_json( # type: ignore[misc] + cls, + resource_name: str, + cloudformation_json: Any, + account_id: str, + region_name: str, + ) -> None: elb_backend = elb_backends[account_id][region_name] try: elb_backend.delete_load_balancer(resource_name) @@ -201,11 +225,11 @@ class FakeLoadBalancer(CloudFormationModel): pass @property - def physical_resource_id(self): + def physical_resource_id(self) -> str: return self.name @classmethod - def has_cfn_attr(cls, attr): + def has_cfn_attr(cls, attr: str) -> bool: return attr in [ "CanonicalHostedZoneName", "CanonicalHostedZoneNameID", @@ -214,7 +238,7 @@ class FakeLoadBalancer(CloudFormationModel): "SourceSecurityGroup.OwnerAlias", ] - def get_cfn_attribute(self, attribute_name): + def get_cfn_attribute(self, attribute_name: str) -> Any: from moto.cloudformation.exceptions import UnformattedGetAttTemplateException if attribute_name == "CanonicalHostedZoneName": @@ -238,8 +262,8 @@ class FakeLoadBalancer(CloudFormationModel): raise UnformattedGetAttTemplateException() @classmethod - def get_default_attributes(cls): - attributes = dict() + def get_default_attributes(cls) -> Dict[str, Any]: # type: ignore[misc] + attributes: Dict[str, Any] = dict() attributes["cross_zone_load_balancing"] = {"enabled": False} attributes["connection_draining"] = {"enabled": False} attributes["access_log"] = {"enabled": False} @@ -247,37 +271,37 @@ class FakeLoadBalancer(CloudFormationModel): 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: raise TooManyTagsError() self.tags[key] = value - def list_tags(self): + def list_tags(self) -> Dict[str, str]: return self.tags - def remove_tag(self, key): + def remove_tag(self, key: str) -> None: if key in self.tags: 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.""" elb_backends[account_id][region].delete_load_balancer(self.name) 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) - self.load_balancers = OrderedDict() + self.load_balancers: Dict[str, FakeLoadBalancer] = OrderedDict() def create_load_balancer( self, - name, - zones, - ports, - scheme="internet-facing", - subnets=None, - security_groups=None, - ): + name: str, + zones: List[str], + ports: List[Dict[str, Any]], + scheme: str = "internet-facing", + subnets: Optional[List[str]] = None, + security_groups: Optional[List[str]] = None, + ) -> FakeLoadBalancer: vpc_id = None ec2_backend = ec2_backends[self.account_id][self.region_name] if subnets: @@ -317,7 +341,9 @@ class ELBBackend(BaseBackend): self.load_balancers[name] = 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) if balancer: for port in ports: @@ -350,19 +376,21 @@ class ELBBackend(BaseBackend): return balancer def describe_load_balancers(self, names: List[str]) -> List[FakeLoadBalancer]: - balancers = self.load_balancers.values() + balancers = list(self.load_balancers.values()) if names: matched_balancers = [ balancer for balancer in balancers if balancer.name in names ] 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) return matched_balancers else: 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] policies = lb.policies if policy_names: @@ -371,7 +399,9 @@ class ELBBackend(BaseBackend): raise PolicyNotFoundError() 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) if elb is None: raise NoActiveLoadBalancerFoundError(name=lb_name) @@ -394,8 +424,10 @@ class ELBBackend(BaseBackend): return instances - def delete_load_balancer_listeners(self, name, ports): - balancer = self.load_balancers.get(name, None) + def delete_load_balancer_listeners( + self, name: str, ports: List[str] + ) -> Optional[FakeLoadBalancer]: + balancer = self.get_load_balancer(name) listeners = [] if balancer: for lb_port in ports: @@ -407,20 +439,20 @@ class ELBBackend(BaseBackend): balancer.listeners = listeners 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) - 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.policies = [p for p in lb.policies if p.policy_name != policy_name] - def get_load_balancer(self, load_balancer_name): - return self.load_balancers.get(load_balancer_name) + def get_load_balancer(self, load_balancer_name: str) -> FakeLoadBalancer: + return self.load_balancers.get(load_balancer_name) # type: ignore[return-value] def apply_security_groups_to_load_balancer( - self, load_balancer_name, security_group_ids - ): - load_balancer = self.load_balancers.get(load_balancer_name) + self, load_balancer_name: str, security_group_ids: List[str] + ) -> None: + load_balancer = self.get_load_balancer(load_balancer_name) ec2_backend = ec2_backends[self.account_id][self.region_name] for security_group_id in security_group_ids: if ec2_backend.get_security_group_from_id(security_group_id) is None: @@ -429,13 +461,13 @@ class ELBBackend(BaseBackend): def configure_health_check( self, - load_balancer_name, - timeout, - healthy_threshold, - unhealthy_threshold, - interval, - target, - ): + load_balancer_name: str, + timeout: str, + healthy_threshold: str, + unhealthy_threshold: str, + interval: str, + target: str, + ) -> FakeHealthCheck: check = FakeHealthCheck( timeout, healthy_threshold, unhealthy_threshold, interval, target ) @@ -444,8 +476,8 @@ class ELBBackend(BaseBackend): return check 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) if balancer: for idx, listener in enumerate(balancer.listeners): @@ -500,12 +532,12 @@ class ELBBackend(BaseBackend): def modify_load_balancer_attributes( self, - load_balancer_name, - cross_zone=None, - connection_settings=None, - connection_draining=None, - access_log=None, - ): + load_balancer_name: str, + cross_zone: Optional[Dict[str, Any]] = None, + connection_settings: Optional[Dict[str, Any]] = None, + connection_draining: Optional[Dict[str, Any]] = None, + access_log: Optional[Dict[str, Any]] = None, + ) -> None: load_balancer = self.get_load_balancer(load_balancer_name) if 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 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) if policy_name not in [p.policy_name for p in load_balancer.policies]: load_balancer.policies.append( @@ -532,24 +568,27 @@ class ELBBackend(BaseBackend): return load_balancer 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) policy = AppCookieStickinessPolicy(policy_name, cookie_name) load_balancer.policies.append(policy) return load_balancer 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) policy = LbCookieStickinessPolicy(policy_name, cookie_expiration_period) load_balancer.policies.append(policy) return load_balancer 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) backend = [ b for b in load_balancer.backends if int(b.instance_port) == instance_port @@ -560,8 +599,8 @@ class ELBBackend(BaseBackend): return load_balancer 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) listener = [ l_listener @@ -573,7 +612,7 @@ class ELBBackend(BaseBackend): load_balancer.listeners[listener_idx] = listener 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 acm_backend = acm_backends[self.account_id][self.region_name] @@ -583,8 +622,8 @@ class ELBBackend(BaseBackend): raise CertificateNotFoundException() 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.zones = sorted( list(set(load_balancer.zones + availability_zones)) @@ -592,8 +631,8 @@ class ELBBackend(BaseBackend): return load_balancer.zones 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.zones = sorted( list( @@ -602,12 +641,16 @@ class ELBBackend(BaseBackend): ) 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.subnets = list(set(load_balancer.subnets + 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.subnets = [s for s in load_balancer.subnets if s not in subnets] return load_balancer.subnets diff --git a/moto/elb/policies.py b/moto/elb/policies.py index b1a6eb39c..0ab1c137d 100644 --- a/moto/elb/policies.py +++ b/moto/elb/policies.py @@ -1,24 +1,30 @@ +from typing import Any, Dict, List, Optional + + 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 class AppCookieStickinessPolicy(Policy): - def __init__(self, policy_name, cookie_name): - super().__init__(policy_type_name="AppCookieStickinessPolicy") - self.policy_name = policy_name + def __init__(self, policy_name: str, cookie_name: str): + super().__init__(policy_name, policy_type_name="AppCookieStickinessPolicy") self.cookie_name = cookie_name class LbCookieStickinessPolicy(Policy): - def __init__(self, policy_name, cookie_expiration_period): - super().__init__(policy_type_name="LbCookieStickinessPolicy") - self.policy_name = policy_name + def __init__(self, policy_name: str, cookie_expiration_period: Optional[int]): + super().__init__(policy_name, policy_type_name="LbCookieStickinessPolicy") self.cookie_expiration_period = cookie_expiration_period class OtherPolicy(Policy): - def __init__(self, policy_name, policy_type_name, policy_attrs): - super().__init__(policy_type_name=policy_type_name) - self.policy_name = policy_name + def __init__( + self, + 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 [] diff --git a/moto/elb/responses.py b/moto/elb/responses.py index 30cbb1393..fe75b49a5 100644 --- a/moto/elb/responses.py +++ b/moto/elb/responses.py @@ -1,17 +1,17 @@ from moto.core.responses import BaseResponse -from .models import elb_backends +from .models import elb_backends, ELBBackend, FakeLoadBalancer from .exceptions import DuplicateTagKeysError, LoadBalancerNotFoundError class ELBResponse(BaseResponse): - def __init__(self): + def __init__(self) -> None: super().__init__(service_name="elb") @property - def elb_backend(self): + def elb_backend(self) -> ELBBackend: 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") availability_zones = self._get_multi_param("AvailabilityZones.member") ports = self._get_list_prefix("Listeners.member") @@ -31,7 +31,7 @@ class ELBResponse(BaseResponse): template = self.response_template(CREATE_LOAD_BALANCER_TEMPLATE) 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") ports = self._get_list_prefix("Listeners.member") @@ -42,7 +42,7 @@ class ELBResponse(BaseResponse): template = self.response_template(CREATE_LOAD_BALANCER_LISTENERS_TEMPLATE) return template.render() - def describe_load_balancers(self): + def describe_load_balancers(self) -> str: names = self._get_multi_param("LoadBalancerNames.member") all_load_balancers = list(self.elb_backend.describe_load_balancers(names)) marker = self._get_param("Marker") @@ -66,7 +66,7 @@ class ELBResponse(BaseResponse): marker=next_marker, ) - def delete_load_balancer_listeners(self): + def delete_load_balancer_listeners(self) -> str: load_balancer_name = self._get_param("LoadBalancerName") ports = self._get_multi_param("LoadBalancerPorts.member") ports = [int(port) for port in ports] @@ -75,14 +75,14 @@ class ELBResponse(BaseResponse): template = self.response_template(DELETE_LOAD_BALANCER_LISTENERS) return template.render() - def delete_load_balancer(self): + def delete_load_balancer(self) -> str: load_balancer_name = self._get_param("LoadBalancerName") self.elb_backend.delete_load_balancer(load_balancer_name) template = self.response_template(DELETE_LOAD_BALANCER_TEMPLATE) return template.render() - def delete_load_balancer_policy(self): - load_balancer_name = self.querystring.get("LoadBalancerName")[0] + def delete_load_balancer_policy(self) -> str: + load_balancer_name = self.querystring.get("LoadBalancerName")[0] # type: ignore names = self._get_param("PolicyName") self.elb_backend.delete_load_balancer_policy( lb_name=load_balancer_name, policy_name=names @@ -91,7 +91,7 @@ class ELBResponse(BaseResponse): template = self.response_template(DELETE_LOAD_BALANCER_POLICY) 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") security_group_ids = self._get_multi_param("SecurityGroups.member") self.elb_backend.apply_security_groups_to_load_balancer( @@ -100,7 +100,7 @@ class ELBResponse(BaseResponse): template = self.response_template(APPLY_SECURITY_GROUPS_TEMPLATE) 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( load_balancer_name=self._get_param("LoadBalancerName"), timeout=self._get_param("HealthCheck.Timeout"), @@ -112,7 +112,7 @@ class ELBResponse(BaseResponse): template = self.response_template(CONFIGURE_HEALTH_CHECK_TEMPLATE) 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") instance_ids = [ list(param.values())[0] @@ -124,7 +124,7 @@ class ELBResponse(BaseResponse): ) 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") ssl_certificate_id = self.querystring["SSLCertificateId"][0] lb_port = self.querystring["LoadBalancerPort"][0] @@ -136,7 +136,7 @@ class ELBResponse(BaseResponse): template = self.response_template(SET_LOAD_BALANCER_SSL_CERTIFICATE) 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") instance_ids = [ list(param.values())[0] @@ -148,13 +148,13 @@ class ELBResponse(BaseResponse): ) 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 = self.elb_backend.get_load_balancer(load_balancer_name) template = self.response_template(DESCRIBE_ATTRIBUTES_TEMPLATE) 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 = self.elb_backend.get_load_balancer(load_balancer_name) @@ -193,7 +193,7 @@ class ELBResponse(BaseResponse): 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") policy_name = self._get_param("PolicyName") @@ -207,7 +207,7 @@ class ELBResponse(BaseResponse): template = self.response_template(CREATE_LOAD_BALANCER_POLICY_TEMPLATE) 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") policy_name = self._get_param("PolicyName") @@ -220,7 +220,7 @@ class ELBResponse(BaseResponse): template = self.response_template(CREATE_APP_COOKIE_STICKINESS_POLICY_TEMPLATE) 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") policy_name = self._get_param("PolicyName") @@ -237,7 +237,7 @@ class ELBResponse(BaseResponse): template = self.response_template(CREATE_LB_COOKIE_STICKINESS_POLICY_TEMPLATE) 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 = self.elb_backend.get_load_balancer(load_balancer_name) load_balancer_port = int(self._get_param("LoadBalancerPort")) @@ -259,10 +259,10 @@ class ELBResponse(BaseResponse): ) return template.render() - def set_load_balancer_policies_for_backend_server(self): - load_balancer_name = self.querystring.get("LoadBalancerName")[0] + def set_load_balancer_policies_for_backend_server(self) -> str: + load_balancer_name = self.querystring.get("LoadBalancerName")[0] # type: ignore 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 = [ b for b in load_balancer.backends if int(b.instance_port) == instance_port @@ -270,7 +270,7 @@ class ELBResponse(BaseResponse): if mb_backend: policies = self._get_multi_param("PolicyNames.member") 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? @@ -279,8 +279,8 @@ class ELBResponse(BaseResponse): ) return template.render() - def describe_load_balancer_policies(self): - load_balancer_name = self.querystring.get("LoadBalancerName")[0] + def describe_load_balancer_policies(self) -> str: + load_balancer_name = self.querystring.get("LoadBalancerName")[0] # type: ignore names = self._get_multi_param("PolicyNames.member.") policies = self.elb_backend.describe_load_balancer_policies( lb_name=load_balancer_name, policy_names=names @@ -289,14 +289,14 @@ class ELBResponse(BaseResponse): template = self.response_template(DESCRIBE_LOAD_BALANCER_POLICIES_TEMPLATE) return template.render(policies=policies) - def describe_instance_health(self): + def describe_instance_health(self) -> str: lb_name = self._get_param("LoadBalancerName") instances = self._get_params().get("Instances", []) instances = self.elb_backend.describe_instance_health(lb_name, instances) template = self.response_template(DESCRIBE_INSTANCE_HEALTH_TEMPLATE) return template.render(instances=instances) - def add_tags(self): + def add_tags(self) -> str: for key, value in self.querystring.items(): if "LoadBalancerNames.member" in key: @@ -310,7 +310,7 @@ class ELBResponse(BaseResponse): template = self.response_template(ADD_TAGS_TEMPLATE) return template.render() - def remove_tags(self): + def remove_tags(self) -> str: for key in self.querystring: if "LoadBalancerNames.member" in key: number = key.split(".")[2] @@ -329,7 +329,7 @@ class ELBResponse(BaseResponse): template = self.response_template(REMOVE_TAGS_TEMPLATE) return template.render() - def describe_tags(self): + def describe_tags(self) -> str: elbs = [] for key in self.querystring: if "LoadBalancerNames.member" in key: @@ -345,7 +345,7 @@ class ELBResponse(BaseResponse): template = self.response_template(DESCRIBE_TAGS_TEMPLATE) return template.render(load_balancers=elbs) - def _add_tags(self, elb): + def _add_tags(self, elb: FakeLoadBalancer) -> None: tag_values = [] tag_keys = [] @@ -356,11 +356,11 @@ class ELBResponse(BaseResponse): elif t_key.split(".")[3] == "Value": tag_values.extend(t_val) - counts = {} + count_dict = {} 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: # We have dupes... @@ -369,14 +369,14 @@ class ELBResponse(BaseResponse): for tag_key, tag_value in zip(tag_keys, tag_values): 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() load_balancer_name = params.get("LoadBalancerName") availability_zones = params.get("AvailabilityZones") availability_zones = ( self.elb_backend.enable_availability_zones_for_load_balancer( - load_balancer_name=load_balancer_name, - availability_zones=availability_zones, + load_balancer_name=load_balancer_name, # type: ignore[arg-type] + availability_zones=availability_zones, # type: ignore[arg-type] ) ) template = self.response_template( @@ -384,14 +384,14 @@ class ELBResponse(BaseResponse): ) 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() load_balancer_name = params.get("LoadBalancerName") availability_zones = params.get("AvailabilityZones") availability_zones = ( self.elb_backend.disable_availability_zones_for_load_balancer( - load_balancer_name=load_balancer_name, - availability_zones=availability_zones, + load_balancer_name=load_balancer_name, # type: ignore[arg-type] + availability_zones=availability_zones, # type: ignore[arg-type] ) ) template = self.response_template( @@ -399,24 +399,24 @@ class ELBResponse(BaseResponse): ) 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() load_balancer_name = params.get("LoadBalancerName") subnets = params.get("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) 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() load_balancer_name = params.get("LoadBalancerName") subnets = params.get("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) return template.render(subnets=all_subnets) diff --git a/moto/elb/urls.py b/moto/elb/urls.py index 8d9db491e..5b2a5f5bc 100644 --- a/moto/elb/urls.py +++ b/moto/elb/urls.py @@ -1,3 +1,4 @@ +from typing import Any from urllib.parse import parse_qs from botocore.awsrequest import AWSPreparedRequest @@ -5,7 +6,7 @@ from moto.elb.responses import ELBResponse 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 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") elif isinstance(request, AWSPreparedRequest): # boto in-memory - version = parse_qs(request.body).get("Version")[0] + version = parse_qs(request.body).get("Version")[0] # type: ignore else: # boto in server mode request.parse_request() diff --git a/setup.cfg b/setup.cfg index 9c85889d8..08b437f69 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,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_error_codes = True disable_error_code=abstract