Browse Source

Fix test_router_is_scheduled_on_l3_agents test

Change-Id: If58535ed365483a33abad7f7afbddb23fcef2fee
changes/59/774059/4
Federico Ressi 6 months ago
parent
commit
7a41559d3b
  1. 3
      tobiko/openstack/neutron/__init__.py
  2. 71
      tobiko/openstack/neutron/_router.py
  3. 19
      tobiko/tests/scenario/neutron/test_router.py

3
tobiko/openstack/neutron/__init__.py

@ -19,6 +19,7 @@ from tobiko.openstack.neutron import _cidr
from tobiko.openstack.neutron import _extension
from tobiko.openstack.neutron import _port
from tobiko.openstack.neutron import _network
from tobiko.openstack.neutron import _router
SERVER = 'neutron-server'
@ -85,3 +86,5 @@ get_network = _network.get_network
find_network = _network.find_network
list_networks = _network.list_networks
list_network_nameservers = _network.list_network_nameservers
wait_for_master_and_backup_agents = _router.wait_for_master_and_backup_agents

71
tobiko/openstack/neutron/_router.py

@ -0,0 +1,71 @@
# Copyright 2019 Red Hat
#
# 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 typing
import json
from oslo_log import log
import tobiko
from tobiko.openstack.neutron import _client
LOG = log.getLogger(__name__)
def wait_for_master_and_backup_agents(
router_id: str,
unique_master: bool = True,
timeout: tobiko.Seconds = None,
interval: tobiko.Seconds = None) -> \
typing.Tuple[typing.Dict, typing.List[typing.Dict]]:
for attempt in tobiko.retry(timeout=timeout,
interval=interval,
default_timeout=300.,
default_interval=5.):
router_agents = _client.list_l3_agent_hosting_routers(router_id)
master_agents = router_agents.with_items(ha_state='active')
if master_agents:
LOG.debug(
f"Router '{router_id}' has {len(master_agents)} master "
"agent(s):\n"
f"{json.dumps(master_agents, indent=4, sort_keys=True)}")
backup_agents = router_agents.with_items(ha_state='standby')
if backup_agents:
LOG.debug(
f"Router '{router_id}' has {len(backup_agents)} backup "
"agent(s)):\n"
f"{json.dumps(backup_agents, indent=4, sort_keys=True)}")
other_agents = [agent
for agent in router_agents
if (agent not in master_agents + backup_agents)]
if other_agents:
LOG.debug(
f"Router '{router_id}' has {len(other_agents)} other "
"agent(s):\n"
f"{json.dumps(master_agents, indent=4, sort_keys=True)}")
try:
if unique_master:
master_agent = master_agents.unique
else:
master_agent = master_agents.first
except (tobiko.MultipleObjectsFound, tobiko.ObjectNotFound):
attempt.check_limits()
else:
break
else:
raise RuntimeError("tobiko retry loop ended before break?")
return master_agent, backup_agents

19
tobiko/tests/scenario/neutron/test_router.py

@ -15,6 +15,7 @@
from __future__ import absolute_import
import pytest
from oslo_log import log
import testtools
import tobiko
@ -26,22 +27,10 @@ from tobiko.openstack import stacks
from tobiko.openstack import topology
LOG = log.getLogger(__name__)
CONF = config.CONF
def get_master_and_backup_agents(router_id, timeout=60., interval=2.):
for attempt in tobiko.retry(timeout=timeout, interval=interval):
try:
router_agents = neutron.list_l3_agent_hosting_routers(
router_id)
master_agent = router_agents.with_items(ha_state='active').unique
backup_agents = router_agents.with_items(ha_state='standby')
return master_agent, backup_agents
except Exception:
attempt.check_limits()
continue
class RouterTest(testtools.TestCase):
"""Test Neutron routers"""
@ -136,8 +125,8 @@ class L3HARouterTest(RouterTest):
@pytest.mark.flaky(reruns=5, reruns_delay=120)
@neutron.skip_if_missing_networking_extensions('l3_agent_scheduler')
def test_router_is_scheduled_on_l3_agents(self):
master_agent, backup_agents = get_master_and_backup_agents(
self.router['id'])
master_agent, backup_agents = (
neutron.wait_for_master_and_backup_agents(self.router['id']))
self.assertGreaterEqual(len(backup_agents), 1)
self._check_routers_namespace_on_host(master_agent['host'])
for backup_agent in backup_agents:

Loading…
Cancel
Save