Define neutron agent service names

Change-Id: I3dd915dacb610bd898f569e1970bc6a2c005958c
This commit is contained in:
Federico Ressi 2020-10-14 09:15:10 +02:00
parent 3df2bb87f1
commit 9ea4160968
2 changed files with 34 additions and 17 deletions

View File

@ -20,6 +20,16 @@ from tobiko.openstack.neutron import _extension
from tobiko.openstack.neutron import _port
DHCP_AGENT = _agent.DHCP_AGENT
L3_AGENT = _agent.L3_AGENT
METADATA_AGENT = _agent.METADATA_AGENT
OPENVSWITCH_AGENT = _agent.OPENVSWITCH_AGENT
OVN_CONTROLLER = _agent.OVN_CONTROLLER
AgentNotFoundOnHost = _agent.AgentNotFoundOnHost
skip_if_missing_networking_agents = _agent.skip_if_missing_networking_agents
list_networking_agents = _agent.list_networking_agents
NeutronClientFixture = _client.NeutronClientFixture
ServiceUnavailable = _client.ServiceUnavailable
neutron_client = _client.neutron_client
@ -45,7 +55,6 @@ NoSuchNetwork = _client.NoSuchNetwork
NoSuchPort = _client.NoSuchPort
NoSuchRouter = _client.NoSuchRouter
NoSuchSubnet = _client.NoSuchSubnet
AgentNotFoundOnHost = _agent.AgentNotFoundOnHost
new_ipv4_cidr = _cidr.new_ipv4_cidr
new_ipv6_cidr = _cidr.new_ipv6_cidr
@ -56,8 +65,6 @@ has_networking_extensions = _extension.has_networking_extensions
skip_if_missing_networking_extensions = (
_extension.skip_if_missing_networking_extensions)
skip_if_missing_networking_agents = _agent.skip_if_missing_networking_agents
get_networking_agents = _agent.get_networking_agents
find_port_ip_address = _port.find_port_ip_address
list_port_ip_addresses = _port.list_port_ip_addresses

View File

@ -13,37 +13,39 @@
# under the License.
from __future__ import absolute_import
import typing # noqa
import tobiko
from tobiko.openstack.neutron import _client
# Agent binary names
DHCP_AGENT = 'neutron-dhcp-agent'
L3_AGENT = 'neutron-l3-agent'
METADATA_AGENT = 'neutron-metadata-agent'
OPENVSWITCH_AGENT = 'neutron-openvswitch-agent'
OVN_CONTROLLER = 'neutron-ovn-controller'
class AgentNotFoundOnHost(tobiko.TobikoException):
message = ("Agent {agent_type!s} not found on the host {host!s}")
class NetworkingAgentFixture(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)
self.agents = _client.list_agents()
def get_networking_agents(**attributes):
agents = tobiko.setup_fixture(NetworkingAgentFixture).agents
return agents.with_items(**attributes)
def list_networking_agents(**attributes):
return tobiko.setup_fixture(
NetworkingAgentFixture).agents.with_items(**attributes)
def missing_networking_agents(count=1, **params):
agents = get_networking_agents(**params)
agents = list_networking_agents(**params)
return max(0, count - len(agents))
@ -51,7 +53,15 @@ def has_networking_agents(count=1, **params):
return not missing_networking_agents(count=count, **params)
def skip_if_missing_networking_agents(count=1, **params):
DecoratorType = typing.Callable[[typing.Union[typing.Callable, typing.Type]],
typing.Union[typing.Callable, typing.Type]]
def skip_if_missing_networking_agents(binary: typing.Optional[str] = None,
count: int = 1, **params) -> \
DecoratorType:
if binary is not None:
params['binary'] = binary
message = "missing {return_value!r} agent(s)"
if params:
message += " with {!s}".format(