Add tags to Elastic IP Addresses (#3310)

* Make ElasticAddress a tagged resource

To be able to filter on tags on ElasticAddresses, I need to have tags.

* remove unneeded commented lines

Was beginning of how to to it before further checking how it is done with other resources.

* do not ignore network-interface-owner-id filter

* add TODO about currently hardcoded region

* remove hardcoding region

* add testing for tags

creating and allocation, adding tags and querying for it

* separate test for tags into own method

* Linting

Co-authored-by: Bert Blommers <info@bertblommers.nl>
This commit is contained in:
Wolfgang Bauer 2020-09-25 16:25:30 +02:00 committed by GitHub
parent ce60e9e3b8
commit a4701dbbe6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 56 additions and 5 deletions

View File

@ -4510,13 +4510,15 @@ class SpotFleetBackend(object):
return True return True
class ElasticAddress(CloudFormationModel): class ElasticAddress(TaggedEC2Resource, CloudFormationModel):
def __init__(self, domain, address=None): def __init__(self, ec2_backend, domain, address=None):
self.ec2_backend = ec2_backend
if address: if address:
self.public_ip = address self.public_ip = address
else: else:
self.public_ip = random_ip() self.public_ip = random_ip()
self.allocation_id = random_eip_allocation_id() if domain == "vpc" else None self.allocation_id = random_eip_allocation_id() if domain == "vpc" else None
self.id = self.allocation_id
self.domain = domain self.domain = domain
self.instance = None self.instance = None
self.eni = None self.eni = None
@ -4578,9 +4580,13 @@ class ElasticAddress(CloudFormationModel):
return self.eni.private_ip_address return self.eni.private_ip_address
elif filter_name == "public-ip": elif filter_name == "public-ip":
return self.public_ip return self.public_ip
else: elif filter_name == "network-interface-owner-id":
# TODO: implement network-interface-owner-id # TODO: implement network-interface-owner-id
raise FilterNotImplementedError(filter_name, "DescribeAddresses") raise FilterNotImplementedError(filter_name, "DescribeAddresses")
else:
return super(ElasticAddress, self).get_filter_value(
filter_name, "DescribeAddresses"
)
class ElasticAddressBackend(object): class ElasticAddressBackend(object):
@ -4592,9 +4598,9 @@ class ElasticAddressBackend(object):
if domain not in ["standard", "vpc"]: if domain not in ["standard", "vpc"]:
raise InvalidDomainError(domain) raise InvalidDomainError(domain)
if address: if address:
address = ElasticAddress(domain, address) address = ElasticAddress(self, domain=domain, address=address)
else: else:
address = ElasticAddress(domain) address = ElasticAddress(self, domain=domain)
self.addresses.append(address) self.addresses.append(address)
return address return address

View File

@ -537,3 +537,48 @@ def test_eip_filters():
service.vpc_addresses.filter(Filters=[{"Name": "domain", "Values": ["vpc"]}]) service.vpc_addresses.filter(Filters=[{"Name": "domain", "Values": ["vpc"]}])
) )
len(addresses).should.equal(3) len(addresses).should.equal(3)
@mock_ec2
def test_eip_tags():
service = boto3.resource("ec2", region_name="us-west-1")
client = boto3.client("ec2", region_name="us-west-1")
# Allocate one address without tags
client.allocate_address(Domain="vpc")
# Allocate one address and add tags
alloc_tags = client.allocate_address(Domain="vpc")
with_tags = client.create_tags(
Resources=[alloc_tags["AllocationId"]],
Tags=[{"Key": "ManagedBy", "Value": "MyCode"}],
)
addresses_with_tags = client.describe_addresses(
Filters=[
{"Name": "domain", "Values": ["vpc"]},
{"Name": "tag:ManagedBy", "Values": ["MyCode"]},
]
)
len(addresses_with_tags["Addresses"]).should.equal(1)
addresses_with_tags = list(
service.vpc_addresses.filter(
Filters=[
{"Name": "domain", "Values": ["vpc"]},
{"Name": "tag:ManagedBy", "Values": ["MyCode"]},
]
)
)
len(addresses_with_tags).should.equal(1)
addresses_with_tags = list(
service.vpc_addresses.filter(
Filters=[
{"Name": "domain", "Values": ["vpc"]},
{"Name": "tag:ManagedBy", "Values": ["SomethingOther"]},
]
)
)
len(addresses_with_tags).should.equal(0)
addresses = list(
service.vpc_addresses.filter(Filters=[{"Name": "domain", "Values": ["vpc"]}])
)
# Expected total is 2, one with and one without tags
len(addresses).should.equal(2)