Add stack properties to get network gateway IPs
Update test_router to use new properties. Change-Id: I73f43c5901cd8ed0e8be01bfeca922a197da5f16
This commit is contained in:
parent
c63fd45ff2
commit
4eb547348c
|
@ -18,6 +18,7 @@ from __future__ import absolute_import
|
||||||
import netaddr
|
import netaddr
|
||||||
from oslo_log import log
|
from oslo_log import log
|
||||||
|
|
||||||
|
import tobiko
|
||||||
from tobiko import config
|
from tobiko import config
|
||||||
from tobiko.openstack import heat
|
from tobiko.openstack import heat
|
||||||
from tobiko.openstack import neutron
|
from tobiko.openstack import neutron
|
||||||
|
@ -108,6 +109,10 @@ class NetworkStackFixture(heat.HeatStackFixture):
|
||||||
def ipv4_subnet_cidr(self):
|
def ipv4_subnet_cidr(self):
|
||||||
return netaddr.IPNetwork(self.ipv4_subnet_details['cidr'])
|
return netaddr.IPNetwork(self.ipv4_subnet_details['cidr'])
|
||||||
|
|
||||||
|
@property
|
||||||
|
def ipv4_subnet_gateway_ip(self):
|
||||||
|
return netaddr.IPAddress(self.ipv4_subnet_details['gateway_ip'])
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def ipv6_subnet_details(self):
|
def ipv6_subnet_details(self):
|
||||||
return neutron.get_subnet(self.ipv6_subnet_id)
|
return neutron.get_subnet(self.ipv6_subnet_id)
|
||||||
|
@ -116,10 +121,21 @@ class NetworkStackFixture(heat.HeatStackFixture):
|
||||||
def ipv6_subnet_cidr(self):
|
def ipv6_subnet_cidr(self):
|
||||||
return netaddr.IPNetwork(self.ipv6_subnet_details['cidr'])
|
return netaddr.IPNetwork(self.ipv6_subnet_details['cidr'])
|
||||||
|
|
||||||
|
@property
|
||||||
|
def ipv6_subnet_gateway_ip(self):
|
||||||
|
return netaddr.IPAddress(self.ipv6_subnet_details['gateway_ip'])
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def gateway_details(self):
|
def gateway_details(self):
|
||||||
return neutron.get_router(self.gateway_id)
|
return neutron.get_router(self.gateway_id)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def external_gateway_ips(self):
|
||||||
|
fixed_ips = self.gateway_details['external_gateway_info'][
|
||||||
|
'external_fixed_ips']
|
||||||
|
return tobiko.select(netaddr.IPAddress(fixed_ip['ip_address'])
|
||||||
|
for fixed_ip in fixed_ips)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def ipv4_gateway_port_details(self):
|
def ipv4_gateway_port_details(self):
|
||||||
return neutron.find_port(
|
return neutron.find_port(
|
||||||
|
|
|
@ -40,13 +40,13 @@ class LegacyRouterTest(testtools.TestCase):
|
||||||
if not self.stack.network_stack.has_gateway:
|
if not self.stack.network_stack.has_gateway:
|
||||||
tobiko.skip('Stack {!s} has no gateway',
|
tobiko.skip('Stack {!s} has no gateway',
|
||||||
self.stack.network_stack.stack_name)
|
self.stack.network_stack.stack_name)
|
||||||
self.router = self.stack.network_stack.gateway_details
|
|
||||||
self.router_ipv4_address = (
|
network_stack = self.stack.network_stack
|
||||||
self.stack.network_stack.ipv4_subnet_details['gateway_ip'])
|
self.router = network_stack.gateway_details
|
||||||
self.router_ipv6_address = (
|
self.router_ipv4_address = network_stack.ipv4_subnet_gateway_ip
|
||||||
self.stack.network_stack.ipv6_subnet_details['gateway_ip'])
|
self.router_ipv6_address = network_stack.ipv6_subnet_gateway_ip
|
||||||
self.router_gw_ip = self.router['external_gateway_info'][
|
self.router_gateway_ip = network_stack.external_gateway_ips.first
|
||||||
'external_fixed_ips'][0]['ip_address']
|
|
||||||
tripleo_topology.setup_tripleo_topology()
|
tripleo_topology.setup_tripleo_topology()
|
||||||
self.topology = topology.get_openstack_topology()
|
self.topology = topology.get_openstack_topology()
|
||||||
|
|
||||||
|
@ -70,7 +70,7 @@ class LegacyRouterTest(testtools.TestCase):
|
||||||
|
|
||||||
def test_router_gateway_is_reachable(self):
|
def test_router_gateway_is_reachable(self):
|
||||||
ping.ping(
|
ping.ping(
|
||||||
self.router_gw_ip,
|
self.router_gateway_ip,
|
||||||
ssh_client=self.stack.ssh_client
|
ssh_client=self.stack.ssh_client
|
||||||
).assert_replied()
|
).assert_replied()
|
||||||
|
|
||||||
|
@ -78,6 +78,24 @@ class LegacyRouterTest(testtools.TestCase):
|
||||||
def test_router_is_scheduled_on_l3_agents(self):
|
def test_router_is_scheduled_on_l3_agents(self):
|
||||||
self._test_router_is_scheduled_on_l3_agents()
|
self._test_router_is_scheduled_on_l3_agents()
|
||||||
|
|
||||||
|
def test_router_ipv4_address(self):
|
||||||
|
self.assertEqual(4, self.router_ipv4_address.version)
|
||||||
|
ips = neutron.list_port_ip_addresses(
|
||||||
|
port=self.stack.network_stack.ipv4_gateway_port_details,
|
||||||
|
ip_version=4)
|
||||||
|
self.assertIn(self.router_ipv4_address, ips)
|
||||||
|
|
||||||
|
def test_router_ipv6_address(self):
|
||||||
|
self.assertEqual(6, self.router_ipv6_address.version)
|
||||||
|
ips = neutron.list_port_ip_addresses(
|
||||||
|
port=self.stack.network_stack.ipv6_gateway_port_details,
|
||||||
|
ip_version=6)
|
||||||
|
self.assertIn(self.router_ipv6_address, ips)
|
||||||
|
|
||||||
|
neutron.find_port_ip_address(
|
||||||
|
port=self.stack.network_stack.ipv6_gateway_port_details,
|
||||||
|
ip_version=6)
|
||||||
|
|
||||||
def _get_l3_agent_nodes(self, hostname):
|
def _get_l3_agent_nodes(self, hostname):
|
||||||
hostname = hostname.split(".")
|
hostname = hostname.split(".")
|
||||||
for host in self.topology.nodes:
|
for host in self.topology.nodes:
|
||||||
|
@ -101,14 +119,14 @@ class LegacyRouterTest(testtools.TestCase):
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
str(self.router_ipv6_address), ns_net_config.stdout)
|
str(self.router_ipv6_address), ns_net_config.stdout)
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
str(self.router_gw_ip), ns_net_config.stdout)
|
str(self.router_gateway_ip), ns_net_config.stdout)
|
||||||
else:
|
else:
|
||||||
self.assertNotIn(
|
self.assertNotIn(
|
||||||
str(self.router_ipv4_address), ns_net_config.stdout)
|
str(self.router_ipv4_address), ns_net_config.stdout)
|
||||||
self.assertNotIn(
|
self.assertNotIn(
|
||||||
str(self.router_ipv6_address), ns_net_config.stdout)
|
str(self.router_ipv6_address), ns_net_config.stdout)
|
||||||
self.assertNotIn(
|
self.assertNotIn(
|
||||||
str(self.router_gw_ip), ns_net_config.stdout)
|
str(self.router_gateway_ip), ns_net_config.stdout)
|
||||||
|
|
||||||
def _test_router_is_scheduled_on_l3_agents(self):
|
def _test_router_is_scheduled_on_l3_agents(self):
|
||||||
router_agent = neutron.find_l3_agent_hosting_router(self.router['id'],
|
router_agent = neutron.find_l3_agent_hosting_router(self.router['id'],
|
||||||
|
|
Loading…
Reference in New Issue