openstack
/
heat
Code Issues Proposed changes
heat/heat/tests/test_vpc.py

853 lines
29 KiB
Python
Raw Blame History

#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import uuid
from heat.common import exception
from heat.common import template_format
from heat.engine.resources.aws.ec2 import subnet as sn
from heat.engine import scheduler
from heat.engine import stack as parser
from heat.engine import template
from heat.tests import common
from heat.tests import utils
try:
from neutronclient.common import exceptions as neutron_exc
from neutronclient.v2_0 import client as neutronclient
except ImportError:
neutronclient = None
class VPCTestBase(common.HeatTestCase):
def setUp(self):
super(VPCTestBase, self).setUp()
self.m.StubOutWithMock(neutronclient.Client, 'add_interface_router')
self.m.StubOutWithMock(neutronclient.Client, 'add_gateway_router')
self.m.StubOutWithMock(neutronclient.Client, 'create_network')
self.m.StubOutWithMock(neutronclient.Client, 'create_port')
self.m.StubOutWithMock(neutronclient.Client, 'create_router')
self.m.StubOutWithMock(neutronclient.Client, 'create_subnet')
self.m.StubOutWithMock(neutronclient.Client, 'delete_network')
self.m.StubOutWithMock(neutronclient.Client, 'delete_port')
self.m.StubOutWithMock(neutronclient.Client, 'delete_router')
self.m.StubOutWithMock(neutronclient.Client, 'delete_subnet')
self.m.StubOutWithMock(neutronclient.Client, 'list_networks')
self.m.StubOutWithMock(neutronclient.Client, 'list_routers')
self.m.StubOutWithMock(neutronclient.Client, 'remove_gateway_router')
self.m.StubOutWithMock(neutronclient.Client, 'remove_interface_router')
self.m.StubOutWithMock(neutronclient.Client, 'show_subnet')
self.m.StubOutWithMock(neutronclient.Client, 'show_network')
self.m.StubOutWithMock(neutronclient.Client, 'show_port')
self.m.StubOutWithMock(neutronclient.Client, 'show_router')
self.m.StubOutWithMock(neutronclient.Client, 'create_security_group')
self.m.StubOutWithMock(neutronclient.Client, 'show_security_group')
self.m.StubOutWithMock(neutronclient.Client, 'list_security_groups')
self.m.StubOutWithMock(neutronclient.Client, 'delete_security_group')
self.m.StubOutWithMock(
neutronclient.Client, 'create_security_group_rule')
self.m.StubOutWithMock(
neutronclient.Client, 'delete_security_group_rule')
def create_stack(self, templ):
t = template_format.parse(templ)
stack = self.parse_stack(t)
self.assertIsNone(stack.validate())
self.assertIsNone(stack.create())
return stack
def parse_stack(self, t):
stack_name = 'test_stack'
tmpl = template.Template(t)
stack = parser.Stack(utils.dummy_context(), stack_name, tmpl)
stack.store()
return stack
def mock_create_network(self):
self.vpc_name = utils.PhysName('test_stack', 'the_vpc')
neutronclient.Client.create_network(
{
'network': {'name': self.vpc_name}
}).AndReturn({'network': {
'status': 'BUILD',
'subnets': [],
'name': 'name',
'admin_state_up': True,
'shared': False,
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'id': 'aaaa'
}})
neutronclient.Client.show_network(
'aaaa'
).AndReturn({"network": {
"status": "BUILD",
"subnets": [],
"name": self.vpc_name,
"admin_state_up": False,
"shared": False,
"tenant_id": "c1210485b2424d48804aad5d39c61b8f",
"id": "aaaa"
}})
neutronclient.Client.show_network(
'aaaa'
).MultipleTimes().AndReturn({"network": {
"status": "ACTIVE",
"subnets": [],
"name": self.vpc_name,
"admin_state_up": False,
"shared": False,
"tenant_id": "c1210485b2424d48804aad5d39c61b8f",
"id": "aaaa"
}})
neutronclient.Client.create_router(
{'router': {'name': self.vpc_name}}).AndReturn({
'router': {
'status': 'BUILD',
'name': self.vpc_name,
'admin_state_up': True,
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'id': 'bbbb'
}})
neutronclient.Client.list_routers(name=self.vpc_name).AndReturn({
"routers": [{
"status": "BUILD",
"external_gateway_info": None,
"name": self.vpc_name,
"admin_state_up": True,
"tenant_id": "3e21026f2dc94372b105808c0e721661",
"routes": [],
"id": "bbbb"
}]
})
self.mock_router_for_vpc()
def mock_create_subnet(self):
self.subnet_name = utils.PhysName('test_stack', 'the_subnet')
neutronclient.Client.create_subnet(
{'subnet': {
'network_id': u'aaaa',
'cidr': u'10.0.0.0/24',
'ip_version': 4,
'name': self.subnet_name}}).AndReturn({
'subnet': {
'status': 'ACTIVE',
'name': self.subnet_name,
'admin_state_up': True,
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'id': 'cccc'}})
self.mock_router_for_vpc()
neutronclient.Client.add_interface_router(
u'bbbb',
{'subnet_id': 'cccc'}).AndReturn(None)
def mock_show_subnet(self):
neutronclient.Client.show_subnet('cccc').AndReturn({
'subnet': {
'name': self.subnet_name,
'network_id': 'aaaa',
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'allocation_pools': [{'start': '10.0.0.2',
'end': '10.0.0.254'}],
'gateway_ip': '10.0.0.1',
'ip_version': 4,
'cidr': '10.0.0.0/24',
'id': 'cccc',
'enable_dhcp': False,
}})
def mock_create_security_group(self):
self.sg_name = utils.PhysName('test_stack', 'the_sg')
neutronclient.Client.create_security_group({
'security_group': {
'name': self.sg_name,
'description': 'SSH access'
}
}).AndReturn({
'security_group': {
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'name': self.sg_name,
'description': 'SSH access',
'security_group_rules': [],
'id': '0389f747-7785-4757-b7bb-2ab07e4b09c3'
}
})
neutronclient.Client.create_security_group_rule({
'security_group_rule': {
'direction': 'ingress',
'remote_group_id': None,
'remote_ip_prefix': '0.0.0.0/0',
'port_range_min': 22,
'ethertype': 'IPv4',
'port_range_max': 22,
'protocol': 'tcp',
'security_group_id': '0389f747-7785-4757-b7bb-2ab07e4b09c3'
}
}).AndReturn({
'security_group_rule': {
'direction': 'ingress',
'remote_group_id': None,
'remote_ip_prefix': '0.0.0.0/0',
'port_range_min': 22,
'ethertype': 'IPv4',
'port_range_max': 22,
'protocol': 'tcp',
'security_group_id': '0389f747-7785-4757-b7bb-2ab07e4b09c3',
'id': 'bbbb'
}
})
def mock_show_security_group(self, group=None):
sg_name = utils.PhysName('test_stack', 'the_sg')
group = group or '0389f747-7785-4757-b7bb-2ab07e4b09c3'
if group == '0389f747-7785-4757-b7bb-2ab07e4b09c3':
neutronclient.Client.show_security_group(group).AndReturn({
'security_group': {
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'name': sg_name,
'description': '',
'security_group_rules': [{
'direction': 'ingress',
'protocol': 'tcp',
'port_range_max': 22,
'id': 'bbbb',
'ethertype': 'IPv4',
'security_group_id': ('0389f747-7785-4757-b7bb-'
'2ab07e4b09c3'),
'remote_group_id': None,
'remote_ip_prefix': '0.0.0.0/0',
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'port_range_min': 22
}],
'id': '0389f747-7785-4757-b7bb-2ab07e4b09c3'}})
elif group == 'INVALID-NO-REF':
neutronclient.Client.show_security_group(group).AndRaise(
neutron_exc.NeutronClientException(status_code=404))
elif group == 'RaiseException':
neutronclient.Client.show_security_group(
'0389f747-7785-4757-b7bb-2ab07e4b09c3').AndRaise(
neutron_exc.NeutronClientException(status_code=403))
def mock_delete_security_group(self):
self.mock_show_security_group()
neutronclient.Client.delete_security_group_rule(
'bbbb').AndReturn(None)
neutronclient.Client.delete_security_group(
'0389f747-7785-4757-b7bb-2ab07e4b09c3').AndReturn(None)
def mock_router_for_vpc(self):
neutronclient.Client.list_routers(name=self.vpc_name).AndReturn({
"routers": [{
"status": "ACTIVE",
"external_gateway_info": {
"network_id": "zzzz",
"enable_snat": True},
"name": self.vpc_name,
"admin_state_up": True,
"tenant_id": "3e21026f2dc94372b105808c0e721661",
"routes": [],
"id": "bbbb"
}]
})
def mock_delete_network(self):
self.mock_router_for_vpc()
neutronclient.Client.delete_router('bbbb').AndReturn(None)
neutronclient.Client.delete_network('aaaa').AndReturn(None)
def mock_delete_subnet(self):
self.mock_router_for_vpc()
neutronclient.Client.remove_interface_router(
u'bbbb',
{'subnet_id': 'cccc'}).AndReturn(None)
neutronclient.Client.delete_subnet('cccc').AndReturn(None)
def mock_create_route_table(self):
self.rt_name = utils.PhysName('test_stack', 'the_route_table')
neutronclient.Client.create_router({
'router': {'name': self.rt_name}}).AndReturn({
'router': {
'status': 'BUILD',
'name': self.rt_name,
'admin_state_up': True,
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'id': 'ffff'
}
})
neutronclient.Client.show_router('ffff').AndReturn({
'router': {
'status': 'BUILD',
'name': self.rt_name,
'admin_state_up': True,
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'id': 'ffff'
}
})
neutronclient.Client.show_router('ffff').AndReturn({
'router': {
'status': 'ACTIVE',
'name': self.rt_name,
'admin_state_up': True,
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'id': 'ffff'
}
})
self.mock_router_for_vpc()
neutronclient.Client.add_gateway_router(
'ffff', {'network_id': 'zzzz'}).AndReturn(None)
def mock_create_association(self):
self.mock_show_subnet()
self.mock_router_for_vpc()
neutronclient.Client.remove_interface_router(
'bbbb',
{'subnet_id': u'cccc'}).AndReturn(None)
neutronclient.Client.add_interface_router(
u'ffff',
{'subnet_id': 'cccc'}).AndReturn(None)
def mock_delete_association(self):
self.mock_show_subnet()
self.mock_router_for_vpc()
neutronclient.Client.remove_interface_router(
'ffff',
{'subnet_id': u'cccc'}).AndReturn(None)
neutronclient.Client.add_interface_router(
u'bbbb',
{'subnet_id': 'cccc'}).AndReturn(None)
def mock_delete_route_table(self):
neutronclient.Client.delete_router('ffff').AndReturn(None)
neutronclient.Client.remove_gateway_router('ffff').AndReturn(None)
def assertResourceState(self, resource, ref_id):
self.assertIsNone(resource.validate())
self.assertEqual((resource.CREATE, resource.COMPLETE), resource.state)
self.assertEqual(ref_id, resource.FnGetRefId())
class VPCTest(VPCTestBase):
test_template = '''
HeatTemplateFormatVersion: '2012-12-12'
Resources:
the_vpc:
Type: AWS::EC2::VPC
Properties: {CidrBlock: '10.0.0.0/16'}
'''
def mock_create_network_failed(self):
self.vpc_name = utils.PhysName('test_stack', 'the_vpc')
neutronclient.Client.create_network(
{
'network': {'name': self.vpc_name}
}).AndRaise(neutron_exc.NeutronClientException())
def test_vpc(self):
self.mock_create_network()
self.mock_delete_network()
self.m.ReplayAll()
stack = self.create_stack(self.test_template)
vpc = stack['the_vpc']
self.assertResourceState(vpc, 'aaaa')
scheduler.TaskRunner(vpc.delete)()
self.m.VerifyAll()
def test_vpc_delete_successful_if_created_failed(self):
self.mock_create_network_failed()
self.m.ReplayAll()
t = template_format.parse(self.test_template)
stack = self.parse_stack(t)
scheduler.TaskRunner(stack.create)()
self.assertEqual((stack.CREATE, stack.FAILED), stack.state)
scheduler.TaskRunner(stack.delete)()
self.m.VerifyAll()
class SubnetTest(VPCTestBase):
test_template = '''
HeatTemplateFormatVersion: '2012-12-12'
Resources:
the_vpc:
Type: AWS::EC2::VPC
Properties: {CidrBlock: '10.0.0.0/16'}
the_subnet:
Type: AWS::EC2::Subnet
Properties:
CidrBlock: 10.0.0.0/24
VpcId: {Ref: the_vpc}
AvailabilityZone: moon
'''
def test_subnet(self):
self.mock_create_network()
self.mock_create_subnet()
self.mock_delete_subnet()
self.mock_delete_network()
# mock delete subnet which is already deleted
self.mock_router_for_vpc()
neutronclient.Client.remove_interface_router(
u'bbbb',
{'subnet_id': 'cccc'}).AndRaise(
neutron_exc.NeutronClientException(status_code=404))
neutronclient.Client.delete_subnet('cccc').AndRaise(
neutron_exc.NeutronClientException(status_code=404))
self.m.ReplayAll()
stack = self.create_stack(self.test_template)
subnet = stack['the_subnet']
self.assertResourceState(subnet, 'cccc')
self.assertRaises(
exception.InvalidTemplateAttribute,
subnet.FnGetAtt,
'Foo')
self.assertEqual('moon', subnet.FnGetAtt('AvailabilityZone'))
scheduler.TaskRunner(subnet.delete)()
subnet.state_set(subnet.CREATE, subnet.COMPLETE, 'to delete again')
scheduler.TaskRunner(subnet.delete)()
scheduler.TaskRunner(stack['the_vpc'].delete)()
self.m.VerifyAll()
def _mock_create_subnet_failed(self):
self.subnet_name = utils.PhysName('test_stack', 'the_subnet')
neutronclient.Client.create_subnet(
{'subnet': {
'network_id': u'aaaa',
'cidr': u'10.0.0.0/24',
'ip_version': 4,
'name': self.subnet_name}}).AndReturn({
'subnet': {
'status': 'ACTIVE',
'name': self.subnet_name,
'admin_state_up': True,
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'id': 'cccc'}})
neutronclient.Client.show_network('aaaa').MultipleTimes().AndRaise(
neutron_exc.NeutronClientException(status_code=404))
def test_create_failed_delete_success(self):
self._mock_create_subnet_failed()
neutronclient.Client.delete_subnet('cccc').AndReturn(None)
self.m.ReplayAll()
t = template_format.parse(self.test_template)
tmpl = template.Template(t)
stack = parser.Stack(utils.dummy_context(), 'test_subnet_', tmpl,
stack_id=str(uuid.uuid4()))
tmpl.t['Resources']['the_subnet']['Properties']['VpcId'] = 'aaaa'
resource_defns = tmpl.resource_definitions(stack)
rsrc = sn.Subnet('the_subnet',
resource_defns['the_subnet'],
stack)
rsrc.validate()
self.assertRaises(exception.ResourceFailure,
scheduler.TaskRunner(rsrc.create))
self.assertEqual((rsrc.CREATE, rsrc.FAILED), rsrc.state)
ref_id = rsrc.FnGetRefId()
self.assertEqual(u'cccc', ref_id)
self.assertIsNone(scheduler.TaskRunner(rsrc.delete)())
self.assertEqual((rsrc.DELETE, rsrc.COMPLETE), rsrc.state)
self.m.VerifyAll()
class NetworkInterfaceTest(VPCTestBase):
test_template = '''
HeatTemplateFormatVersion: '2012-12-12'
Resources:
the_sg:
Type: AWS::EC2::SecurityGroup
Properties:
VpcId: {Ref: the_vpc}
GroupDescription: SSH access
SecurityGroupIngress:
- IpProtocol: tcp
FromPort: "22"
ToPort: "22"
CidrIp: 0.0.0.0/0
the_vpc:
Type: AWS::EC2::VPC
Properties: {CidrBlock: '10.0.0.0/16'}
the_subnet:
Type: AWS::EC2::Subnet
Properties:
CidrBlock: 10.0.0.0/24
VpcId: {Ref: the_vpc}
AvailabilityZone: moon
the_nic:
Type: AWS::EC2::NetworkInterface
Properties:
PrivateIpAddress: 10.0.0.100
SubnetId: {Ref: the_subnet}
GroupSet:
- Ref: the_sg
'''
test_template_no_groupset = '''
HeatTemplateFormatVersion: '2012-12-12'
Resources:
the_vpc:
Type: AWS::EC2::VPC
Properties: {CidrBlock: '10.0.0.0/16'}
the_subnet:
Type: AWS::EC2::Subnet
Properties:
CidrBlock: 10.0.0.0/24
VpcId: {Ref: the_vpc}
AvailabilityZone: moon
the_nic:
Type: AWS::EC2::NetworkInterface
Properties:
PrivateIpAddress: 10.0.0.100
SubnetId: {Ref: the_subnet}
'''
test_template_error = '''
HeatTemplateFormatVersion: '2012-12-12'
Resources:
the_sg:
Type: AWS::EC2::SecurityGroup
Properties:
VpcId: {Ref: the_vpc}
GroupDescription: SSH access
SecurityGroupIngress:
- IpProtocol: tcp
FromPort: "22"
ToPort: "22"
CidrIp: 0.0.0.0/0
the_vpc:
Type: AWS::EC2::VPC
Properties: {CidrBlock: '10.0.0.0/16'}
the_subnet:
Type: AWS::EC2::Subnet
Properties:
CidrBlock: 10.0.0.0/24
VpcId: {Ref: the_vpc}
AvailabilityZone: moon
the_nic:
Type: AWS::EC2::NetworkInterface
Properties:
PrivateIpAddress: 10.0.0.100
SubnetId: {Ref: the_subnet}
GroupSet:
- Ref: INVALID-REF-IN-TEMPLATE
'''
test_template_error_no_ref = '''
HeatTemplateFormatVersion: '2012-12-12'
Resources:
the_vpc:
Type: AWS::EC2::VPC
Properties: {CidrBlock: '10.0.0.0/16'}
the_subnet:
Type: AWS::EC2::Subnet
Properties:
CidrBlock: 10.0.0.0/24
VpcId: {Ref: the_vpc}
AvailabilityZone: moon
the_nic:
Type: AWS::EC2::NetworkInterface
Properties:
PrivateIpAddress: 10.0.0.100
SubnetId: {Ref: the_subnet}
GroupSet:
- INVALID-NO-REF
'''
def mock_create_network_interface(
self, security_groups=['0389f747-7785-4757-b7bb-2ab07e4b09c3']):
self.nic_name = utils.PhysName('test_stack', 'the_nic')
port = {'network_id': 'aaaa',
'fixed_ips': [{
'subnet_id': u'cccc',
'ip_address': u'10.0.0.100'
}],
'name': self.nic_name,
'admin_state_up': True}
if security_groups:
port['security_groups'] = security_groups
neutronclient.Client.create_port({'port': port}).AndReturn({
'port': {
'admin_state_up': True,
'device_id': '',
'device_owner': '',
'fixed_ips': [
{
'ip_address': '10.0.0.100',
'subnet_id': 'cccc'
}
],
'id': 'dddd',
'mac_address': 'fa:16:3e:25:32:5d',
'name': self.nic_name,
'network_id': 'aaaa',
'status': 'ACTIVE',
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f'
}
})
def mock_show_network_interface(self):
self.nic_name = utils.PhysName('test_stack', 'the_nic')
neutronclient.Client.show_port('dddd').AndReturn({
'port': {
'admin_state_up': True,
'device_id': '',
'device_owner': '',
'fixed_ips': [
{
'ip_address': '10.0.0.100',
'subnet_id': 'cccc'
}
],
'id': 'dddd',
'mac_address': 'fa:16:3e:25:32:5d',
'name': self.nic_name,
'network_id': 'aaaa',
'security_groups': ['0389f747-7785-4757-b7bb-2ab07e4b09c3'],
'status': 'ACTIVE',
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f'
}
})
def mock_delete_network_interface(self):
neutronclient.Client.delete_port('dddd').AndReturn(None)
def test_network_interface(self):
self.mock_create_security_group()
self.mock_create_network()
self.mock_create_subnet()
self.mock_show_subnet()
self.stub_SubnetConstraint_validate()
self.mock_create_network_interface()
self.mock_show_network_interface()
self.mock_delete_network_interface()
self.mock_delete_subnet()
self.mock_delete_network()
self.mock_delete_security_group()
self.m.ReplayAll()
stack = self.create_stack(self.test_template)
try:
self.assertEqual((stack.CREATE, stack.COMPLETE), stack.state)
rsrc = stack['the_nic']
self.assertResourceState(rsrc, 'dddd')
self.assertEqual('10.0.0.100', rsrc.FnGetAtt('PrivateIpAddress'))
finally:
scheduler.TaskRunner(stack.delete)()
self.m.VerifyAll()
def test_network_interface_existing_groupset(self):
self.m.StubOutWithMock(parser.Stack, 'resource_by_refid')
self.mock_create_security_group()
self.mock_create_network()
self.mock_create_subnet()
self.mock_show_subnet()
self.stub_SubnetConstraint_validate()
self.mock_create_network_interface()
self.mock_delete_network_interface()
self.mock_delete_subnet()
self.mock_delete_network()
self.mock_delete_security_group()
self.m.ReplayAll()
stack = self.create_stack(self.test_template)
try:
self.assertEqual((stack.CREATE, stack.COMPLETE), stack.state)
rsrc = stack['the_nic']
self.assertResourceState(rsrc, 'dddd')
finally:
stack.delete()
self.m.VerifyAll()
def test_network_interface_no_groupset(self):
self.mock_create_network()
self.mock_create_subnet()
self.mock_show_subnet()
self.stub_SubnetConstraint_validate()
self.mock_create_network_interface(security_groups=None)
self.mock_delete_network_interface()
self.mock_delete_subnet()
self.mock_delete_network()
self.m.ReplayAll()
stack = self.create_stack(self.test_template_no_groupset)
stack.delete()
self.m.VerifyAll()
def test_network_interface_error(self):
self.assertRaises(
exception.StackValidationFailed,
self.create_stack,
self.test_template_error)
class InternetGatewayTest(VPCTestBase):
test_template = '''
HeatTemplateFormatVersion: '2012-12-12'
Resources:
the_gateway:
Type: AWS::EC2::InternetGateway
the_vpc:
Type: AWS::EC2::VPC
Properties:
CidrBlock: '10.0.0.0/16'
the_subnet:
Type: AWS::EC2::Subnet
Properties:
CidrBlock: 10.0.0.0/24
VpcId: {Ref: the_vpc}
AvailabilityZone: moon
the_attachment:
Type: AWS::EC2::VPCGatewayAttachment
Properties:
VpcId: {Ref: the_vpc}
InternetGatewayId: {Ref: the_gateway}
the_route_table:
Type: AWS::EC2::RouteTable
Properties:
VpcId: {Ref: the_vpc}
the_association:
Type: AWS::EC2::SubnetRouteTableAssociation
Properties:
RouteTableId: {Ref: the_route_table}
SubnetId: {Ref: the_subnet}
'''
def mock_create_internet_gateway(self):
neutronclient.Client.list_networks(
**{'router:external': True}).AndReturn({'networks': [{
'status': 'ACTIVE',
'subnets': [],
'name': 'nova',
'router:external': True,
'tenant_id': 'c1210485b2424d48804aad5d39c61b8f',
'admin_state_up': True,
'shared': True,
'id': '0389f747-7785-4757-b7bb-2ab07e4b09c3'
}]})
def mock_create_gateway_attachment(self):
neutronclient.Client.add_gateway_router(
'ffff', {'network_id': '0389f747-7785-4757-b7bb-2ab07e4b09c3'}
).AndReturn(None)
def mock_delete_gateway_attachment(self):
neutronclient.Client.remove_gateway_router('ffff').AndReturn(None)
def test_internet_gateway(self):
self.mock_create_internet_gateway()
self.mock_create_network()
self.mock_create_subnet()
self.mock_create_route_table()
self.stub_SubnetConstraint_validate()
self.mock_create_association()
self.mock_create_gateway_attachment()
self.mock_delete_gateway_attachment()
self.mock_delete_association()
self.mock_delete_route_table()
self.mock_delete_subnet()
self.mock_delete_network()
self.m.ReplayAll()
stack = self.create_stack(self.test_template)
gateway = stack['the_gateway']
self.assertResourceState(gateway, gateway.physical_resource_name())
attachment = stack['the_attachment']
self.assertResourceState(attachment, 'the_attachment')
route_table = stack['the_route_table']
self.assertEqual([route_table], list(attachment._vpc_route_tables()))
stack.delete()
self.m.VerifyAll()
class RouteTableTest(VPCTestBase):
test_template = '''
HeatTemplateFormatVersion: '2012-12-12'
Resources:
the_vpc:
Type: AWS::EC2::VPC
Properties:
CidrBlock: '10.0.0.0/16'
the_subnet:
Type: AWS::EC2::Subnet
Properties:
CidrBlock: 10.0.0.0/24
VpcId: {Ref: the_vpc}
AvailabilityZone: moon
the_route_table:
Type: AWS::EC2::RouteTable
Properties:
VpcId: {Ref: the_vpc}
the_association:
Type: AWS::EC2::SubnetRouteTableAssociation
Properties:
RouteTableId: {Ref: the_route_table}
SubnetId: {Ref: the_subnet}
'''
def test_route_table(self):
self.mock_create_network()
self.mock_create_subnet()
self.mock_create_route_table()
self.stub_SubnetConstraint_validate()
self.mock_create_association()
self.mock_delete_association()
self.mock_delete_route_table()
self.mock_delete_subnet()
self.mock_delete_network()
self.m.ReplayAll()
stack = self.create_stack(self.test_template)
route_table = stack['the_route_table']
self.assertResourceState(route_table, 'ffff')
association = stack['the_association']
self.assertResourceState(association, 'the_association')
scheduler.TaskRunner(association.delete)()
scheduler.TaskRunner(route_table.delete)()
stack.delete()
self.m.VerifyAll()
View Git Blame Copy Permalink