diff --git a/tobiko/openstack/neutron/__init__.py b/tobiko/openstack/neutron/__init__.py index 4dc39a285..eb75afe46 100644 --- a/tobiko/openstack/neutron/__init__.py +++ b/tobiko/openstack/neutron/__init__.py @@ -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 diff --git a/tobiko/openstack/neutron/_agent.py b/tobiko/openstack/neutron/_agent.py index 0e94202b2..890b9b99f 100644 --- a/tobiko/openstack/neutron/_agent.py +++ b/tobiko/openstack/neutron/_agent.py @@ -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(