EC2 - Improve/Extend RouteTable unit tests (#4165)

This commit is contained in:
Bert Blommers 2021-08-11 18:50:15 +01:00 committed by GitHub
parent 6f361e6afb
commit 574053cb27
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 257 additions and 21 deletions

View File

@ -437,6 +437,9 @@ class NetworkInterfaceBackend(object):
return deleted
def describe_network_interfaces(self, filters=None):
# Note: This is only used in EC2Backend#do_resources_exist
# Client-calls use #get_all_network_interfaces()
# We should probably merge these at some point..
enis = self.enis.values()
if filters:
@ -446,22 +449,6 @@ class NetworkInterfaceBackend(object):
enis = [
eni for eni in enis if getattr(eni, _filter) in _filter_value
]
elif _filter == "group-id":
original_enis = enis
enis = []
for eni in original_enis:
for group in eni.group_set:
if group.id in _filter_value:
enis.append(eni)
break
elif _filter == "private-ip-address:":
enis = [
eni for eni in enis if eni.private_ip_address in _filter_value
]
elif _filter == "subnet-id":
enis = [eni for eni in enis if eni.subnet.id in _filter_value]
elif _filter == "description":
enis = [eni for eni in enis if eni.description in _filter_value]
else:
self.raise_not_implemented_error(
"The filter '{0}' for DescribeNetworkInterfaces".format(_filter)
@ -2134,11 +2121,11 @@ class SecurityGroup(TaggedEC2Resource, CloudFormationModel):
return attr
if key.startswith("ip-permission"):
match = re.search(r"ip-permission.(*)", key)
match = re.search(r"ip-permission.(.*)", key)
ingress_attr = to_attr(match.groups()[0])
for ingress in self.ingress_rules:
if getattr(ingress, ingress_attr) in filter_value:
if str(getattr(ingress, ingress_attr)) in filter_value:
return True
elif is_tag_filter(key):
tag_value = self.get_filter_value(key)
@ -6736,7 +6723,7 @@ class TransitGatewayRelationsBackend(object):
def disassociate_transit_gateway_route_table(self, tgw_attach_id, tgw_rt_id):
tgw_association = self.transit_gateway_associations.pop(tgw_attach_id)
tgw_association.state == "disassociated"
tgw_association.state = "disassociated"
self.unset_route_table_association(tgw_rt_id)
self.unset_attachment_association(tgw_attach_id)

View File

@ -95,7 +95,6 @@ class TransitGatewayAttachment(BaseResponse):
tgw_association = self.ec2_backend.disassociate_transit_gateway_route_table(
tgw_attach_id, tgw_rt_id
)
tgw_association.state == "disassociated"
template = self.response_template(TRANSIT_GATEWAY_DISASSOCIATION)
return template.render(tgw_association=tgw_association)

View File

@ -413,8 +413,14 @@ def test_elastic_network_interfaces_describe_network_interfaces_with_filter():
VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-2a"
)
sg = ec2_client.create_security_group(Description="test", GroupName="test_sg")
sg_id = sg["GroupId"]
eni1 = ec2.create_network_interface(
SubnetId=subnet.id, PrivateIpAddress="10.0.10.5", Description="test interface"
SubnetId=subnet.id,
PrivateIpAddress="10.0.10.5",
Description="test interface",
Groups=[sg_id],
)
# The status of the new interface should be 'available'
@ -432,6 +438,13 @@ def test_elastic_network_interfaces_describe_network_interfaces_with_filter():
)
response["NetworkInterfaces"][0]["Description"].should.equal(eni1.description)
# Filter by network-interface-id
response = ec2_client.describe_network_interfaces(
Filters=[{"Name": "group-id", "Values": [sg_id]}]
)
response["NetworkInterfaces"].should.have.length_of(1)
response["NetworkInterfaces"][0]["NetworkInterfaceId"].should.equal(eni1.id)
response = ec2_client.describe_network_interfaces(
Filters=[{"Name": "network-interface-id", "Values": ["bad-id"]}]
)
@ -560,3 +573,29 @@ def test_elastic_network_interfaces_filter_by_tag():
Filters=[{"Name": "tag:environment", "Values": ["dev", "prod"]}]
)
resp["NetworkInterfaces"].should.have.length_of(2)
@mock_ec2
def test_elastic_network_interfaces_auto_create_securitygroup():
ec2 = boto3.resource("ec2", region_name="us-west-2")
ec2_client = boto3.client("ec2", region_name="us-west-2")
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
subnet = ec2.create_subnet(
VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-2a"
)
eni1 = ec2.create_network_interface(
SubnetId=subnet.id, PrivateIpAddress="10.0.10.5", Groups=["testgroup"]
)
# The status of the new interface should be 'available'
waiter = ec2_client.get_waiter("network_interface_available")
waiter.wait(NetworkInterfaceIds=[eni1.id])
sgs = ec2_client.describe_security_groups()["SecurityGroups"]
found_sg = [sg for sg in sgs if sg["GroupId"] == "testgroup"]
found_sg.should.have.length_of(1)
found_sg[0]["GroupName"].should.equal("testgroup")
found_sg[0]["Description"].should.equal("testgroup")

View File

@ -807,6 +807,30 @@ def test_security_group_wildcard_tag_filter_boto3():
tag["Key"].should.equal("Test")
@mock_ec2
def test_security_group_filter_ip_permission():
ec2 = boto3.resource("ec2", region_name="us-east-1")
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
conn = boto3.client("ec2", region_name="us-east-1")
sg = ec2.create_security_group(
GroupName="test-sg", Description="Test SG", VpcId=vpc.id
)
ip_permissions = [
{"IpProtocol": "tcp", "FromPort": 27017, "ToPort": 27017, "IpRanges": [],},
]
sg.authorize_ingress(IpPermissions=ip_permissions)
describe = conn.describe_security_groups(
Filters=[{"Name": "ip-permission.from-port", "Values": ["27017"]}]
)["SecurityGroups"]
describe.should.have.length_of(1)
describe[0]["GroupName"].should.equal("test-sg")
@mock_ec2
def test_authorize_and_revoke_in_bulk():
ec2 = boto3.resource("ec2", region_name="us-west-1")

View File

@ -117,6 +117,34 @@ def test_describe_transit_gateway_attachments():
response.should.have.key("TransitGatewayAttachments").equal([])
@mock_ec2
def test_create_transit_gateway_vpn_attachment():
ec2 = boto3.client("ec2", region_name="us-west-1")
vpn_gateway = ec2.create_vpn_gateway(Type="ipsec.1").get("VpnGateway", {})
customer_gateway = ec2.create_customer_gateway(
Type="ipsec.1", PublicIp="205.251.242.54", BgpAsn=65534,
).get("CustomerGateway", {})
vpn_connection = ec2.create_vpn_connection(
Type="ipsec.1",
VpnGatewayId=vpn_gateway["VpnGatewayId"],
CustomerGatewayId=customer_gateway["CustomerGatewayId"],
TransitGatewayId="gateway_id",
).get("VpnConnection", {})
#
# Verify we can retrieve it as a general attachment
attachments = ec2.describe_transit_gateway_attachments()[
"TransitGatewayAttachments"
]
attachments.should.have.length_of(1)
attachments[0].should.have.key("ResourceType").equal("vpn")
attachments[0].should.have.key("ResourceId").equal(
vpn_connection["VpnConnectionId"]
)
@mock_ec2
def test_create_transit_gateway_vpc_attachment():
ec2 = boto3.client("ec2", region_name="us-west-1")
@ -564,6 +592,50 @@ def test_associate_transit_gateway_route_table():
)
@mock_ec2
def test_disassociate_transit_gateway_route_table():
ec2 = boto3.client("ec2", region_name="us-west-1")
gateway_id = ec2.create_transit_gateway(Description="g")["TransitGateway"][
"TransitGatewayId"
]
attchmnt = ec2.create_transit_gateway_vpc_attachment(
TransitGatewayId=gateway_id, VpcId="vpc-id", SubnetIds=["sub1"]
)["TransitGatewayVpcAttachment"]
table = ec2.create_transit_gateway_route_table(TransitGatewayId=gateway_id)[
"TransitGatewayRouteTable"
]
initial = ec2.get_transit_gateway_route_table_associations(
TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"]
)["Associations"][0]
initial["TransitGatewayAttachmentId"].should.equal("")
ec2.associate_transit_gateway_route_table(
TransitGatewayAttachmentId=attchmnt["TransitGatewayAttachmentId"],
TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"],
)
updated = ec2.get_transit_gateway_route_table_associations(
TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"]
)["Associations"][0]
updated["TransitGatewayAttachmentId"].should.equal(
attchmnt["TransitGatewayAttachmentId"]
)
updated["State"].should.equal("associated")
dis = ec2.disassociate_transit_gateway_route_table(
TransitGatewayAttachmentId=attchmnt["TransitGatewayAttachmentId"],
TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"],
)["Association"]
dis["State"].should.equal("disassociated")
updated = ec2.get_transit_gateway_route_table_associations(
TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"]
)["Associations"][0]
updated["TransitGatewayAttachmentId"].should.equal("")
updated["State"].should.equal("")
@mock_ec2
def test_enable_transit_gateway_route_table_propagation():
ec2 = boto3.client("ec2", region_name="us-west-1")

View File

@ -1,9 +1,11 @@
from __future__ import unicode_literals
import boto
import boto3
import pytest
import sure # noqa
from moto import mock_ec2_deprecated, mock_ec2
from botocore.exceptions import ClientError
@mock_ec2_deprecated
@ -34,6 +36,49 @@ def test_describe_vpn_gateway():
vpn_gateway.availability_zone.should.equal("us-east-1a")
@mock_ec2
def test_attach_unknown_vpn_gateway():
"""describe_vpn_gateways attachment.vpc-id filter"""
ec2 = boto3.client("ec2", region_name="us-east-1")
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
with pytest.raises(ClientError) as ex:
ec2.attach_vpn_gateway(VpcId=vpc["VpcId"], VpnGatewayId="?")
err = ex.value.response["Error"]
err["Message"].should.equal("The virtual private gateway ID '?' does not exist")
err["Code"].should.equal("InvalidVpnGatewayID.NotFound")
@mock_ec2
def test_delete_unknown_vpn_gateway():
"""describe_vpn_gateways attachment.vpc-id filter"""
ec2 = boto3.client("ec2", region_name="us-east-1")
with pytest.raises(ClientError) as ex:
ec2.delete_vpn_gateway(VpnGatewayId="?")
err = ex.value.response["Error"]
err["Message"].should.equal("The virtual private gateway ID '?' does not exist")
err["Code"].should.equal("InvalidVpnGatewayID.NotFound")
@mock_ec2
def test_detach_unknown_vpn_gateway():
"""describe_vpn_gateways attachment.vpc-id filter"""
ec2 = boto3.client("ec2", region_name="us-east-1")
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
with pytest.raises(ClientError) as ex:
ec2.detach_vpn_gateway(VpcId=vpc["VpcId"], VpnGatewayId="?")
err = ex.value.response["Error"]
err["Message"].should.equal("The virtual private gateway ID '?' does not exist")
err["Code"].should.equal("InvalidVpnGatewayID.NotFound")
@mock_ec2
def test_describe_vpn_connections_attachment_vpc_id_filter():
"""describe_vpn_gateways attachment.vpc-id filter"""

View File

@ -316,6 +316,31 @@ def test_vpc_dedicated_tenancy():
vpc.instance_tenancy.should.equal("dedicated")
@mock_ec2
def test_vpc_modify_tenancy_unknown():
ec2 = boto3.resource("ec2", region_name="us-west-1")
ec2_client = boto3.client("ec2", region_name="us-west-1")
# Create the default VPC
ec2.create_vpc(CidrBlock="172.31.0.0/16")
# Create the non default VPC
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16", InstanceTenancy="dedicated")
vpc.instance_tenancy.should.equal("dedicated")
with pytest.raises(ClientError) as ex:
ec2_client.modify_vpc_tenancy(VpcId=vpc.id, InstanceTenancy="unknown")
err = ex.value.response["Error"]
err["Message"].should.equal("The tenancy value unknown is not supported.")
err["Code"].should.equal("UnsupportedTenancy")
ec2_client.modify_vpc_tenancy(VpcId=vpc.id, InstanceTenancy="default")
vpc.reload()
vpc.instance_tenancy.should.equal("default")
@mock_ec2
def test_vpc_modify_enable_dns_support():
ec2 = boto3.resource("ec2", region_name="us-west-1")

View File

@ -5,6 +5,7 @@ import boto3
import pytest
import sure # noqa
from boto.exception import EC2ResponseError
from botocore.client import ClientError
from moto import mock_ec2, mock_ec2_deprecated
@ -73,3 +74,47 @@ def test_create_vpn_connection_with_vpn_gateway():
vpn_connection["CustomerGatewayId"].should.equal(
customer_gateway["CustomerGatewayId"]
)
@mock_ec2
def test_describe_vpn_connections():
client = boto3.client("ec2", region_name="us-east-1")
vpn_gateway = client.create_vpn_gateway(Type="ipsec.1").get("VpnGateway", {})
customer_gateway = client.create_customer_gateway(
Type="ipsec.1", PublicIp="205.251.242.54", BgpAsn=65534,
).get("CustomerGateway", {})
client.create_vpn_connection(
Type="ipsec.1",
VpnGatewayId=vpn_gateway["VpnGatewayId"],
CustomerGatewayId=customer_gateway["CustomerGatewayId"],
)["VpnConnection"]
vpn_connection2 = client.create_vpn_connection(
Type="ipsec.1",
VpnGatewayId=vpn_gateway["VpnGatewayId"],
CustomerGatewayId=customer_gateway["CustomerGatewayId"],
)["VpnConnection"]
conns = client.describe_vpn_connections()["VpnConnections"]
conns.should.have.length_of(2)
conns = client.describe_vpn_connections(
VpnConnectionIds=[vpn_connection2["VpnConnectionId"]]
)["VpnConnections"]
conns[0]["VpnConnectionId"].should.equal(vpn_connection2["VpnConnectionId"])
conns[0]["VpnGatewayId"].should.equal(vpn_gateway["VpnGatewayId"])
conns[0]["Type"].should.equal("ipsec.1")
conns[0]["CustomerGatewayId"].should.equal(customer_gateway["CustomerGatewayId"])
conns[0]["State"].should.equal("available")
@mock_ec2
def test_describe_vpn_connections_unknown():
client = boto3.client("ec2", region_name="us-east-1")
with pytest.raises(ClientError) as ex:
client.describe_vpn_connections(VpnConnectionIds=["?"])
err = ex.value.response["Error"]
err["Message"].should.equal("The vpnConnection ID '?' does not exist")
err["Code"].should.equal("InvalidVpnConnectionID.NotFound")