Use tobiko selection API for filtering L3 router agents

This also fixes handling of unique parameter in below functions:
 - find_network
 - find_port
 - find_subnet

Rename wrapper of neutron 'list_l3_agent_hosting_routers'
Neutron client method with the same name.

Add 'find_l3_agent_hosting_router' function

Change-Id: I03a4a4351ec349eb818861d3f43fc075cb6537d9
This commit is contained in:
Federico Ressi 2019-10-16 13:20:57 +02:00
parent d47cf1bd76
commit c63fd45ff2
3 changed files with 34 additions and 24 deletions

View File

@ -35,7 +35,8 @@ 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
list_l3_agent_hosting_routers = _client.list_l3_agent_hosting_routers
find_l3_agent_hosting_router = _client.find_l3_agent_hosting_router
new_ipv4_cidr = _cidr.new_ipv4_cidr
new_ipv6_cidr = _cidr.new_ipv6_cidr

View File

@ -70,9 +70,9 @@ def find_network(client=None, unique=False, default=_RAISE_ERROR,
networks = list_networks(client=client, **attributes)
if default is _RAISE_ERROR or networks:
if unique:
return networks.first
else:
return networks.unique
else:
return networks.first
else:
return default
@ -82,9 +82,9 @@ def find_port(client=None, unique=False, default=_RAISE_ERROR, **attributes):
ports = list_ports(client=client, **attributes)
if default is _RAISE_ERROR or ports:
if unique:
return ports.first
else:
return ports.unique
else:
return ports.first
else:
return default
@ -94,9 +94,9 @@ def find_subnet(client=None, unique=False, default=_RAISE_ERROR, **attributes):
subnets = list_subnets(client=client, **attributes)
if default is _RAISE_ERROR or subnets:
if unique:
return subnets.first
else:
return subnets.unique
else:
return subnets.first
else:
return default
@ -146,9 +146,22 @@ 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):
def list_l3_agent_hosting_routers(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)
def find_l3_agent_hosting_router(router, client=None, unique=False,
default=_RAISE_ERROR, **params):
agents = list_l3_agent_hosting_routers(router=router, client=client,
**params)
if default is _RAISE_ERROR or agents:
if unique:
return agents.unique
else:
return agents.first
else:
return default

View File

@ -41,12 +41,10 @@ class LegacyRouterTest(testtools.TestCase):
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_ipv4_address = (
self.stack.network_stack.ipv4_subnet_details['gateway_ip'])
self.router_ipv6_address = (
self.stack.network_stack.ipv6_subnet_details['gateway_ip'])
self.router_gw_ip = self.router['external_gateway_info'][
'external_fixed_ips'][0]['ip_address']
tripleo_topology.setup_tripleo_topology()
@ -113,9 +111,9 @@ class LegacyRouterTest(testtools.TestCase):
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'])
router_agent = neutron.find_l3_agent_hosting_router(self.router['id'],
unique=True)
self._check_routers_namespace_on_host(router_agent['host'])
@neutron.skip_if_missing_networking_extensions('l3-ha')
@ -135,14 +133,12 @@ class HaRouterTest(LegacyRouterTest):
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))
router_agents = neutron.list_l3_agent_hosting_routers(
self.router['id'])
master_agent = router_agents.with_items(ha_state='active').unique
backup_agents = router_agents.with_items(ha_state='standby')
self.assertGreaterEqual(len(backup_agents), 1)
self._check_routers_namespace_on_host(master_agents[0]['host'])
self._check_routers_namespace_on_host(master_agent['host'])
for backup_agent in backup_agents:
self._check_routers_namespace_on_host(
backup_agent['host'], state="backup")