from __future__ import unicode_literals
import boto3
import botocore
from botocore.exceptions import ClientError
from nose.tools import assert_raises
import sure # noqa
from moto import mock_elbv2, mock_ec2
@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.192/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.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'}])
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.192/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.192/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.192/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_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')
# 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_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.192/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.192/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.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'}])
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_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.192/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.192/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_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.192/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'])