Enhancement/describe network acls (#3003)

* update describe_network_acls and create unit test

* add fail test case

* adjustment after feedback

* fix result test
This commit is contained in:
André Nardy 2020-05-26 07:04:59 -03:00 committed by GitHub
parent 4f42ba93d8
commit 97a6e8d9e8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 51 additions and 19 deletions

View File

@ -2738,7 +2738,7 @@
- [ ] describe_local_gateways
- [ ] describe_moving_addresses
- [ ] describe_nat_gateways
- [ ] describe_network_acls
- [X] describe_network_acls
- [ ] describe_network_interface_attribute
- [ ] describe_network_interface_permissions
- [X] describe_network_interfaces

View File

@ -4750,23 +4750,7 @@ class NetworkAclBackend(object):
)
def get_all_network_acls(self, network_acl_ids=None, filters=None):
network_acls = self.network_acls.values()
if network_acl_ids:
network_acls = [
network_acl
for network_acl in network_acls
if network_acl.id in network_acl_ids
]
if len(network_acls) != len(network_acl_ids):
invalid_id = list(
set(network_acl_ids).difference(
set([network_acl.id for network_acl in network_acls])
)
)[0]
raise InvalidRouteTableIdError(invalid_id)
return generic_filter(filters, network_acls)
self.describe_network_acls(network_acl_ids, filters)
def delete_network_acl(self, network_acl_id):
deleted = self.network_acls.pop(network_acl_id, None)
@ -4886,6 +4870,25 @@ class NetworkAclBackend(object):
self, association_id, subnet_id, acl.id
)
def describe_network_acls(self, network_acl_ids=None, filters=None):
network_acls = self.network_acls.values()
if network_acl_ids:
network_acls = [
network_acl
for network_acl in network_acls
if network_acl.id in network_acl_ids
]
if len(network_acls) != len(network_acl_ids):
invalid_id = list(
set(network_acl_ids).difference(
set([network_acl.id for network_acl in network_acls])
)
)[0]
raise InvalidRouteTableIdError(invalid_id)
return generic_filter(filters, network_acls)
class NetworkAclAssociation(object):
def __init__(self, ec2_backend, new_association_id, subnet_id, network_acl_id):

View File

@ -83,7 +83,7 @@ class NetworkACLs(BaseResponse):
def describe_network_acls(self):
network_acl_ids = self._get_multi_param("NetworkAclId")
filters = filters_from_querystring(self.querystring)
network_acls = self.ec2_backend.get_all_network_acls(network_acl_ids, filters)
network_acls = self.ec2_backend.describe_network_acls(network_acl_ids, filters)
template = self.response_template(DESCRIBE_NETWORK_ACL_RESPONSE)
return template.render(network_acls=network_acls)

View File

@ -275,3 +275,32 @@ def test_duplicate_network_acl_entry():
rule_number
)
)
@mock_ec2
def test_describe_network_acls():
conn = boto3.client("ec2", region_name="us-west-2")
vpc = conn.create_vpc(CidrBlock="10.0.0.0/16")
vpc_id = vpc["Vpc"]["VpcId"]
network_acl = conn.create_network_acl(VpcId=vpc_id)
network_acl_id = network_acl["NetworkAcl"]["NetworkAclId"]
resp = conn.describe_network_acls(NetworkAclIds=[network_acl_id])
result = resp["NetworkAcls"]
result.should.have.length_of(1)
result[0]["NetworkAclId"].should.equal(network_acl_id)
resp2 = conn.describe_network_acls()["NetworkAcls"]
resp2.should.have.length_of(3)
with assert_raises(ClientError) as ex:
conn.describe_network_acls(NetworkAclIds=["1"])
str(ex.exception).should.equal(
"An error occurred (InvalidRouteTableID.NotFound) when calling the "
"DescribeNetworkAcls operation: The routeTable ID '1' does not exist"
)