From 22dc8bbbf5606de3af52abd131c386188262d65f Mon Sep 17 00:00:00 2001 From: Federico Ressi Date: Tue, 22 Oct 2019 20:10:43 +0200 Subject: [PATCH] Improve the use of tobiko framework in test_routers Change-Id: I2ddf46a9dced575e3af9935771cfc7bcd3efad86 --- tobiko/tests/scenario/neutron/test_router.py | 120 ++++++++----------- 1 file changed, 53 insertions(+), 67 deletions(-) diff --git a/tobiko/tests/scenario/neutron/test_router.py b/tobiko/tests/scenario/neutron/test_router.py index f9751e50f..7dce31389 100644 --- a/tobiko/tests/scenario/neutron/test_router.py +++ b/tobiko/tests/scenario/neutron/test_router.py @@ -23,120 +23,106 @@ from tobiko.shell import ip 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): +class RouterTest(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() + super(RouterTest, self).setUp() if not self.stack.network_stack.has_gateway: tobiko.skip('Stack {!s} has no gateway', self.stack.network_stack.stack_name) - self.network_stack = self.stack.network_stack - self.router = self.network_stack.gateway_details - self.router_ipv4_address = self.network_stack.ipv4_subnet_gateway_ip - self.router_ipv6_address = self.network_stack.ipv6_subnet_gateway_ip - self.router_gateway_ip = self.network_stack.external_gateway_ips.first + @property + def router(self): + return self.stack.network_stack.gateway_details - tripleo_topology.setup_tripleo_topology() - self.topology = topology.get_openstack_topology() - - def test_internal_router_ipv4_interface_is_reachable(self): + @property + def ipv4_subnet_gateway_ip(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() + return self.stack.network_stack.ipv4_subnet_gateway_ip - def test_internal_router_ipv6_interface_is_reachable(self): + @property + def ipv6_subnet_gateway_ip(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() + return self.stack.network_stack.ipv6_subnet_gateway_ip + + @property + def external_gateway_ips(self): + return self.stack.network_stack.external_gateway_ips + + @property + def router_ips(self): + return tobiko.select([self.ipv4_subnet_gateway_ip, + self.ipv6_subnet_gateway_ip] + + self.external_gateway_ips) + + def test_internal_router_ipv4_interface_is_reachable(self): + ping.assert_reachable_ips([self.ipv4_subnet_gateway_ip], + ssh_client=self.stack.ssh_client) + + def test_internal_router_ipv6_interface_is_reachable(self): + ping.assert_reachable_ips([self.ipv6_subnet_gateway_ip], + ssh_client=self.stack.ssh_client) def test_router_gateway_is_reachable(self): - ping.ping( - self.router_gateway_ip, - ssh_client=self.stack.ssh_client - ).assert_replied() + ping.assert_reachable_ips(self.external_gateway_ips, + ssh_client=self.stack.ssh_client) + + def test_ipv4_subnet_gateway_ip(self): + self.assertEqual(4, self.ipv4_subnet_gateway_ip.version) + self.assertIn(self.ipv4_subnet_gateway_ip, + self.stack.network_stack.ipv4_gateway_addresses) + + def test_ipv6_subnet_gateway_ip(self): + self.assertEqual(6, self.ipv6_subnet_gateway_ip.version) + self.assertIn(self.ipv6_subnet_gateway_ip, + self.stack.network_stack.ipv6_gateway_addresses) @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 test_router_ipv4_address(self): - self.assertEqual(4, self.router_ipv4_address.version) - self.assertIn(self.router_ipv4_address, - self.network_stack.ipv4_gateway_addresses) - - def test_router_ipv6_address(self): - self.assertEqual(6, self.router_ipv6_address.version) - self.assertIn(self.router_ipv6_address, - self.network_stack.ipv6_gateway_addresses) - - 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) + router_agent = neutron.find_l3_agent_hosting_router(self.router['id'], + unique=True) + self._check_routers_namespace_on_host(router_agent['host']) 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) + agent_host = topology.get_openstack_node(hostname=hostname) namespaces = ip.list_network_namespaces( ssh_client=agent_host.ssh_client) self.assertIn(router_namespace, namespaces) namespace_ips = ip.list_ip_addresses( scope='global', network_namespace=router_namespace, ssh_client=agent_host.ssh_client) + missing_ips = set(self.router_ips) - set(namespace_ips) if state == "master": - self.assertIn(self.router_ipv4_address, namespace_ips) - self.assertIn(self.router_ipv6_address, namespace_ips) - self.assertIn(self.router_gateway_ip, namespace_ips) + self.assertFalse(missing_ips) else: - self.assertNotIn(self.router_ipv4_address, namespace_ips) - self.assertNotIn(self.router_ipv6_address, namespace_ips) - self.assertNotIn(self.router_gateway_ip, namespace_ips) - - def _test_router_is_scheduled_on_l3_agents(self): - router_agent = neutron.find_l3_agent_hosting_router(self.router['id'], - unique=True) - self._check_routers_namespace_on_host(router_agent['host']) + self.assertTrue(missing_ips) @neutron.skip_if_missing_networking_extensions('l3-ha') -class HaRouterTest(LegacyRouterTest): +@neutron.skip_if_missing_networking_agents(binary='neutron-l3-agent', + count=2) +class L3HARouterTest(RouterTest): """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): + @neutron.skip_if_missing_networking_extensions('l3_agent_scheduler') + def test_router_is_scheduled_on_l3_agents(self): router_agents = neutron.list_l3_agent_hosting_routers( self.router['id']) master_agent = router_agents.with_items(ha_state='active').unique