Add tests for routers

This patch adds basic tests for router, like:
* test if router's internal IPv4 address is reachable from VM,
* test if router's internal IPv6 address is reachable from VM,
* test if external gateway IP address is reachable from VM,
* test if router is scheduled on proper number of L3 agents (if
  agents scheduler extension is available)
* test if qrouter- namespaces are created properly on hosts where
  router is scheduled (if agents scheduler extension is available)

To make those tests, additional method for neutron client is added to
list l3 agents hosting given router.

Change-Id: I1ce358eb7a48c1343f626f6bee01d9b26f79473f

tobiko/openstack/neutron/__init__.py

@@ -35,6 +35,7 @@ get_network = _client.get_network
 get_router = _client.get_router
 get_port = _client.get_port
 get_subnet = _client.get_subnet
+list_agents_hosting_router = _client.list_agents_hosting_router
 
 new_ipv4_cidr = _cidr.new_ipv4_cidr
 new_ipv6_cidr = _cidr.new_ipv6_cidr

tobiko/openstack/neutron/_client.py

@@ -144,3 +144,11 @@ def get_router(router, client=None, **params):
 
 def get_subnet(subnet, client=None, **params):
     return neutron_client(client).show_subnet(subnet, **params)['subnet']
+
+
+def list_agents_hosting_router(router, client=None, **params):
+    agents = neutron_client(client).list_l3_agent_hosting_routers(
+        router, **params)
+    if isinstance(agents, collections.Mapping):
+        agents = agents['agents']
+    return tobiko.select(agents)

tobiko/tests/scenario/neutron/test_router.py

@@ -0,0 +1,148 @@
+# Copyright (c) 2019 Red Hat
+# 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.
+from __future__ import absolute_import
+
+import testtools
+
+import tobiko
+from tobiko import config
+from tobiko.shell import ping
+from tobiko.shell import sh
+from tobiko.openstack import neutron
+from tobiko.openstack import stacks
+from tobiko.openstack import topology
+from tobiko.tripleo import topology as tripleo_topology
+
+
+CONF = config.CONF
+
+
+class LegacyRouterTest(testtools.TestCase):
+    """Test Neutron routers"""
+
+    #: Resources stack with Nova server to send messages to
+    stack = tobiko.required_setup_fixture(stacks.CirrosServerStackFixture)
+
+    def setUp(self):
+        super(LegacyRouterTest, self).setUp()
+        if not self.stack.network_stack.has_gateway:
+            tobiko.skip('Stack {!s} has no gateway',
+                        self.stack.network_stack.stack_name)
+        self.router = self.stack.network_stack.gateway_details
+        self.router_ipv4_address = neutron.find_port_ip_address(
+             port=self.stack.network_stack.ipv4_gateway_port_details,
+             ip_version=4)
+        self.router_ipv6_address = neutron.find_port_ip_address(
+             port=self.stack.network_stack.ipv6_gateway_port_details,
+             ip_version=6)
+        self.router_gw_ip = self.router['external_gateway_info'][
+            'external_fixed_ips'][0]['ip_address']
+        tripleo_topology.setup_tripleo_topology()
+        self.topology = topology.get_openstack_topology()
+
+    def test_internal_router_ipv4_interface_is_reachable(self):
+        if not self.stack.network_stack.has_ipv4:
+            tobiko.skip('Stack {!s} has no ipv4 subnet',
+                        self.stack.network_stack.stack_name)
+        ping.ping(
+            self.router_ipv4_address,
+            ssh_client=self.stack.ssh_client
+        ).assert_replied()
+
+    def test_internal_router_ipv6_interface_is_reachable(self):
+        if not self.stack.network_stack.has_ipv6:
+            tobiko.skip('Stack {!s} has no ipv6 subnet',
+                        self.stack.network_stack.stack_name)
+        ping.ping(
+            self.router_ipv6_address,
+            ssh_client=self.stack.ssh_client
+        ).assert_replied()
+
+    def test_router_gateway_is_reachable(self):
+        ping.ping(
+            self.router_gw_ip,
+            ssh_client=self.stack.ssh_client
+        ).assert_replied()
+
+    @neutron.skip_if_missing_networking_extensions('l3_agent_scheduler')
+    def test_router_is_scheduled_on_l3_agents(self):
+        self._test_router_is_scheduled_on_l3_agents()
+
+    def _get_l3_agent_nodes(self, hostname):
+        hostname = hostname.split(".")
+        for host in self.topology.nodes:
+            if host.name in hostname:
+                return host
+        self.fail("Node with hostname %s not found in cloud topology" %
+                  hostname)
+
+    def _check_routers_namespace_on_host(self, hostname, state="master"):
+        router_namespace = "qrouter-%s" % self.router['id']
+        agent_host = self._get_l3_agent_nodes(hostname)
+        ns_list = sh.execute(
+            "sudo ip netns list", ssh_client=agent_host.ssh_client)
+        self.assertIn(router_namespace, ns_list.stdout)
+        ns_net_config = sh.execute(
+            "sudo ip netns exec %s ip -o addr" % router_namespace,
+            ssh_client=agent_host.ssh_client)
+        if state == "master":
+            self.assertIn(
+                str(self.router_ipv4_address), ns_net_config.stdout)
+            self.assertIn(
+                str(self.router_ipv6_address), ns_net_config.stdout)
+            self.assertIn(
+                str(self.router_gw_ip), ns_net_config.stdout)
+        else:
+            self.assertNotIn(
+                str(self.router_ipv4_address), ns_net_config.stdout)
+            self.assertNotIn(
+                str(self.router_ipv6_address), ns_net_config.stdout)
+            self.assertNotIn(
+                str(self.router_gw_ip), ns_net_config.stdout)
+
+    def _test_router_is_scheduled_on_l3_agents(self):
+        router_agents = neutron.list_agents_hosting_router(self.router['id'])
+        self.assertEqual(1, len(router_agents))
+        self._check_routers_namespace_on_host(router_agents[0]['host'])
+
+
+@neutron.skip_if_missing_networking_extensions('l3-ha')
+class HaRouterTest(LegacyRouterTest):
+    """Test Neutron HA routers"""
+
+    #: Resources stack with Nova server to send messages to
+    stack = tobiko.required_setup_fixture(stacks.L3haServerStackFixture)
+
+    def setUp(self):
+        l3_agents = neutron.list_agents(agent_type="L3 agent")
+        if len(l3_agents) < 2:
+            neutron_extensions = neutron.get_networking_extensions()
+            if "l3_agent_scheduler" in neutron_extensions:
+                tobiko.skip("Ha router tests requires at least 2 L3 agents in "
+                            "the cloud.")
+        super(HaRouterTest, self).setUp()
+
+    def _test_router_is_scheduled_on_l3_agents(self):
+        router_agents = neutron.list_agents_hosting_router(self.router['id'])
+        master_agents = [
+            agent for agent in router_agents if agent['ha_state'] == 'active']
+        backup_agents = [
+            agent for agent in router_agents if agent['ha_state'] == 'standby']
+        self.assertEqual(1, len(master_agents))
+        self.assertGreaterEqual(len(backup_agents), 1)
+        self._check_routers_namespace_on_host(master_agents[0]['host'])
+        for backup_agent in backup_agents:
+            self._check_routers_namespace_on_host(
+                backup_agent['host'], state="backup")