Add API to list neutron agents and skipp tests if missing some
Change-Id: Iba4491fae85d3f4ed0e09518e20fe3ffabfde1f3
This commit is contained in:
parent
147752e53a
commit
3a6569200c
|
@ -13,6 +13,7 @@
|
||||||
# under the License.
|
# under the License.
|
||||||
from __future__ import absolute_import
|
from __future__ import absolute_import
|
||||||
|
|
||||||
|
from tobiko.openstack.neutron import _agent
|
||||||
from tobiko.openstack.neutron import _client
|
from tobiko.openstack.neutron import _client
|
||||||
from tobiko.openstack.neutron import _cidr
|
from tobiko.openstack.neutron import _cidr
|
||||||
from tobiko.openstack.neutron import _extension
|
from tobiko.openstack.neutron import _extension
|
||||||
|
@ -28,6 +29,7 @@ find_port = _client.find_port
|
||||||
list_ports = _client.list_ports
|
list_ports = _client.list_ports
|
||||||
list_subnets = _client.list_subnets
|
list_subnets = _client.list_subnets
|
||||||
list_subnet_cidrs = _client.list_subnet_cidrs
|
list_subnet_cidrs = _client.list_subnet_cidrs
|
||||||
|
list_agents = _client.list_agents
|
||||||
show_network = _client.show_network
|
show_network = _client.show_network
|
||||||
show_router = _client.show_router
|
show_router = _client.show_router
|
||||||
show_port = _client.show_port
|
show_port = _client.show_port
|
||||||
|
@ -39,5 +41,7 @@ new_ipv6_cidr = _cidr.new_ipv6_cidr
|
||||||
get_networking_extensions = _extension.get_networking_extensions
|
get_networking_extensions = _extension.get_networking_extensions
|
||||||
missing_networking_extensions = _extension.missing_networking_extensions
|
missing_networking_extensions = _extension.missing_networking_extensions
|
||||||
has_networking_extensions = _extension.has_networking_extensions
|
has_networking_extensions = _extension.has_networking_extensions
|
||||||
|
|
||||||
skip_if_missing_networking_extensions = (
|
skip_if_missing_networking_extensions = (
|
||||||
_extension.skip_if_missing_networking_extensions)
|
_extension.skip_if_missing_networking_extensions)
|
||||||
|
skip_if_missing_networking_agents = _agent.skip_if_missing_networking_agents
|
||||||
|
|
|
@ -0,0 +1,66 @@
|
||||||
|
# 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 tobiko
|
||||||
|
|
||||||
|
from tobiko.openstack.neutron import _client
|
||||||
|
|
||||||
|
|
||||||
|
class NetworkingAgentaFixture(tobiko.SharedFixture):
|
||||||
|
|
||||||
|
client = None
|
||||||
|
agents = None
|
||||||
|
|
||||||
|
def setup_fixture(self):
|
||||||
|
self.setup_client()
|
||||||
|
self.get_agents()
|
||||||
|
|
||||||
|
def setup_client(self):
|
||||||
|
self.client = _client.get_neutron_client()
|
||||||
|
|
||||||
|
def get_agents(self):
|
||||||
|
self.agents = _client.list_agents(client=self.client)
|
||||||
|
|
||||||
|
|
||||||
|
def get_networking_agents(**params):
|
||||||
|
agents = tobiko.setup_fixture(NetworkingAgentaFixture).agents
|
||||||
|
if params:
|
||||||
|
selected_agents = []
|
||||||
|
for agent in agents:
|
||||||
|
for key, value in params.items():
|
||||||
|
if value != agent[key]:
|
||||||
|
break
|
||||||
|
else:
|
||||||
|
selected_agents.append(agent)
|
||||||
|
agents = selected_agents
|
||||||
|
return agents
|
||||||
|
|
||||||
|
|
||||||
|
def missing_networking_agents(count=1, **params):
|
||||||
|
agents = get_networking_agents(**params)
|
||||||
|
return max(0, count - len(agents))
|
||||||
|
|
||||||
|
|
||||||
|
def has_networking_agents(count=1, **params):
|
||||||
|
return not missing_networking_agents(count=count, **params)
|
||||||
|
|
||||||
|
|
||||||
|
def skip_if_missing_networking_agents(count=1, **params):
|
||||||
|
message = "missing {return_value!r} agent(s)"
|
||||||
|
if params:
|
||||||
|
message += " with {!s}".format(
|
||||||
|
', '.join("{!s}={!r}".format(k, v) for k, v in params.items()))
|
||||||
|
return tobiko.skip_if(message, missing_networking_agents, count=count,
|
||||||
|
**params)
|
|
@ -118,6 +118,13 @@ def list_subnets(show=False, client=None, **params):
|
||||||
return subnets
|
return subnets
|
||||||
|
|
||||||
|
|
||||||
|
def list_agents(client=None, **params):
|
||||||
|
agents = neutron_client(client).list_agents(**params)
|
||||||
|
if isinstance(agents, collections.Mapping):
|
||||||
|
agents = agents['agents']
|
||||||
|
return agents
|
||||||
|
|
||||||
|
|
||||||
def list_subnet_cidrs(client=None, **params):
|
def list_subnet_cidrs(client=None, **params):
|
||||||
return [netaddr.IPNetwork(subnet['cidr'])
|
return [netaddr.IPNetwork(subnet['cidr'])
|
||||||
for subnet in list_subnets(client=client, **params)]
|
for subnet in list_subnets(client=client, **params)]
|
||||||
|
|
|
@ -103,3 +103,41 @@ class NeutronApiTestCase(testtools.TestCase):
|
||||||
subnet = neutron.show_subnet(self.stack.ipv6_subnet_id)
|
subnet = neutron.show_subnet(self.stack.ipv6_subnet_id)
|
||||||
self.assertEqual(self.stack.ipv6_subnet_id, subnet['id'])
|
self.assertEqual(self.stack.ipv6_subnet_id, subnet['id'])
|
||||||
self.assertEqual(self.stack.ipv6_subnet_details, subnet)
|
self.assertEqual(self.stack.ipv6_subnet_details, subnet)
|
||||||
|
|
||||||
|
def test_find_agents_with_binary_id(self):
|
||||||
|
agents = neutron.list_agents(binary='neutron-l3-agent')
|
||||||
|
self.assertTrue(agents)
|
||||||
|
|
||||||
|
|
||||||
|
class AgentTest(testtools.TestCase):
|
||||||
|
|
||||||
|
def test_skip_if_missing_agents(self, count=1, should_skip=False,
|
||||||
|
**params):
|
||||||
|
if should_skip:
|
||||||
|
expected_exeption = self.skipException
|
||||||
|
else:
|
||||||
|
expected_exeption = self.failureException
|
||||||
|
|
||||||
|
@neutron.skip_if_missing_networking_agents(count=count, **params)
|
||||||
|
def method():
|
||||||
|
raise self.fail('Not skipped')
|
||||||
|
|
||||||
|
exception = self.assertRaises(expected_exeption, method)
|
||||||
|
if should_skip:
|
||||||
|
agents = neutron.list_agents(**params)
|
||||||
|
message = "missing {!r} agent(s)".format(count - len(agents))
|
||||||
|
if params:
|
||||||
|
message += " with {!s}".format(
|
||||||
|
','.join('{!s}={!r}'.format(k, v)
|
||||||
|
for k, v in params.items()))
|
||||||
|
self.assertEqual(message, str(exception))
|
||||||
|
else:
|
||||||
|
self.assertEqual('Not skipped', str(exception))
|
||||||
|
|
||||||
|
def test_skip_if_missing_agents_with_no_agents(self):
|
||||||
|
self.test_skip_if_missing_agents(binary='never-never-land',
|
||||||
|
should_skip=True)
|
||||||
|
|
||||||
|
def test_skip_if_missing_agents_with_big_count(self):
|
||||||
|
self.test_skip_if_missing_agents(count=1000000,
|
||||||
|
should_skip=True)
|
||||||
|
|
Loading…
Reference in New Issue