Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ELBv2: added validation for target group (#6808)

Browse Source
This commit is contained in:
Macwan Nevil 2023-09-14 18:22:14 +05:30 committed by GitHub
parent 9d8c11fdc8
commit c1a6609f55
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 297 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
moto/elbv2/exceptions.py
View File

@ -184,3 +184,8 @@ class InvalidStatusCodeActionTypeError(ELBClientError):
class InvalidLoadBalancerActionException(ELBClientError):
def __init__(self, msg: str):
super().__init__("InvalidLoadBalancerAction", msg)
class ValidationError(ELBClientError):
def __init__(self, msg: str):
super().__init__("ValidationError", msg)

171
moto/elbv2/models.py
View File

@ -35,6 +35,7 @@ from .exceptions import (
InvalidModifyRuleArgumentsError,
InvalidStatusCodeActionTypeError,
InvalidLoadBalancerActionException,
ValidationError,
)
ALLOWED_ACTIONS = [
@ -79,42 +80,49 @@ class FakeTargetGroup(CloudFormationModel):
healthcheck_port: Optional[str] = None,
healthcheck_path: Optional[str] = None,
healthcheck_interval_seconds: Optional[str] = None,
healthcheck_timeout_seconds: Optional[int] = None,
healthcheck_timeout_seconds: Optional[str] = None,
healthcheck_enabled: Optional[str] = None,
healthy_threshold_count: Optional[str] = None,
unhealthy_threshold_count: Optional[str] = None,
matcher: Optional[Dict[str, Any]] = None,
target_type: Optional[str] = None,
ip_address_type: Optional[str] = None,
):
# TODO: default values differs when you add Network Load balancer
self.name = name
self.arn = arn
self.vpc_id = vpc_id
self.protocol = protocol
self.protocol_version = protocol_version or "HTTP1"
if target_type == "lambda":
self.protocol = None
self.protocol_version = None
elif target_type == "alb":
self.protocol = "TCP"
self.protocol_version = None
else:
self.protocol = protocol
self.protocol_version = protocol_version
self.port = port
self.healthcheck_protocol = healthcheck_protocol or self.protocol
self.healthcheck_port = healthcheck_port
self.healthcheck_port = healthcheck_port or "traffic-port"
self.healthcheck_path = healthcheck_path
self.healthcheck_interval_seconds = healthcheck_interval_seconds or 30
self.healthcheck_timeout_seconds = healthcheck_timeout_seconds
if not healthcheck_timeout_seconds:
# Default depends on protocol
if protocol in ["TCP", "TLS"]:
self.healthcheck_timeout_seconds = 6
elif protocol in ["HTTP", "HTTPS", "GENEVE"]:
self.healthcheck_timeout_seconds = 5
else:
self.healthcheck_timeout_seconds = 30
self.healthcheck_enabled = healthcheck_enabled
self.healthy_threshold_count = healthy_threshold_count or 5
self.unhealthy_threshold_count = unhealthy_threshold_count or 2
self.healthcheck_interval_seconds = healthcheck_interval_seconds or "30"
self.healthcheck_timeout_seconds = healthcheck_timeout_seconds or "10"
self.ip_address_type = (
ip_address_type or "ipv4" if self.protocol != "GENEVE" else None
)
self.healthcheck_enabled = (
healthcheck_enabled.lower() == "true"
if healthcheck_enabled in ["true", "false"]
else True
)
self.healthy_threshold_count = healthy_threshold_count or "5"
self.unhealthy_threshold_count = unhealthy_threshold_count or "2"
self.load_balancer_arns: List[str] = []
if self.healthcheck_protocol != "TCP":
self.matcher: Dict[str, Any] = matcher or {"HttpCode": "200"}
self.healthcheck_path = self.healthcheck_path or "/"
self.healthcheck_path = self.healthcheck_path
self.healthcheck_port = self.healthcheck_port or str(self.port)
self.target_type = target_type
self.target_type = target_type or "instance"
self.attributes = {
"deregistration_delay.timeout_seconds": 300,
@ -1030,6 +1038,9 @@ Member must satisfy regular expression pattern: {expression}"
)
def create_target_group(self, name: str, **kwargs: Any) -> FakeTargetGroup:
protocol = kwargs.get("protocol")
target_type = kwargs.get("target_type")
if len(name) > 32:
raise InvalidTargetGroupNameError(
f"Target group name '{name}' cannot be longer than '32' characters"
@ -1080,6 +1091,124 @@ Member must satisfy regular expression pattern: {expression}"
"HttpCode must be like 200 | 200-399 | 200,201 ...",
)
if target_type in ("instance", "ip", "alb"):
for param in ("protocol", "port", "vpc_id"):
if not kwargs.get(param):
param = "VPC ID" if param == "vpc_id" else param.lower()
raise ValidationError(f"A {param} must be specified")
if kwargs.get("vpc_id"):
from moto.ec2.exceptions import InvalidVPCIdError
try:
self.ec2_backend.get_vpc(kwargs.get("vpc_id"))
except InvalidVPCIdError:
raise ValidationError(
f"The VPC ID '{kwargs.get('vpc_id')}' is not found"
)
kwargs_patch = {}
conditions: Dict[str, Any] = {
"target_lambda": {
"healthcheck_interval_seconds": 35,
"healthcheck_timeout_seconds": 30,
"unhealthy_threshold_count": 2,
"healthcheck_enabled": "false",
"healthcheck_path": "/",
},
"target_alb": {
"healthcheck_protocol": "HTTP",
"healthcheck_path": "/",
"healthcheck_timeout_seconds": 6,
"matcher": {"HttpCode": "200-399"},
},
"protocol_GENEVE": {
"healthcheck_interval_seconds": 10,
"healthcheck_port": 80,
"healthcheck_timeout_seconds": 5,
"healthcheck_protocol": "TCP",
"unhealthy_threshold_count": 2,
},
"protocol_HTTP_HTTPS": {
"healthcheck_timeout_seconds": 5,
"protocol_version": "HTTP1",
"healthcheck_path": "/",
"unhealthy_threshold_count": 2,
"healthcheck_interval_seconds": 30,
},
"protocol_TCP": {
"healthcheck_timeout_seconds": 10,
},
"protocol_TCP_TCP_UDP_UDP_TLS": {
"healthcheck_protocol": "TCP",
"unhealthy_threshold_count": 2,
"healthcheck_interval_seconds": 30,
},
}
if target_type == "lambda":
kwargs_patch.update(
{k: kwargs.get(k) or v for k, v in conditions["target_lambda"].items()}
)
if protocol == "GENEVE":
kwargs_patch.update(
{
k: kwargs.get(k) or v
for k, v in conditions["protocol_GENEVE"].items()
}
)
if protocol in ("HTTP", "HTTPS"):
kwargs_patch.update(
{
k: kwargs.get(k) or v
for k, v in conditions["protocol_HTTP_HTTPS"].items()
}
)
if protocol == "TCP":
kwargs_patch.update(
{k: kwargs.get(k) or v for k, v in conditions["protocol_TCP"].items()}
)
if protocol in ("TCP", "TCP_UDP", "UDP", "TLS"):
kwargs_patch.update(
{
k: kwargs.get(k) or v
for k, v in conditions["protocol_TCP_TCP_UDP_UDP_TLS"].items()
}
)
if target_type == "alb":
kwargs_patch.update(
{k: kwargs.get(k) or v for k, v in conditions["target_alb"].items()}
)
kwargs.update(kwargs_patch)
healthcheck_timeout_seconds = int(
str(kwargs.get("healthcheck_timeout_seconds") or "10")
)
healthcheck_interval_seconds = int(
str(kwargs.get("healthcheck_interval_seconds") or "30")
)
if (
healthcheck_timeout_seconds is not None
and healthcheck_interval_seconds is not None
):
if healthcheck_interval_seconds < healthcheck_timeout_seconds:
raise ValidationError(
"Health check interval must be greater than the timeout."
)
if healthcheck_interval_seconds == healthcheck_timeout_seconds:
raise ValidationError(
f"Health check timeout '{healthcheck_timeout_seconds}' must be smaller than the interval '{healthcheck_interval_seconds}'"
)
arn = make_arn_for_target_group(
account_id=self.account_id, name=name, region_name=self.region_name
)
@ -1524,11 +1653,11 @@ Member must satisfy regular expression pattern: {expression}"
health_check_port: Optional[str] = None,
health_check_path: Optional[str] = None,
health_check_interval: Optional[str] = None,
health_check_timeout: Optional[int] = None,
health_check_timeout: Optional[str] = None,
healthy_threshold_count: Optional[str] = None,
unhealthy_threshold_count: Optional[str] = None,
http_codes: Optional[str] = None,
health_check_enabled: Optional[str] = None,
health_check_enabled: Optional[bool] = None,
) -> FakeTargetGroup:
target_group = self.target_groups.get(arn)
if target_group is None:

26
moto/elbv2/responses.py
View File

@ -185,7 +185,7 @@ class ELBV2Response(BaseResponse):
name = params.get("Name")
vpc_id = params.get("VpcId")
protocol = params.get("Protocol")
protocol_version = params.get("ProtocolVersion", "HTTP1")
protocol_version = params.get("ProtocolVersion")
port = params.get("Port")
healthcheck_protocol = self._get_param("HealthCheckProtocol")
healthcheck_port = self._get_param("HealthCheckPort")
@ -196,7 +196,8 @@ class ELBV2Response(BaseResponse):
healthy_threshold_count = self._get_param("HealthyThresholdCount")
unhealthy_threshold_count = self._get_param("UnhealthyThresholdCount")
matcher = params.get("Matcher")
target_type = params.get("TargetType")
target_type = params.get("TargetType", "instance")
ip_address_type = params.get("IpAddressType")
tags = params.get("Tags")
target_group = self.elbv2_backend.create_target_group(
@ -214,6 +215,7 @@ class ELBV2Response(BaseResponse):
healthy_threshold_count=healthy_threshold_count,
unhealthy_threshold_count=unhealthy_threshold_count,
matcher=matcher,
ip_address_type=ip_address_type,
target_type=target_type,
tags=tags,
)
@ -797,6 +799,9 @@ CREATE_TARGET_GROUP_TEMPLATE = """<CreateTargetGroupResponse xmlns="http://elast
<TargetGroupName>{{ target_group.name }}</TargetGroupName>
{% if target_group.protocol %}
<Protocol>{{ target_group.protocol }}</Protocol>
{% if target_group.protocol_version %}
<ProtocolVersion>{{ target_group.protocol_version }}</ProtocolVersion>
{% endif %}
{% endif %}
{% if target_group.port %}
<Port>{{ target_group.port }}</Port>
@ -804,14 +809,22 @@ CREATE_TARGET_GROUP_TEMPLATE = """<CreateTargetGroupResponse xmlns="http://elast
{% if target_group.vpc_id %}
<VpcId>{{ target_group.vpc_id }}</VpcId>
{% endif %}
<HealthCheckProtocol>{{ target_group.healthcheck_protocol }}</HealthCheckProtocol>
{% if target_group.healthcheck_port %}<HealthCheckPort>{{ target_group.healthcheck_port }}</HealthCheckPort>{% endif %}
{% if target_group.healthcheck_enabled %}
{% if target_group.healthcheck_port %}
<HealthCheckPort>{{ target_group.healthcheck_port }}</HealthCheckPort>
{% endif %}
{% if target_group.healthcheck_protocol %}
<HealthCheckProtocol>{{ target_group.healthcheck_protocol or "None" }}</HealthCheckProtocol>
{% endif %}
{% endif %}
{% if target_group.healthcheck_path %}
<HealthCheckPath>{{ target_group.healthcheck_path or '' }}</HealthCheckPath>
{% endif %}
<HealthCheckIntervalSeconds>{{ target_group.healthcheck_interval_seconds }}</HealthCheckIntervalSeconds>
<HealthCheckTimeoutSeconds>{{ target_group.healthcheck_timeout_seconds }}</HealthCheckTimeoutSeconds>
<HealthCheckEnabled>{{ target_group.healthcheck_enabled and 'true' or 'false' }}</HealthCheckEnabled>
<HealthyThresholdCount>{{ target_group.healthy_threshold_count }}</HealthyThresholdCount>
<UnhealthyThresholdCount>{{ target_group.unhealthy_threshold_count }}</UnhealthyThresholdCount>
<HealthCheckEnabled>{{ target_group.healthcheck_enabled and 'true' or 'false' }}</HealthCheckEnabled>
{% if target_group.matcher %}
<Matcher>
<HttpCode>{{ target_group.matcher['HttpCode'] }}</HttpCode>
@ -820,6 +833,9 @@ CREATE_TARGET_GROUP_TEMPLATE = """<CreateTargetGroupResponse xmlns="http://elast
{% if target_group.target_type %}
<TargetType>{{ target_group.target_type }}</TargetType>
{% endif %}
{% if target_group.ip_address_type %}
<IpAddressType>{{ target_group.ip_address_type }}</IpAddressType>
{% endif %}
</member>
</TargetGroups>
</CreateTargetGroupResult>

2
tests/test_autoscaling/test_elbv2.py
View File

@ -27,7 +27,7 @@ class TestAutoscalignELBv2(unittest.TestCase):
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},

34
tests/test_elbv2/test_elbv2.py
View File

@ -266,7 +266,7 @@ def test_create_listeners_without_port():
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -445,7 +445,7 @@ def test_register_targets():
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -551,7 +551,7 @@ def test_stopped_instance_target():
HealthCheckProtocol="HTTP",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -580,7 +580,7 @@ def test_stopped_instance_target():
target_health_description = response["TargetHealthDescriptions"][0]
assert target_health_description["Target"] == target_dict
assert target_health_description["HealthCheckPort"] == str(target_group_port)
assert target_health_description["HealthCheckPort"] == "traffic-port"
assert target_health_description["TargetHealth"] == {"State": "healthy"}
instance.stop()
@ -591,7 +591,7 @@ def test_stopped_instance_target():
assert len(response["TargetHealthDescriptions"]) == 1
target_health_description = response["TargetHealthDescriptions"][0]
assert target_health_description["Target"] == target_dict
assert target_health_description["HealthCheckPort"] == str(target_group_port)
assert target_health_description["HealthCheckPort"] == "traffic-port"
assert target_health_description["TargetHealth"] == {
"State": "unused",
"Reason": "Target.InvalidState",
@ -634,7 +634,7 @@ def test_terminated_instance_target():
HealthCheckProtocol="HTTP",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -663,7 +663,7 @@ def test_terminated_instance_target():
target_health_description = response["TargetHealthDescriptions"][0]
assert target_health_description["Target"] == target_dict
assert target_health_description["HealthCheckPort"] == str(target_group_port)
assert target_health_description["HealthCheckPort"] == "traffic-port"
assert target_health_description["TargetHealth"] == {"State": "healthy"}
instance.terminate()
@ -772,7 +772,7 @@ def test_handle_listener_rules():
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -1323,7 +1323,7 @@ def test_modify_listener_http_to_https():
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -1523,7 +1523,9 @@ def test_add_listener_certificate():
load_balancer_arn = response["LoadBalancers"][0]["LoadBalancerArn"]
response = client.create_target_group(Name="a-target", Protocol="HTTPS", Port=8443)
response = client.create_target_group(
Name="a-target", Protocol="HTTPS", Port=8443, VpcId=vpc.id
)
target_group_arn = response["TargetGroups"][0]["TargetGroupArn"]
# HTTPS listener
@ -1564,10 +1566,12 @@ def test_add_listener_certificate():
@mock_elbv2
@mock_ec2
def test_forward_config_action():
response, _, _, _, _, conn = create_load_balancer()
response, vpc, _, _, _, conn = create_load_balancer()
load_balancer_arn = response["LoadBalancers"][0]["LoadBalancerArn"]
response = conn.create_target_group(Name="a-target", Protocol="HTTPS", Port=8443)
response = conn.create_target_group(
Name="a-target", Protocol="HTTPS", Port=8443, VpcId=vpc.id
)
target_group_arn = response["TargetGroups"][0]["TargetGroupArn"]
action = {
@ -1600,10 +1604,12 @@ def test_forward_config_action():
@mock_elbv2
@mock_ec2
def test_forward_config_action__with_stickiness():
response, _, _, _, _, conn = create_load_balancer()
response, vpc, _, _, _, conn = create_load_balancer()
load_balancer_arn = response["LoadBalancers"][0]["LoadBalancerArn"]
response = conn.create_target_group(Name="a-target", Protocol="HTTPS", Port=8443)
response = conn.create_target_group(
Name="a-target", Protocol="HTTPS", Port=8443, VpcId=vpc.id
)
target_group_arn = response["TargetGroups"][0]["TargetGroupArn"]
action = {

119
tests/test_elbv2/test_elbv2_target_groups.py
View File

@ -23,7 +23,7 @@ def test_create_target_group_with_invalid_healthcheck_protocol():
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -50,7 +50,7 @@ def test_create_target_group_with_tags():
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -93,7 +93,7 @@ def test_create_target_group_and_listeners():
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -245,7 +245,7 @@ def test_create_invalid_target_group_long_name():
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -277,7 +277,7 @@ def test_create_invalid_target_group_invalid_characters(name):
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -309,7 +309,7 @@ def test_create_invalid_target_group_alphanumeric_characters(name):
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -340,7 +340,7 @@ def test_create_valid_target_group_valid_names(name):
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -361,7 +361,7 @@ def test_target_group_attributes():
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -433,7 +433,7 @@ def test_create_target_group_invalid_protocol():
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -474,7 +474,7 @@ def test_describe_target_groups_no_arguments():
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "201"},
@ -502,7 +502,7 @@ def test_modify_target_group():
HealthCheckPort="8080",
HealthCheckPath="/",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthCheckTimeoutSeconds=3,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"},
@ -515,7 +515,7 @@ def test_modify_target_group():
HealthCheckPort="8081",
HealthCheckPath="/status",
HealthCheckIntervalSeconds=10,
HealthCheckTimeoutSeconds=10,
HealthCheckTimeoutSeconds=8,
HealthyThresholdCount=10,
UnhealthyThresholdCount=4,
Matcher={"HttpCode": "200-399"},
@ -527,7 +527,7 @@ def test_modify_target_group():
assert response["TargetGroups"][0]["HealthCheckPath"] == "/status"
assert response["TargetGroups"][0]["HealthCheckPort"] == "8081"
assert response["TargetGroups"][0]["HealthCheckProtocol"] == "HTTPS"
assert response["TargetGroups"][0]["HealthCheckTimeoutSeconds"] == 10
assert response["TargetGroups"][0]["HealthCheckTimeoutSeconds"] == 8
assert response["TargetGroups"][0]["HealthyThresholdCount"] == 10
assert response["TargetGroups"][0]["Protocol"] == "HTTP"
assert response["TargetGroups"][0]["ProtocolVersion"] == "HTTP1"
@ -538,23 +538,35 @@ def test_modify_target_group():
@mock_ec2
@pytest.mark.parametrize("target_type", ["instance", "ip", "lambda", "alb", "other"])
def test_create_target_group_with_target_type(target_type):
response, _, _, _, _, conn = create_load_balancer()
response, vpc, _, _, _, conn = create_load_balancer()
response = conn.create_target_group(Name="a-target", TargetType=target_type)
args = {
"Name": "a-target",
"TargetType": target_type,
}
if target_type != "lambda":
args["Protocol"] = "HTTP"
args["Port"] = 80
args["VpcId"] = vpc.id
response = conn.create_target_group(**args)
group = response["TargetGroups"][0]
assert "TargetGroupArn" in group
assert group["TargetGroupName"] == "a-target"
assert group["TargetType"] == target_type
assert "Protocol" not in group
assert "VpcId" not in group
if target_type != "lambda":
assert "Protocol" in group
assert "VpcId" in group
group = conn.describe_target_groups()["TargetGroups"][0]
assert "TargetGroupArn" in group
assert group["TargetGroupName"] == "a-target"
assert group["TargetType"] == target_type
assert "Protocol" not in group
assert "VpcId" not in group
if target_type != "lambda":
assert "Protocol" in group
assert "VpcId" in group
@mock_elbv2
@ -721,3 +733,72 @@ def test_delete_target_group_while_listener_still_exists():
# Deletion does succeed now that the listener is deleted
client.delete_target_group(TargetGroupArn=target_group_arn1)
@mock_elbv2
def test_create_target_group_validation_error():
elbv2 = boto3.client("elbv2", region_name="us-east-1")
with pytest.raises(ClientError) as ex:
elbv2.create_target_group(
Name="a-target",
Protocol="HTTP",
)
err = ex.value.response["Error"]
assert err["Code"] == "ValidationError"
assert err["Message"] == "A port must be specified"
with pytest.raises(ClientError) as ex:
elbv2.create_target_group(
Name="a-target",
Protocol="HTTP",
)
err = ex.value.response["Error"]
assert err["Code"] == "ValidationError"
assert err["Message"] == "A port must be specified"
with pytest.raises(ClientError) as ex:
elbv2.create_target_group(
Name="a-target",
Protocol="HTTP",
Port=8080,
)
err = ex.value.response["Error"]
assert err["Code"] == "ValidationError"
assert err["Message"] == "A VPC ID must be specified"
with pytest.raises(ClientError) as ex:
elbv2.create_target_group(
Name="a-target",
Protocol="HTTP",
Port=8080,
VpcId="non-existing",
)
err = ex.value.response["Error"]
assert err["Code"] == "ValidationError"
assert err["Message"] == "The VPC ID 'non-existing' is not found"
with pytest.raises(ClientError) as ex:
elbv2.create_target_group(
Name="a-target",
TargetType="lambda",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
)
err = ex.value.response["Error"]
assert err["Code"] == "ValidationError"
assert (
err["Message"]
== "Health check timeout '5' must be smaller than the interval '5'"
)
with pytest.raises(ClientError) as ex:
elbv2.create_target_group(
Name="a-target",
TargetType="lambda",
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=6,
)
err = ex.value.response["Error"]
assert err["Code"] == "ValidationError"
assert err["Message"] == "Health check interval must be greater than the timeout."

2
tests/test_sqs/test_sqs.py
View File

@ -3176,8 +3176,6 @@ def test_message_delay_is_more_than_15_minutes():
assert sorted([entry["Id"] for entry in response["Failed"]]) == ["id_2"]
# print(response)
time.sleep(4)
response = client.receive_message(