1932 lines
64 KiB
Python
1932 lines
64 KiB
Python
import copy
|
|
import os
|
|
import boto3
|
|
from botocore.exceptions import ClientError
|
|
import pytest
|
|
import sure # noqa # pylint: disable=unused-import
|
|
|
|
from moto import mock_elbv2, mock_ec2, mock_acm
|
|
from moto.elbv2 import elbv2_backends
|
|
from moto.core import ACCOUNT_ID
|
|
from tests import EXAMPLE_AMI_ID
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_create_load_balancer():
|
|
response, _, security_group, subnet1, subnet2, conn = create_load_balancer()
|
|
|
|
lb = response.get("LoadBalancers")[0]
|
|
lb.get("DNSName").should.equal("my-lb-1.us-east-1.elb.amazonaws.com")
|
|
lb.get("LoadBalancerArn").should.equal(
|
|
f"arn:aws:elasticloadbalancing:us-east-1:{ACCOUNT_ID}:loadbalancer/app/my-lb/50dc6c495c0c9188"
|
|
)
|
|
lb.get("SecurityGroups").should.equal([security_group.id])
|
|
lb.get("AvailabilityZones").should.equal(
|
|
[
|
|
{"SubnetId": subnet1.id, "ZoneName": "us-east-1a"},
|
|
{"SubnetId": subnet2.id, "ZoneName": "us-east-1b"},
|
|
]
|
|
)
|
|
lb.get("CreatedTime").tzinfo.should_not.be.none
|
|
lb.get("State").get("Code").should.equal("provisioning")
|
|
lb_arn = lb.get("LoadBalancerArn")
|
|
|
|
# Ensure the tags persisted
|
|
tag_desc = conn.describe_tags(ResourceArns=[lb_arn])["TagDescriptions"][0]
|
|
tag_desc.should.have.key("ResourceArn").equals(lb_arn)
|
|
tags = {d["Key"]: d["Value"] for d in tag_desc["Tags"]}
|
|
tags.should.equal({"key_name": "a_value"})
|
|
|
|
|
|
def create_load_balancer():
|
|
conn = boto3.client("elbv2", region_name="us-east-1")
|
|
ec2 = boto3.resource("ec2", region_name="us-east-1")
|
|
|
|
security_group = ec2.create_security_group(
|
|
GroupName="a-security-group", Description="First One"
|
|
)
|
|
vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
|
|
subnet1 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.192/26", AvailabilityZone="us-east-1a"
|
|
)
|
|
subnet2 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.0/26", AvailabilityZone="us-east-1b"
|
|
)
|
|
|
|
response = conn.create_load_balancer(
|
|
Name="my-lb",
|
|
Subnets=[subnet1.id, subnet2.id],
|
|
SecurityGroups=[security_group.id],
|
|
Scheme="internal",
|
|
Tags=[{"Key": "key_name", "Value": "a_value"}],
|
|
)
|
|
return response, vpc, security_group, subnet1, subnet2, conn
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_create_elb_using_subnetmapping():
|
|
region = "us-west-1"
|
|
conn = boto3.client("elbv2", region_name=region)
|
|
ec2 = boto3.resource("ec2", region_name=region)
|
|
|
|
security_group = ec2.create_security_group(
|
|
GroupName="a-security-group", Description="First One"
|
|
)
|
|
vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
|
|
subnet1 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.0/26", AvailabilityZone=region + "a"
|
|
)
|
|
subnet2 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.192/26", AvailabilityZone=region + "b"
|
|
)
|
|
|
|
conn.create_load_balancer(
|
|
Name="my-lb",
|
|
SubnetMappings=[{"SubnetId": subnet1.id}, {"SubnetId": subnet2.id}],
|
|
SecurityGroups=[security_group.id],
|
|
Scheme="internal",
|
|
Tags=[{"Key": "key_name", "Value": "a_value"}],
|
|
)
|
|
|
|
lb = conn.describe_load_balancers()["LoadBalancers"][0]
|
|
lb.should.have.key("AvailabilityZones").length_of(2)
|
|
lb["AvailabilityZones"].should.contain(
|
|
{"ZoneName": "us-west-1a", "SubnetId": subnet1.id}
|
|
)
|
|
lb["AvailabilityZones"].should.contain(
|
|
{"ZoneName": "us-west-1b", "SubnetId": subnet2.id}
|
|
)
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_describe_load_balancers():
|
|
response, _, _, _, _, conn = create_load_balancer()
|
|
|
|
response = conn.describe_load_balancers()
|
|
|
|
response.get("LoadBalancers").should.have.length_of(1)
|
|
lb = response.get("LoadBalancers")[0]
|
|
lb.get("LoadBalancerName").should.equal("my-lb")
|
|
lb.get("State").get("Code").should.equal("active")
|
|
|
|
response = conn.describe_load_balancers(
|
|
LoadBalancerArns=[lb.get("LoadBalancerArn")]
|
|
)
|
|
response.get("LoadBalancers")[0].get("LoadBalancerName").should.equal("my-lb")
|
|
|
|
response = conn.describe_load_balancers(Names=["my-lb"])
|
|
response.get("LoadBalancers")[0].get("LoadBalancerName").should.equal("my-lb")
|
|
|
|
with pytest.raises(ClientError):
|
|
conn.describe_load_balancers(LoadBalancerArns=["not-a/real/arn"])
|
|
with pytest.raises(ClientError):
|
|
conn.describe_load_balancers(Names=["nope"])
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_describe_listeners():
|
|
conn = boto3.client("elbv2", region_name="us-east-1")
|
|
|
|
with pytest.raises(ClientError) as exc:
|
|
conn.describe_listeners()
|
|
err = exc.value.response["Error"]
|
|
err["Code"].should.equal("ValidationError")
|
|
err["Message"].should.equal(
|
|
"You must specify either listener ARNs or a load balancer ARN"
|
|
)
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_add_remove_tags():
|
|
_, _, _, _, _, conn = create_load_balancer()
|
|
|
|
lbs = conn.describe_load_balancers()["LoadBalancers"]
|
|
lbs.should.have.length_of(1)
|
|
lb = lbs[0]
|
|
|
|
with pytest.raises(ClientError):
|
|
conn.add_tags(ResourceArns=["missing-arn"], Tags=[{"Key": "a", "Value": "b"}])
|
|
|
|
conn.add_tags(
|
|
ResourceArns=[lb.get("LoadBalancerArn")], Tags=[{"Key": "a", "Value": "b"}]
|
|
)
|
|
|
|
tags = {
|
|
d["Key"]: d["Value"]
|
|
for d in conn.describe_tags(ResourceArns=[lb.get("LoadBalancerArn")])[
|
|
"TagDescriptions"
|
|
][0]["Tags"]
|
|
}
|
|
tags.should.have.key("a").which.should.equal("b")
|
|
|
|
conn.add_tags(
|
|
ResourceArns=[lb.get("LoadBalancerArn")],
|
|
Tags=[
|
|
{"Key": "a", "Value": "b"},
|
|
{"Key": "b", "Value": "b"},
|
|
{"Key": "c", "Value": "b"},
|
|
{"Key": "d", "Value": "b"},
|
|
{"Key": "e", "Value": "b"},
|
|
{"Key": "f", "Value": "b"},
|
|
{"Key": "g", "Value": "b"},
|
|
{"Key": "h", "Value": "b"},
|
|
{"Key": "j", "Value": "b"},
|
|
],
|
|
)
|
|
|
|
with pytest.raises(ClientError) as exc:
|
|
conn.add_tags(
|
|
ResourceArns=[lb.get("LoadBalancerArn")], Tags=[{"Key": "k", "Value": "b"}]
|
|
)
|
|
err = exc.value.response["Error"]
|
|
err["Code"].should.equal("TooManyTagsError")
|
|
|
|
conn.add_tags(
|
|
ResourceArns=[lb.get("LoadBalancerArn")], Tags=[{"Key": "j", "Value": "c"}]
|
|
)
|
|
|
|
tags = {
|
|
d["Key"]: d["Value"]
|
|
for d in conn.describe_tags(ResourceArns=[lb.get("LoadBalancerArn")])[
|
|
"TagDescriptions"
|
|
][0]["Tags"]
|
|
}
|
|
|
|
tags.should.have.key("a").which.should.equal("b")
|
|
tags.should.have.key("b").which.should.equal("b")
|
|
tags.should.have.key("c").which.should.equal("b")
|
|
tags.should.have.key("d").which.should.equal("b")
|
|
tags.should.have.key("e").which.should.equal("b")
|
|
tags.should.have.key("f").which.should.equal("b")
|
|
tags.should.have.key("g").which.should.equal("b")
|
|
tags.should.have.key("h").which.should.equal("b")
|
|
tags.should.have.key("j").which.should.equal("c")
|
|
tags.shouldnt.have.key("k")
|
|
|
|
conn.remove_tags(ResourceArns=[lb.get("LoadBalancerArn")], TagKeys=["a"])
|
|
|
|
tags = {
|
|
d["Key"]: d["Value"]
|
|
for d in conn.describe_tags(ResourceArns=[lb.get("LoadBalancerArn")])[
|
|
"TagDescriptions"
|
|
][0]["Tags"]
|
|
}
|
|
|
|
tags.shouldnt.have.key("a")
|
|
tags.should.have.key("b").which.should.equal("b")
|
|
tags.should.have.key("c").which.should.equal("b")
|
|
tags.should.have.key("d").which.should.equal("b")
|
|
tags.should.have.key("e").which.should.equal("b")
|
|
tags.should.have.key("f").which.should.equal("b")
|
|
tags.should.have.key("g").which.should.equal("b")
|
|
tags.should.have.key("h").which.should.equal("b")
|
|
tags.should.have.key("j").which.should.equal("c")
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_create_elb_in_multiple_region():
|
|
for region in ["us-west-1", "us-west-2"]:
|
|
conn = boto3.client("elbv2", region_name=region)
|
|
ec2 = boto3.resource("ec2", region_name=region)
|
|
|
|
security_group = ec2.create_security_group(
|
|
GroupName="a-security-group", Description="First One"
|
|
)
|
|
vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
|
|
subnet1 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.0/26", AvailabilityZone=region + "a"
|
|
)
|
|
subnet2 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.192/26", AvailabilityZone=region + "b"
|
|
)
|
|
|
|
conn.create_load_balancer(
|
|
Name="my-lb",
|
|
Subnets=[subnet1.id, subnet2.id],
|
|
SecurityGroups=[security_group.id],
|
|
Scheme="internal",
|
|
Tags=[{"Key": "key_name", "Value": "a_value"}],
|
|
)
|
|
|
|
list(
|
|
boto3.client("elbv2", region_name="us-west-1")
|
|
.describe_load_balancers()
|
|
.get("LoadBalancers")
|
|
).should.have.length_of(1)
|
|
list(
|
|
boto3.client("elbv2", region_name="us-west-2")
|
|
.describe_load_balancers()
|
|
.get("LoadBalancers")
|
|
).should.have.length_of(1)
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_create_listeners_without_port():
|
|
response, vpc, _, _, _, conn = create_load_balancer()
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
response = conn.create_target_group(
|
|
Name="a-target",
|
|
Protocol="HTTP",
|
|
Port=8080,
|
|
VpcId=vpc.id,
|
|
HealthCheckProtocol="HTTP",
|
|
HealthCheckPort="8080",
|
|
HealthCheckPath="/",
|
|
HealthCheckIntervalSeconds=5,
|
|
HealthCheckTimeoutSeconds=5,
|
|
HealthyThresholdCount=5,
|
|
UnhealthyThresholdCount=2,
|
|
Matcher={"HttpCode": "200"},
|
|
)
|
|
target_group = response.get("TargetGroups")[0]
|
|
target_group_arn = target_group["TargetGroupArn"]
|
|
response = conn.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTP",
|
|
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
|
|
)
|
|
|
|
listener = response.get("Listeners")[0]
|
|
listener.get("Port").should.equal(None)
|
|
listener.get("Protocol").should.equal("HTTP")
|
|
listener.get("DefaultActions").should.equal(
|
|
[{"TargetGroupArn": target_group_arn, "Type": "forward"}]
|
|
)
|
|
|
|
|
|
@mock_ec2
|
|
@mock_elbv2
|
|
def test_create_rule_forward_config_as_second_arg():
|
|
# https://github.com/spulec/moto/issues/4123
|
|
# Necessary because there was some convoluted way of parsing arguments
|
|
# Actions with type=forward had to be the first action specified
|
|
response, vpc, _, _, _, elbv2 = create_load_balancer()
|
|
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
response = elbv2.create_listener(
|
|
LoadBalancerArn=load_balancer_arn, Protocol="HTTP", Port=80, DefaultActions=[]
|
|
)
|
|
http_listener_arn = response.get("Listeners")[0]["ListenerArn"]
|
|
|
|
priority = 100
|
|
|
|
response = elbv2.create_target_group(
|
|
Name="a-target",
|
|
Protocol="HTTP",
|
|
Port=8080,
|
|
VpcId=vpc.id,
|
|
HealthCheckProtocol="HTTP",
|
|
HealthCheckPort="8080",
|
|
HealthCheckPath="/",
|
|
Matcher={"HttpCode": "200"},
|
|
)
|
|
target_group = response.get("TargetGroups")[0]
|
|
|
|
# No targets registered yet
|
|
target_group_arn = target_group.get("TargetGroupArn")
|
|
elbv2.create_rule(
|
|
ListenerArn=http_listener_arn,
|
|
Conditions=[
|
|
{"Field": "path-pattern", "PathPatternConfig": {"Values": ["/sth*"]}}
|
|
],
|
|
Priority=priority,
|
|
Actions=[
|
|
{
|
|
"Type": "authenticate-cognito",
|
|
"Order": 1,
|
|
"AuthenticateCognitoConfig": {
|
|
"UserPoolArn": "?1",
|
|
"UserPoolClientId": "?2",
|
|
"UserPoolDomain": "?2",
|
|
"SessionCookieName": "AWSELBAuthSessionCookie",
|
|
"Scope": "openid",
|
|
"SessionTimeout": 604800,
|
|
"OnUnauthenticatedRequest": "authenticate",
|
|
},
|
|
},
|
|
{
|
|
"Type": "forward",
|
|
"Order": 2,
|
|
"ForwardConfig": {
|
|
"TargetGroups": [
|
|
{"TargetGroupArn": target_group_arn, "Weight": 1},
|
|
],
|
|
"TargetGroupStickinessConfig": {"Enabled": False},
|
|
},
|
|
},
|
|
],
|
|
)
|
|
all_rules = elbv2.describe_rules(ListenerArn=http_listener_arn)["Rules"]
|
|
our_rule = all_rules[0]
|
|
actions = our_rule["Actions"]
|
|
forward_action = [a for a in actions if "ForwardConfig" in a.keys()][0]
|
|
forward_action.should.equal(
|
|
{
|
|
"ForwardConfig": {
|
|
"TargetGroups": [{"TargetGroupArn": target_group_arn, "Weight": 1}],
|
|
"TargetGroupStickinessConfig": {"Enabled": False},
|
|
},
|
|
"Type": "forward",
|
|
"Order": 2,
|
|
}
|
|
)
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_describe_paginated_balancers():
|
|
conn = boto3.client("elbv2", region_name="us-east-1")
|
|
ec2 = boto3.resource("ec2", region_name="us-east-1")
|
|
|
|
security_group = ec2.create_security_group(
|
|
GroupName="a-security-group", Description="First One"
|
|
)
|
|
vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
|
|
subnet1 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.192/26", AvailabilityZone="us-east-1a"
|
|
)
|
|
subnet2 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.0/26", AvailabilityZone="us-east-1b"
|
|
)
|
|
|
|
for i in range(51):
|
|
conn.create_load_balancer(
|
|
Name="my-lb%d" % i,
|
|
Subnets=[subnet1.id, subnet2.id],
|
|
SecurityGroups=[security_group.id],
|
|
Scheme="internal",
|
|
Tags=[{"Key": "key_name", "Value": "a_value"}],
|
|
)
|
|
|
|
resp = conn.describe_load_balancers()
|
|
resp["LoadBalancers"].should.have.length_of(50)
|
|
resp["NextMarker"].should.equal(resp["LoadBalancers"][-1]["LoadBalancerName"])
|
|
resp2 = conn.describe_load_balancers(Marker=resp["NextMarker"])
|
|
resp2["LoadBalancers"].should.have.length_of(1)
|
|
assert "NextToken" not in resp2.keys()
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_delete_load_balancer():
|
|
response, _, _, _, _, conn = create_load_balancer()
|
|
|
|
response.get("LoadBalancers").should.have.length_of(1)
|
|
lb = response.get("LoadBalancers")[0]
|
|
|
|
conn.delete_load_balancer(LoadBalancerArn=lb.get("LoadBalancerArn"))
|
|
balancers = conn.describe_load_balancers().get("LoadBalancers")
|
|
balancers.should.have.length_of(0)
|
|
|
|
|
|
@mock_ec2
|
|
@mock_elbv2
|
|
def test_register_targets():
|
|
conn = boto3.client("elbv2", region_name="us-east-1")
|
|
ec2 = boto3.resource("ec2", region_name="us-east-1")
|
|
|
|
security_group = ec2.create_security_group(
|
|
GroupName="a-security-group", Description="First One"
|
|
)
|
|
vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
|
|
subnet1 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.192/26", AvailabilityZone="us-east-1a"
|
|
)
|
|
subnet2 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.0/26", AvailabilityZone="us-east-1b"
|
|
)
|
|
|
|
conn.create_load_balancer(
|
|
Name="my-lb",
|
|
Subnets=[subnet1.id, subnet2.id],
|
|
SecurityGroups=[security_group.id],
|
|
Scheme="internal",
|
|
Tags=[{"Key": "key_name", "Value": "a_value"}],
|
|
)
|
|
|
|
response = conn.create_target_group(
|
|
Name="a-target",
|
|
Protocol="HTTP",
|
|
Port=8080,
|
|
VpcId=vpc.id,
|
|
HealthCheckProtocol="HTTP",
|
|
HealthCheckPort="8080",
|
|
HealthCheckPath="/",
|
|
HealthCheckIntervalSeconds=5,
|
|
HealthCheckTimeoutSeconds=5,
|
|
HealthyThresholdCount=5,
|
|
UnhealthyThresholdCount=2,
|
|
Matcher={"HttpCode": "200"},
|
|
)
|
|
target_group = response.get("TargetGroups")[0]
|
|
|
|
# No targets registered yet
|
|
response = conn.describe_target_health(
|
|
TargetGroupArn=target_group.get("TargetGroupArn")
|
|
)
|
|
response.get("TargetHealthDescriptions").should.have.length_of(0)
|
|
|
|
response = ec2.create_instances(ImageId=EXAMPLE_AMI_ID, MinCount=2, MaxCount=2)
|
|
instance_id1 = response[0].id
|
|
instance_id2 = response[1].id
|
|
|
|
response = conn.register_targets(
|
|
TargetGroupArn=target_group.get("TargetGroupArn"),
|
|
Targets=[
|
|
{"Id": instance_id1, "Port": 5060},
|
|
{"Id": instance_id2, "Port": 4030},
|
|
],
|
|
)
|
|
|
|
response = conn.describe_target_health(
|
|
TargetGroupArn=target_group.get("TargetGroupArn")
|
|
)
|
|
response.get("TargetHealthDescriptions").should.have.length_of(2)
|
|
|
|
response = conn.deregister_targets(
|
|
TargetGroupArn=target_group.get("TargetGroupArn"),
|
|
Targets=[{"Id": instance_id2}],
|
|
)
|
|
|
|
response = conn.describe_target_health(
|
|
TargetGroupArn=target_group.get("TargetGroupArn")
|
|
)
|
|
response.get("TargetHealthDescriptions").should.have.length_of(1)
|
|
|
|
|
|
@mock_ec2
|
|
@mock_elbv2
|
|
def test_stopped_instance_target():
|
|
target_group_port = 8080
|
|
|
|
conn = boto3.client("elbv2", region_name="us-east-1")
|
|
ec2 = boto3.resource("ec2", region_name="us-east-1")
|
|
|
|
security_group = ec2.create_security_group(
|
|
GroupName="a-security-group", Description="First One"
|
|
)
|
|
vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
|
|
subnet1 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.192/26", AvailabilityZone="us-east-1a"
|
|
)
|
|
subnet2 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.0/26", AvailabilityZone="us-east-1b"
|
|
)
|
|
|
|
conn.create_load_balancer(
|
|
Name="my-lb",
|
|
Subnets=[subnet1.id, subnet2.id],
|
|
SecurityGroups=[security_group.id],
|
|
Scheme="internal",
|
|
Tags=[{"Key": "key_name", "Value": "a_value"}],
|
|
)
|
|
|
|
response = conn.create_target_group(
|
|
Name="a-target",
|
|
Protocol="HTTP",
|
|
Port=target_group_port,
|
|
VpcId=vpc.id,
|
|
HealthCheckProtocol="HTTP",
|
|
HealthCheckPath="/",
|
|
HealthCheckIntervalSeconds=5,
|
|
HealthCheckTimeoutSeconds=5,
|
|
HealthyThresholdCount=5,
|
|
UnhealthyThresholdCount=2,
|
|
Matcher={"HttpCode": "200"},
|
|
)
|
|
target_group = response.get("TargetGroups")[0]
|
|
|
|
# No targets registered yet
|
|
response = conn.describe_target_health(
|
|
TargetGroupArn=target_group.get("TargetGroupArn")
|
|
)
|
|
response.get("TargetHealthDescriptions").should.have.length_of(0)
|
|
|
|
response = ec2.create_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1)
|
|
instance = response[0]
|
|
|
|
target_dict = {"Id": instance.id, "Port": 500}
|
|
|
|
response = conn.register_targets(
|
|
TargetGroupArn=target_group.get("TargetGroupArn"), Targets=[target_dict]
|
|
)
|
|
|
|
response = conn.describe_target_health(
|
|
TargetGroupArn=target_group.get("TargetGroupArn")
|
|
)
|
|
response.get("TargetHealthDescriptions").should.have.length_of(1)
|
|
target_health_description = response.get("TargetHealthDescriptions")[0]
|
|
|
|
target_health_description["Target"].should.equal(target_dict)
|
|
target_health_description["HealthCheckPort"].should.equal(str(target_group_port))
|
|
target_health_description["TargetHealth"].should.equal({"State": "healthy"})
|
|
|
|
instance.stop()
|
|
|
|
response = conn.describe_target_health(
|
|
TargetGroupArn=target_group.get("TargetGroupArn")
|
|
)
|
|
response.get("TargetHealthDescriptions").should.have.length_of(1)
|
|
target_health_description = response.get("TargetHealthDescriptions")[0]
|
|
target_health_description["Target"].should.equal(target_dict)
|
|
target_health_description["HealthCheckPort"].should.equal(str(target_group_port))
|
|
target_health_description["TargetHealth"].should.equal(
|
|
{
|
|
"State": "unused",
|
|
"Reason": "Target.InvalidState",
|
|
"Description": "Target is in the stopped state",
|
|
}
|
|
)
|
|
|
|
|
|
@mock_ec2
|
|
@mock_elbv2
|
|
def test_terminated_instance_target():
|
|
target_group_port = 8080
|
|
|
|
conn = boto3.client("elbv2", region_name="us-east-1")
|
|
ec2 = boto3.resource("ec2", region_name="us-east-1")
|
|
|
|
security_group = ec2.create_security_group(
|
|
GroupName="a-security-group", Description="First One"
|
|
)
|
|
vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
|
|
subnet1 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.192/26", AvailabilityZone="us-east-1a"
|
|
)
|
|
subnet2 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.0/26", AvailabilityZone="us-east-1b"
|
|
)
|
|
|
|
conn.create_load_balancer(
|
|
Name="my-lb",
|
|
Subnets=[subnet1.id, subnet2.id],
|
|
SecurityGroups=[security_group.id],
|
|
Scheme="internal",
|
|
Tags=[{"Key": "key_name", "Value": "a_value"}],
|
|
)
|
|
|
|
response = conn.create_target_group(
|
|
Name="a-target",
|
|
Protocol="HTTP",
|
|
Port=target_group_port,
|
|
VpcId=vpc.id,
|
|
HealthCheckProtocol="HTTP",
|
|
HealthCheckPath="/",
|
|
HealthCheckIntervalSeconds=5,
|
|
HealthCheckTimeoutSeconds=5,
|
|
HealthyThresholdCount=5,
|
|
UnhealthyThresholdCount=2,
|
|
Matcher={"HttpCode": "200"},
|
|
)
|
|
target_group = response.get("TargetGroups")[0]
|
|
|
|
# No targets registered yet
|
|
response = conn.describe_target_health(
|
|
TargetGroupArn=target_group.get("TargetGroupArn")
|
|
)
|
|
response.get("TargetHealthDescriptions").should.have.length_of(0)
|
|
|
|
response = ec2.create_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1)
|
|
instance = response[0]
|
|
|
|
target_dict = {"Id": instance.id, "Port": 500}
|
|
|
|
response = conn.register_targets(
|
|
TargetGroupArn=target_group.get("TargetGroupArn"), Targets=[target_dict]
|
|
)
|
|
|
|
response = conn.describe_target_health(
|
|
TargetGroupArn=target_group.get("TargetGroupArn")
|
|
)
|
|
response.get("TargetHealthDescriptions").should.have.length_of(1)
|
|
target_health_description = response.get("TargetHealthDescriptions")[0]
|
|
|
|
target_health_description["Target"].should.equal(target_dict)
|
|
target_health_description["HealthCheckPort"].should.equal(str(target_group_port))
|
|
target_health_description["TargetHealth"].should.equal({"State": "healthy"})
|
|
|
|
instance.terminate()
|
|
|
|
response = conn.describe_target_health(
|
|
TargetGroupArn=target_group.get("TargetGroupArn")
|
|
)
|
|
response.get("TargetHealthDescriptions").should.have.length_of(0)
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_create_rule_priority_in_use():
|
|
response, _, _, _, _, elbv2 = create_load_balancer()
|
|
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
response = elbv2.create_listener(
|
|
LoadBalancerArn=load_balancer_arn, Protocol="HTTP", Port=80, DefaultActions=[]
|
|
)
|
|
http_listener_arn = response.get("Listeners")[0]["ListenerArn"]
|
|
|
|
priority = 100
|
|
elbv2.create_rule(
|
|
ListenerArn=http_listener_arn, Priority=priority, Conditions=[], Actions=[]
|
|
)
|
|
|
|
# test for PriorityInUse
|
|
with pytest.raises(ClientError) as ex:
|
|
elbv2.create_rule(
|
|
ListenerArn=http_listener_arn, Priority=priority, Conditions=[], Actions=[]
|
|
)
|
|
err = ex.value.response["Error"]
|
|
err["Code"].should.equal("PriorityInUse")
|
|
err["Message"].should.equal("The specified priority is in use.")
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_modify_rule_conditions():
|
|
response, _, _, _, _, elbv2 = create_load_balancer()
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
action = {
|
|
"Type": "redirect",
|
|
"RedirectConfig": {
|
|
"Protocol": "HTTPS",
|
|
"Port": "443",
|
|
"StatusCode": "HTTP_301",
|
|
},
|
|
}
|
|
condition = {"Field": "path-pattern", "PathPatternConfig": {"Values": ["/sth*"]}}
|
|
|
|
response = elbv2.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTP",
|
|
Port=80,
|
|
DefaultActions=[action],
|
|
)
|
|
http_listener_arn = response.get("Listeners")[0]["ListenerArn"]
|
|
|
|
response = elbv2.create_rule(
|
|
ListenerArn=http_listener_arn, Priority=100, Conditions=[], Actions=[]
|
|
)
|
|
rule = response["Rules"][0]
|
|
|
|
assert len(rule["Actions"]) == 0
|
|
assert len(rule["Conditions"]) == 0
|
|
|
|
response = elbv2.modify_rule(RuleArn=rule["RuleArn"], Actions=[action])
|
|
rule = response["Rules"][0]
|
|
|
|
assert len(rule["Actions"]) == 1
|
|
assert len(rule["Conditions"]) == 0
|
|
|
|
response = elbv2.modify_rule(RuleArn=rule["RuleArn"], Conditions=[condition])
|
|
rule = response["Rules"][0]
|
|
|
|
assert len(rule["Actions"]) == 1
|
|
assert len(rule["Conditions"]) == 1
|
|
|
|
response = elbv2.modify_rule(
|
|
RuleArn=rule["RuleArn"],
|
|
Conditions=[condition, condition],
|
|
Actions=[action, action],
|
|
)
|
|
rule = response["Rules"][0]
|
|
|
|
assert len(rule["Actions"]) == 2
|
|
assert len(rule["Conditions"]) == 2
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_handle_listener_rules():
|
|
response, vpc, _, _, _, conn = create_load_balancer()
|
|
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
response = conn.create_target_group(
|
|
Name="a-target",
|
|
Protocol="HTTP",
|
|
Port=8080,
|
|
VpcId=vpc.id,
|
|
HealthCheckProtocol="HTTP",
|
|
HealthCheckPort="8080",
|
|
HealthCheckPath="/",
|
|
HealthCheckIntervalSeconds=5,
|
|
HealthCheckTimeoutSeconds=5,
|
|
HealthyThresholdCount=5,
|
|
UnhealthyThresholdCount=2,
|
|
Matcher={"HttpCode": "200"},
|
|
)
|
|
target_group = response.get("TargetGroups")[0]
|
|
|
|
# Plain HTTP listener
|
|
response = conn.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTP",
|
|
Port=80,
|
|
DefaultActions=[
|
|
{"Type": "forward", "TargetGroupArn": target_group.get("TargetGroupArn")}
|
|
],
|
|
)
|
|
listener = response.get("Listeners")[0]
|
|
listener.get("Port").should.equal(80)
|
|
listener.get("Protocol").should.equal("HTTP")
|
|
listener.get("DefaultActions").should.equal(
|
|
[{"TargetGroupArn": target_group.get("TargetGroupArn"), "Type": "forward"}]
|
|
)
|
|
http_listener_arn = listener.get("ListenerArn")
|
|
|
|
# create first rule
|
|
priority = 100
|
|
host = "xxx.example.com"
|
|
path_pattern = "foobar"
|
|
pathpatternconfig_pattern = "foobar2"
|
|
created_rule = conn.create_rule(
|
|
ListenerArn=http_listener_arn,
|
|
Priority=priority,
|
|
Conditions=[
|
|
{"Field": "host-header", "Values": [host]},
|
|
{"Field": "path-pattern", "Values": [path_pattern]},
|
|
{
|
|
"Field": "path-pattern",
|
|
"PathPatternConfig": {"Values": [pathpatternconfig_pattern]},
|
|
},
|
|
],
|
|
Actions=[
|
|
{"TargetGroupArn": target_group.get("TargetGroupArn"), "Type": "forward"}
|
|
],
|
|
)
|
|
rule = created_rule.get("Rules")[0]
|
|
rule["Priority"].should.equal("100")
|
|
|
|
# check if rules is sorted by priority
|
|
priority = 500
|
|
host = "yyy.example.com"
|
|
path_pattern = "foobar"
|
|
rules = conn.create_rule(
|
|
ListenerArn=http_listener_arn,
|
|
Priority=priority,
|
|
Conditions=[
|
|
{"Field": "host-header", "Values": [host]},
|
|
{"Field": "path-pattern", "Values": [path_pattern]},
|
|
{
|
|
"Field": "path-pattern",
|
|
"PathPatternConfig": {"Values": [pathpatternconfig_pattern]},
|
|
},
|
|
],
|
|
Actions=[
|
|
{"TargetGroupArn": target_group.get("TargetGroupArn"), "Type": "forward"}
|
|
],
|
|
)
|
|
|
|
# add rule that uses forward_config
|
|
priority = 550
|
|
host = "aaa.example.com"
|
|
path_pattern = "barfoo"
|
|
rules = conn.create_rule(
|
|
ListenerArn=http_listener_arn,
|
|
Priority=priority,
|
|
Conditions=[
|
|
{"Field": "host-header", "Values": [host]},
|
|
{"Field": "path-pattern", "Values": [path_pattern]},
|
|
{
|
|
"Field": "path-pattern",
|
|
"PathPatternConfig": {"Values": [pathpatternconfig_pattern]},
|
|
},
|
|
],
|
|
Actions=[
|
|
{
|
|
"Type": "forward",
|
|
"ForwardConfig": {
|
|
"TargetGroups": [
|
|
{
|
|
"TargetGroupArn": target_group.get("TargetGroupArn"),
|
|
"Weight": 1,
|
|
},
|
|
{
|
|
"TargetGroupArn": target_group.get("TargetGroupArn"),
|
|
"Weight": 2,
|
|
},
|
|
]
|
|
},
|
|
},
|
|
],
|
|
)
|
|
|
|
# test for PriorityInUse
|
|
with pytest.raises(ClientError):
|
|
conn.create_rule(
|
|
ListenerArn=http_listener_arn,
|
|
Priority=priority,
|
|
Conditions=[
|
|
{"Field": "host-header", "Values": [host]},
|
|
{"Field": "path-pattern", "Values": [path_pattern]},
|
|
{
|
|
"Field": "path-pattern",
|
|
"PathPatternConfig": {"Values": [pathpatternconfig_pattern]},
|
|
},
|
|
],
|
|
Actions=[
|
|
{
|
|
"TargetGroupArn": target_group.get("TargetGroupArn"),
|
|
"Type": "forward",
|
|
}
|
|
],
|
|
)
|
|
|
|
# test for describe listeners
|
|
obtained_rules = conn.describe_rules(ListenerArn=http_listener_arn)
|
|
obtained_rules["Rules"].should.have.length_of(4)
|
|
priorities = [rule["Priority"] for rule in obtained_rules["Rules"]]
|
|
priorities.should.equal(["100", "500", "550", "default"])
|
|
|
|
first_rule = obtained_rules["Rules"][0]
|
|
second_rule = obtained_rules["Rules"][1]
|
|
third_rule = obtained_rules["Rules"][2]
|
|
default_rule = obtained_rules["Rules"][3]
|
|
first_rule["IsDefault"].should.equal(False)
|
|
default_rule["IsDefault"].should.equal(True)
|
|
obtained_rules = conn.describe_rules(RuleArns=[first_rule["RuleArn"]])
|
|
obtained_rules["Rules"].should.equal([first_rule])
|
|
|
|
# test for pagination
|
|
obtained_rules = conn.describe_rules(ListenerArn=http_listener_arn, PageSize=1)
|
|
len(obtained_rules["Rules"]).should.equal(1)
|
|
obtained_rules.should.have.key("NextMarker")
|
|
next_marker = obtained_rules["NextMarker"]
|
|
|
|
following_rules = conn.describe_rules(
|
|
ListenerArn=http_listener_arn, PageSize=1, Marker=next_marker
|
|
)
|
|
len(following_rules["Rules"]).should.equal(1)
|
|
following_rules["Rules"][0]["RuleArn"].should_not.equal(
|
|
obtained_rules["Rules"][0]["RuleArn"]
|
|
)
|
|
|
|
# test for invalid describe rule request
|
|
with pytest.raises(ClientError):
|
|
conn.describe_rules()
|
|
with pytest.raises(ClientError):
|
|
conn.describe_rules(RuleArns=[])
|
|
with pytest.raises(ClientError):
|
|
conn.describe_rules(
|
|
ListenerArn=http_listener_arn, RuleArns=[first_rule["RuleArn"]]
|
|
)
|
|
|
|
# modify rule partially
|
|
new_host = "new.example.com"
|
|
new_path_pattern = "new_path"
|
|
new_pathpatternconfig_pattern = "new_path2"
|
|
conn.modify_rule(
|
|
RuleArn=first_rule["RuleArn"],
|
|
Conditions=[
|
|
{"Field": "host-header", "Values": [new_host]},
|
|
{"Field": "path-pattern", "Values": [new_path_pattern]},
|
|
{
|
|
"Field": "path-pattern",
|
|
"PathPatternConfig": {"Values": [new_pathpatternconfig_pattern]},
|
|
},
|
|
],
|
|
)
|
|
|
|
rules = conn.describe_rules(ListenerArn=http_listener_arn)
|
|
obtained_rule = rules["Rules"][0]
|
|
obtained_rule["Conditions"][0]["Values"][0].should.equal(new_host)
|
|
obtained_rule["Conditions"][1]["Values"][0].should.equal(new_path_pattern)
|
|
obtained_rule["Conditions"][2]["PathPatternConfig"]["Values"][0].should.equal(
|
|
new_pathpatternconfig_pattern
|
|
)
|
|
obtained_rule["Actions"][0]["TargetGroupArn"].should.equal(
|
|
target_group.get("TargetGroupArn")
|
|
)
|
|
|
|
# modify priority
|
|
new_priority = int(first_rule["Priority"]) - 1
|
|
updated_rule = conn.set_rule_priorities(
|
|
RulePriorities=[
|
|
{
|
|
"RuleArn": first_rule["RuleArn"],
|
|
"Priority": new_priority,
|
|
}
|
|
]
|
|
)
|
|
|
|
# assert response of SetRulePriorities operation
|
|
updated_rule["Rules"].should.have.length_of(1)
|
|
updated_rule["Rules"][0]["RuleArn"].should.equal(first_rule["RuleArn"])
|
|
updated_rule["Rules"][0]["Priority"].should.equal(str(new_priority))
|
|
updated_rule["Rules"][0]["Conditions"].should.have.length_of(3)
|
|
updated_rule["Rules"][0]["Actions"].should.have.length_of(1)
|
|
|
|
# modify forward_config rule partially rule
|
|
new_host_2 = "new.examplewebsite.com"
|
|
new_path_pattern_2 = "new_path_2"
|
|
new_pathpatternconfig_pattern_2 = "new_path_2"
|
|
conn.modify_rule(
|
|
RuleArn=third_rule["RuleArn"],
|
|
Conditions=[
|
|
{"Field": "host-header", "Values": [new_host_2]},
|
|
{"Field": "path-pattern", "Values": [new_path_pattern_2]},
|
|
{
|
|
"Field": "path-pattern",
|
|
"PathPatternConfig": {"Values": [new_pathpatternconfig_pattern_2]},
|
|
},
|
|
],
|
|
Actions=[
|
|
{"TargetGroupArn": target_group.get("TargetGroupArn"), "Type": "forward"}
|
|
],
|
|
)
|
|
|
|
rules = conn.describe_rules(ListenerArn=http_listener_arn)
|
|
obtained_rule = rules["Rules"][2]
|
|
obtained_rule["Conditions"][0]["Values"][0].should.equal(new_host_2)
|
|
obtained_rule["Conditions"][1]["Values"][0].should.equal(new_path_pattern_2)
|
|
obtained_rule["Conditions"][2]["PathPatternConfig"]["Values"][0].should.equal(
|
|
new_pathpatternconfig_pattern_2
|
|
)
|
|
obtained_rule["Actions"][0]["TargetGroupArn"].should.equal(
|
|
target_group.get("TargetGroupArn")
|
|
)
|
|
|
|
# modify priority
|
|
conn.set_rule_priorities(
|
|
RulePriorities=[
|
|
{
|
|
"RuleArn": third_rule["RuleArn"],
|
|
"Priority": int(third_rule["Priority"]) - 1,
|
|
}
|
|
]
|
|
)
|
|
|
|
with pytest.raises(ClientError):
|
|
conn.set_rule_priorities(
|
|
RulePriorities=[
|
|
{"RuleArn": first_rule["RuleArn"], "Priority": 999},
|
|
{"RuleArn": second_rule["RuleArn"], "Priority": 999},
|
|
{"RuleArn": third_rule["RuleArn"], "Priority": 999},
|
|
]
|
|
)
|
|
|
|
# delete
|
|
arn = first_rule["RuleArn"]
|
|
conn.delete_rule(RuleArn=arn)
|
|
rules = conn.describe_rules(ListenerArn=http_listener_arn)["Rules"]
|
|
len(rules).should.equal(3)
|
|
|
|
# test for invalid action type
|
|
safe_priority = 2
|
|
with pytest.raises(ClientError):
|
|
conn.create_rule(
|
|
ListenerArn=http_listener_arn,
|
|
Priority=safe_priority,
|
|
Conditions=[
|
|
{"Field": "host-header", "Values": [host]},
|
|
{"Field": "path-pattern", "Values": [path_pattern]},
|
|
],
|
|
Actions=[
|
|
{
|
|
"TargetGroupArn": target_group.get("TargetGroupArn"),
|
|
"Type": "forward2",
|
|
}
|
|
],
|
|
)
|
|
|
|
# test for invalid action type
|
|
safe_priority = 2
|
|
invalid_target_group_arn = target_group.get("TargetGroupArn") + "x"
|
|
with pytest.raises(ClientError):
|
|
conn.create_rule(
|
|
ListenerArn=http_listener_arn,
|
|
Priority=safe_priority,
|
|
Conditions=[
|
|
{"Field": "host-header", "Values": [host]},
|
|
{"Field": "path-pattern", "Values": [path_pattern]},
|
|
],
|
|
Actions=[{"TargetGroupArn": invalid_target_group_arn, "Type": "forward"}],
|
|
)
|
|
|
|
# test for invalid condition field_name
|
|
safe_priority = 2
|
|
with pytest.raises(ClientError):
|
|
conn.create_rule(
|
|
ListenerArn=http_listener_arn,
|
|
Priority=safe_priority,
|
|
Conditions=[{"Field": "xxxxxxx", "Values": [host]}],
|
|
Actions=[
|
|
{
|
|
"TargetGroupArn": target_group.get("TargetGroupArn"),
|
|
"Type": "forward",
|
|
}
|
|
],
|
|
)
|
|
|
|
# test for emptry condition value
|
|
safe_priority = 2
|
|
with pytest.raises(ClientError):
|
|
conn.create_rule(
|
|
ListenerArn=http_listener_arn,
|
|
Priority=safe_priority,
|
|
Conditions=[{"Field": "host-header", "Values": []}],
|
|
Actions=[
|
|
{
|
|
"TargetGroupArn": target_group.get("TargetGroupArn"),
|
|
"Type": "forward",
|
|
}
|
|
],
|
|
)
|
|
|
|
# test for multiple condition value
|
|
safe_priority = 2
|
|
with pytest.raises(ClientError):
|
|
conn.create_rule(
|
|
ListenerArn=http_listener_arn,
|
|
Priority=safe_priority,
|
|
Conditions=[{"Field": "host-header", "Values": [host, host]}],
|
|
Actions=[
|
|
{
|
|
"TargetGroupArn": target_group.get("TargetGroupArn"),
|
|
"Type": "forward",
|
|
}
|
|
],
|
|
)
|
|
|
|
|
|
@mock_elbv2
|
|
def test_describe_account_limits():
|
|
client = boto3.client("elbv2", region_name="eu-central-1")
|
|
|
|
resp = client.describe_account_limits()
|
|
resp["Limits"][0].should.contain("Name")
|
|
resp["Limits"][0].should.contain("Max")
|
|
|
|
|
|
@mock_elbv2
|
|
def test_describe_ssl_policies():
|
|
client = boto3.client("elbv2", region_name="eu-central-1")
|
|
|
|
resp = client.describe_ssl_policies()
|
|
len(resp["SslPolicies"]).should.equal(6)
|
|
|
|
resp = client.describe_ssl_policies(
|
|
Names=["ELBSecurityPolicy-TLS-1-2-2017-01", "ELBSecurityPolicy-2016-08"]
|
|
)
|
|
len(resp["SslPolicies"]).should.equal(2)
|
|
|
|
resp = client.describe_ssl_policies(
|
|
Names=[
|
|
"ELBSecurityPolicy-TLS-1-2-2017-01",
|
|
"ELBSecurityPolicy-2016-08",
|
|
"ELBSecurityPolicy-2016-08",
|
|
]
|
|
)
|
|
len(resp["SslPolicies"]).should.equal(2)
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_set_ip_address_type():
|
|
response, _, security_group, subnet1, subnet2, client = create_load_balancer()
|
|
arn = response["LoadBalancers"][0]["LoadBalancerArn"]
|
|
|
|
# Internal LBs cant be dualstack yet
|
|
with pytest.raises(ClientError):
|
|
client.set_ip_address_type(LoadBalancerArn=arn, IpAddressType="dualstack")
|
|
|
|
# Create internet facing one
|
|
response = client.create_load_balancer(
|
|
Name="my-lb2",
|
|
Subnets=[subnet1.id, subnet2.id],
|
|
SecurityGroups=[security_group.id],
|
|
Scheme="internet-facing",
|
|
Tags=[{"Key": "key_name", "Value": "a_value"}],
|
|
)
|
|
arn = response["LoadBalancers"][0]["LoadBalancerArn"]
|
|
|
|
client.set_ip_address_type(LoadBalancerArn=arn, IpAddressType="dualstack")
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_set_security_groups():
|
|
client = boto3.client("elbv2", region_name="us-east-1")
|
|
ec2 = boto3.resource("ec2", region_name="us-east-1")
|
|
|
|
security_group = ec2.create_security_group(
|
|
GroupName="a-security-group", Description="First One"
|
|
)
|
|
security_group2 = ec2.create_security_group(
|
|
GroupName="b-security-group", Description="Second One"
|
|
)
|
|
vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
|
|
subnet1 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.192/26", AvailabilityZone="us-east-1a"
|
|
)
|
|
subnet2 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.0/26", AvailabilityZone="us-east-1b"
|
|
)
|
|
|
|
response = client.create_load_balancer(
|
|
Name="my-lb",
|
|
Subnets=[subnet1.id, subnet2.id],
|
|
SecurityGroups=[security_group.id],
|
|
Scheme="internal",
|
|
Tags=[{"Key": "key_name", "Value": "a_value"}],
|
|
)
|
|
arn = response["LoadBalancers"][0]["LoadBalancerArn"]
|
|
|
|
client.set_security_groups(
|
|
LoadBalancerArn=arn, SecurityGroups=[security_group.id, security_group2.id]
|
|
)
|
|
|
|
resp = client.describe_load_balancers(LoadBalancerArns=[arn])
|
|
len(resp["LoadBalancers"][0]["SecurityGroups"]).should.equal(2)
|
|
|
|
with pytest.raises(ClientError):
|
|
client.set_security_groups(LoadBalancerArn=arn, SecurityGroups=["non_existent"])
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_modify_load_balancer_attributes_idle_timeout():
|
|
response, _, _, _, _, client = create_load_balancer()
|
|
arn = response["LoadBalancers"][0]["LoadBalancerArn"]
|
|
|
|
client.modify_load_balancer_attributes(
|
|
LoadBalancerArn=arn,
|
|
Attributes=[{"Key": "idle_timeout.timeout_seconds", "Value": "600"}],
|
|
)
|
|
|
|
# Check its 600 not 60
|
|
response = client.describe_load_balancer_attributes(LoadBalancerArn=arn)
|
|
idle_timeout = list(
|
|
filter(
|
|
lambda item: item["Key"] == "idle_timeout.timeout_seconds",
|
|
response["Attributes"],
|
|
)
|
|
)[0]
|
|
idle_timeout["Value"].should.equal("600")
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_modify_load_balancer_attributes_routing_http2_enabled():
|
|
response, _, _, _, _, client = create_load_balancer()
|
|
arn = response["LoadBalancers"][0]["LoadBalancerArn"]
|
|
|
|
client.modify_load_balancer_attributes(
|
|
LoadBalancerArn=arn,
|
|
Attributes=[{"Key": "routing.http2.enabled", "Value": "false"}],
|
|
)
|
|
|
|
response = client.describe_load_balancer_attributes(LoadBalancerArn=arn)
|
|
routing_http2_enabled = list(
|
|
filter(
|
|
lambda item: item["Key"] == "routing.http2.enabled", response["Attributes"]
|
|
)
|
|
)[0]
|
|
routing_http2_enabled["Value"].should.equal("false")
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_modify_load_balancer_attributes_crosszone_enabled():
|
|
response, _, _, _, _, client = create_load_balancer()
|
|
arn = response["LoadBalancers"][0]["LoadBalancerArn"]
|
|
|
|
client.modify_load_balancer_attributes(
|
|
LoadBalancerArn=arn,
|
|
Attributes=[
|
|
{"Key": "load_balancing.cross_zone.enabled", "Value": "false"},
|
|
{"Key": "deletion_protection.enabled", "Value": "false"},
|
|
],
|
|
)
|
|
|
|
attrs = client.describe_load_balancer_attributes(LoadBalancerArn=arn)["Attributes"]
|
|
attrs.should.contain({"Key": "deletion_protection.enabled", "Value": "false"})
|
|
attrs.should.contain({"Key": "load_balancing.cross_zone.enabled", "Value": "false"})
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_modify_load_balancer_attributes_routing_http_drop_invalid_header_fields_enabled():
|
|
response, _, _, _, _, client = create_load_balancer()
|
|
arn = response["LoadBalancers"][0]["LoadBalancerArn"]
|
|
|
|
client.modify_load_balancer_attributes(
|
|
LoadBalancerArn=arn,
|
|
Attributes=[
|
|
{"Key": "routing.http.drop_invalid_header_fields.enabled", "Value": "false"}
|
|
],
|
|
)
|
|
|
|
response = client.describe_load_balancer_attributes(LoadBalancerArn=arn)
|
|
routing_http_drop_invalid_header_fields_enabled = list(
|
|
filter(
|
|
lambda item: item["Key"]
|
|
== "routing.http.drop_invalid_header_fields.enabled",
|
|
response["Attributes"],
|
|
)
|
|
)[0]
|
|
routing_http_drop_invalid_header_fields_enabled["Value"].should.equal("false")
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
@mock_acm
|
|
def test_modify_listener_http_to_https():
|
|
client = boto3.client("elbv2", region_name="eu-central-1")
|
|
acm = boto3.client("acm", region_name="eu-central-1")
|
|
ec2 = boto3.resource("ec2", region_name="eu-central-1")
|
|
|
|
security_group = ec2.create_security_group(
|
|
GroupName="a-security-group", Description="First One"
|
|
)
|
|
vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
|
|
subnet1 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.192/26", AvailabilityZone="eu-central-1a"
|
|
)
|
|
subnet2 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.0/26", AvailabilityZone="eu-central-1b"
|
|
)
|
|
|
|
response = client.create_load_balancer(
|
|
Name="my-lb",
|
|
Subnets=[subnet1.id, subnet2.id],
|
|
SecurityGroups=[security_group.id],
|
|
Scheme="internal",
|
|
Tags=[{"Key": "key_name", "Value": "a_value"}],
|
|
)
|
|
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
response = client.create_target_group(
|
|
Name="a-target",
|
|
Protocol="HTTP",
|
|
Port=8080,
|
|
VpcId=vpc.id,
|
|
HealthCheckProtocol="HTTP",
|
|
HealthCheckPort="8080",
|
|
HealthCheckPath="/",
|
|
HealthCheckIntervalSeconds=5,
|
|
HealthCheckTimeoutSeconds=5,
|
|
HealthyThresholdCount=5,
|
|
UnhealthyThresholdCount=2,
|
|
Matcher={"HttpCode": "200"},
|
|
)
|
|
target_group = response.get("TargetGroups")[0]
|
|
target_group_arn = target_group["TargetGroupArn"]
|
|
|
|
# Plain HTTP listener
|
|
response = client.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTP",
|
|
Port=80,
|
|
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
|
|
)
|
|
listener_arn = response["Listeners"][0]["ListenerArn"]
|
|
|
|
# No default cert
|
|
with pytest.raises(ClientError) as ex:
|
|
client.modify_listener(
|
|
ListenerArn=listener_arn,
|
|
Port=443,
|
|
Protocol="HTTPS",
|
|
SslPolicy="ELBSecurityPolicy-TLS-1-2-2017-01",
|
|
Certificates=[],
|
|
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
|
|
)
|
|
err = ex.value.response["Error"]
|
|
err["Code"].should.equal("CertificateWereNotPassed")
|
|
err["Message"].should.equal(
|
|
"You must provide a list containing exactly one certificate if the listener protocol is HTTPS."
|
|
)
|
|
|
|
acm.request_certificate(
|
|
DomainName="google.com",
|
|
SubjectAlternativeNames=["google.com", "www.google.com", "mail.google.com"],
|
|
)
|
|
response = acm.request_certificate(
|
|
DomainName="yahoo.com",
|
|
SubjectAlternativeNames=["yahoo.com", "www.yahoo.com", "mail.yahoo.com"],
|
|
)
|
|
yahoo_arn = response["CertificateArn"]
|
|
|
|
response = client.modify_listener(
|
|
ListenerArn=listener_arn,
|
|
Port=443,
|
|
Protocol="HTTPS",
|
|
SslPolicy="ELBSecurityPolicy-TLS-1-2-2017-01",
|
|
Certificates=[{"CertificateArn": yahoo_arn}],
|
|
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
|
|
)
|
|
response["Listeners"][0]["Port"].should.equal(443)
|
|
response["Listeners"][0]["Protocol"].should.equal("HTTPS")
|
|
response["Listeners"][0]["SslPolicy"].should.equal(
|
|
"ELBSecurityPolicy-TLS-1-2-2017-01"
|
|
)
|
|
len(response["Listeners"][0]["Certificates"]).should.equal(1)
|
|
|
|
# Check default cert, can't do this in server mode
|
|
if os.environ.get("TEST_SERVER_MODE", "false").lower() == "false":
|
|
listener = (
|
|
elbv2_backends["eu-central-1"]
|
|
.load_balancers[load_balancer_arn]
|
|
.listeners[listener_arn]
|
|
)
|
|
listener.certificate.should.equal(yahoo_arn)
|
|
|
|
# Bad cert
|
|
with pytest.raises(ClientError) as exc:
|
|
client.modify_listener(
|
|
ListenerArn=listener_arn,
|
|
Port=443,
|
|
Protocol="HTTPS",
|
|
SslPolicy="ELBSecurityPolicy-TLS-1-2-2017-01",
|
|
Certificates=[{"CertificateArn": "lalala", "IsDefault": True}],
|
|
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
|
|
)
|
|
err = exc.value.response["Error"]
|
|
err["Message"].should.equal("Certificate lalala not found")
|
|
|
|
# Unknown protocol
|
|
with pytest.raises(ClientError) as exc:
|
|
client.modify_listener(
|
|
ListenerArn=listener_arn,
|
|
Port=443,
|
|
Protocol="HTP",
|
|
SslPolicy="ELBSecurityPolicy-TLS-1-2-2017-01",
|
|
Certificates=[{"CertificateArn": yahoo_arn, "IsDefault": True}],
|
|
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
|
|
)
|
|
err = exc.value.response["Error"]
|
|
err["Message"].should.equal("Protocol HTP is not supported")
|
|
|
|
|
|
@mock_acm
|
|
@mock_ec2
|
|
@mock_elbv2
|
|
def test_modify_listener_of_https_target_group():
|
|
# Verify we can add a listener for a TargetGroup that is already HTTPS
|
|
client = boto3.client("elbv2", region_name="eu-central-1")
|
|
acm = boto3.client("acm", region_name="eu-central-1")
|
|
ec2 = boto3.resource("ec2", region_name="eu-central-1")
|
|
|
|
security_group = ec2.create_security_group(
|
|
GroupName="a-security-group", Description="First One"
|
|
)
|
|
vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
|
|
subnet1 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.192/26", AvailabilityZone="eu-central-1a"
|
|
)
|
|
|
|
response = client.create_load_balancer(
|
|
Name="my-lb",
|
|
Subnets=[subnet1.id],
|
|
SecurityGroups=[security_group.id],
|
|
Scheme="internal",
|
|
Tags=[{"Key": "key_name", "Value": "a_value"}],
|
|
)
|
|
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
response = client.create_target_group(
|
|
Name="a-target", Protocol="HTTPS", Port=8443, VpcId=vpc.id
|
|
)
|
|
target_group = response.get("TargetGroups")[0]
|
|
target_group_arn = target_group["TargetGroupArn"]
|
|
|
|
# HTTPS listener
|
|
response = acm.request_certificate(
|
|
DomainName="google.com", SubjectAlternativeNames=["google.com"]
|
|
)
|
|
google_arn = response["CertificateArn"]
|
|
response = client.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTPS",
|
|
Port=443,
|
|
Certificates=[{"CertificateArn": google_arn}],
|
|
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
|
|
)
|
|
listener_arn = response["Listeners"][0]["ListenerArn"]
|
|
|
|
# Now modify the HTTPS listener with a different certificate
|
|
response = acm.request_certificate(
|
|
DomainName="yahoo.com", SubjectAlternativeNames=["yahoo.com"]
|
|
)
|
|
yahoo_arn = response["CertificateArn"]
|
|
|
|
listener = client.modify_listener(
|
|
ListenerArn=listener_arn,
|
|
Certificates=[{"CertificateArn": yahoo_arn}],
|
|
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
|
|
)["Listeners"][0]
|
|
listener["Certificates"].should.equal([{"CertificateArn": yahoo_arn}])
|
|
|
|
listener = client.describe_listeners(ListenerArns=[listener_arn])["Listeners"][0]
|
|
listener["Certificates"].should.equal([{"CertificateArn": yahoo_arn}])
|
|
|
|
|
|
@mock_elbv2
|
|
def test_add_unknown_listener_certificate():
|
|
client = boto3.client("elbv2", region_name="eu-central-1")
|
|
with pytest.raises(ClientError) as exc:
|
|
client.add_listener_certificates(
|
|
ListenerArn="unknown", Certificates=[{"CertificateArn": "google_arn"}]
|
|
)
|
|
err = exc.value.response["Error"]
|
|
err["Code"].should.equal("ListenerNotFound")
|
|
|
|
|
|
@mock_elbv2
|
|
def test_describe_unknown_listener_certificate():
|
|
client = boto3.client("elbv2", region_name="eu-central-1")
|
|
with pytest.raises(ClientError) as exc:
|
|
client.describe_listener_certificates(ListenerArn="unknown")
|
|
err = exc.value.response["Error"]
|
|
err["Code"].should.equal("ListenerNotFound")
|
|
|
|
|
|
@mock_acm
|
|
@mock_ec2
|
|
@mock_elbv2
|
|
def test_add_listener_certificate():
|
|
# Verify we can add a listener for a TargetGroup that is already HTTPS
|
|
client = boto3.client("elbv2", region_name="eu-central-1")
|
|
acm = boto3.client("acm", region_name="eu-central-1")
|
|
ec2 = boto3.resource("ec2", region_name="eu-central-1")
|
|
|
|
vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
|
|
subnet1 = ec2.create_subnet(
|
|
VpcId=vpc.id, CidrBlock="172.28.7.192/26", AvailabilityZone="eu-central-1a"
|
|
)
|
|
|
|
response = client.create_load_balancer(
|
|
Name="my-lb",
|
|
Subnets=[subnet1.id],
|
|
Scheme="internal",
|
|
Tags=[{"Key": "key_name", "Value": "a_value"}],
|
|
)
|
|
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
response = client.create_target_group(Name="a-target", Protocol="HTTPS", Port=8443)
|
|
target_group_arn = response.get("TargetGroups")[0]["TargetGroupArn"]
|
|
|
|
# HTTPS listener
|
|
response = acm.request_certificate(
|
|
DomainName="google.com", SubjectAlternativeNames=["google.com"]
|
|
)
|
|
google_arn = response["CertificateArn"]
|
|
response = client.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTPS",
|
|
Port=443,
|
|
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
|
|
)
|
|
listener_arn = response["Listeners"][0]["ListenerArn"]
|
|
|
|
certs = client.add_listener_certificates(
|
|
ListenerArn=listener_arn, Certificates=[{"CertificateArn": google_arn}]
|
|
)["Certificates"]
|
|
certs.should.have.length_of(1)
|
|
certs[0].should.have.key("CertificateArn").equals(google_arn)
|
|
|
|
certs = client.describe_listener_certificates(ListenerArn=listener_arn)[
|
|
"Certificates"
|
|
]
|
|
certs.should.have.length_of(1)
|
|
certs[0].should.have.key("CertificateArn").equals(google_arn)
|
|
|
|
client.remove_listener_certificates(
|
|
ListenerArn=listener_arn, Certificates=[{"CertificateArn": google_arn}]
|
|
)
|
|
|
|
certs = client.describe_listener_certificates(ListenerArn=listener_arn)[
|
|
"Certificates"
|
|
]
|
|
certs.should.have.length_of(0)
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_forward_config_action():
|
|
response, _, _, _, _, conn = create_load_balancer()
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
response = conn.create_target_group(Name="a-target", Protocol="HTTPS", Port=8443)
|
|
target_group_arn = response.get("TargetGroups")[0]["TargetGroupArn"]
|
|
|
|
action = {
|
|
"Type": "forward",
|
|
"ForwardConfig": {
|
|
"TargetGroups": [{"TargetGroupArn": target_group_arn, "Weight": 1}],
|
|
},
|
|
}
|
|
expected_action = copy.deepcopy(action)
|
|
expected_action["ForwardConfig"]["TargetGroupStickinessConfig"] = {"Enabled": False}
|
|
|
|
response = conn.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTP",
|
|
Port=80,
|
|
DefaultActions=[action],
|
|
)
|
|
|
|
listener = response.get("Listeners")[0]
|
|
listener.get("DefaultActions").should.equal([expected_action])
|
|
listener_arn = listener.get("ListenerArn")
|
|
|
|
describe_listener_response = conn.describe_listeners(ListenerArns=[listener_arn])
|
|
describe_listener_actions = describe_listener_response["Listeners"][0][
|
|
"DefaultActions"
|
|
]
|
|
describe_listener_actions.should.equal([expected_action])
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_forward_config_action__with_stickiness():
|
|
response, _, _, _, _, conn = create_load_balancer()
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
response = conn.create_target_group(Name="a-target", Protocol="HTTPS", Port=8443)
|
|
target_group_arn = response.get("TargetGroups")[0]["TargetGroupArn"]
|
|
|
|
action = {
|
|
"Type": "forward",
|
|
"ForwardConfig": {
|
|
"TargetGroups": [{"TargetGroupArn": target_group_arn, "Weight": 1}],
|
|
"TargetGroupStickinessConfig": {"Enabled": True},
|
|
},
|
|
}
|
|
|
|
response = conn.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTP",
|
|
Port=80,
|
|
DefaultActions=[action],
|
|
)
|
|
|
|
listener = response.get("Listeners")[0]
|
|
listener.get("DefaultActions").should.equal([action])
|
|
listener_arn = listener.get("ListenerArn")
|
|
|
|
describe_listener_response = conn.describe_listeners(ListenerArns=[listener_arn])
|
|
describe_listener_actions = describe_listener_response["Listeners"][0][
|
|
"DefaultActions"
|
|
]
|
|
describe_listener_actions.should.equal([action])
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_redirect_action_listener_rule():
|
|
response, _, _, _, _, conn = create_load_balancer()
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
action = {
|
|
"Type": "redirect",
|
|
"RedirectConfig": {
|
|
"Protocol": "HTTPS",
|
|
"Port": "443",
|
|
"StatusCode": "HTTP_301",
|
|
"Host": "h",
|
|
"Path": "p",
|
|
"Query": "q",
|
|
},
|
|
"Order": 1,
|
|
}
|
|
|
|
response = conn.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTP",
|
|
Port=80,
|
|
DefaultActions=[action],
|
|
)
|
|
|
|
listener = response.get("Listeners")[0]
|
|
listener.get("DefaultActions").should.equal([action])
|
|
listener_arn = listener.get("ListenerArn")
|
|
|
|
conn.create_rule(
|
|
ListenerArn=listener_arn,
|
|
Conditions=[{"Field": "path-pattern", "Values": ["/*"]}],
|
|
Priority=3,
|
|
Actions=[action],
|
|
)
|
|
describe_rules_response = conn.describe_rules(ListenerArn=listener_arn)
|
|
describe_rules_response["Rules"][0]["Actions"].should.equal([action])
|
|
|
|
describe_listener_response = conn.describe_listeners(ListenerArns=[listener_arn])
|
|
describe_listener_actions = describe_listener_response["Listeners"][0][
|
|
"DefaultActions"
|
|
]
|
|
describe_listener_actions.should.equal([action])
|
|
|
|
modify_listener_response = conn.modify_listener(ListenerArn=listener_arn, Port=81)
|
|
modify_listener_actions = modify_listener_response["Listeners"][0]["DefaultActions"]
|
|
modify_listener_actions.should.equal([action])
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_cognito_action_listener_rule():
|
|
response, _, _, _, _, conn = create_load_balancer()
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
action = {
|
|
"Type": "authenticate-cognito",
|
|
"AuthenticateCognitoConfig": {
|
|
"UserPoolArn": "arn:aws:cognito-idp:us-east-1:{}:userpool/us-east-1_ABCD1234".format(
|
|
ACCOUNT_ID
|
|
),
|
|
"UserPoolClientId": "abcd1234abcd",
|
|
"UserPoolDomain": "testpool",
|
|
"AuthenticationRequestExtraParams": {"param": "test"},
|
|
},
|
|
}
|
|
response = conn.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTP",
|
|
Port=80,
|
|
DefaultActions=[action],
|
|
)
|
|
|
|
listener = response.get("Listeners")[0]
|
|
listener.get("DefaultActions")[0].should.equal(action)
|
|
listener_arn = listener.get("ListenerArn")
|
|
|
|
conn.create_rule(
|
|
ListenerArn=listener_arn,
|
|
Conditions=[{"Field": "path-pattern", "Values": ["/*"]}],
|
|
Priority=3,
|
|
Actions=[action],
|
|
)
|
|
describe_rules_response = conn.describe_rules(ListenerArn=listener_arn)
|
|
describe_rules_response["Rules"][0]["Actions"][0].should.equal(action)
|
|
|
|
describe_listener_response = conn.describe_listeners(ListenerArns=[listener_arn])
|
|
describe_listener_actions = describe_listener_response["Listeners"][0][
|
|
"DefaultActions"
|
|
][0]
|
|
describe_listener_actions.should.equal(action)
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_oidc_action_listener__simple():
|
|
response, _, _, _, _, conn = create_load_balancer()
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
action = {
|
|
"Type": "authenticate-oidc",
|
|
"AuthenticateOidcConfig": {
|
|
"AuthorizationEndpoint": "ae",
|
|
"ClientId": "ci",
|
|
"TokenEndpoint": "te",
|
|
"UserInfoEndpoint": "uie",
|
|
"Issuer": "is",
|
|
},
|
|
}
|
|
response = conn.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTP",
|
|
Port=80,
|
|
DefaultActions=[action],
|
|
)
|
|
|
|
listener = response.get("Listeners")[0]
|
|
listener.get("DefaultActions")[0].should.equal(action)
|
|
listener_arn = listener.get("ListenerArn")
|
|
|
|
conn.create_rule(
|
|
ListenerArn=listener_arn,
|
|
Conditions=[{"Field": "path-pattern", "Values": ["/*"]}],
|
|
Priority=3,
|
|
Actions=[action],
|
|
)
|
|
describe_rules_response = conn.describe_rules(ListenerArn=listener_arn)
|
|
describe_rules_response["Rules"][0]["Actions"][0].should.equal(action)
|
|
|
|
describe_listener_response = conn.describe_listeners(ListenerArns=[listener_arn])
|
|
describe_listener_actions = describe_listener_response["Listeners"][0][
|
|
"DefaultActions"
|
|
][0]
|
|
describe_listener_actions.should.equal(action)
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
@pytest.mark.parametrize("use_secret", [True, False])
|
|
def test_oidc_action_listener(use_secret):
|
|
response, _, _, _, _, conn = create_load_balancer()
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
action = {
|
|
"Type": "authenticate-oidc",
|
|
"AuthenticateOidcConfig": {
|
|
"Issuer": "is",
|
|
"AuthorizationEndpoint": "ae",
|
|
"TokenEndpoint": "te",
|
|
"UserInfoEndpoint": "uie",
|
|
"ClientId": "ci",
|
|
"ClientSecret": "cs",
|
|
"SessionCookieName": "scn",
|
|
"Scope": "s",
|
|
"SessionTimeout": 42,
|
|
"AuthenticationRequestExtraParams": {"param": "test"},
|
|
"OnUnauthenticatedRequest": "our",
|
|
"UseExistingClientSecret": use_secret,
|
|
},
|
|
}
|
|
response = conn.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTP",
|
|
Port=80,
|
|
DefaultActions=[action],
|
|
)
|
|
|
|
listener = response.get("Listeners")[0]
|
|
listener.get("DefaultActions")[0].should.equal(action)
|
|
listener_arn = listener.get("ListenerArn")
|
|
|
|
conn.create_rule(
|
|
ListenerArn=listener_arn,
|
|
Conditions=[{"Field": "path-pattern", "Values": ["/*"]}],
|
|
Priority=3,
|
|
Actions=[action],
|
|
)
|
|
describe_rules_response = conn.describe_rules(ListenerArn=listener_arn)
|
|
describe_rules_response["Rules"][0]["Actions"][0].should.equal(action)
|
|
|
|
describe_listener_response = conn.describe_listeners(ListenerArns=[listener_arn])
|
|
describe_listener_actions = describe_listener_response["Listeners"][0][
|
|
"DefaultActions"
|
|
][0]
|
|
describe_listener_actions.should.equal(action)
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_fixed_response_action_listener_rule():
|
|
response, _, _, _, _, conn = create_load_balancer()
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
action = {
|
|
"Type": "fixed-response",
|
|
"FixedResponseConfig": {
|
|
"ContentType": "text/plain",
|
|
"MessageBody": "This page does not exist",
|
|
"StatusCode": "404",
|
|
},
|
|
}
|
|
response = conn.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTP",
|
|
Port=80,
|
|
DefaultActions=[action],
|
|
)
|
|
|
|
listener = response.get("Listeners")[0]
|
|
listener.get("DefaultActions")[0].should.equal(action)
|
|
listener_arn = listener.get("ListenerArn")
|
|
|
|
conn.create_rule(
|
|
ListenerArn=listener_arn,
|
|
Conditions=[{"Field": "path-pattern", "Values": ["/*"]}],
|
|
Priority=3,
|
|
Actions=[action],
|
|
)
|
|
describe_rules_response = conn.describe_rules(ListenerArn=listener_arn)
|
|
describe_rules_response["Rules"][0]["Actions"][0].should.equal(action)
|
|
|
|
describe_listener_response = conn.describe_listeners(ListenerArns=[listener_arn])
|
|
describe_listener_actions = describe_listener_response["Listeners"][0][
|
|
"DefaultActions"
|
|
][0]
|
|
describe_listener_actions.should.equal(action)
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_fixed_response_action_listener_rule_validates_status_code():
|
|
response, _, _, _, _, conn = create_load_balancer()
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
invalid_status_code_action = {
|
|
"Type": "fixed-response",
|
|
"FixedResponseConfig": {
|
|
"ContentType": "text/plain",
|
|
"MessageBody": "This page does not exist",
|
|
"StatusCode": "100",
|
|
},
|
|
}
|
|
|
|
with pytest.raises(ClientError) as invalid_status_code_exception:
|
|
conn.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTP",
|
|
Port=80,
|
|
DefaultActions=[invalid_status_code_action],
|
|
)
|
|
|
|
invalid_status_code_exception.value.response["Error"]["Code"].should.equal(
|
|
"ValidationError"
|
|
)
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_fixed_response_action_listener_rule_validates_content_type():
|
|
response, _, _, _, _, conn = create_load_balancer()
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
invalid_content_type_action = {
|
|
"Type": "fixed-response",
|
|
"FixedResponseConfig": {
|
|
"ContentType": "Fake content type",
|
|
"MessageBody": "This page does not exist",
|
|
"StatusCode": "200",
|
|
},
|
|
}
|
|
with pytest.raises(ClientError) as invalid_content_type_exception:
|
|
conn.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTP",
|
|
Port=80,
|
|
DefaultActions=[invalid_content_type_action],
|
|
)
|
|
invalid_content_type_exception.value.response["Error"]["Code"].should.equal(
|
|
"InvalidLoadBalancerAction"
|
|
)
|
|
|
|
|
|
@mock_elbv2
|
|
@mock_ec2
|
|
def test_create_listener_with_alpn_policy():
|
|
response, _, _, _, _, conn = create_load_balancer()
|
|
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
|
|
|
|
response = conn.create_listener(
|
|
LoadBalancerArn=load_balancer_arn,
|
|
Protocol="HTTP",
|
|
Port=80,
|
|
DefaultActions=[],
|
|
AlpnPolicy=["pol1", "pol2"],
|
|
)
|
|
|
|
listener = response.get("Listeners")[0]
|
|
listener_arn = listener["ListenerArn"]
|
|
listener.get("AlpnPolicy").should.equal(["pol1", "pol2"])
|
|
|
|
describe = conn.describe_listeners(ListenerArns=[listener_arn])["Listeners"][0]
|
|
describe.should.have.key("AlpnPolicy").should.equal(["pol1", "pol2"])
|