from __future__ import unicode_literals
import json
import os
import boto3
import botocore
from botocore.exceptions import ClientError, ParamValidationError
from nose.tools import assert_raises
import sure # noqa
from moto import mock_elbv2, mock_ec2, mock_acm, mock_cloudformation
from moto.elbv2 import elbv2_backends
@mock_elbv2
@mock_ec2
def test_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'}])
lb = response.get('LoadBalancers')[0]
lb.get('DNSName').should.equal("my-lb-1.us-east-1.elb.amazonaws.com")
lb.get('LoadBalancerArn').should.equal(
'arn:aws:elasticloadbalancing:us-east-1:1:loadbalancer/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'}])
# Ensure the tags persisted
response = conn.describe_tags(ResourceArns=[lb.get('LoadBalancerArn')])
tags = {d['Key']: d['Value']
for d in response['TagDescriptions'][0]['Tags']}
tags.should.equal({'key_name': 'a_value'})
@mock_elbv2
@mock_ec2
def test_describe_load_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')
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.describe_load_balancers()
response.get('LoadBalancers').should.have.length_of(1)
lb = response.get('LoadBalancers')[0]
lb.get('LoadBalancerName').should.equal('my-lb')
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 assert_raises(ClientError):
conn.describe_load_balancers(LoadBalancerArns=['not-a/real/arn'])
with assert_raises(ClientError):
conn.describe_load_balancers(Names=['nope'])
@mock_elbv2
@mock_ec2
def test_add_remove_tags():
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.0/26',
AvailabilityZone='us-east-1a')
subnet2 = ec2.create_subnet(
VpcId=vpc.id,
CidrBlock='172.28.7.192/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'}])
lbs = conn.describe_load_balancers()['LoadBalancers']
lbs.should.have.length_of(1)
lb = lbs[0]
with assert_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'
}])
conn.add_tags.when.called_with(ResourceArns=[lb.get('LoadBalancerArn')],
Tags=[{
'Key': 'k',
'Value': 'b'
}]).should.throw(botocore.exceptions.ClientError)
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_target_group_and_listeners():
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'}])
load_balancer_arn = response.get('LoadBalancers')[0].get('LoadBalancerArn')
# Can't create a target group with an invalid protocol
with assert_raises(ClientError):
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'})
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']
# Add tags to the target group
conn.add_tags(ResourceArns=[target_group_arn], Tags=[
{'Key': 'target', 'Value': 'group'}])
conn.describe_tags(ResourceArns=[target_group_arn])['TagDescriptions'][0]['Tags'].should.equal(
[{'Key': 'target', 'Value': 'group'}])
# Check it's in the describe_target_groups response
response = conn.describe_target_groups()
response.get('TargetGroups').should.have.length_of(1)
# 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')
response = conn.describe_target_groups(LoadBalancerArn=load_balancer_arn,
Names=['a-target'])
response.get('TargetGroups').should.have.length_of(1)
# And another with SSL
response = conn.create_listener(
LoadBalancerArn=load_balancer_arn,
Protocol='HTTPS',
Port=443,
Certificates=[
{'CertificateArn': 'arn:aws:iam:123456789012:server-certificate/test-cert'}],
DefaultActions=[{'Type': 'forward', 'TargetGroupArn': target_group.get('TargetGroupArn')}])
listener = response.get('Listeners')[0]
listener.get('Port').should.equal(443)
listener.get('Protocol').should.equal('HTTPS')
listener.get('Certificates').should.equal([{
'CertificateArn': 'arn:aws:iam:123456789012:server-certificate/test-cert',
}])
listener.get('DefaultActions').should.equal([{
'TargetGroupArn': target_group.get('TargetGroupArn'),
'Type': 'forward'}])
https_listener_arn = listener.get('ListenerArn')
response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn)
response.get('Listeners').should.have.length_of(2)
response = conn.describe_listeners(ListenerArns=[https_listener_arn])
response.get('Listeners').should.have.length_of(1)
listener = response.get('Listeners')[0]
listener.get('Port').should.equal(443)
listener.get('Protocol').should.equal('HTTPS')
response = conn.describe_listeners(
ListenerArns=[
http_listener_arn,
https_listener_arn])
response.get('Listeners').should.have.length_of(2)
# Try to delete the target group and it fails because there's a
# listener referencing it
with assert_raises(ClientError) as e:
conn.delete_target_group(
TargetGroupArn=target_group.get('TargetGroupArn'))
e.exception.operation_name.should.equal('DeleteTargetGroup')
e.exception.args.should.equal(("An error occurred (ResourceInUse) when calling the DeleteTargetGroup operation: The target group 'arn:aws:elasticloadbalancing:us-east-1:1:targetgroup/a-target/50dc6c495c0c9188' is currently in use by a listener or a rule", )) # NOQA
# Delete one listener
response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn)
response.get('Listeners').should.have.length_of(2)
conn.delete_listener(ListenerArn=http_listener_arn)
response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn)
response.get('Listeners').should.have.length_of(1)
# Then delete the load balancer
conn.delete_load_balancer(LoadBalancerArn=load_balancer_arn)
# It's gone
response = conn.describe_load_balancers()
response.get('LoadBalancers').should.have.length_of(0)
# And it deleted the remaining listener
response = conn.describe_listeners(
ListenerArns=[
http_listener_arn,
https_listener_arn])
response.get('Listeners').should.have.length_of(0)
# But not the target groups
response = conn.describe_target_groups()
response.get('TargetGroups').should.have.length_of(1)
# Which we'll now delete
conn.delete_target_group(TargetGroupArn=target_group.get('TargetGroupArn'))
response = conn.describe_target_groups()
response.get('TargetGroups').should.have.length_of(0)
@mock_elbv2
@mock_ec2
def test_create_target_group_without_non_required_parameters():
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'}])
# request without HealthCheckIntervalSeconds parameter
# which is default to 30 seconds
response = conn.create_target_group(
Name='a-target',
Protocol='HTTP',
Port=8080,
VpcId=vpc.id,
HealthCheckProtocol='HTTP',
HealthCheckPort='8080'
)
target_group = response.get('TargetGroups')[0]
target_group.should_not.be.none
@mock_elbv2
@mock_ec2
def test_create_invalid_target_group():
conn = boto3.client('elbv2', region_name='us-east-1')
ec2 = boto3.resource('ec2', region_name='us-east-1')
vpc = ec2.create_vpc(CidrBlock='172.28.7.0/24', InstanceTenancy='default')
# Fail to create target group with name which length is 33
long_name = 'A' * 33
with assert_raises(ClientError):
conn.create_target_group(
Name=long_name,
Protocol='HTTP',
Port=8080,
VpcId=vpc.id,
HealthCheckProtocol='HTTP',
HealthCheckPort='8080',
HealthCheckPath='/',
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={'HttpCode': '200'})
invalid_names = [
'-name',
'name-',
'-name-',
'example.com',
'test@test',
'Na--me']
for name in invalid_names:
with assert_raises(ClientError):
conn.create_target_group(
Name=name,
Protocol='HTTP',
Port=8080,
VpcId=vpc.id,
HealthCheckProtocol='HTTP',
HealthCheckPort='8080',
HealthCheckPath='/',
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={'HttpCode': '200'})
valid_names = ['name', 'Name', '000']
for name in valid_names:
conn.create_target_group(
Name=name,
Protocol='HTTP',
Port=8080,
VpcId=vpc.id,
HealthCheckProtocol='HTTP',
HealthCheckPort='8080',
HealthCheckPath='/',
HealthCheckIntervalSeconds=5,
HealthCheckTimeoutSeconds=5,
HealthyThresholdCount=5,
UnhealthyThresholdCount=2,
Matcher={'HttpCode': '200'})
@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():
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'}])
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='ami-1234abcd', 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='ami-1234abcd', 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='ami-1234abcd', 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_ec2
@mock_elbv2
def test_target_group_attributes():
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'}])
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]
# Check it's in the describe_target_groups response
response = conn.describe_target_groups()
response.get('TargetGroups').should.have.length_of(1)
target_group_arn = target_group['TargetGroupArn']
# check if Names filter works
response = conn.describe_target_groups(Names=[])
response = conn.describe_target_groups(Names=['a-target'])
response.get('TargetGroups').should.have.length_of(1)
target_group_arn = target_group['TargetGroupArn']
# The attributes should start with the two defaults
response = conn.describe_target_group_attributes(
TargetGroupArn=target_group_arn)
response['Attributes'].should.have.length_of(2)
attributes = {attr['Key']: attr['Value']
for attr in response['Attributes']}
attributes['deregistration_delay.timeout_seconds'].should.equal('300')
attributes['stickiness.enabled'].should.equal('false')
# Add cookie stickiness
response = conn.modify_target_group_attributes(
TargetGroupArn=target_group_arn,
Attributes=[
{
'Key': 'stickiness.enabled',
'Value': 'true',
},
{
'Key': 'stickiness.type',
'Value': 'lb_cookie',
},
])
# The response should have only the keys updated
response['Attributes'].should.have.length_of(2)
attributes = {attr['Key']: attr['Value']
for attr in response['Attributes']}
attributes['stickiness.type'].should.equal('lb_cookie')
attributes['stickiness.enabled'].should.equal('true')
# These new values should be in the full attribute list
response = conn.describe_target_group_attributes(
TargetGroupArn=target_group_arn)
response['Attributes'].should.have.length_of(3)
attributes = {attr['Key']: attr['Value']
for attr in response['Attributes']}
attributes['stickiness.type'].should.equal('lb_cookie')
attributes['stickiness.enabled'].should.equal('true')
@mock_elbv2
@mock_ec2
def test_handle_listener_rules():
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'}])
load_balancer_arn = response.get('LoadBalancers')[0].get('LoadBalancerArn')
# Can't create a target group with an invalid protocol
with assert_raises(ClientError):
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'})
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'
created_rule = conn.create_rule(
ListenerArn=http_listener_arn,
Priority=priority,
Conditions=[{
'Field': 'host-header',
'Values': [host]
},
{
'Field': 'path-pattern',
'Values': [path_pattern]
}],
Actions=[{
'TargetGroupArn': target_group.get('TargetGroupArn'),
'Type': 'forward'
}]
)['Rules'][0]
created_rule['Priority'].should.equal('100')
# check if rules is sorted by priority
priority = 50
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]
}],
Actions=[{
'TargetGroupArn': target_group.get('TargetGroupArn'),
'Type': 'forward'
}]
)
# test for PriorityInUse
with assert_raises(ClientError):
conn.create_rule(
ListenerArn=http_listener_arn,
Priority=priority,
Conditions=[{
'Field': 'host-header',
'Values': [host]
},
{
'Field': 'path-pattern',
'Values': [path_pattern]
}],
Actions=[{
'TargetGroupArn': target_group.get('TargetGroupArn'),
'Type': 'forward'
}]
)
# test for describe listeners
obtained_rules = conn.describe_rules(ListenerArn=http_listener_arn)
len(obtained_rules['Rules']).should.equal(3)
priorities = [rule['Priority'] for rule in obtained_rules['Rules']]
priorities.should.equal(['50', '100', 'default'])
first_rule = obtained_rules['Rules'][0]
second_rule = obtained_rules['Rules'][1]
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.should.have.key('NextMarker')
following_rules['Rules'][0]['RuleArn'].should_not.equal(
obtained_rules['Rules'][0]['RuleArn'])
# test for invalid describe rule request
with assert_raises(ClientError):
conn.describe_rules()
with assert_raises(ClientError):
conn.describe_rules(RuleArns=[])
with assert_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'
modified_rule = conn.modify_rule(
RuleArn=first_rule['RuleArn'],
Conditions=[{
'Field': 'host-header',
'Values': [new_host]
},
{
'Field': 'path-pattern',
'Values': [new_path_pattern]
}]
)['Rules'][0]
rules = conn.describe_rules(ListenerArn=http_listener_arn)
obtained_rule = rules['Rules'][0]
modified_rule.should.equal(obtained_rule)
obtained_rule['Conditions'][0]['Values'][0].should.equal(new_host)
obtained_rule['Conditions'][1]['Values'][0].should.equal(new_path_pattern)
obtained_rule['Actions'][0]['TargetGroupArn'].should.equal(
target_group.get('TargetGroupArn'))
# modify priority
conn.set_rule_priorities(
RulePriorities=[
{'RuleArn': first_rule['RuleArn'],
'Priority': int(first_rule['Priority']) - 1}
]
)
with assert_raises(ClientError):
conn.set_rule_priorities(
RulePriorities=[
{'RuleArn': first_rule['RuleArn'], 'Priority': 999},
{'RuleArn': second_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(2)
# test for invalid action type
safe_priority = 2
with assert_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 assert_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 assert_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 assert_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 assert_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
@mock_ec2
def test_describe_invalid_target_group():
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'}])
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'})
# Check error raises correctly
with assert_raises(ClientError):
conn.describe_target_groups(Names=['invalid'])
@mock_elbv2
@mock_ec2
def test_describe_target_groups_no_arguments():
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'}])
response.get('LoadBalancers')[0].get('LoadBalancerArn')
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'})
assert len(conn.describe_target_groups()['TargetGroups']) == 1
@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(5)
resp = client.describe_ssl_policies(Names=['ELBSecurityPolicy-TLS-1-2-2017-01', 'ELBSecurityPolicy-2016-08'])
len(resp['SslPolicies']).should.equal(2)
@mock_elbv2
@mock_ec2
def test_set_ip_address_type():
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')
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']
# Internal LBs cant be dualstack yet
with assert_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 assert_raises(ClientError):
client.set_security_groups(
LoadBalancerArn=arn,
SecurityGroups=['non_existant']
)
@mock_elbv2
@mock_ec2
def test_set_subnets():
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')
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='us-east-1a')
subnet2 = ec2.create_subnet(
VpcId=vpc.id,
CidrBlock='172.28.7.64/26',
AvailabilityZone='us-east-1b')
subnet3 = ec2.create_subnet(
VpcId=vpc.id,
CidrBlock='172.28.7.192/26',
AvailabilityZone='us-east-1c')
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_subnets(
LoadBalancerArn=arn,
Subnets=[subnet1.id, subnet2.id, subnet3.id]
)
resp = client.describe_load_balancers(LoadBalancerArns=[arn])
len(resp['LoadBalancers'][0]['AvailabilityZones']).should.equal(3)
# Only 1 AZ
with assert_raises(ClientError):
client.set_subnets(
LoadBalancerArn=arn,
Subnets=[subnet1.id]
)
# Multiple subnets in same AZ
with assert_raises(ClientError):
client.set_subnets(
LoadBalancerArn=arn,
Subnets=[subnet1.id, subnet2.id, subnet2.id]
)
@mock_elbv2
@mock_ec2
def test_set_subnets():
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')
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.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_target_group():
client = boto3.client('elbv2', region_name='us-east-1')
ec2 = boto3.resource('ec2', region_name='us-east-1')
vpc = ec2.create_vpc(CidrBlock='172.28.7.0/24', InstanceTenancy='default')
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'})
arn = response.get('TargetGroups')[0]['TargetGroupArn']
client.modify_target_group(
TargetGroupArn=arn,
HealthCheckProtocol='HTTPS',
HealthCheckPort='8081',
HealthCheckPath='/status',
HealthCheckIntervalSeconds=10,
HealthCheckTimeoutSeconds=10,
HealthyThresholdCount=10,
UnhealthyThresholdCount=4,
Matcher={'HttpCode': '200-399'}
)
response = client.describe_target_groups(
TargetGroupArns=[arn]
)
response['TargetGroups'][0]['Matcher']['HttpCode'].should.equal('200-399')
response['TargetGroups'][0]['HealthCheckIntervalSeconds'].should.equal(10)
response['TargetGroups'][0]['HealthCheckPath'].should.equal('/status')
response['TargetGroups'][0]['HealthCheckPort'].should.equal('8081')
response['TargetGroups'][0]['HealthCheckProtocol'].should.equal('HTTPS')
response['TargetGroups'][0]['HealthCheckTimeoutSeconds'].should.equal(10)
response['TargetGroups'][0]['HealthyThresholdCount'].should.equal(10)
response['TargetGroups'][0]['UnhealthyThresholdCount'].should.equal(4)
@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']
response = acm.request_certificate(
DomainName='google.com',
SubjectAlternativeNames=['google.com', 'www.google.com', 'mail.google.com'],
)
google_arn = response['CertificateArn']
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': google_arn, 'IsDefault': False},
{'CertificateArn': yahoo_arn, 'IsDefault': True}
],
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(2)
# 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)
# No default cert
with assert_raises(ClientError):
client.modify_listener(
ListenerArn=listener_arn,
Port=443,
Protocol='HTTPS',
SslPolicy='ELBSecurityPolicy-TLS-1-2-2017-01',
Certificates=[
{'CertificateArn': google_arn, 'IsDefault': False}
],
DefaultActions=[
{'Type': 'forward', 'TargetGroupArn': target_group_arn}
]
)
# Bad cert
with assert_raises(ClientError):
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}
]
)
@mock_ec2
@mock_elbv2
@mock_cloudformation
def test_create_target_groups_through_cloudformation():
cfn_conn = boto3.client('cloudformation', region_name='us-east-1')
elbv2_client = boto3.client('elbv2', region_name='us-east-1')
# test that setting a name manually as well as letting cloudformation create a name both work
# this is a special case because test groups have a name length limit of 22 characters, and must be unique
# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-elasticloadbalancingv2-targetgroup.html#cfn-elasticloadbalancingv2-targetgroup-name
template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Description": "ECS Cluster Test CloudFormation",
"Resources": {
"testVPC": {
"Type": "AWS::EC2::VPC",
"Properties": {
"CidrBlock": "10.0.0.0/16",
},
},
"testGroup1": {
"Type": "AWS::ElasticLoadBalancingV2::TargetGroup",
"Properties": {
"Port": 80,
"Protocol": "HTTP",
"VpcId": {"Ref": "testVPC"},
},
},
"testGroup2": {
"Type": "AWS::ElasticLoadBalancingV2::TargetGroup",
"Properties": {
"Port": 90,
"Protocol": "HTTP",
"VpcId": {"Ref": "testVPC"},
},
},
"testGroup3": {
"Type": "AWS::ElasticLoadBalancingV2::TargetGroup",
"Properties": {
"Name": "MyTargetGroup",
"Port": 70,
"Protocol": "HTTPS",
"VpcId": {"Ref": "testVPC"},
},
},
}
}
template_json = json.dumps(template)
cfn_conn.create_stack(
StackName="test-stack",
TemplateBody=template_json,
)
describe_target_groups_response = elbv2_client.describe_target_groups()
target_group_dicts = describe_target_groups_response['TargetGroups']
assert len(target_group_dicts) == 3
# there should be 2 target groups with the same prefix of 10 characters (since the random suffix is 12)
# and one named MyTargetGroup
assert len([tg for tg in target_group_dicts if tg['TargetGroupName'] == 'MyTargetGroup']) == 1
assert len(
[tg for tg in target_group_dicts if tg['TargetGroupName'].startswith('test-stack')]
) == 2
@mock_elbv2
@mock_ec2
def test_redirect_action_listener_rule():
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.128/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'}])
load_balancer_arn = response.get('LoadBalancers')[0].get('LoadBalancerArn')
response = conn.create_listener(LoadBalancerArn=load_balancer_arn,
Protocol='HTTP',
Port=80,
DefaultActions=[
{'Type': 'redirect',
'RedirectConfig': {
'Protocol': 'HTTPS',
'Port': '443',
'StatusCode': 'HTTP_301'
}}])
listener = response.get('Listeners')[0]
expected_default_actions = [{
'Type': 'redirect',
'RedirectConfig': {
'Protocol': 'HTTPS',
'Port': '443',
'StatusCode': 'HTTP_301'
}
}]
listener.get('DefaultActions').should.equal(expected_default_actions)
listener_arn = listener.get('ListenerArn')
describe_rules_response = conn.describe_rules(ListenerArn=listener_arn)
describe_rules_response['Rules'][0]['Actions'].should.equal(expected_default_actions)
describe_listener_response = conn.describe_listeners(ListenerArns=[listener_arn, ])
describe_listener_actions = describe_listener_response['Listeners'][0]['DefaultActions']
describe_listener_actions.should.equal(expected_default_actions)
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(expected_default_actions)
@mock_elbv2
@mock_cloudformation
def test_redirect_action_listener_rule_cloudformation():
cnf_conn = boto3.client('cloudformation', region_name='us-east-1')
elbv2_client = boto3.client('elbv2', region_name='us-east-1')
template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Description": "ECS Cluster Test CloudFormation",
"Resources": {
"testVPC": {
"Type": "AWS::EC2::VPC",
"Properties": {
"CidrBlock": "10.0.0.0/16",
},
},
"subnet1": {
"Type": "AWS::EC2::Subnet",
"Properties": {
"CidrBlock": "10.0.0.0/24",
"VpcId": {"Ref": "testVPC"},
"AvalabilityZone": "us-east-1b",
},
},
"subnet2": {
"Type": "AWS::EC2::Subnet",
"Properties": {
"CidrBlock": "10.0.1.0/24",
"VpcId": {"Ref": "testVPC"},
"AvalabilityZone": "us-east-1b",
},
},
"testLb": {
"Type": "AWS::ElasticLoadBalancingV2::LoadBalancer",
"Properties": {
"Name": "my-lb",
"Subnets": [{"Ref": "subnet1"}, {"Ref": "subnet2"}],
"Type": "application",
"SecurityGroups": [],
}
},
"testListener": {
"Type": "AWS::ElasticLoadBalancingV2::Listener",
"Properties": {
"LoadBalancerArn": {"Ref": "testLb"},
"Port": 80,
"Protocol": "HTTP",
"DefaultActions": [{
"Type": "redirect",
"RedirectConfig": {
"Port": "443",
"Protocol": "HTTPS",
"StatusCode": "HTTP_301",
}
}]
}
}
}
}
template_json = json.dumps(template)
cnf_conn.create_stack(StackName="test-stack", TemplateBody=template_json)
describe_load_balancers_response = elbv2_client.describe_load_balancers(Names=['my-lb',])
describe_load_balancers_response['LoadBalancers'].should.have.length_of(1)
load_balancer_arn = describe_load_balancers_response['LoadBalancers'][0]['LoadBalancerArn']
describe_listeners_response = elbv2_client.describe_listeners(LoadBalancerArn=load_balancer_arn)
describe_listeners_response['Listeners'].should.have.length_of(1)
describe_listeners_response['Listeners'][0]['DefaultActions'].should.equal([{
'Type': 'redirect',
'RedirectConfig': {
'Port': '443', 'Protocol': 'HTTPS', 'StatusCode': 'HTTP_301',
}
},])
@mock_elbv2
@mock_ec2
def test_cognito_action_listener_rule():
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.128/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'}])
load_balancer_arn = response.get('LoadBalancers')[0].get('LoadBalancerArn')
action = {
'Type': 'authenticate-cognito',
'AuthenticateCognitoConfig': {
'UserPoolArn': 'arn:aws:cognito-idp:us-east-1:123456789012:userpool/us-east-1_ABCD1234',
'UserPoolClientId': 'abcd1234abcd',
'UserPoolDomain': 'testpool',
}
}
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')
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_cloudformation
def test_cognito_action_listener_rule_cloudformation():
cnf_conn = boto3.client('cloudformation', region_name='us-east-1')
elbv2_client = boto3.client('elbv2', region_name='us-east-1')
template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Description": "ECS Cluster Test CloudFormation",
"Resources": {
"testVPC": {
"Type": "AWS::EC2::VPC",
"Properties": {
"CidrBlock": "10.0.0.0/16",
},
},
"subnet1": {
"Type": "AWS::EC2::Subnet",
"Properties": {
"CidrBlock": "10.0.0.0/24",
"VpcId": {"Ref": "testVPC"},
"AvalabilityZone": "us-east-1b",
},
},
"subnet2": {
"Type": "AWS::EC2::Subnet",
"Properties": {
"CidrBlock": "10.0.1.0/24",
"VpcId": {"Ref": "testVPC"},
"AvalabilityZone": "us-east-1b",
},
},
"testLb": {
"Type": "AWS::ElasticLoadBalancingV2::LoadBalancer",
"Properties": {
"Name": "my-lb",
"Subnets": [{"Ref": "subnet1"}, {"Ref": "subnet2"}],
"Type": "application",
"SecurityGroups": [],
}
},
"testListener": {
"Type": "AWS::ElasticLoadBalancingV2::Listener",
"Properties": {
"LoadBalancerArn": {"Ref": "testLb"},
"Port": 80,
"Protocol": "HTTP",
"DefaultActions": [{
"Type": "authenticate-cognito",
"AuthenticateCognitoConfig": {
'UserPoolArn': 'arn:aws:cognito-idp:us-east-1:123456789012:userpool/us-east-1_ABCD1234',
'UserPoolClientId': 'abcd1234abcd',
'UserPoolDomain': 'testpool',
}
}]
}
}
}
}
template_json = json.dumps(template)
cnf_conn.create_stack(StackName="test-stack", TemplateBody=template_json)
describe_load_balancers_response = elbv2_client.describe_load_balancers(Names=['my-lb',])
load_balancer_arn = describe_load_balancers_response['LoadBalancers'][0]['LoadBalancerArn']
describe_listeners_response = elbv2_client.describe_listeners(LoadBalancerArn=load_balancer_arn)
describe_listeners_response['Listeners'].should.have.length_of(1)
describe_listeners_response['Listeners'][0]['DefaultActions'].should.equal([{
'Type': 'authenticate-cognito',
"AuthenticateCognitoConfig": {
'UserPoolArn': 'arn:aws:cognito-idp:us-east-1:123456789012:userpool/us-east-1_ABCD1234',
'UserPoolClientId': 'abcd1234abcd',
'UserPoolDomain': 'testpool',
}
},])
@mock_elbv2
@mock_ec2
def test_fixed_response_action_listener_rule():
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.128/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'}])
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')
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_cloudformation
def test_fixed_response_action_listener_rule_cloudformation():
cnf_conn = boto3.client('cloudformation', region_name='us-east-1')
elbv2_client = boto3.client('elbv2', region_name='us-east-1')
template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Description": "ECS Cluster Test CloudFormation",
"Resources": {
"testVPC": {
"Type": "AWS::EC2::VPC",
"Properties": {
"CidrBlock": "10.0.0.0/16",
},
},
"subnet1": {
"Type": "AWS::EC2::Subnet",
"Properties": {
"CidrBlock": "10.0.0.0/24",
"VpcId": {"Ref": "testVPC"},
"AvalabilityZone": "us-east-1b",
},
},
"subnet2": {
"Type": "AWS::EC2::Subnet",
"Properties": {
"CidrBlock": "10.0.1.0/24",
"VpcId": {"Ref": "testVPC"},
"AvalabilityZone": "us-east-1b",
},
},
"testLb": {
"Type": "AWS::ElasticLoadBalancingV2::LoadBalancer",
"Properties": {
"Name": "my-lb",
"Subnets": [{"Ref": "subnet1"}, {"Ref": "subnet2"}],
"Type": "application",
"SecurityGroups": [],
}
},
"testListener": {
"Type": "AWS::ElasticLoadBalancingV2::Listener",
"Properties": {
"LoadBalancerArn": {"Ref": "testLb"},
"Port": 80,
"Protocol": "HTTP",
"DefaultActions": [{
"Type": "fixed-response",
"FixedResponseConfig": {
'ContentType': 'text/plain',
'MessageBody': 'This page does not exist',
'StatusCode': '404',
}
}]
}
}
}
}
template_json = json.dumps(template)
cnf_conn.create_stack(StackName="test-stack", TemplateBody=template_json)
describe_load_balancers_response = elbv2_client.describe_load_balancers(Names=['my-lb',])
load_balancer_arn = describe_load_balancers_response['LoadBalancers'][0]['LoadBalancerArn']
describe_listeners_response = elbv2_client.describe_listeners(LoadBalancerArn=load_balancer_arn)
describe_listeners_response['Listeners'].should.have.length_of(1)
describe_listeners_response['Listeners'][0]['DefaultActions'].should.equal([{
'Type': 'fixed-response',
"FixedResponseConfig": {
'ContentType': 'text/plain',
'MessageBody': 'This page does not exist',
'StatusCode': '404',
}
},])
@mock_elbv2
@mock_ec2
def test_fixed_response_action_listener_rule_validates_status_code():
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.128/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'}])
load_balancer_arn = response.get('LoadBalancers')[0].get('LoadBalancerArn')
missing_status_code_action = {
'Type': 'fixed-response',
'FixedResponseConfig': {
'ContentType': 'text/plain',
'MessageBody': 'This page does not exist',
}
}
with assert_raises(ParamValidationError):
conn.create_listener(LoadBalancerArn=load_balancer_arn,
Protocol='HTTP',
Port=80,
DefaultActions=[missing_status_code_action])
invalid_status_code_action = {
'Type': 'fixed-response',
'FixedResponseConfig': {
'ContentType': 'text/plain',
'MessageBody': 'This page does not exist',
'StatusCode': '100'
}
}
@mock_elbv2
@mock_ec2
def test_fixed_response_action_listener_rule_validates_status_code():
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.128/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'}])
load_balancer_arn = response.get('LoadBalancers')[0].get('LoadBalancerArn')
missing_status_code_action = {
'Type': 'fixed-response',
'FixedResponseConfig': {
'ContentType': 'text/plain',
'MessageBody': 'This page does not exist',
}
}
with assert_raises(ParamValidationError):
conn.create_listener(LoadBalancerArn=load_balancer_arn,
Protocol='HTTP',
Port=80,
DefaultActions=[missing_status_code_action])
invalid_status_code_action = {
'Type': 'fixed-response',
'FixedResponseConfig': {
'ContentType': 'text/plain',
'MessageBody': 'This page does not exist',
'StatusCode': '100'
}
}
with assert_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.exception.response['Error']['Code'].should.equal('ValidationError')
@mock_elbv2
@mock_ec2
def test_fixed_response_action_listener_rule_validates_content_type():
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.128/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'}])
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 assert_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.exception.response['Error']['Code'].should.equal('InvalidLoadBalancerAction')