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"])