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

Merge pull request #482 from silveregg/master

Browse Source 
Add new endpoints to ELB (fix #479)
This commit is contained in:
Steve Pulec 2015-12-13 21:50:26 -05:00
commit cac88cf3e3
6 changed files with 260 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
moto/elb/exceptions.py Normal file
View File

@ -0,0 +1,29 @@
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, cidr):
super(LoadBalancerNotFoundError, self).__init__(
"LoadBalancerNotFound",
"The specified load balancer does not exist: {0}"
.format(cidr))
class TooManyTagsError(ELBClientError):
def __init__(self):
super(TooManyTagsError, self).__init__(
"LoadBalancerNotFound",
"The quota for the number of tags that can be assigned to a load balancer has been reached")

14
moto/elb/models.py
View File

@ -10,6 +10,7 @@ from boto.ec2.elb.attributes import (
)
from boto.ec2.elb.policies import Policies
from moto.core import BaseBackend
from .exceptions import TooManyTagsError
class FakeHealthCheck(object):
@ -57,6 +58,7 @@ class FakeLoadBalancer(object):
self.policies.other_policies = []
self.policies.app_cookie_stickiness_policies = []
self.policies.lb_cookie_stickiness_policies = []
self.tags = {}
for port in ports:
listener = FakeListener(
@ -130,6 +132,18 @@ class FakeLoadBalancer(object):
return attributes
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]
class ELBBackend(BaseBackend):

104
moto/elb/responses.py
View File

@ -12,6 +12,8 @@ from boto.ec2.elb.policies import (
from moto.core.responses import BaseResponse
from .models import elb_backends
from .exceptions import DuplicateTagKeysError, LoadBalancerNotFoundError, \
TooManyTagsError
class ELBResponse(BaseResponse):
@ -218,6 +220,108 @@ class ELBResponse(BaseResponse):
template = self.response_template(DESCRIBE_INSTANCE_HEALTH_TEMPLATE)
return template.render(instance_ids=instance_ids)
def add_tags(self):
for key, value in self.querystring.items():
if "LoadBalancerNames.member" in key:
number = key.split('.')[2]
load_balancer_name = value[0]
elb = self.elb_backend.get_load_balancer(load_balancer_name)
if not elb:
raise LoadBalancerNotFoundError(load_balancer_name)
value = 'Tags.member.{0}.Value'.format(number)
key = 'Tags.member.{0}.Key'.format(number)
tag_values = []
tag_keys = []
for t_key, t_val in 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)
template = self.response_template(ADD_TAGS_TEMPLATE)
return template.render()
def remove_tags(self):
for key, value in self.querystring.items():
if "LoadBalancerNames.member" in key:
number = key.split('.')[2]
load_balancer_name = self._get_param('LoadBalancerNames.member.{0}'.format(number))
elb = self.elb_backend.get_load_balancer(load_balancer_name)
if not elb:
raise LoadBalancerNotFound(load_balancer_name)
key = 'Tag.member.{0}.Key'.format(number)
for t_key, t_val in self.querystring.items():
if t_key.startswith('Tags.member.'):
if t_key.split('.')[3] == 'Key':
elb.remove_tag(t_val[0])
template = self.response_template(REMOVE_TAGS_TEMPLATE)
return template.render()
def describe_tags(self):
for key, value in self.querystring.items():
if "LoadBalancerNames.member" in key:
number = key.split('.')[2]
load_balancer_name = self._get_param('LoadBalancerNames.member.{0}'.format(number))
elb = self.elb_backend.get_load_balancer(load_balancer_name)
if not elb:
raise LoadBalancerNotFound(load_balancer_name)
template = self.response_template(DESCRIBE_TAGS_TEMPLATE)
return template.render(tags=elb.tags)
ADD_TAGS_TEMPLATE = """<AddTagsResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2012-06-01/">
<AddTagsResult/>
<ResponseMetadata>
<RequestId>360e81f7-1100-11e4-b6ed-0f30EXAMPLE</RequestId>
</ResponseMetadata>
</AddTagsResponse>"""
REMOVE_TAGS_TEMPLATE = """<RemoveTagsResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2012-06-01/">
<RemoveTagsResult/>
<ResponseMetadata>
<RequestId>360e81f7-1100-11e4-b6ed-0f30EXAMPLE</RequestId>
</ResponseMetadata>
</RemoveTagsResponse>"""
DESCRIBE_TAGS_TEMPLATE = """<DescribeTagsResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2012-06-01/">
<DescribeTagsResult>
<TagDescriptions>
<member>
<Tags>
{% for key, value in tags.items() %}
<member>
<Value>{{ value }}</Value>
<Key>{{ key }}</Key>
</member>
{% endfor %}
</Tags>
</member>
</TagDescriptions>
</DescribeTagsResult>
<ResponseMetadata>
<RequestId>360e81f7-1100-11e4-b6ed-0f30EXAMPLE</RequestId>
</ResponseMetadata>
</DescribeTagsResponse>"""
CREATE_LOAD_BALANCER_TEMPLATE = """<CreateLoadBalancerResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2012-06-01/">
<CreateLoadBalancerResult>

2
requirements-dev.txt
View File

@ -1,7 +1,7 @@
-r requirements.txt
mock
nose
sure<1.2.4
sure>=1.2.24
coverage
freezegun
flask

110
tests/test_elb/test_elb.py
View File

@ -1,5 +1,6 @@
from __future__ import unicode_literals
import boto3
import botocore
import boto
import boto.ec2.elb
from boto.ec2.elb import HealthCheck
@ -18,7 +19,6 @@ import sure # noqa
from moto import mock_elb, mock_ec2
@mock_elb
def test_create_load_balancer():
conn = boto.connect_elb()
@ -583,3 +583,111 @@ def test_describe_instance_health():
instances_health.should.have.length_of(1)
instances_health[0].instance_id.should.equal(instance_id1)
instances_health[0].state.should.equal('InService')
@mock_elb
def test_add_remove_tags():
client = boto3.client('elb', region_name='us-east-1')
client.add_tags.when.called_with(LoadBalancerNames=['my-lb'],
Tags=[{
'Key': 'a',
'Value': 'b'
}]).should.throw(botocore.exceptions.ClientError)
client.create_load_balancer(
LoadBalancerName='my-lb',
Listeners=[{'Protocol':'tcp', 'LoadBalancerPort':80, 'InstancePort':8080}],
AvailabilityZones=['us-east-1a', 'us-east-1b']
)
list(client.describe_load_balancers()['LoadBalancerDescriptions']).should.have.length_of(1)
client.add_tags(LoadBalancerNames=['my-lb'],
Tags=[{
'Key': 'a',
'Value': 'a'
}])
tags = dict([(d['Key'], d['Value']) for d in client.describe_tags(LoadBalancerNames=['my-lb'])['TagDescriptions'][0]['Tags']])
tags.should.have('a').should.equal('a')
client.add_tags(LoadBalancerNames=['my-lb'],
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': 'i',
'Value': 'b'
}, {
'Key': 'j',
'Value': 'b'
}])
client.add_tags.when.called_with(LoadBalancerNames=['my-lb'],
Tags=[{
'Key': 'k',
'Value': 'b'
}]).should.throw(botocore.exceptions.ClientError)
client.add_tags(LoadBalancerNames=['my-lb'],
Tags=[{
'Key': 'j',
'Value': 'c'
}])
tags = dict([(d['Key'], d['Value']) for d in client.describe_tags(LoadBalancerNames=['my-lb'])['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('i').which.should.equal('b')
tags.should.have.key('j').which.should.equal('c')
tags.shouldnt.have.key('k')
client.remove_tags(LoadBalancerNames=['my-lb'],
Tags=[{
'Key': 'a'
}])
tags = dict([(d['Key'], d['Value']) for d in client.describe_tags(LoadBalancerNames=['my-lb'])['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('i').which.should.equal('b')
tags.should.have.key('j').which.should.equal('c')

6
tests/test_swf/responses/test_timeouts.py
View File

@ -31,7 +31,7 @@ def test_activity_task_heartbeat_timeout():
attrs = resp["events"][-2]["activityTaskTimedOutEventAttributes"]
attrs["timeoutType"].should.equal("HEARTBEAT")
# checks that event has been emitted at 12:05:00, not 12:05:30
resp["events"][-2]["eventTimestamp"].should.equal(1420113900)
resp["events"][-2]["eventTimestamp"].should.equal(1420113900.0)
resp["events"][-1]["eventType"].should.equal("DecisionTaskScheduled")
@ -66,7 +66,7 @@ def test_decision_task_start_to_close_timeout():
"scheduledEventId": 2, "startedEventId": 3, "timeoutType": "START_TO_CLOSE"
})
# checks that event has been emitted at 12:05:00, not 12:05:30
resp["events"][-2]["eventTimestamp"].should.equal(1420113900)
resp["events"][-2]["eventTimestamp"].should.equal(1420113900.0)
# Workflow Execution Start to Close timeout
# Default value in workflow helpers: 2 hours
@ -97,4 +97,4 @@ def test_workflow_execution_start_to_close_timeout():
"childPolicy": "ABANDON", "timeoutType": "START_TO_CLOSE"
})
# checks that event has been emitted at 14:00:00, not 14:00:30
resp["events"][-1]["eventTimestamp"].should.equal(1420120800)
resp["events"][-1]["eventTimestamp"].should.equal(1420120800.0)