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>

diff --git a/neutron/tests/functional/services/__init__.py b/neutron/tests/functional/services/__init__.py
new file mode 100644 (file)
index 0000000..e69de29

diff --git a/neutron/tests/functional/services/l3_router/__init__.py b/neutron/tests/functional/services/l3_router/__init__.py
new file mode 100644 (file)
index 0000000..e69de29

diff --git a/neutron/tests/functional/services/l3_router/test_l3_dvr_router_plugin.py b/neutron/tests/functional/services/l3_router/test_l3_dvr_router_plugin.py
new file mode 100644 (file)
index 0000000..473133b
--- /dev/null
+++ b/neutron/tests/functional/services/l3_router/test_l3_dvr_router_plugin.py
@@ -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)))

diff --git a/neutron/tests/unit/plugins/ml2/base.py b/neutron/tests/unit/plugins/ml2/base.py
new file mode 100644 (file)
index 0000000..6c193a4
--- /dev/null
+++ b/neutron/tests/unit/plugins/ml2/base.py
@@ -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}})

diff --git a/neutron/tests/unit/plugins/ml2/test_plugin.py b/neutron/tests/unit/plugins/ml2/test_plugin.py
index 2e5c7392719d9d3aedee35678251544ca93feed1..56c0a3d270be1a1d20c1fd0fa4e74685326774e3 100644 (file)
--- a/neutron/tests/unit/plugins/ml2/test_plugin.py
+++ b/neutron/tests/unit/plugins/ml2/test_plugin.py
@@ -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(