adding Address reallocate capability for EIP

This commit is contained in:
zane 2018-05-10 23:39:19 -07:00
parent cb364eedc6
commit 12188733b7
3 changed files with 27 additions and 6 deletions

View File

@ -3159,8 +3159,11 @@ class SpotFleetBackend(object):
class ElasticAddress(object):
def __init__(self, domain):
self.public_ip = random_ip()
def __init__(self, domain, address=None):
if address:
self.public_ip = address
else:
self.public_ip = random_ip()
self.allocation_id = random_eip_allocation_id() if domain == "vpc" else None
self.domain = domain
self.instance = None
@ -3222,11 +3225,13 @@ class ElasticAddressBackend(object):
self.addresses = []
super(ElasticAddressBackend, self).__init__()
def allocate_address(self, domain):
def allocate_address(self, domain, address=None):
if domain not in ['standard', 'vpc']:
raise InvalidDomainError(domain)
address = ElasticAddress(domain)
if address:
address = ElasticAddress(domain, address)
else:
address = ElasticAddress(domain)
self.addresses.append(address)
return address

View File

@ -7,8 +7,13 @@ class ElasticIPAddresses(BaseResponse):
def allocate_address(self):
domain = self._get_param('Domain', if_none='standard')
reallocate_address = self._get_param('Address', if_none=None)
if self.is_not_dryrun('AllocateAddress'):
address = self.ec2_backend.allocate_address(domain)
if reallocate_address:
address = self.ec2_backend.allocate_address(
domain, address=reallocate_address)
else:
address = self.ec2_backend.allocate_address(domain)
template = self.response_template(ALLOCATE_ADDRESS_RESPONSE)
return template.render(address=address)

View File

@ -62,6 +62,17 @@ def test_eip_allocate_vpc():
logging.debug("vpc alloc_id:".format(vpc.allocation_id))
vpc.release()
@mock_ec2
def test_specific_eip_allocate_vpc():
"""Allocate VPC EIP with specific address"""
service = boto3.resource('ec2', region_name='us-west-1')
client = boto3.client('ec2', region_name='us-west-1')
vpc = client.allocate_address(Domain="vpc", Address="127.38.43.222")
vpc['Domain'].should.be.equal("vpc")
vpc['PublicIp'].should.be.equal("127.38.43.222")
logging.debug("vpc alloc_id:".format(vpc['AllocationId']))
@mock_ec2_deprecated
def test_eip_allocate_invalid_domain():