Techdebt: MyPy ELBv2 (#6001)

This commit is contained in:
Bert Blommers 2023-03-01 10:39:12 -01:00 committed by GitHub
parent 490e631245
commit a59f921036
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 382 additions and 294 deletions

View File

@ -1,46 +1,47 @@
from typing import Any, Optional
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 LoadBalancerNotFoundError(ELBClientError):
def __init__(self):
def __init__(self) -> None:
super().__init__(
"LoadBalancerNotFound", "The specified load balancer does not exist."
)
class ListenerNotFoundError(ELBClientError):
def __init__(self):
def __init__(self) -> None:
super().__init__("ListenerNotFound", "The specified listener does not exist.")
class SubnetNotFoundError(ELBClientError):
def __init__(self):
def __init__(self) -> None:
super().__init__("SubnetNotFound", "The specified subnet does not exist.")
class TargetGroupNotFoundError(ELBClientError):
def __init__(self):
def __init__(self) -> None:
super().__init__(
"TargetGroupNotFound", "The specified target group does not exist."
)
class TooManyTagsError(ELBClientError):
def __init__(self):
def __init__(self) -> None:
super().__init__(
"TooManyTagsError",
"The quota for the number of tags that can be assigned to a load balancer has been reached",
@ -48,7 +49,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",
@ -56,14 +57,14 @@ class BadHealthCheckDefinition(ELBClientError):
class DuplicateListenerError(ELBClientError):
def __init__(self):
def __init__(self) -> None:
super().__init__(
"DuplicateListener", "A listener with the specified port already exists."
)
class DuplicateLoadBalancerName(ELBClientError):
def __init__(self):
def __init__(self) -> None:
super().__init__(
"DuplicateLoadBalancerName",
"A load balancer with the specified name already exists.",
@ -71,7 +72,7 @@ class DuplicateLoadBalancerName(ELBClientError):
class DuplicateTargetGroupName(ELBClientError):
def __init__(self):
def __init__(self) -> None:
super().__init__(
"DuplicateTargetGroupName",
"A target group with the specified name already exists.",
@ -79,7 +80,7 @@ class DuplicateTargetGroupName(ELBClientError):
class InvalidTargetError(ELBClientError):
def __init__(self):
def __init__(self) -> None:
super().__init__(
"InvalidTarget",
"The specified target does not exist or is not in the same VPC as the target group.",
@ -87,12 +88,12 @@ class InvalidTargetError(ELBClientError):
class EmptyListenersError(ELBClientError):
def __init__(self):
def __init__(self) -> None:
super().__init__("ValidationError", "Listeners cannot be empty")
class PriorityInUseError(ELBClientError):
def __init__(self):
def __init__(self) -> None:
super().__init__("PriorityInUse", "The specified priority is in use.")
@ -106,7 +107,7 @@ class InvalidConditionFieldError(ELBClientError):
"source-ip",
]
def __init__(self, invalid_name):
def __init__(self, invalid_name: str):
valid = ",".join(self.VALID_FIELDS)
super().__init__(
"ValidationError",
@ -115,12 +116,12 @@ class InvalidConditionFieldError(ELBClientError):
class InvalidConditionValueError(ELBClientError):
def __init__(self, msg):
def __init__(self, msg: str):
super().__init__("ValidationError", msg)
class InvalidActionTypeError(ELBClientError):
def __init__(self, invalid_name, index):
def __init__(self, invalid_name: Any, index: int):
super().__init__(
"ValidationError",
f"1 validation error detected: Value '{invalid_name}' at 'actions.{index}.member.type' failed to satisfy constraint: Member must satisfy enum value set: [forward, redirect, fixed-response]",
@ -128,12 +129,12 @@ class InvalidActionTypeError(ELBClientError):
class ActionTargetGroupNotFoundError(ELBClientError):
def __init__(self, arn):
def __init__(self, arn: str):
super().__init__("TargetGroupNotFound", f"Target group '{arn}' not found")
class ListenerOrBalancerMissingError(ELBClientError):
def __init__(self):
def __init__(self) -> None:
super().__init__(
"ValidationError",
"You must specify either listener ARNs or a load balancer ARN",
@ -141,45 +142,45 @@ class ListenerOrBalancerMissingError(ELBClientError):
class InvalidDescribeRulesRequest(ELBClientError):
def __init__(self, msg):
def __init__(self, msg: str):
super().__init__("ValidationError", msg)
class ResourceInUseError(ELBClientError):
def __init__(self, msg="A specified resource is in use"):
def __init__(self, msg: str = "A specified resource is in use"):
super().__init__("ResourceInUse", msg)
class RuleNotFoundError(ELBClientError):
def __init__(self, msg=None):
def __init__(self, msg: Optional[str] = None):
msg = msg or "The specified rule does not exist."
super().__init__("RuleNotFound", msg)
class DuplicatePriorityError(ELBClientError):
def __init__(self, invalid_value):
def __init__(self, invalid_value: str):
super().__init__(
"ValidationError", f"Priority '{invalid_value}' was provided multiple times"
)
class InvalidTargetGroupNameError(ELBClientError):
def __init__(self, msg):
def __init__(self, msg: str):
super().__init__("ValidationError", msg)
class InvalidModifyRuleArgumentsError(ELBClientError):
def __init__(self):
def __init__(self) -> None:
super().__init__(
"ValidationError", "Either conditions or actions must be specified"
)
class InvalidStatusCodeActionTypeError(ELBClientError):
def __init__(self, msg):
def __init__(self, msg: str):
super().__init__("ValidationError", msg)
class InvalidLoadBalancerActionException(ELBClientError):
def __init__(self, msg):
def __init__(self, msg: str):
super().__init__("InvalidLoadBalancerAction", msg)

File diff suppressed because it is too large Load Diff

View File

@ -1,7 +1,7 @@
from moto.core.exceptions import RESTError
from moto.core.responses import BaseResponse
from moto.utilities.aws_headers import amzn_request_id
from .models import elbv2_backends
from .models import elbv2_backends, ELBv2Backend
from .exceptions import TargetGroupNotFoundError
from .exceptions import ListenerOrBalancerMissingError
@ -135,15 +135,15 @@ SSL_POLICIES = [
class ELBV2Response(BaseResponse):
def __init__(self):
def __init__(self) -> None:
super().__init__(service_name="elbv2")
@property
def elbv2_backend(self):
def elbv2_backend(self) -> ELBv2Backend:
return elbv2_backends[self.current_account][self.region]
@amzn_request_id
def create_load_balancer(self):
def create_load_balancer(self) -> str:
params = self._get_params()
load_balancer_name = params.get("Name")
subnet_ids = self._get_multi_param("Subnets.member")
@ -154,11 +154,11 @@ class ELBV2Response(BaseResponse):
tags = params.get("Tags")
load_balancer = self.elbv2_backend.create_load_balancer(
name=load_balancer_name,
name=load_balancer_name, # type: ignore
security_groups=security_groups,
subnet_ids=subnet_ids,
subnet_mappings=subnet_mappings,
scheme=scheme,
scheme=scheme, # type: ignore
loadbalancer_type=loadbalancer_type,
tags=tags,
)
@ -166,7 +166,7 @@ class ELBV2Response(BaseResponse):
return template.render(load_balancer=load_balancer)
@amzn_request_id
def create_rule(self):
def create_rule(self) -> str:
params = self._get_params()
rules = self.elbv2_backend.create_rule(
listener_arn=params["ListenerArn"],
@ -180,7 +180,7 @@ class ELBV2Response(BaseResponse):
return template.render(rules=rules)
@amzn_request_id
def create_target_group(self):
def create_target_group(self) -> str:
params = self._get_params()
name = params.get("Name")
vpc_id = params.get("VpcId")
@ -200,7 +200,7 @@ class ELBV2Response(BaseResponse):
tags = params.get("Tags")
target_group = self.elbv2_backend.create_target_group(
name,
name, # type: ignore
vpc_id=vpc_id,
protocol=protocol,
protocol_version=protocol_version,
@ -222,7 +222,7 @@ class ELBV2Response(BaseResponse):
return template.render(target_group=target_group)
@amzn_request_id
def create_listener(self):
def create_listener(self) -> str:
params = self._get_params()
load_balancer_arn = self._get_param("LoadBalancerArn")
protocol = self._get_param("Protocol")
@ -243,7 +243,7 @@ class ELBV2Response(BaseResponse):
port=port,
ssl_policy=ssl_policy,
certificate=certificate,
default_actions=default_actions,
actions=default_actions,
alpn_policy=alpn_policy,
tags=tags,
)
@ -252,7 +252,7 @@ class ELBV2Response(BaseResponse):
return template.render(listener=listener)
@amzn_request_id
def describe_load_balancers(self):
def describe_load_balancers(self) -> str:
arns = self._get_multi_param("LoadBalancerArns.member")
names = self._get_multi_param("Names.member")
all_load_balancers = list(
@ -276,7 +276,7 @@ class ELBV2Response(BaseResponse):
return template.render(load_balancers=load_balancers_resp, marker=next_marker)
@amzn_request_id
def describe_rules(self):
def describe_rules(self) -> str:
listener_arn = self._get_param("ListenerArn")
rule_arns = (
self._get_multi_param("RuleArns.member")
@ -305,7 +305,7 @@ class ELBV2Response(BaseResponse):
return template.render(rules=rules_resp, marker=next_marker)
@amzn_request_id
def describe_target_groups(self):
def describe_target_groups(self) -> str:
load_balancer_arn = self._get_param("LoadBalancerArn")
target_group_arns = self._get_multi_param("TargetGroupArns.member")
names = self._get_multi_param("Names.member")
@ -317,7 +317,7 @@ class ELBV2Response(BaseResponse):
return template.render(target_groups=target_groups)
@amzn_request_id
def describe_target_group_attributes(self):
def describe_target_group_attributes(self) -> str:
target_group_arn = self._get_param("TargetGroupArn")
target_group = self.elbv2_backend.target_groups.get(target_group_arn)
if not target_group:
@ -326,7 +326,7 @@ class ELBV2Response(BaseResponse):
return template.render(attributes=target_group.attributes)
@amzn_request_id
def describe_listeners(self):
def describe_listeners(self) -> str:
load_balancer_arn = self._get_param("LoadBalancerArn")
listener_arns = self._get_multi_param("ListenerArns.member")
if not load_balancer_arn and not listener_arns:
@ -339,35 +339,35 @@ class ELBV2Response(BaseResponse):
return template.render(listeners=listeners)
@amzn_request_id
def delete_load_balancer(self):
def delete_load_balancer(self) -> str:
arn = self._get_param("LoadBalancerArn")
self.elbv2_backend.delete_load_balancer(arn)
template = self.response_template(DELETE_LOAD_BALANCER_TEMPLATE)
return template.render()
@amzn_request_id
def delete_rule(self):
def delete_rule(self) -> str:
arn = self._get_param("RuleArn")
self.elbv2_backend.delete_rule(arn)
template = self.response_template(DELETE_RULE_TEMPLATE)
return template.render()
@amzn_request_id
def delete_target_group(self):
def delete_target_group(self) -> str:
arn = self._get_param("TargetGroupArn")
self.elbv2_backend.delete_target_group(arn)
template = self.response_template(DELETE_TARGET_GROUP_TEMPLATE)
return template.render()
@amzn_request_id
def delete_listener(self):
def delete_listener(self) -> str:
arn = self._get_param("ListenerArn")
self.elbv2_backend.delete_listener(arn)
template = self.response_template(DELETE_LISTENER_TEMPLATE)
return template.render()
@amzn_request_id
def modify_rule(self):
def modify_rule(self) -> str:
rule_arn = self._get_param("RuleArn")
params = self._get_params()
conditions = params.get("Conditions", [])
@ -379,17 +379,17 @@ class ELBV2Response(BaseResponse):
return template.render(rules=rules)
@amzn_request_id
def modify_target_group_attributes(self):
def modify_target_group_attributes(self) -> str:
target_group_arn = self._get_param("TargetGroupArn")
attributes = self._get_list_prefix("Attributes.member")
attributes = {attr["key"]: attr["value"] for attr in attributes}
attrs = self._get_list_prefix("Attributes.member")
attributes = {attr["key"]: attr["value"] for attr in attrs}
self.elbv2_backend.modify_target_group_attributes(target_group_arn, attributes)
template = self.response_template(MODIFY_TARGET_GROUP_ATTRIBUTES_TEMPLATE)
return template.render(attributes=attributes)
@amzn_request_id
def register_targets(self):
def register_targets(self) -> str:
target_group_arn = self._get_param("TargetGroupArn")
targets = self._get_list_prefix("Targets.member")
self.elbv2_backend.register_targets(target_group_arn, targets)
@ -398,7 +398,7 @@ class ELBV2Response(BaseResponse):
return template.render()
@amzn_request_id
def deregister_targets(self):
def deregister_targets(self) -> str:
target_group_arn = self._get_param("TargetGroupArn")
targets = self._get_list_prefix("Targets.member")
self.elbv2_backend.deregister_targets(target_group_arn, targets)
@ -407,7 +407,7 @@ class ELBV2Response(BaseResponse):
return template.render()
@amzn_request_id
def describe_target_health(self):
def describe_target_health(self) -> str:
target_group_arn = self._get_param("TargetGroupArn")
targets = self._get_list_prefix("Targets.member")
target_health_descriptions = self.elbv2_backend.describe_target_health(
@ -418,7 +418,7 @@ class ELBV2Response(BaseResponse):
return template.render(target_health_descriptions=target_health_descriptions)
@amzn_request_id
def set_rule_priorities(self):
def set_rule_priorities(self) -> str:
rule_priorities = self._get_list_prefix("RulePriorities.member")
for rule_priority in rule_priorities:
rule_priority["priority"] = int(rule_priority["priority"])
@ -427,18 +427,17 @@ class ELBV2Response(BaseResponse):
return template.render(rules=rules)
@amzn_request_id
def add_tags(self):
def add_tags(self) -> str:
resource_arns = self._get_multi_param("ResourceArns.member")
tags = self._get_params().get("Tags")
tags = self._get_params().get("Tags")
self.elbv2_backend.add_tags(resource_arns, tags)
self.elbv2_backend.add_tags(resource_arns, tags) # type: ignore
template = self.response_template(ADD_TAGS_TEMPLATE)
return template.render()
@amzn_request_id
def remove_tags(self):
def remove_tags(self) -> str:
resource_arns = self._get_multi_param("ResourceArns.member")
tag_keys = self._get_multi_param("TagKeys.member")
@ -448,7 +447,7 @@ class ELBV2Response(BaseResponse):
return template.render()
@amzn_request_id
def describe_tags(self):
def describe_tags(self) -> str:
resource_arns = self._get_multi_param("ResourceArns.member")
resource_tags = self.elbv2_backend.describe_tags(resource_arns)
@ -456,7 +455,7 @@ class ELBV2Response(BaseResponse):
return template.render(resource_tags=resource_tags)
@amzn_request_id
def describe_account_limits(self):
def describe_account_limits(self) -> str:
# Supports paging but not worth implementing yet
# marker = self._get_param('Marker')
# page_size = self._get_int_param('PageSize')
@ -476,7 +475,7 @@ class ELBV2Response(BaseResponse):
return template.render(limits=limits)
@amzn_request_id
def describe_ssl_policies(self):
def describe_ssl_policies(self) -> str:
names = self._get_multi_param("Names.member.")
# Supports paging but not worth implementing yet
# marker = self._get_param('Marker')
@ -484,13 +483,13 @@ class ELBV2Response(BaseResponse):
policies = SSL_POLICIES
if names:
policies = filter(lambda policy: policy["name"] in names, policies)
policies = filter(lambda policy: policy["name"] in names, policies) # type: ignore
template = self.response_template(DESCRIBE_SSL_POLICIES_TEMPLATE)
return template.render(policies=policies)
@amzn_request_id
def set_ip_address_type(self):
def set_ip_address_type(self) -> str:
arn = self._get_param("LoadBalancerArn")
ip_type = self._get_param("IpAddressType")
@ -500,7 +499,7 @@ class ELBV2Response(BaseResponse):
return template.render(ip_type=ip_type)
@amzn_request_id
def set_security_groups(self):
def set_security_groups(self) -> str:
arn = self._get_param("LoadBalancerArn")
sec_groups = self._get_multi_param("SecurityGroups.member.")
@ -510,7 +509,7 @@ class ELBV2Response(BaseResponse):
return template.render(sec_groups=sec_groups)
@amzn_request_id
def set_subnets(self):
def set_subnets(self) -> str:
arn = self._get_param("LoadBalancerArn")
subnets = self._get_multi_param("Subnets.member.")
subnet_mappings = self._get_params().get("SubnetMappings", [])
@ -521,7 +520,7 @@ class ELBV2Response(BaseResponse):
return template.render(subnets=subnet_zone_list)
@amzn_request_id
def modify_load_balancer_attributes(self):
def modify_load_balancer_attributes(self) -> str:
arn = self._get_param("LoadBalancerArn")
attrs = self._get_map_prefix(
"Attributes.member", key_end="Key", value_end="Value"
@ -533,7 +532,7 @@ class ELBV2Response(BaseResponse):
return template.render(attrs=all_attrs)
@amzn_request_id
def describe_load_balancer_attributes(self):
def describe_load_balancer_attributes(self) -> str:
arn = self._get_param("LoadBalancerArn")
attrs = self.elbv2_backend.describe_load_balancer_attributes(arn)
@ -541,7 +540,7 @@ class ELBV2Response(BaseResponse):
return template.render(attrs=attrs)
@amzn_request_id
def modify_target_group(self):
def modify_target_group(self) -> str:
arn = self._get_param("TargetGroupArn")
health_check_proto = self._get_param(
@ -573,7 +572,7 @@ class ELBV2Response(BaseResponse):
return template.render(target_group=target_group)
@amzn_request_id
def modify_listener(self):
def modify_listener(self) -> str:
arn = self._get_param("ListenerArn")
port = self._get_param("Port")
protocol = self._get_param("Protocol")
@ -595,16 +594,18 @@ class ELBV2Response(BaseResponse):
return template.render(listener=listener)
@amzn_request_id
def add_listener_certificates(self):
def add_listener_certificates(self) -> str:
arn = self._get_param("ListenerArn")
certificates = self._get_list_prefix("Certificates.member")
certificates = self.elbv2_backend.add_listener_certificates(arn, certificates)
certificate_arns = self.elbv2_backend.add_listener_certificates(
arn, certificates
)
template = self.response_template(ADD_LISTENER_CERTIFICATES_TEMPLATE)
return template.render(certificates=certificates)
return template.render(certificates=certificate_arns)
@amzn_request_id
def describe_listener_certificates(self):
def describe_listener_certificates(self) -> str:
arn = self._get_param("ListenerArn")
certificates = self.elbv2_backend.describe_listener_certificates(arn)
@ -612,15 +613,13 @@ class ELBV2Response(BaseResponse):
return template.render(certificates=certificates)
@amzn_request_id
def remove_listener_certificates(self):
def remove_listener_certificates(self) -> str:
arn = self._get_param("ListenerArn")
certificates = self._get_list_prefix("Certificates.member")
certificates = self.elbv2_backend.remove_listener_certificates(
arn, certificates
)
self.elbv2_backend.remove_listener_certificates(arn, certificates)
template = self.response_template(REMOVE_LISTENER_CERTIFICATES_TEMPLATE)
return template.render(certificates=certificates)
return template.render()
ADD_TAGS_TEMPLATE = """<AddTagsResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
@ -1580,7 +1579,7 @@ SET_SECURITY_GROUPS_TEMPLATE = """<SetSecurityGroupsResponse xmlns="http://elast
SET_SUBNETS_TEMPLATE = """<SetSubnetsResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<SetSubnetsResult>
<AvailabilityZones>
{% for zone_id, subnet_id in subnets %}
{% for zone_id, subnet_id in subnets.items() %}
<member>
<SubnetId>{{ subnet_id }}</SubnetId>
<ZoneName>{{ zone_id }}</ZoneName>

View File

@ -1,6 +1,6 @@
def make_arn_for_load_balancer(account_id, name, region_name):
def make_arn_for_load_balancer(account_id: str, name: str, region_name: str) -> str:
return f"arn:aws:elasticloadbalancing:{region_name}:{account_id}:loadbalancer/app/{name}/50dc6c495c0c9188"
def make_arn_for_target_group(account_id, name, region_name):
def make_arn_for_target_group(account_id: str, name: str, region_name: str) -> str:
return f"arn:aws:elasticloadbalancing:{region_name}:{account_id}:targetgroup/{name}/50dc6c495c0c9188"

View File

@ -1,6 +1,6 @@
"""Tag functionality contained in class TaggingService."""
import re
from typing import Dict, List
from typing import Dict, List, Optional
class TaggingService:
@ -43,7 +43,7 @@ class TaggingService:
"""Return True if the ARN has any associated tags, False otherwise."""
return arn in self.tags
def tag_resource(self, arn: str, tags: List[Dict[str, str]]) -> None:
def tag_resource(self, arn: str, tags: Optional[List[Dict[str, str]]]) -> None:
"""Store associated list of dicts with ARN.
Note: the storage is internal to this class instance.

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
[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/elb,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/elbv2,moto/es,moto/moto_api,moto/neptune
show_column_numbers=True
show_error_codes = True
disable_error_code=abstract