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

Merge pull request #1019 from JackDanger/elbv2

Browse Source 
Implemented core endpoints of ELBv2
This commit is contained in:
Jack Danger 2017-08-02 13:54:20 -07:00 committed by GitHub
commit b061ac4d28
12 changed files with 1585 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/index.rst
View File

@ -43,6 +43,7 @@ Currently implemented Services:
| ECS | @mock_ecs | basic endpoints done |
+-----------------------+---------------------+-----------------------------------+
| ELB | @mock_elb | core endpoints done |
| | @mock_elbv2 | core endpoints done |
+-----------------------+---------------------+-----------------------------------+
| EMR | @mock_emr | core endpoints done |
+-----------------------+---------------------+-----------------------------------+

1
moto/__init__.py
View File

@ -17,6 +17,7 @@ from .ec2 import mock_ec2, mock_ec2_deprecated # flake8: noqa
from .ecr import mock_ecr, mock_ecr_deprecated # flake8: noqa
from .ecs import mock_ecs, mock_ecs_deprecated # flake8: noqa
from .elb import mock_elb, mock_elb_deprecated # flake8: noqa
from .elbv2 import mock_elbv2 # flake8: noqa
from .emr import mock_emr, mock_emr_deprecated # flake8: noqa
from .events import mock_events # flake8: noqa
from .glacier import mock_glacier, mock_glacier_deprecated # flake8: noqa

2
moto/backends.py
View File

@ -13,6 +13,7 @@ from moto.ec2 import ec2_backends
from moto.ecr import ecr_backends
from moto.ecs import ecs_backends
from moto.elb import elb_backends
from moto.elbv2 import elbv2_backends
from moto.emr import emr_backends
from moto.events import events_backends
from moto.glacier import glacier_backends
@ -43,6 +44,7 @@ BACKENDS = {
'ecr': ecr_backends,
'ecs': ecs_backends,
'elb': elb_backends,
'elbv2': elbv2_backends,
'events': events_backends,
'emr': emr_backends,
'glacier': glacier_backends,

38
moto/elb/urls.py
View File

@ -1,10 +1,44 @@
from __future__ import unicode_literals
from .responses import ELBResponse
from six.moves.urllib.parse import parse_qs
from botocore.awsrequest import AWSPreparedRequest
from moto.elb.responses import ELBResponse
from moto.elbv2.responses import ELBV2Response
def api_version_elb_backend(*args, **kwargs):
"""
ELB and ELBV2 (Classic and Application load balancers) use the same
hostname and url space. To differentiate them we must read the
`Version` parameter out of the url-encoded request body. TODO: There
has _got_ to be a better way to do this. Please help us think of
one.
"""
request = args[0]
if hasattr(request, 'values'):
# boto3
version = request.values.get('Version')
elif isinstance(request, AWSPreparedRequest):
# boto in-memory
version = parse_qs(request.body).get('Version')[0]
else:
# boto in server mode
request.parse_request()
version = request.querystring.get('Version')[0]
if '2012-06-01' == version:
return ELBResponse.dispatch(*args, **kwargs)
elif '2015-12-01' == version:
return ELBV2Response.dispatch(*args, **kwargs)
else:
raise Exception("Unknown ELB API version: {}".format(version))
url_bases = [
"https?://elasticloadbalancing.(.+).amazonaws.com",
]
url_paths = {
'{0}/$': ELBResponse.dispatch,
'{0}/$': api_version_elb_backend,
}

6
moto/elbv2/__init__.py Normal file
View File

@ -0,0 +1,6 @@
from __future__ import unicode_literals
from .models import elbv2_backends
from ..core.models import base_decorator
elb_backend = elbv2_backends['us-east-1']
mock_elbv2 = base_decorator(elbv2_backends)

103
moto/elbv2/exceptions.py Normal file
View File

@ -0,0 +1,103 @@
from __future__ import unicode_literals
from moto.core.exceptions import RESTError
class ELBClientError(RESTError):
code = 400
class DuplicateTagKeysError(ELBClientError):
def __init__(self, cidr):
super(DuplicateTagKeysError, self).__init__(
"DuplicateTagKeys",
"Tag key was specified more than once: {0}"
.format(cidr))
class LoadBalancerNotFoundError(ELBClientError):
def __init__(self):
super(LoadBalancerNotFoundError, self).__init__(
"LoadBalancerNotFound",
"The specified load balancer does not exist.")
class ListenerNotFoundError(ELBClientError):
def __init__(self):
super(ListenerNotFoundError, self).__init__(
"ListenerNotFound",
"The specified listener does not exist.")
class SubnetNotFoundError(ELBClientError):
def __init__(self):
super(SubnetNotFoundError, self).__init__(
"SubnetNotFound",
"The specified subnet does not exist.")
class TargetGroupNotFoundError(ELBClientError):
def __init__(self):
super(TooManyTagsError, self).__init__(
"TargetGroupNotFound",
"The specified target group does not exist.")
class TooManyTagsError(ELBClientError):
def __init__(self):
super(TooManyTagsError, self).__init__(
"TooManyTagsError",
"The quota for the number of tags that can be assigned to a load balancer has been reached")
class BadHealthCheckDefinition(ELBClientError):
def __init__(self):
super(BadHealthCheckDefinition, self).__init__(
"ValidationError",
"HealthCheck Target must begin with one of HTTP, TCP, HTTPS, SSL")
class DuplicateListenerError(ELBClientError):
def __init__(self):
super(DuplicateListenerError, self).__init__(
"DuplicateListener",
"A listener with the specified port already exists.")
class DuplicateLoadBalancerName(ELBClientError):
def __init__(self):
super(DuplicateLoadBalancerName, self).__init__(
"DuplicateLoadBalancerName",
"A load balancer with the specified name already exists.")
class DuplicateTargetGroupName(ELBClientError):
def __init__(self):
super(DuplicateTargetGroupName, self).__init__(
"DuplicateTargetGroupName",
"A target group with the specified name already exists.")
class InvalidTargetError(ELBClientError):
def __init__(self):
super(InvalidTargetError, self).__init__(
"InvalidTarget",
"The specified target does not exist or is not in the same VPC as the target group.")
class EmptyListenersError(ELBClientError):
def __init__(self):
super(EmptyListenersError, self).__init__(
"ValidationError",
"Listeners cannot be empty")

312
moto/elbv2/models.py Normal file
View File

@ -0,0 +1,312 @@
from __future__ import unicode_literals
import datetime
from moto.compat import OrderedDict
from moto.core import BaseBackend, BaseModel
from moto.ec2.models import ec2_backends
from .exceptions import (
DuplicateLoadBalancerName,
DuplicateListenerError,
DuplicateTargetGroupName,
InvalidTargetError,
ListenerNotFoundError,
LoadBalancerNotFoundError,
SubnetNotFoundError,
TargetGroupNotFoundError,
TooManyTagsError,
)
class FakeHealthStatus(BaseModel):
def __init__(self, instance_id, port, health_port, status, reason=None):
self.instance_id = instance_id
self.port = port
self.health_port = health_port
self.status = status
self.reason = reason
class FakeTargetGroup(BaseModel):
def __init__(self,
name,
arn,
vpc_id,
protocol,
port,
healthcheck_protocol,
healthcheck_port,
healthcheck_path,
healthcheck_interval_seconds,
healthcheck_timeout_seconds,
healthy_threshold_count,
unhealthy_threshold_count):
self.name = name
self.arn = arn
self.vpc_id = vpc_id
self.protocol = protocol
self.port = port
self.healthcheck_protocol = healthcheck_protocol
self.healthcheck_port = healthcheck_port
self.healthcheck_path = healthcheck_path
self.healthcheck_interval_seconds = healthcheck_interval_seconds
self.healthcheck_timeout_seconds = healthcheck_timeout_seconds
self.healthy_threshold_count = healthy_threshold_count
self.unhealthy_threshold_count = unhealthy_threshold_count
self.load_balancer_arns = []
self.targets = OrderedDict()
def register(self, targets):
for target in targets:
self.targets[target['id']] = {
'id': target['id'],
'port': target.get('port', self.port),
}
def deregister(self, targets):
for target in targets:
t = self.targets.pop(target['id'])
if not t:
raise InvalidTargetError()
def health_for(self, target):
t = self.targets.get(target['id'])
if t is None:
raise InvalidTargetError()
return FakeHealthStatus(t['id'], t['port'], self.healthcheck_port, 'healthy')
class FakeListener(BaseModel):
def __init__(self, load_balancer_arn, arn, protocol, port, ssl_policy, certificate, default_actions):
self.load_balancer_arn = load_balancer_arn
self.arn = arn
self.protocol = protocol.upper()
self.port = port
self.ssl_policy = ssl_policy
self.certificate = certificate
self.default_actions = default_actions
class FakeBackend(BaseModel):
def __init__(self, instance_port):
self.instance_port = instance_port
self.policy_names = []
def __repr__(self):
return "FakeBackend(inp: %s, policies: %s)" % (self.instance_port, self.policy_names)
class FakeLoadBalancer(BaseModel):
def __init__(self, name, security_groups, subnets, vpc_id, arn, dns_name, scheme='internet-facing'):
self.name = name
self.created_time = datetime.datetime.now()
self.scheme = scheme
self.security_groups = security_groups
self.subnets = subnets or []
self.vpc_id = vpc_id
self.listeners = OrderedDict()
self.tags = {}
self.arn = arn
self.dns_name = dns_name
@property
def physical_resource_id(self):
return self.name
def add_tag(self, key, value):
if len(self.tags) >= 10 and key not in self.tags:
raise TooManyTagsError()
self.tags[key] = value
def list_tags(self):
return self.tags
def remove_tag(self, key):
if key in self.tags:
del self.tags[key]
def delete(self, region):
''' Not exposed as part of the ELB API - used for CloudFormation. '''
elbv2_backends[region].delete_load_balancer(self.arn)
class ELBv2Backend(BaseBackend):
def __init__(self, region_name=None):
self.region_name = region_name
self.target_groups = OrderedDict()
self.load_balancers = OrderedDict()
def reset(self):
region_name = self.region_name
self.__dict__ = {}
self.__init__(region_name)
def create_load_balancer(self, name, security_groups, subnet_ids, scheme='internet-facing'):
vpc_id = None
ec2_backend = ec2_backends[self.region_name]
subnets = []
if not subnet_ids:
raise SubnetNotFoundError()
for subnet_id in subnet_ids:
subnet = ec2_backend.get_subnet(subnet_id)
if subnet is None:
raise SubnetNotFoundError()
subnets.append(subnet)
vpc_id = subnets[0].vpc_id
arn = "arn:aws:elasticloadbalancing:%s:1:loadbalancer/%s/50dc6c495c0c9188" % (self.region_name, name)
dns_name = "%s-1.%s.elb.amazonaws.com" % (name, self.region_name)
if arn in self.load_balancers:
raise DuplicateLoadBalancerName()
new_load_balancer = FakeLoadBalancer(
name=name,
security_groups=security_groups,
arn=arn,
scheme=scheme,
subnets=subnets,
vpc_id=vpc_id,
dns_name=dns_name)
self.load_balancers[arn] = new_load_balancer
return new_load_balancer
def create_target_group(self, name, **kwargs):
for target_group in self.target_groups.values():
if target_group.name == name:
raise DuplicateTargetGroupName()
arn = "arn:aws:elasticloadbalancing:%s:1:targetgroup/%s/50dc6c495c0c9188" % (self.region_name, name)
target_group = FakeTargetGroup(name, arn, **kwargs)
self.target_groups[target_group.arn] = target_group
return target_group
def create_listener(self, load_balancer_arn, protocol, port, ssl_policy, certificate, default_actions):
balancer = self.load_balancers.get(load_balancer_arn)
if balancer is None:
raise LoadBalancerNotFoundError()
if port in balancer.listeners:
raise DuplicateListenerError()
arn = load_balancer_arn.replace(':loadbalancer/', ':listener/') + "/%s%s" % (port, id(self))
listener = FakeListener(load_balancer_arn, arn, protocol, port, ssl_policy, certificate, default_actions)
balancer.listeners[listener.arn] = listener
return listener
def describe_load_balancers(self, arns, names):
balancers = self.load_balancers.values()
arns = arns or []
names = names or []
if not arns and not names:
return balancers
matched_balancers = []
matched_balancer = None
for arn in arns:
for balancer in balancers:
if balancer.arn == arn:
matched_balancer = balancer
if matched_balancer is None:
raise LoadBalancerNotFoundError()
elif matched_balancer not in matched_balancers:
matched_balancers.append(matched_balancer)
for name in names:
for balancer in balancers:
if balancer.name == name:
matched_balancer = balancer
if matched_balancer is None:
raise LoadBalancerNotFoundError()
elif matched_balancer not in matched_balancers:
matched_balancers.append(matched_balancer)
return matched_balancers
def describe_target_groups(self, load_balancer_arn, target_group_arns, names):
if load_balancer_arn:
if load_balancer_arn not in self.load_balancers:
raise LoadBalancerNotFoundError()
return [tg for tg in self.target_groups.values()
if load_balancer_arn in tg.load_balancer_arns]
if target_group_arns:
try:
return [self.target_groups[arn] for arn in target_group_arns]
except KeyError:
raise TargetGroupNotFoundError()
if names:
matched = []
for name in names:
found = None
for target_group in self.target_groups:
if target_group.name == name:
found = target_group
if not found:
raise TargetGroupNotFoundError()
matched.append(found)
return matched
return self.target_groups.values()
def describe_listeners(self, load_balancer_arn, listener_arns):
if load_balancer_arn:
if load_balancer_arn not in self.load_balancers:
raise LoadBalancerNotFoundError()
return self.load_balancers.get(load_balancer_arn).listeners.values()
matched = []
for load_balancer in self.load_balancers.values():
for listener_arn in listener_arns:
listener = load_balancer.listeners.get(listener_arn)
if not listener:
raise ListenerNotFoundError()
matched.append(listener)
return matched
def delete_load_balancer(self, arn):
self.load_balancers.pop(arn, None)
def delete_target_group(self, target_group_arn):
target_group = self.target_groups.pop(target_group_arn)
if target_group:
return target_group
raise TargetGroupNotFoundError()
def delete_listener(self, listener_arn):
for load_balancer in self.load_balancers.values():
listener = load_balancer.listeners.pop(listener_arn)
if listener:
return listener
raise ListenerNotFoundError()
def register_targets(self, target_group_arn, instances):
target_group = self.target_groups.get(target_group_arn)
if target_group is None:
raise TargetGroupNotFoundError()
target_group.register(instances)
def deregister_targets(self, target_group_arn, instances):
target_group = self.target_groups.get(target_group_arn)
if target_group is None:
raise TargetGroupNotFoundError()
target_group.deregister(instances)
def describe_target_health(self, target_group_arn, targets):
target_group = self.target_groups.get(target_group_arn)
if target_group is None:
raise TargetGroupNotFoundError()
if not targets:
targets = target_group.targets.values()
return [target_group.health_for(target) for target in targets]
elbv2_backends = {}
for region in ec2_backends.keys():
elbv2_backends[region] = ELBv2Backend(region)

649
moto/elbv2/responses.py Normal file
View File

@ -0,0 +1,649 @@
from __future__ import unicode_literals
from moto.core.responses import BaseResponse
from .models import elbv2_backends
from .exceptions import DuplicateTagKeysError, LoadBalancerNotFoundError
class ELBV2Response(BaseResponse):
@property
def elbv2_backend(self):
return elbv2_backends[self.region]
def create_load_balancer(self):
load_balancer_name = self._get_param('Name')
subnet_ids = self._get_multi_param("Subnets.member")
security_groups = self._get_multi_param("SecurityGroups.member")
scheme = self._get_param('Scheme')
load_balancer = self.elbv2_backend.create_load_balancer(
name=load_balancer_name,
security_groups=security_groups,
subnet_ids=subnet_ids,
scheme=scheme,
)
self._add_tags(load_balancer)
template = self.response_template(CREATE_LOAD_BALANCER_TEMPLATE)
return template.render(load_balancer=load_balancer)
def create_target_group(self):
name = self._get_param('Name')
vpc_id = self._get_param('VpcId')
protocol = self._get_param('Protocol')
port = self._get_param('Port')
healthcheck_protocol = self._get_param('HealthCheckProtocol', 'HTTP')
healthcheck_port = self._get_param('HealthCheckPort', 'traffic-port')
healthcheck_path = self._get_param('HealthCheckPath', '/')
healthcheck_interval_seconds = self._get_param('HealthCheckIntervalSeconds', '30')
healthcheck_timeout_seconds = self._get_param('HealthCheckTimeoutSeconds', '5')
healthy_threshold_count = self._get_param('HealthyThresholdCount', '5')
unhealthy_threshold_count = self._get_param('UnhealthyThresholdCount', '2')
target_group = self.elbv2_backend.create_target_group(
name,
vpc_id=vpc_id,
protocol=protocol,
port=port,
healthcheck_protocol=healthcheck_protocol,
healthcheck_port=healthcheck_port,
healthcheck_path=healthcheck_path,
healthcheck_interval_seconds=healthcheck_interval_seconds,
healthcheck_timeout_seconds=healthcheck_timeout_seconds,
healthy_threshold_count=healthy_threshold_count,
unhealthy_threshold_count=unhealthy_threshold_count,
)
template = self.response_template(CREATE_TARGET_GROUP_TEMPLATE)
return template.render(target_group=target_group)
def create_listener(self):
load_balancer_arn = self._get_param('LoadBalancerArn')
protocol = self._get_param('Protocol')
port = self._get_param('Port')
ssl_policy = self._get_param('SslPolicy', 'ELBSecurityPolicy-2016-08')
certificates = self._get_list_prefix('Certificates.member')
if certificates:
certificate = certificates[0].get('certificate_arn')
else:
certificate = None
default_actions = self._get_list_prefix('DefaultActions.member')
listener = self.elbv2_backend.create_listener(
load_balancer_arn=load_balancer_arn,
protocol=protocol,
port=port,
ssl_policy=ssl_policy,
certificate=certificate,
default_actions=default_actions)
template = self.response_template(CREATE_LISTENER_TEMPLATE)
return template.render(listener=listener)
def describe_load_balancers(self):
arns = self._get_multi_param("LoadBalancerArns.member")
names = self._get_multi_param("Names.member")
all_load_balancers = list(self.elbv2_backend.describe_load_balancers(arns, names))
marker = self._get_param('Marker')
all_names = [balancer.name for balancer in all_load_balancers]
if marker:
start = all_names.index(marker) + 1
else:
start = 0
page_size = self._get_param('PageSize', 50) # the default is 400, but using 50 to make testing easier
load_balancers_resp = all_load_balancers[start:start + page_size]
next_marker = None
if len(all_load_balancers) > start + page_size:
next_marker = load_balancers_resp[-1].name
template = self.response_template(DESCRIBE_LOAD_BALANCERS_TEMPLATE)
return template.render(load_balancers=load_balancers_resp, marker=next_marker)
def describe_target_groups(self):
load_balancer_arn = self._get_param('LoadBalancerArn')
target_group_arns = self._get_multi_param('TargetGroupArns.member')
names = self._get_multi_param('Names.member')
target_groups = self.elbv2_backend.describe_target_groups(load_balancer_arn, target_group_arns, names)
template = self.response_template(DESCRIBE_TARGET_GROUPS_TEMPLATE)
return template.render(target_groups=target_groups)
def describe_listeners(self):
load_balancer_arn = self._get_param('LoadBalancerArn')
listener_arns = self._get_multi_param('ListenerArns.member')
if not load_balancer_arn and not listener_arns:
raise LoadBalancerNotFoundError()
listeners = self.elbv2_backend.describe_listeners(load_balancer_arn, listener_arns)
template = self.response_template(DESCRIBE_LISTENERS_TEMPLATE)
return template.render(listeners=listeners)
def delete_load_balancer(self):
arn = self._get_param('LoadBalancerArn')
self.elbv2_backend.delete_load_balancer(arn)
template = self.response_template(DELETE_LOAD_BALANCER_TEMPLATE)
return template.render()
def delete_target_group(self):
arn = self._get_param('TargetGroupArn')
self.elbv2_backend.delete_target_group(arn)
template = self.response_template(DELETE_TARGET_GROUP_TEMPLATE)
return template.render()
def delete_listener(self):
arn = self._get_param('ListenerArn')
self.elbv2_backend.delete_listener(arn)
template = self.response_template(DELETE_LISTENER_TEMPLATE)
return template.render()
def register_targets(self):
target_group_arn = self._get_param('TargetGroupArn')
targets = self._get_list_prefix('Targets.member')
self.elbv2_backend.register_targets(target_group_arn, targets)
template = self.response_template(REGISTER_TARGETS_TEMPLATE)
return template.render()
def deregister_targets(self):
target_group_arn = self._get_param('TargetGroupArn')
targets = self._get_list_prefix('Targets.member')
self.elbv2_backend.deregister_targets(target_group_arn, targets)
template = self.response_template(DEREGISTER_TARGETS_TEMPLATE)
return template.render()
def describe_target_health(self):
target_group_arn = self._get_param('TargetGroupArn')
targets = self._get_list_prefix('Targets.member')
target_health_descriptions = self.elbv2_backend.describe_target_health(target_group_arn, targets)
template = self.response_template(DESCRIBE_TARGET_HEALTH_TEMPLATE)
return template.render(target_health_descriptions=target_health_descriptions)
def add_tags(self):
resource_arns = self._get_multi_param('ResourceArns.member')
for arn in resource_arns:
load_balancer = self.elbv2_backend.load_balancers.get(arn)
if not load_balancer:
raise LoadBalancerNotFoundError()
self._add_tags(load_balancer)
template = self.response_template(ADD_TAGS_TEMPLATE)
return template.render()
def remove_tags(self):
resource_arns = self._get_multi_param('ResourceArns.member')
tag_keys = self._get_multi_param('TagKeys.member')
for arn in resource_arns:
load_balancer = self.elbv2_backend.load_balancers.get(arn)
if not load_balancer:
raise LoadBalancerNotFoundError()
[load_balancer.remove_tag(key) for key in tag_keys]
template = self.response_template(REMOVE_TAGS_TEMPLATE)
return template.render()
def describe_tags(self):
elbs = []
for key, value in self.querystring.items():
if "ResourceArns.member" in key:
number = key.split('.')[2]
load_balancer_arn = self._get_param(
'ResourceArns.member.{0}'.format(number))
elb = self.elbv2_backend.load_balancers.get(load_balancer_arn)
if not elb:
raise LoadBalancerNotFoundError()
elbs.append(elb)
template = self.response_template(DESCRIBE_TAGS_TEMPLATE)
return template.render(load_balancers=elbs)
def _add_tags(self, elb):
tag_values = []
tag_keys = []
for t_key, t_val in sorted(self.querystring.items()):
if t_key.startswith('Tags.member.'):
if t_key.split('.')[3] == 'Key':
tag_keys.extend(t_val)
elif t_key.split('.')[3] == 'Value':
tag_values.extend(t_val)
counts = {}
for i in tag_keys:
counts[i] = tag_keys.count(i)
counts = sorted(counts.items(), key=lambda i: i[1], reverse=True)
if counts and counts[0][1] > 1:
# We have dupes...
raise DuplicateTagKeysError(counts[0])
for tag_key, tag_value in zip(tag_keys, tag_values):
elb.add_tag(tag_key, tag_value)
ADD_TAGS_TEMPLATE = """<AddTagsResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<AddTagsResult/>
<ResponseMetadata>
<RequestId>360e81f7-1100-11e4-b6ed-0f30EXAMPLE</RequestId>
</ResponseMetadata>
</AddTagsResponse>"""
REMOVE_TAGS_TEMPLATE = """<RemoveTagsResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<RemoveTagsResult/>
<ResponseMetadata>
<RequestId>360e81f7-1100-11e4-b6ed-0f30EXAMPLE</RequestId>
</ResponseMetadata>
</RemoveTagsResponse>"""
DESCRIBE_TAGS_TEMPLATE = """<DescribeTagsResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<DescribeTagsResult>
<TagDescriptions>
{% for load_balancer in load_balancers %}
<member>
<ResourceArn>{{ load_balancer.arn }}</ResourceArn>
<Tags>
{% for key, value in load_balancer.tags.items() %}
<member>
<Value>{{ value }}</Value>
<Key>{{ key }}</Key>
</member>
{% endfor %}
</Tags>
</member>
{% endfor %}
</TagDescriptions>
</DescribeTagsResult>
<ResponseMetadata>
<RequestId>360e81f7-1100-11e4-b6ed-0f30EXAMPLE</RequestId>
</ResponseMetadata>
</DescribeTagsResponse>"""
CREATE_LOAD_BALANCER_TEMPLATE = """<CreateLoadBalancerResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<CreateLoadBalancerResult>
<LoadBalancers>
<member>
<LoadBalancerArn>{{ load_balancer.arn }}</LoadBalancerArn>
<Scheme>{{ load_balancer.scheme }}</Scheme>
<LoadBalancerName>{{ load_balancer.name }}</LoadBalancerName>
<VpcId>{{ load_balancer.vpc_id }}</VpcId>
<CanonicalHostedZoneId>Z2P70J7EXAMPLE</CanonicalHostedZoneId>
<CreatedTime>{{ load_balancer.created_time }}</CreatedTime>
<AvailabilityZones>
{% for subnet in load_balancer.subnets %}
<member>
<SubnetId>{{ subnet.id }}</SubnetId>
<ZoneName>{{ subnet.availability_zone }}</ZoneName>
</member>
{% endfor %}
</AvailabilityZones>
<SecurityGroups>
{% for security_group in load_balancer.security_groups %}
<member>{{ security_group }}</member>
{% endfor %}
</SecurityGroups>
<DNSName>{{ load_balancer.dns_name }}</DNSName>
<State>
<Code>provisioning</Code>
</State>
<Type>application</Type>
</member>
</LoadBalancers>
</CreateLoadBalancerResult>
<ResponseMetadata>
<RequestId>32d531b2-f2d0-11e5-9192-3fff33344cfa</RequestId>
</ResponseMetadata>
</CreateLoadBalancerResponse>"""
CREATE_TARGET_GROUP_TEMPLATE = """<CreateTargetGroupResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<CreateTargetGroupResult>
<TargetGroups>
<member>
<TargetGroupArn>{{ target_group.arn }}</TargetGroupArn>
<TargetGroupName>{{ target_group.name }}</TargetGroupName>
<Protocol>{{ target_group.protocol }}</Protocol>
<Port>{{ target_group.port }}</Port>
<VpcId>{{ target_group.vpc_id }}</VpcId>
<HealthCheckProtocol>{{ target_group.health_check_protocol }}</HealthCheckProtocol>
<HealthCheckPort>{{ target_group.healthcheck_port }}</HealthCheckPort>
<HealthCheckPath>{{ target_group.healthcheck_path }}</HealthCheckPath>
<HealthCheckIntervalSeconds>{{ target_group.healthcheck_interval_seconds }}</HealthCheckIntervalSeconds>
<HealthCheckTimeoutSeconds>{{ target_group.healthcheck_timeout_seconds }}</HealthCheckTimeoutSeconds>
<HealthyThresholdCount>{{ target_group.healthy_threshold_count }}</HealthyThresholdCount>
<UnhealthyThresholdCount>{{ target_group.unhealthy_threshold_count }}</UnhealthyThresholdCount>
<Matcher>
<HttpCode>200</HttpCode>
</Matcher>
</member>
</TargetGroups>
</CreateTargetGroupResult>
<ResponseMetadata>
<RequestId>b83fe90e-f2d5-11e5-b95d-3b2c1831fc26</RequestId>
</ResponseMetadata>
</CreateTargetGroupResponse>"""
CREATE_LISTENER_TEMPLATE = """<CreateListenerResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<CreateListenerResult>
<Listeners>
<member>
<LoadBalancerArn>{{ listener.load_balancer_arn }}</LoadBalancerArn>
<Protocol>{{ listener.protocol }}</Protocol>
{% if listener.certificate %}
<Certificates>
<member>
<CertificateArn>{{ listener.certificate }}</CertificateArn>
</member>
</Certificates>
{% endif %}
<Port>{{ listener.port }}</Port>
<SslPolicy>{{ listener.ssl_policy }}</SslPolicy>
<ListenerArn>{{ listener.arn }}</ListenerArn>
<DefaultActions>
{% for action in listener.default_actions %}
<member>
<Type>{{ action.type }}</Type>
<TargetGroupArn>{{ action.target_group_arn }}</TargetGroupArn>
</member>
{% endfor %}
</DefaultActions>
</member>
</Listeners>
</CreateListenerResult>
<ResponseMetadata>
<RequestId>97f1bb38-f390-11e5-b95d-3b2c1831fc26</RequestId>
</ResponseMetadata>
</CreateListenerResponse>"""
DELETE_LOAD_BALANCER_TEMPLATE = """<DeleteLoadBalancerResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<DeleteLoadBalancerResult/>
<ResponseMetadata>
<RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId>
</ResponseMetadata>
</DeleteLoadBalancerResponse>"""
DELETE_TARGET_GROUP_TEMPLATE = """<DeleteTargetGroupResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<DeleteTargetGroupResult/>
<ResponseMetadata>
<RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId>
</ResponseMetadata>
</DeleteTargetGroupResponse>"""
DELETE_LISTENER_TEMPLATE = """<DeleteListenerResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<DeleteListenerResult/>
<ResponseMetadata>
<RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId>
</ResponseMetadata>
</DeleteListenerResponse>"""
DESCRIBE_LOAD_BALANCERS_TEMPLATE = """<DescribeLoadBalancersResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<DescribeLoadBalancersResult>
<LoadBalancers>
{% for load_balancer in load_balancers %}
<member>
<LoadBalancerArn>{{ load_balancer.arn }}</LoadBalancerArn>
<Scheme>{{ load_balancer.scheme }}</Scheme>
<LoadBalancerName>{{ load_balancer.name }}</LoadBalancerName>
<VpcId>{{ load_balancer.vpc_id }}</VpcId>
<CanonicalHostedZoneId>Z2P70J7EXAMPLE</CanonicalHostedZoneId>
<CreatedTime>{{ load_balancer.created_time }}</CreatedTime>
<AvailabilityZones>
{% for subnet in load_balancer.subnets %}
<member>
<SubnetId>{{ subnet.id }}</SubnetId>
<ZoneName>{{ subnet.availability_zone }}</ZoneName>
</member>
{% endfor %}
</AvailabilityZones>
<SecurityGroups>
{% for security_group in load_balancer.security_groups %}
<member>{{ security_group }}</member>
{% endfor %}
</SecurityGroups>
<DNSName>{{ load_balancer.dns_name }}</DNSName>
<State>
<Code>provisioning</Code>
</State>
<Type>application</Type>
</member>
{% endfor %}
</LoadBalancers>
{% if marker %}
<NextMarker>{{ marker }}</NextMarker>
{% endif %}
</DescribeLoadBalancersResult>
<ResponseMetadata>
<RequestId>f9880f01-7852-629d-a6c3-3ae2-666a409287e6dc0c</RequestId>
</ResponseMetadata>
</DescribeLoadBalancersResponse>"""
DESCRIBE_TARGET_GROUPS_TEMPLATE = """<DescribeTargetGroupsResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<DescribeTargetGroupsResult>
<TargetGroups>
{% for target_group in target_groups %}
<member>
<TargetGroupArn>{{ target_group.arn }}</TargetGroupArn>
<TargetGroupName>{{ target_group.name }}</TargetGroupName>
<Protocol>{{ target_group.protocol }}</Protocol>
<Port>{{ target_group.port }}</Port>
<VpcId>{{ target_group.vpc_id }}</VpcId>
<HealthCheckProtocol>{{ target_group.health_check_protocol }}</HealthCheckProtocol>
<HealthCheckPort>{{ target_group.healthcheck_port }}</HealthCheckPort>
<HealthCheckPath>{{ target_group.healthcheck_path }}</HealthCheckPath>
<HealthCheckIntervalSeconds>{{ target_group.healthcheck_interval_seconds }}</HealthCheckIntervalSeconds>
<HealthCheckTimeoutSeconds>{{ target_group.healthcheck_timeout_seconds }}</HealthCheckTimeoutSeconds>
<HealthyThresholdCount>{{ target_group.healthy_threshold_count }}</HealthyThresholdCount>
<UnhealthyThresholdCount>{{ target_group.unhealthy_threshold_count }}</UnhealthyThresholdCount>
<Matcher>
<HttpCode>200</HttpCode>
</Matcher>
<LoadBalancerArns>
{% for load_balancer_arn in target_group.load_balancer_arns %}
<member>{{ load_balancer_arn }}</member>
{% endfor %}
</LoadBalancerArns>
</member>
{% endfor %}
</TargetGroups>
</DescribeTargetGroupsResult>
<ResponseMetadata>
<RequestId>70092c0e-f3a9-11e5-ae48-cff02092876b</RequestId>
</ResponseMetadata>
</DescribeTargetGroupsResponse>"""
DESCRIBE_LISTENERS_TEMPLATE = """<DescribeLoadBalancersResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<DescribeListenersResult>
<Listeners>
{% for listener in listeners %}
<member>
<LoadBalancerArn>{{ listener.load_balancer_arn }}</LoadBalancerArn>
<Protocol>{{ listener.protocol }}</Protocol>
{% if listener.certificate %}
<Certificates>
<member>
<CertificateArn>{{ listener.certificate }} </CertificateArn>
</member>
</Certificates>
{% endif %}
<Port>{{ listener.port }}</Port>
<SslPolicy>{{ listener.ssl_policy }}</SslPolicy>
<ListenerArn>{{ listener.arn }}</ListenerArn>
<DefaultActions>
{% for action in listener.default_actions %}
<member>
<Type>{{ action.type }}</Type>
<TargetGroupArn>{{ action.target_group_arn }}</TargetGroupArn>
</member>
{% endfor %}
</DefaultActions>
</member>
{% endfor %}
</Listeners>
</DescribeListenersResult>
<ResponseMetadata>
<RequestId>65a3a7ea-f39c-11e5-b543-9f2c3fbb9bee</RequestId>
</ResponseMetadata>
</DescribeLoadBalancersResponse>"""
CONFIGURE_HEALTH_CHECK_TEMPLATE = """<ConfigureHealthCheckResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<ConfigureHealthCheckResult>
<HealthCheck>
<Interval>{{ check.interval }}</Interval>
<Target>{{ check.target }}</Target>
<HealthyThreshold>{{ check.healthy_threshold }}</HealthyThreshold>
<Timeout>{{ check.timeout }}</Timeout>
<UnhealthyThreshold>{{ check.unhealthy_threshold }}</UnhealthyThreshold>
</HealthCheck>
</ConfigureHealthCheckResult>
<ResponseMetadata>
<RequestId>f9880f01-7852-629d-a6c3-3ae2-666a409287e6dc0c</RequestId>
</ResponseMetadata>
</ConfigureHealthCheckResponse>"""
REGISTER_TARGETS_TEMPLATE = """<RegisterTargetsResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<RegisterTargetsResult>
</RegisterTargetsResult>
<ResponseMetadata>
<RequestId>f9880f01-7852-629d-a6c3-3ae2-666a409287e6dc0c</RequestId>
</ResponseMetadata>
</RegisterTargetsResponse>"""
DEREGISTER_TARGETS_TEMPLATE = """<DeregisterTargetsResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<DeregisterTargetsResult>
</DeregisterTargetsResult>
<ResponseMetadata>
<RequestId>f9880f01-7852-629d-a6c3-3ae2-666a409287e6dc0c</RequestId>
</ResponseMetadata>
</DeregisterTargetsResponse>"""
SET_LOAD_BALANCER_SSL_CERTIFICATE = """<SetLoadBalancerListenerSSLCertificateResponse xmlns="http://elasticloadbalan cing.amazonaws.com/doc/2015-12-01/">
<SetLoadBalancerListenerSSLCertificateResult/>
<ResponseMetadata>
<RequestId>83c88b9d-12b7-11e3-8b82-87b12EXAMPLE</RequestId>
</ResponseMetadata>
</SetLoadBalancerListenerSSLCertificateResponse>"""
DELETE_LOAD_BALANCER_LISTENERS = """<DeleteLoadBalancerListenersResponse xmlns="http://elasticloadbalan cing.amazonaws.com/doc/2015-12-01/">
<DeleteLoadBalancerListenersResult/>
<ResponseMetadata>
<RequestId>83c88b9d-12b7-11e3-8b82-87b12EXAMPLE</RequestId>
</ResponseMetadata>
</DeleteLoadBalancerListenersResponse>"""
DESCRIBE_ATTRIBUTES_TEMPLATE = """<DescribeLoadBalancerAttributesResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<DescribeLoadBalancerAttributesResult>
<LoadBalancerAttributes>
<AccessLog>
<Enabled>{{ attributes.access_log.enabled }}</Enabled>
{% if attributes.access_log.enabled %}
<S3BucketName>{{ attributes.access_log.s3_bucket_name }}</S3BucketName>
<S3BucketPrefix>{{ attributes.access_log.s3_bucket_prefix }}</S3BucketPrefix>
<EmitInterval>{{ attributes.access_log.emit_interval }}</EmitInterval>
{% endif %}
</AccessLog>
<ConnectionSettings>
<IdleTimeout>{{ attributes.connecting_settings.idle_timeout }}</IdleTimeout>
</ConnectionSettings>
<CrossZoneLoadBalancing>
<Enabled>{{ attributes.cross_zone_load_balancing.enabled }}</Enabled>
</CrossZoneLoadBalancing>
<ConnectionDraining>
{% if attributes.connection_draining.enabled %}
<Enabled>true</Enabled>
<Timeout>{{ attributes.connection_draining.timeout }}</Timeout>
{% else %}
<Enabled>false</Enabled>
{% endif %}
</ConnectionDraining>
</LoadBalancerAttributes>
</DescribeLoadBalancerAttributesResult>
<ResponseMetadata>
<RequestId>83c88b9d-12b7-11e3-8b82-87b12EXAMPLE</RequestId>
</ResponseMetadata>
</DescribeLoadBalancerAttributesResponse>
"""
MODIFY_ATTRIBUTES_TEMPLATE = """<ModifyLoadBalancerAttributesResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<ModifyLoadBalancerAttributesResult>
<LoadBalancerName>{{ load_balancer.name }}</LoadBalancerName>
<LoadBalancerAttributes>
<AccessLog>
<Enabled>{{ attributes.access_log.enabled }}</Enabled>
{% if attributes.access_log.enabled %}
<S3BucketName>{{ attributes.access_log.s3_bucket_name }}</S3BucketName>
<S3BucketPrefix>{{ attributes.access_log.s3_bucket_prefix }}</S3BucketPrefix>
<EmitInterval>{{ attributes.access_log.emit_interval }}</EmitInterval>
{% endif %}
</AccessLog>
<ConnectionSettings>
<IdleTimeout>{{ attributes.connecting_settings.idle_timeout }}</IdleTimeout>
</ConnectionSettings>
<CrossZoneLoadBalancing>
<Enabled>{{ attributes.cross_zone_load_balancing.enabled }}</Enabled>
</CrossZoneLoadBalancing>
<ConnectionDraining>
{% if attributes.connection_draining.enabled %}
<Enabled>true</Enabled>
<Timeout>{{ attributes.connection_draining.timeout }}</Timeout>
{% else %}
<Enabled>false</Enabled>
{% endif %}
</ConnectionDraining>
</LoadBalancerAttributes>
</ModifyLoadBalancerAttributesResult>
<ResponseMetadata>
<RequestId>83c88b9d-12b7-11e3-8b82-87b12EXAMPLE</RequestId>
</ResponseMetadata>
</ModifyLoadBalancerAttributesResponse>
"""
CREATE_LOAD_BALANCER_POLICY_TEMPLATE = """<CreateLoadBalancerPolicyResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<CreateLoadBalancerPolicyResult/>
<ResponseMetadata>
<RequestId>83c88b9d-12b7-11e3-8b82-87b12EXAMPLE</RequestId>
</ResponseMetadata>
</CreateLoadBalancerPolicyResponse>
"""
SET_LOAD_BALANCER_POLICIES_OF_LISTENER_TEMPLATE = """<SetLoadBalancerPoliciesOfListenerResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<SetLoadBalancerPoliciesOfListenerResult/>
<ResponseMetadata>
<RequestId>07b1ecbc-1100-11e3-acaf-dd7edEXAMPLE</RequestId>
</ResponseMetadata>
</SetLoadBalancerPoliciesOfListenerResponse>
"""
SET_LOAD_BALANCER_POLICIES_FOR_BACKEND_SERVER_TEMPLATE = """<SetLoadBalancerPoliciesForBackendServerResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<SetLoadBalancerPoliciesForBackendServerResult/>
<ResponseMetadata>
<RequestId>0eb9b381-dde0-11e2-8d78-6ddbaEXAMPLE</RequestId>
</ResponseMetadata>
</SetLoadBalancerPoliciesForBackendServerResponse>
"""
DESCRIBE_TARGET_HEALTH_TEMPLATE = """<DescribeTargetHealthResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<DescribeTargetHealthResult>
<TargetHealthDescriptions>
{% for target_health in target_health_descriptions %}
<member>
<HealthCheckPort>{{ target_health.health_port }}</HealthCheckPort>
<TargetHealth>
<State>{{ target_health.status }}</State>
</TargetHealth>
<Target>
<Port>{{ target_health.port }}</Port>
<Id>{{ target_health.instance_id }}</Id>
</Target>
</member>
{% endfor %}
</TargetHealthDescriptions>
</DescribeTargetHealthResult>
<ResponseMetadata>
<RequestId>c534f810-f389-11e5-9192-3fff33344cfa</RequestId>
</ResponseMetadata>
</DescribeTargetHealthResponse>"""

10
moto/elbv2/urls.py Normal file
View File

@ -0,0 +1,10 @@
from __future__ import unicode_literals
from .responses import ELBV2Response
url_bases = [
"https?://elasticloadbalancing.(.+).amazonaws.com",
]
url_paths = {
'{0}/$': ELBV2Response.dispatch,
}

2
tests/test_elb/test_server.py
View File

@ -12,6 +12,6 @@ def test_elb_describe_instances():
backend = server.create_backend_app("elb")
test_client = backend.test_client()
res = test_client.get('/?Action=DescribeLoadBalancers')
res = test_client.get('/?Action=DescribeLoadBalancers&Version=2015-12-01')
res.data.should.contain(b'DescribeLoadBalancersResponse')

447
tests/test_elbv2/test_elbv2.py Normal file
View File

@ -0,0 +1,447 @@
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]
# 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)
# 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_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)

17
tests/test_elbv2/test_server.py Normal file
View File

@ -0,0 +1,17 @@
from __future__ import unicode_literals
import sure # noqa
import moto.server as server
'''
Test the different server responses
'''
def test_elbv2_describe_load_balancers():
backend = server.create_backend_app("elbv2")
test_client = backend.test_client()
res = test_client.get('/?Action=DescribeLoadBalancers&Version=2015-12-01')
res.data.should.contain(b'DescribeLoadBalancersResponse')