Add functional test for l3_agent

In order to insert the HA related code into the L3
agent, as part of blueprint l3-high-availability,
it's important to add functional tests for the L3 agent.

The L3 HA patch will use the framework provided here
to implement additional HA specific tests.

Implements: blueprint l3-high-availability
Change-Id: I49ddc95a0c41330580fcec6ba05c72684248af5e

neutron/agent/l3_agent.py

@@ -1899,8 +1899,7 @@ class L3NATAgentWithStateReport(L3NATAgent):
         LOG.info(_("agent_updated by server side %s!"), payload)
 
 
-def main(manager='neutron.agent.l3_agent.L3NATAgentWithStateReport'):
+def _register_opts(conf):
-    conf = cfg.CONF
     conf.register_opts(L3NATAgent.OPTS)
     config.register_interface_driver_opts_helper(conf)
     config.register_use_namespaces_opts_helper(conf)
@@ -1908,8 +1907,12 @@ def main(manager='neutron.agent.l3_agent.L3NATAgentWithStateReport'):
     config.register_root_helper(conf)
     conf.register_opts(interface.OPTS)
     conf.register_opts(external_process.OPTS)
+
+
+def main(manager='neutron.agent.l3_agent.L3NATAgentWithStateReport'):
+    _register_opts(cfg.CONF)
     common_config.init(sys.argv[1:])
-    config.setup_logging(conf)
+    config.setup_logging(cfg.CONF)
     server = neutron_service.Service.create(
         binary='neutron-l3-agent',
         topic=topics.L3_AGENT,

neutron/agent/linux/iptables_manager.py

@@ -236,11 +236,17 @@ class IptablesTable(object):
                      {'chain': chain, 'rule': rule,
                       'top': top, 'wrap': wrap})
 
+    def _get_chain_rules(self, chain, wrap):
+        chain = get_chain_name(chain, wrap)
+        return [rule for rule in self.rules
+                if rule.chain == chain and rule.wrap == wrap]
+
+    def is_chain_empty(self, chain, wrap=True):
+        return not self._get_chain_rules(chain, wrap)
+
     def empty_chain(self, chain, wrap=True):
         """Remove all rules from a chain."""
-        chain = get_chain_name(chain, wrap)
+        chained_rules = self._get_chain_rules(chain, wrap)
-        chained_rules = [rule for rule in self.rules
-                         if rule.chain == chain and rule.wrap == wrap]
         for rule in chained_rules:
             self.rules.remove(rule)
 
@@ -349,6 +355,13 @@ class IptablesManager(object):
             self.ipv4['nat'].add_chain('float-snat')
             self.ipv4['nat'].add_rule('snat', '-j $float-snat')
 
+    def is_chain_empty(self, table, chain, ip_version=4, wrap=True):
+        try:
+            requested_table = {4: self.ipv4, 6: self.ipv6}[ip_version][table]
+        except KeyError:
+            return True
+        return requested_table.is_chain_empty(chain, wrap)
+
     def defer_apply_on(self):
         self.iptables_apply_deferred = True
 

neutron/tests/functional/agent/test_l3_agent.py

@@ -0,0 +1,156 @@
+# Copyright (c) 2014 Red Hat, Inc.
+# 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 mock
+from oslo.config import cfg
+
+from neutron.agent.common import config
+from neutron.agent import l3_agent
+from neutron.agent.linux import external_process
+from neutron.agent.linux import ip_lib
+from neutron.common import constants as l3_constants
+from neutron.openstack.common import log as logging
+from neutron.openstack.common import uuidutils
+from neutron.tests.functional.agent.linux import base
+from neutron.tests.unit import test_l3_agent
+
+LOG = logging.getLogger(__name__)
+_uuid = uuidutils.generate_uuid
+
+
+class L3AgentTestFramework(base.BaseOVSLinuxTestCase):
+    def setUp(self):
+        super(L3AgentTestFramework, self).setUp()
+        self.check_sudo_enabled()
+        self._configure()
+
+    def _configure(self):
+        l3_agent._register_opts(cfg.CONF)
+        cfg.CONF.set_override('debug', True)
+        config.setup_logging(cfg.CONF)
+        cfg.CONF.set_override(
+            'interface_driver',
+            'neutron.agent.linux.interface.OVSInterfaceDriver')
+        cfg.CONF.set_override('router_delete_namespaces', True)
+        cfg.CONF.set_override('root_helper', self.root_helper, group='AGENT')
+        cfg.CONF.set_override('use_namespaces', True)
+        cfg.CONF.set_override('enable_metadata_proxy', True)
+
+        br_int = self.create_ovs_bridge()
+        cfg.CONF.set_override('ovs_integration_bridge', br_int.br_name)
+        br_ex = self.create_ovs_bridge()
+        cfg.CONF.set_override('external_network_bridge', br_ex.br_name)
+
+        mock.patch('neutron.common.rpc.RpcProxy.cast').start()
+        mock.patch('neutron.common.rpc.RpcProxy.call').start()
+        mock.patch('neutron.common.rpc.RpcProxy.fanout_cast').start()
+        self.agent = l3_agent.L3NATAgent('localhost', cfg.CONF)
+
+        mock.patch.object(self.agent, '_send_gratuitous_arp_packet').start()
+
+    def manage_router(self):
+        router = test_l3_agent.prepare_router_data(enable_snat=True,
+                                                   enable_floating_ip=True)
+        self.addCleanup(self._delete_router, router['id'])
+        ri = self._create_router(router)
+        return ri
+
+    def _create_router(self, router):
+        self.agent._router_added(router['id'], router)
+        ri = self.agent.router_info[router['id']]
+        ri.router = router
+        self.agent.process_router(ri)
+        return ri
+
+    def _delete_router(self, router_id):
+        self.agent._router_removed(router_id)
+
+    def _namespace_exists(self, router):
+        ip = ip_lib.IPWrapper(self.root_helper, router.ns_name)
+        return ip.netns.exists(router.ns_name)
+
+    def _metadata_proxy_exists(self, router):
+        pm = external_process.ProcessManager(
+            cfg.CONF,
+            router.router_id,
+            self.root_helper,
+            router.ns_name)
+        return pm.active
+
+    def device_exists_with_ip_mac(self, expected_device, name_getter,
+                                  namespace):
+        return ip_lib.device_exists_with_ip_mac(
+            name_getter(expected_device['id']),
+            expected_device['ip_cidr'],
+            expected_device['mac_address'],
+            namespace, self.root_helper)
+
+
+class L3AgentTestCase(L3AgentTestFramework):
+    def test_router_lifecycle(self):
+        router = self.manage_router()
+
+        self.assertTrue(self._namespace_exists(router))
+        self.assertTrue(self._metadata_proxy_exists(router))
+        self._assert_internal_devices(router)
+        self._assert_external_device(router)
+        self._assert_gateway(router)
+        self._assert_snat_chains(router)
+        self._assert_floating_ip_chains(router)
+
+        self._delete_router(router.router_id)
+        self._assert_router_does_not_exist(router)
+
+    def _assert_internal_devices(self, router):
+        internal_devices = router.router[l3_constants.INTERFACE_KEY]
+        self.assertTrue(len(internal_devices))
+        for device in internal_devices:
+            self.assertTrue(self.device_exists_with_ip_mac(
+                device, self.agent.get_internal_device_name, router.ns_name))
+
+    def _assert_external_device(self, router):
+        external_port = self.agent._get_ex_gw_port(router)
+        self.assertTrue(self.device_exists_with_ip_mac(
+            external_port, self.agent.get_external_device_name,
+            router.ns_name))
+
+    def _assert_gateway(self, router):
+        external_port = self.agent._get_ex_gw_port(router)
+        external_device_name = self.agent.get_external_device_name(
+            external_port['id'])
+        external_device = ip_lib.IPDevice(external_device_name,
+                                          self.root_helper,
+                                          router.ns_name)
+        existing_gateway = (
+            external_device.route.get_gateway().get('gateway'))
+        expected_gateway = external_port['subnet']['gateway_ip']
+        self.assertEqual(expected_gateway, existing_gateway)
+
+    def _assert_snat_chains(self, router):
+        self.assertFalse(router.iptables_manager.is_chain_empty(
+            'nat', 'snat'))
+        self.assertFalse(router.iptables_manager.is_chain_empty(
+            'nat', 'POSTROUTING'))
+
+    def _assert_floating_ip_chains(self, router):
+        self.assertFalse(router.iptables_manager.is_chain_empty(
+            'nat', 'float-snat'))
+
+    def _assert_router_does_not_exist(self, router):
+        # If the namespace assertion succeeds
+        # then the devices and iptable rules have also been deleted,
+        # so there's no need to check that explicitly.
+        self.assertFalse(self._namespace_exists(router))
+        self.assertFalse(self._metadata_proxy_exists(router))

neutron/tests/unit/test_l3_agent.py

@@ -229,7 +229,8 @@ def router_append_interface(router, count=1, ip_version=4, ra_mode=None,
                         'ipv6_address_mode': addr_mode}})
 
 
-def prepare_router_data(ip_version=4, enable_snat=None, num_internal_ports=1):
+def prepare_router_data(ip_version=4, enable_snat=None, num_internal_ports=1,
+                        enable_floating_ip=False):
     if ip_version == 4:
         ip_addr = '19.4.4.4'
         cidr = '19.4.4.0/24'
@@ -243,6 +244,7 @@ def prepare_router_data(ip_version=4, enable_snat=None, num_internal_ports=1):
 
     router_id = _uuid()
     ex_gw_port = {'id': _uuid(),
+                  'mac_address': 'ca:fe:de:ad:be:ef',
                   'network_id': _uuid(),
                   'fixed_ips': [{'ip_address': ip_addr,
                                  'subnet_id': _uuid()}],
@@ -255,6 +257,14 @@ def prepare_router_data(ip_version=4, enable_snat=None, num_internal_ports=1):
         l3_constants.INTERFACE_KEY: [],
         'routes': [],
         'gw_port': ex_gw_port}
+
+    if enable_floating_ip:
+        router[l3_constants.FLOATINGIP_KEY] = [{
+            'id': _uuid(),
+            'port_id': _uuid(),
+            'floating_ip_address': '19.4.4.2',
+            'fixed_ip_address': '10.0.0.1'}]
+
     router_append_interface(router, count=num_internal_ports,
                             ip_version=ip_version)