diff --git a/neutron_lib/exceptions/__init__.py b/neutron_lib/exceptions/__init__.py index 32db10320..54847de83 100644 --- a/neutron_lib/exceptions/__init__.py +++ b/neutron_lib/exceptions/__init__.py @@ -781,3 +781,8 @@ class FilterIDForIPNotFound(NotFound): class FailedToAddQdiscToDevice(NeutronException): message = _("Failed to add %(direction)s qdisc " "to device %(device)s.") + + +class PortBindingNotFound(NotFound): + message = _("Binding for port %(port_id)s for host %(host)s could not be " + "found.") diff --git a/neutron_lib/plugins/utils.py b/neutron_lib/plugins/utils.py index 1582d889d..2fa32a891 100644 --- a/neutron_lib/plugins/utils.py +++ b/neutron_lib/plugins/utils.py @@ -26,6 +26,7 @@ from neutron_lib._i18n import _ from neutron_lib.api import attributes from neutron_lib.api.definitions import network as net_apidef from neutron_lib.api.definitions import port as port_apidef +from neutron_lib.api.definitions import portbindings_extended as pb_ext from neutron_lib.api.definitions import subnet as subnet_apidef from neutron_lib import constants from neutron_lib import exceptions @@ -348,3 +349,36 @@ def get_deployment_physnet_mtu(): :returns: The global_physnet_mtu from the global CONF. """ return cfg.CONF.global_physnet_mtu + + +def get_port_binding_by_status_and_host(bindings, status, host='', + raise_if_not_found=False, + port_id=None): + """Returns from an iterable the binding with the specified status and host. + + The input iterable can contain zero or one binding in status ACTIVE + and zero or many bindings in status INACTIVE. As a consequence, to + unequivocally retrieve an inactive binding, the caller must specify a non + empty value for host. If host is the empty string, the first binding + satisfying the specified status will be returned. If no binding is found + with the specified status and host, None is returned or PortBindingNotFound + is raised if raise_if_not_found is True + + :param bindings: An iterable containing port bindings + :param status: The status of the port binding to return. Possible values + are ACTIVE or INACTIVE as defined in + :file:`neutron_lib/constants.py`. + :param host: str representing the host of the binding to return. + :param raise_if_not_found: If a binding is not found and this parameter is + True, a PortBindingNotFound exception is raised + :param port_id: The id of the binding's port + :returns: The searched for port binding or None if it is not found + :raises: PortBindingNotFound if the binding is not found and + raise_if_not_found is True + """ + for binding in bindings: + if binding[pb_ext.STATUS] == status: + if not host or binding[pb_ext.HOST] == host: + return binding + if raise_if_not_found: + raise exceptions.PortBindingNotFound(port_id=port_id, host=host) diff --git a/neutron_lib/tests/unit/plugins/test_utils.py b/neutron_lib/tests/unit/plugins/test_utils.py index 9567278d3..fd28afe40 100644 --- a/neutron_lib/tests/unit/plugins/test_utils.py +++ b/neutron_lib/tests/unit/plugins/test_utils.py @@ -18,6 +18,7 @@ import mock from oslo_utils import excutils from oslo_utils import uuidutils +from neutron_lib.api.definitions import portbindings_extended as pb_ext from neutron_lib import constants from neutron_lib import exceptions from neutron_lib.plugins import utils @@ -248,3 +249,26 @@ class TestUtils(base.BaseTestCase): self.assertEqual('p1', port['port']['project_id']) self.assertEqual('aport', port['port']['name']) self.assertEqual(net_id, port['port']['network_id']) + + def test_get_port_binding_by_status_and_host(self): + bindings = [] + self.assertIsNone(utils.get_port_binding_by_status_and_host( + bindings, constants.INACTIVE)) + bindings.extend([{pb_ext.STATUS: constants.INACTIVE, + pb_ext.HOST: 'host-1'}, + {pb_ext.STATUS: constants.INACTIVE, + pb_ext.HOST: 'host-2'}]) + self.assertEqual( + 'host-1', utils.get_port_binding_by_status_and_host( + bindings, + constants.INACTIVE)[pb_ext.HOST]) + self.assertEqual( + 'host-2', utils.get_port_binding_by_status_and_host( + bindings, + constants.INACTIVE, + host='host-2')[pb_ext.HOST]) + self.assertIsNone(utils.get_port_binding_by_status_and_host( + bindings, constants.ACTIVE)) + self.assertRaises(exceptions.PortBindingNotFound, + utils.get_port_binding_by_status_and_host, bindings, + constants.ACTIVE, 'host', True, 'port_id') diff --git a/releasenotes/notes/rehome-get-port-binding-98765e77c627e57d.yaml b/releasenotes/notes/rehome-get-port-binding-98765e77c627e57d.yaml new file mode 100644 index 000000000..5ff7fd6c0 --- /dev/null +++ b/releasenotes/notes/rehome-get-port-binding-98765e77c627e57d.yaml @@ -0,0 +1,4 @@ +--- +features: + - The ``get_port_binding_by_status_and_host`` function is now available + in ``neutron_lib.plugins.utils``.