Add high-level functional/integration DVR tests

* A note from the legal team: These tests in no way replace
  any existing tests. I would never dream of such a thing. Nor
  would anyone ever consider calling these 'unit' tests. That
  would be mad!

Change-Id: I73c2b2096e767575a196bf08e7d4cc7ec52fdfa3
Co-Authored-By: Lynn Li <lynn.li@hp.com>
This commit is contained in:
Assaf Muller 2015-04-28 10:47:13 -04:00
parent e0bc5239e0
commit b1fe6426e3
5 changed files with 210 additions and 3 deletions

View File

@ -0,0 +1,168 @@
# Copyright (c) 2015 Red Hat, Inc.
#
# 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.
from neutron.api.v2 import attributes
from neutron.common import constants as l3_const
from neutron.db import l3_dvr_db
from neutron.extensions import external_net
from neutron.tests.common import helpers
from neutron.tests.unit.plugins.ml2 import base as ml2_test_base
class L3DvrTestCase(ml2_test_base.ML2TestFramework):
def setUp(self):
super(L3DvrTestCase, self).setUp()
self.l3_agent = helpers.register_l3_agent(
agent_mode=l3_const.L3_AGENT_MODE_DVR_SNAT)
def _create_router(self, distributed=True):
return (super(L3DvrTestCase, self).
_create_router(distributed=distributed))
def test_update_router_db_centralized_to_distributed(self):
router = self._create_router(distributed=False)
self.assertFalse(router['distributed'])
self.l3_plugin.update_router(
self.context, router['id'], {'router': {'distributed': True}})
router = self.l3_plugin.get_router(self.context, router['id'])
self.assertTrue(router['distributed'])
def test_get_device_owner_distributed_router_object(self):
router = self._create_router()
self.assertEqual(
l3_dvr_db.DEVICE_OWNER_DVR_INTERFACE,
self.l3_plugin._get_device_owner(self.context, router))
def test_get_device_owner_distributed_router_id(self):
router = self._create_router()
self.assertEqual(
l3_dvr_db.DEVICE_OWNER_DVR_INTERFACE,
self.l3_plugin._get_device_owner(self.context, router['id']))
def test_get_device_owner_centralized(self):
router = self._create_router(distributed=False)
self.assertEqual(
l3_const.DEVICE_OWNER_ROUTER_INTF,
self.l3_plugin._get_device_owner(self.context, router['id']))
def test_get_agent_gw_ports_exist_for_network_no_port(self):
self.assertIsNone(
self.l3_plugin._get_agent_gw_ports_exist_for_network(
self.context, 'network_id', 'host', 'agent_id'))
def _test_remove_router_interface_leaves_snat_intact(self, by_subnet):
with self.subnet() as subnet1, \
self.subnet(cidr='20.0.0.0/24') as subnet2:
kwargs = {'arg_list': (external_net.EXTERNAL,),
external_net.EXTERNAL: True}
with self.network(**kwargs) as ext_net, \
self.subnet(network=ext_net,
cidr='30.0.0.0/24'):
router = self._create_router()
self.l3_plugin.add_router_interface(
self.context, router['id'],
{'subnet_id': subnet1['subnet']['id']})
self.l3_plugin.add_router_interface(
self.context, router['id'],
{'subnet_id': subnet2['subnet']['id']})
self.l3_plugin._update_router_gw_info(
self.context, router['id'],
{'network_id': ext_net['network']['id']})
snat_router_intfs = self.l3_plugin._get_snat_sync_interfaces(
self.context, [router['id']])
self.assertEqual(
2, len(snat_router_intfs[router['id']]))
if by_subnet:
self.l3_plugin.remove_router_interface(
self.context, router['id'],
{'subnet_id': subnet1['subnet']['id']})
else:
port = self.core_plugin.get_ports(
self.context, filters={
'network_id': [subnet1['subnet']['network_id']],
'device_owner':
[l3_const.DEVICE_OWNER_DVR_INTERFACE]})[0]
self.l3_plugin.remove_router_interface(
self.context, router['id'],
{'port_id': port['id']})
self.assertEqual(
1, len(self.l3_plugin._get_snat_sync_interfaces(
self.context, [router['id']])))
def test_remove_router_interface_by_subnet_leaves_snat_intact(self):
self._test_remove_router_interface_leaves_snat_intact(by_subnet=True)
def test_remove_router_interface_by_port_leaves_snat_intact(self):
self._test_remove_router_interface_leaves_snat_intact(
by_subnet=False)
def setup_create_agent_gw_port_for_network(self):
network = self._make_network(self.fmt, '', True)
network_id = network['network']['id']
port = self.core_plugin.create_port(
self.context,
{'port': {'tenant_id': '',
'network_id': network_id,
'mac_address': attributes.ATTR_NOT_SPECIFIED,
'fixed_ips': attributes.ATTR_NOT_SPECIFIED,
'device_id': self.l3_agent['id'],
'device_owner': l3_dvr_db.DEVICE_OWNER_AGENT_GW,
'binding:host_id': '',
'admin_state_up': True,
'name': ''}})
return network_id, port
def test_get_agent_gw_port_for_network(self):
network_id, port = (
self.setup_create_agent_gw_port_for_network())
self.assertEqual(
port['id'],
self.l3_plugin._get_agent_gw_ports_exist_for_network(
self.context, network_id, None, self.l3_agent['id'])['id'])
def test_delete_agent_gw_port_for_network(self):
network_id, port = (
self.setup_create_agent_gw_port_for_network())
self.l3_plugin._delete_floatingip_agent_gateway_port(
self.context, "", network_id)
self.assertIsNone(
self.l3_plugin._get_agent_gw_ports_exist_for_network(
self.context, network_id, "", self.l3_agent['id']))
def test_get_fip_sync_interfaces(self):
self.setup_create_agent_gw_port_for_network()
self.assertEqual(
1, len(self.l3_plugin._get_fip_sync_interfaces(
self.context, self.l3_agent['id'])))
def test_process_routers(self):
router = self._create_router()
result = self.l3_plugin._process_routers(self.context, [router])
self.assertEqual(
router['id'], result[router['id']]['id'])
def test_get_router_ids(self):
router = self._create_router()
self.assertEqual(
router['id'],
self.l3_plugin._get_router_ids(self.context)[0])
self._create_router()
self.assertEqual(
2, len(self.l3_plugin._get_router_ids(self.context)))

View File

@ -0,0 +1,39 @@
# Copyright (c) 2015 Red Hat, Inc.
#
# 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.
from neutron import manager
from neutron.plugins.common import constants as plugin_constants
from neutron.tests.unit.plugins.ml2 import test_plugin
class ML2TestFramework(test_plugin.Ml2PluginV2TestCase):
l3_plugin = ('neutron.services.l3_router.l3_router_plugin.'
'L3RouterPlugin')
_mechanism_drivers = ['openvswitch']
def setUp(self):
super(ML2TestFramework, self).setUp()
self.core_plugin = manager.NeutronManager.get_instance().get_plugin()
self.l3_plugin = manager.NeutronManager.get_service_plugins().get(
plugin_constants.L3_ROUTER_NAT)
def _create_router(self, distributed=False, ha=False):
return self.l3_plugin.create_router(
self.context,
{'router':
{'name': 'router',
'admin_state_up': True,
'tenant_id': self._tenant_id,
'ha': ha,
'distributed': distributed}})

View File

@ -95,12 +95,12 @@ class Ml2ConfFixture(PluginConfFixture):
class Ml2PluginV2TestCase(test_plugin.NeutronDbPluginV2TestCase):
_mechanism_drivers = ['logger', 'test']
l3_plugin = ('neutron.tests.unit.extensions.test_l3.'
'TestL3NatServicePlugin')
def setup_parent(self):
"""Perform parent setup with the common plugin configuration class."""
l3_plugin = ('neutron.tests.unit.extensions.test_l3.'
'TestL3NatServicePlugin')
service_plugins = {'l3_plugin_name': l3_plugin}
service_plugins = {'l3_plugin_name': self.l3_plugin}
# Ensure that the parent setup can be called without arguments
# by the common configuration setUp.
parent_setup = functools.partial(