deb-heat/heat/tests/neutron/test_neutron_network_gateway.py
Zane Bitter 203b8e8ecf Retain references to stacks in all unit tests
This will allow Resources to hold weak references to Stacks (so as to avoid
circular references) without causing the Stack object to be prematurely
deleted.

Change-Id: Ia76da7bc51042fb3598ef2a660d6fbf78137a37b
Related-Bug: #1454873
2015-05-14 16:55:12 -04:00

580 lines
22 KiB
Python

#
# Copyright 2013 NTT Corp.
# All Rights Reserved.
#
# 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 mox
from neutronclient.common import exceptions as qe
from neutronclient.neutron import v2_0 as neutronV20
from neutronclient.v2_0 import client as neutronclient
import six
from heat.common import exception
from heat.common import template_format
from heat.engine.resources.openstack.neutron import network_gateway
from heat.engine import rsrc_defn
from heat.engine import scheduler
from heat.tests import common
from heat.tests import utils
gw_template_deprecated = '''
heat_template_version: 2015-04-30
description: Template to test Network Gateway resource
resources:
NetworkGateway:
type: OS::Neutron::NetworkGateway
properties:
name: NetworkGateway
devices:
- id: e52148ca-7db9-4ec3-abe6-2c7c0ff316eb
interface_name: breth1
connections:
- network_id: 6af055d3-26f6-48dd-a597-7611d7e58d35
segmentation_type: vlan
segmentation_id: 10
'''
gw_template = '''
heat_template_version: 2015-04-30
description: Template to test Network Gateway resource
resources:
NetworkGateway:
type: OS::Neutron::NetworkGateway
properties:
name: NetworkGateway
devices:
- id: e52148ca-7db9-4ec3-abe6-2c7c0ff316eb
interface_name: breth1
connections:
- network: 6af055d3-26f6-48dd-a597-7611d7e58d35
segmentation_type: vlan
segmentation_id: 10
'''
sng = {
'network_gateway': {
'id': 'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37',
'name': 'NetworkGateway',
'default': False,
'tenant_id': '96ba52dc-c5c5-44c6-9a9d-d3ba1a03f77f',
'devices': [{
'id': 'e52148ca-7db9-4ec3-abe6-2c7c0ff316eb',
'interface_name': 'breth1'}],
'ports': [{
'segmentation_type': 'vlan',
'port_id': '32acc49c-899e-44ea-8177-6f4157e12eb4',
'segmentation_id': 10}]
}
}
class NeutronNetworkGatewayTest(common.HeatTestCase):
def setUp(self):
super(NeutronNetworkGatewayTest, self).setUp()
self.m.StubOutWithMock(neutronclient.Client, 'create_network_gateway')
self.m.StubOutWithMock(neutronclient.Client, 'show_network_gateway')
self.m.StubOutWithMock(neutronclient.Client, 'delete_network_gateway')
self.m.StubOutWithMock(neutronclient.Client, 'connect_network_gateway')
self.m.StubOutWithMock(neutronclient.Client, 'update_network_gateway')
self.m.StubOutWithMock(neutronclient.Client,
'disconnect_network_gateway')
self.m.StubOutWithMock(neutronclient.Client, 'list_networks')
self.m.StubOutWithMock(neutronV20, 'find_resourceid_by_name_or_id')
def mock_create_fail_network_not_found_delete_success(self):
neutronclient.Client.create_network_gateway({
'network_gateway': {
'name': u'NetworkGateway',
'devices': [{'id': u'e52148ca-7db9-4ec3-abe6-2c7c0ff316eb',
'interface_name': u'breth1'}]
}
}
).AndReturn({
'network_gateway': {
'id': 'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37',
'name': 'NetworkGateway',
'default': False,
'tenant_id': '96ba52dc-c5c5-44c6-9a9d-d3ba1a03f77f',
'devices': [{
'id': 'e52148ca-7db9-4ec3-abe6-2c7c0ff316eb',
'interface_name': 'breth1'}]
}
}
)
neutronV20.find_resourceid_by_name_or_id(
mox.IsA(neutronclient.Client),
'network',
'6af055d3-26f6-48dd-a597-7611d7e58d35'
).MultipleTimes().AndRaise(qe.NeutronClientException(status_code=404))
# mock successful to delete the network_gateway
neutronclient.Client.delete_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37'
).AndReturn(None)
neutronclient.Client.show_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37'
).AndRaise(qe.NeutronClientException(status_code=404))
t = template_format.parse(gw_template)
self.stack = utils.parse_stack(t)
resource_defns = self.stack.t.resource_definitions(self.stack)
rsrc = network_gateway.NetworkGateway(
'test_network_gateway',
resource_defns['NetworkGateway'], self.stack)
return rsrc
def prepare_create_network_gateway(self, resolve_neutron=True):
neutronclient.Client.create_network_gateway({
'network_gateway': {
'name': u'NetworkGateway',
'devices': [{'id': u'e52148ca-7db9-4ec3-abe6-2c7c0ff316eb',
'interface_name': u'breth1'}]
}
}
).AndReturn({
'network_gateway': {
'id': 'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37',
'name': 'NetworkGateway',
'default': False,
'tenant_id': '96ba52dc-c5c5-44c6-9a9d-d3ba1a03f77f',
'devices': [{
'id': 'e52148ca-7db9-4ec3-abe6-2c7c0ff316eb',
'interface_name': 'breth1'}]
}
}
)
neutronclient.Client.connect_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37', {
'network_id': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'segmentation_id': 10,
'segmentation_type': u'vlan'
}
).AndReturn({
'connection_info': {
'network_gateway_id': u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37',
'network_id': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'port_id': u'32acc49c-899e-44ea-8177-6f4157e12eb4'
}
})
self.stub_NetworkConstraint_validate()
if resolve_neutron:
neutronV20.find_resourceid_by_name_or_id(
mox.IsA(neutronclient.Client),
'network',
'6af055d3-26f6-48dd-a597-7611d7e58d35'
).AndReturn('6af055d3-26f6-48dd-a597-7611d7e58d35')
t = template_format.parse(gw_template)
else:
t = template_format.parse(gw_template_deprecated)
self.stack = utils.parse_stack(t)
resource_defns = self.stack.t.resource_definitions(self.stack)
rsrc = network_gateway.NetworkGateway(
'test_network_gateway',
resource_defns['NetworkGateway'], self.stack)
return rsrc
def _test_network_gateway_create(self, resolve_neutron=True):
rsrc = self.prepare_create_network_gateway(resolve_neutron)
if resolve_neutron:
neutronV20.find_resourceid_by_name_or_id(
mox.IsA(neutronclient.Client),
'network',
'6af055d3-26f6-48dd-a597-7611d7e58d35'
).MultipleTimes().AndReturn(
'6af055d3-26f6-48dd-a597-7611d7e58d35')
neutronclient.Client.disconnect_network_gateway(
'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37', {
'network_id': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'segmentation_id': 10,
'segmentation_type': u'vlan'
}
).AndReturn(None)
neutronclient.Client.disconnect_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37', {
'network_id': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'segmentation_id': 10,
'segmentation_type': u'vlan'
}
).AndReturn(qe.NeutronClientException(status_code=404))
neutronclient.Client.delete_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37'
).AndReturn(None)
neutronclient.Client.show_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37'
).AndReturn(sng)
neutronclient.Client.show_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37'
).AndRaise(qe.NeutronClientException(status_code=404))
neutronclient.Client.delete_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37'
).AndRaise(qe.NeutronClientException(status_code=404))
self.m.ReplayAll()
rsrc.validate()
scheduler.TaskRunner(rsrc.create)()
self.assertEqual((rsrc.CREATE, rsrc.COMPLETE), rsrc.state)
ref_id = rsrc.FnGetRefId()
self.assertEqual(u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37', ref_id)
self.assertRaises(
exception.InvalidTemplateAttribute, rsrc.FnGetAtt, 'Foo')
self.assertIsNone(scheduler.TaskRunner(rsrc.delete)())
self.assertEqual((rsrc.DELETE, rsrc.COMPLETE), rsrc.state)
rsrc.state_set(rsrc.CREATE, rsrc.COMPLETE, 'to delete again')
scheduler.TaskRunner(rsrc.delete)()
self.assertEqual((rsrc.DELETE, rsrc.COMPLETE), rsrc.state)
self.m.VerifyAll()
def test_network_gateway_create_deprecated(self):
self._test_network_gateway_create(resolve_neutron=False)
def test_network_gateway_create(self):
self._test_network_gateway_create()
def test_network_gateway_create_fail_delete_success(self):
# if network_gateway created successful, but didn't to connect with
# network, then can delete the network_gateway successful
# without residue network_gateway
rsrc = self.mock_create_fail_network_not_found_delete_success()
self.stub_NetworkConstraint_validate()
self.m.ReplayAll()
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'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37', ref_id)
self.assertIsNone(scheduler.TaskRunner(rsrc.delete)())
self.assertEqual((rsrc.DELETE, rsrc.COMPLETE), rsrc.state)
self.m.VerifyAll()
def test_network_gateway_update(self):
rsrc = self.prepare_create_network_gateway()
neutronV20.find_resourceid_by_name_or_id(
mox.IsA(neutronclient.Client),
'network',
'6af055d3-26f6-48dd-a597-7611d7e58d35'
).AndReturn('6af055d3-26f6-48dd-a597-7611d7e58d35')
neutronV20.find_resourceid_by_name_or_id(
mox.IsA(neutronclient.Client),
'network',
'6af055d3-26f6-48dd-a597-7611d7e58d35'
).AndReturn('6af055d3-26f6-48dd-a597-7611d7e58d35')
neutronV20.find_resourceid_by_name_or_id(
mox.IsA(neutronclient.Client),
'network',
'6af055d3-26f6-48dd-a597-7611d7e58d35'
).AndReturn('6af055d3-26f6-48dd-a597-7611d7e58d35')
neutronV20.find_resourceid_by_name_or_id(
mox.IsA(neutronclient.Client),
'network',
'6af055d3-26f6-48dd-a597-7611d7e58d35'
).AndReturn('6af055d3-26f6-48dd-a597-7611d7e58d35')
neutronV20.find_resourceid_by_name_or_id(
mox.IsA(neutronclient.Client),
'network',
'6af055d3-26f6-48dd-a597-7611d7e58d35'
).AndReturn('6af055d3-26f6-48dd-a597-7611d7e58d35')
neutronV20.find_resourceid_by_name_or_id(
mox.IsA(neutronclient.Client),
'network',
'6af055d3-26f6-48dd-a597-7611d7e58d35'
).AndReturn('6af055d3-26f6-48dd-a597-7611d7e58d35')
neutronclient.Client.update_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37', {
'network_gateway': {
'name': u'NetworkGatewayUpdate'
}
}
).AndReturn(None)
neutronclient.Client.disconnect_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37', {
'network_id': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'segmentation_id': 10,
'segmentation_type': u'vlan'
}
).AndReturn(None)
neutronclient.Client.connect_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37', {
'network_id': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'segmentation_id': 0,
'segmentation_type': u'flat'
}
).AndReturn({
'connection_info': {
'network_gateway_id': u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37',
'network_id': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'port_id': u'aa800972-f6be-4c65-8453-9ab31834bf80'
}
})
neutronclient.Client.disconnect_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37', {
'network_id': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'segmentation_id': 10,
'segmentation_type': u'vlan'
}
).AndRaise(qe.NeutronClientException(status_code=404))
neutronclient.Client.connect_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37', {
'network_id': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'segmentation_id': 0,
'segmentation_type': u'flat'
}
).AndReturn({
'connection_info': {
'network_gateway_id': u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37',
'network_id': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'port_id': u'aa800972-f6be-4c65-8453-9ab31834bf80'
}
})
neutronclient.Client.disconnect_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37', {
'network_id': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'segmentation_id': 10,
'segmentation_type': u'vlan'
}
).AndReturn(None)
neutronclient.Client.delete_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37'
).AndReturn(None)
neutronclient.Client.create_network_gateway({
'network_gateway': {
'name': u'NetworkGateway',
'devices': [{'id': u'e52148ca-7db9-4ec3-abe6-2c7c0ff316eb',
'interface_name': u'breth2'}]
}
}
).AndReturn({
'network_gateway': {
'id': 'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37',
'name': 'NetworkGateway',
'default': False,
'tenant_id': '96ba52dc-c5c5-44c6-9a9d-d3ba1a03f77f',
'devices': [{
'id': 'e52148ca-7db9-4ec3-abe6-2c7c0ff316eb',
'interface_name': 'breth2'}]
}
}
)
neutronclient.Client.connect_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37', {
'network_id': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'segmentation_id': 10,
'segmentation_type': u'vlan'
}
).AndReturn({
'connection_info': {
'network_gateway_id': u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37',
'network_id': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'port_id': u'aa800972-f6be-4c65-8453-9ab31834bf80'
}
})
self.m.ReplayAll()
rsrc.validate()
scheduler.TaskRunner(rsrc.create)()
self.assertEqual((rsrc.CREATE, rsrc.COMPLETE), rsrc.state)
# update name
snippet_for_update = rsrc_defn.ResourceDefinition(
rsrc.name,
rsrc.type(),
{
'name': u'NetworkGatewayUpdate',
'devices': [{
'id': u'e52148ca-7db9-4ec3-abe6-2c7c0ff316eb',
'interface_name': u'breth1'}],
'connections': [{
'network': '6af055d3-26f6-48dd-a597-7611d7e58d35',
'segmentation_type': 'vlan',
'segmentation_id': 10}]
})
prop_diff = {'name': u'NetworkGatewayUpdate'}
self.assertIsNone(rsrc.handle_update(snippet_for_update,
mox.IgnoreArg(),
prop_diff))
# update connections
snippet_for_update = rsrc_defn.ResourceDefinition(
rsrc.name,
rsrc.type(),
{
'name': u'NetworkGateway',
'devices': [{
'id': u'e52148ca-7db9-4ec3-abe6-2c7c0ff316eb',
'interface_name': u'breth1'}],
'connections': [{
'network': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'segmentation_type': u'flat',
'segmentation_id': 0}]
})
prop_diff = {
'connections': [{
'network': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'segmentation_type': u'flat',
'segmentation_id': 0}]
}
self.assertIsNone(rsrc.handle_update(snippet_for_update,
mox.IgnoreArg(),
prop_diff))
# update connections once more
self.assertIsNone(rsrc.handle_update(snippet_for_update,
mox.IgnoreArg(),
prop_diff))
# update devices
snippet_for_update = rsrc_defn.ResourceDefinition(
rsrc.name,
rsrc.type(),
{
'name': u'NetworkGateway',
'devices': [{
'id': u'e52148ca-7db9-4ec3-abe6-2c7c0ff316eb',
'interface_name': u'breth2'}],
'connections': [{
'network_id': u'6af055d3-26f6-48dd-a597-7611d7e58d35',
'segmentation_type': u'vlan',
'segmentation_id': 10}]
})
prop_diff = {
'devices': [{
'id': u'e52148ca-7db9-4ec3-abe6-2c7c0ff316eb',
'interface_name': u'breth2'}]
}
self.assertIsNone(rsrc.handle_update(snippet_for_update,
mox.IgnoreArg(),
prop_diff))
self.m.VerifyAll()
def test_network_gatway_create_failed(self):
neutronclient.Client.create_network_gateway({
'network_gateway': {
'name': u'NetworkGateway',
'devices': [{
'id': u'e52148ca-7db9-4ec3-abe6-2c7c0ff316eb',
'interface_name': u'breth1'}]
}
}
).AndRaise(qe.NeutronClientException)
self.stub_NetworkConstraint_validate()
self.m.ReplayAll()
t = template_format.parse(gw_template)
stack = utils.parse_stack(t)
resource_defns = stack.t.resource_definitions(stack)
rsrc = network_gateway.NetworkGateway(
'network_gateway', resource_defns['NetworkGateway'], stack)
error = self.assertRaises(exception.ResourceFailure,
scheduler.TaskRunner(rsrc.create))
self.assertEqual(
'NeutronClientException: An unknown exception occurred.',
six.text_type(error))
self.assertEqual((rsrc.CREATE, rsrc.FAILED), rsrc.state)
self.assertIsNone(scheduler.TaskRunner(rsrc.delete)())
self.assertEqual((rsrc.DELETE, rsrc.COMPLETE), rsrc.state)
self.m.VerifyAll()
def test_gateway_validate_failed_with_vlan(self):
t = template_format.parse(gw_template)
del t['resources']['NetworkGateway']['properties'][
'connections'][0]['segmentation_id']
stack = utils.parse_stack(t)
resource_defns = stack.t.resource_definitions(stack)
rsrc = network_gateway.NetworkGateway(
'test_network_gateway',
resource_defns['NetworkGateway'], stack)
self.stub_NetworkConstraint_validate()
self.m.ReplayAll()
error = self.assertRaises(exception.StackValidationFailed,
scheduler.TaskRunner(rsrc.validate))
self.assertEqual(
'segmentation_id must be specified for using vlan',
six.text_type(error))
self.m.VerifyAll()
def test_gateway_validate_failed_with_flat(self):
t = template_format.parse(gw_template)
t['resources']['NetworkGateway']['properties'][
'connections'][0]['segmentation_type'] = 'flat'
stack = utils.parse_stack(t)
resource_defns = stack.t.resource_definitions(stack)
rsrc = network_gateway.NetworkGateway(
'test_network_gateway',
resource_defns['NetworkGateway'], stack)
self.stub_NetworkConstraint_validate()
self.m.ReplayAll()
error = self.assertRaises(exception.StackValidationFailed,
scheduler.TaskRunner(rsrc.validate))
self.assertEqual(
'segmentation_id cannot be specified except 0 for using flat',
six.text_type(error))
self.m.VerifyAll()
def test_network_gateway_attribute(self):
neutronclient.Client.show_network_gateway(
u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37'
).MultipleTimes().AndReturn(sng)
rsrc = self.prepare_create_network_gateway()
self.m.ReplayAll()
scheduler.TaskRunner(rsrc.create)()
self.assertEqual(u'ed4c03b9-8251-4c09-acc4-e59ee9e6aa37',
rsrc.FnGetRefId())
self.assertEqual(False, rsrc.FnGetAtt('default'))
error = self.assertRaises(exception.InvalidTemplateAttribute,
rsrc.FnGetAtt, 'hoge')
self.assertEqual(
'The Referenced Attribute (test_network_gateway hoge) is '
'incorrect.', six.text_type(error))
self.m.VerifyAll()