openstack/neutron
Code Issues Proposed changes

Merge "Remove compat logic for neutron-lib < 3.9.0"

Browse Source
This commit is contained in:
Zuul
2025-06-24 10:21:27 +00:00
committed by Gerrit Code Review
parent b8e21fd54e addd8da5c4
commit a1ae35bed1
5 changed files with 48 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
neutron/tests/tools.py
View File

@@ -19,7 +19,6 @@ import unittest
import fixtures import fixtures
import netaddr import netaddr
from neutron_lib.callbacks import priority_group
from neutron_lib import constants from neutron_lib import constants
from neutron_lib.services.logapi import constants as log_const from neutron_lib.services.logapi import constants as log_const
from neutron_lib.utils import helpers from neutron_lib.utils import helpers
@@ -28,16 +27,6 @@ from oslo_utils import netutils
from oslo_utils import timeutils from oslo_utils import timeutils
# NOTE(ykarel): from neutron-lib 3.9.0, cancellable flag was added
# test the existence of the is_cancellable_event function to check if
# cancellable flag is supported or not. This compatibility check can
# be removed once neutron-lib >= 3.9.0 in requirements.txt.
_CANCELLABLE_FLAG_SUPPORTED = True
try:
from neutron_lib.callbacks.events import is_cancellable_event # noqa
except ImportError:
_CANCELLABLE_FLAG_SUPPORTED = False
LAST_RANDOM_PORT_RANGE_GENERATED = 1 LAST_RANDOM_PORT_RANGE_GENERATED = 1
@@ -146,18 +135,6 @@ def make_mock_plugin_json_encodable(plugin_instance_mock):
method_mock._get_child_mock = _get_child_mock method_mock._get_child_mock = _get_child_mock
def get_subscribe_args(*args):
args = list(args) # don't modify original list
args.append(priority_group.PRIORITY_DEFAULT)
# NOTE(ykarel): from neutron-lib 3.9.0, cancellable flag was added.
# old signature: (callback, resource, event, priority=PRIORITY_DEFAULT)
# new signature: (callback, resource, event, priority=PRIORITY_DEFAULT,
# cancellable=False)
if len(args) == 4 and _CANCELLABLE_FLAG_SUPPORTED:
args.append(False)
return args
def fail(msg=None): def fail(msg=None):
"""Fail immediately, with the given message. """Fail immediately, with the given message.

8
neutron/tests/unit/plugins/ml2/drivers/agent/test_capabilities.py
View File

@@ -14,11 +14,11 @@
from unittest import mock from unittest import mock
from neutron_lib.callbacks import events from neutron_lib.callbacks import events
from neutron_lib.callbacks import priority_group
from neutron_lib import fixture from neutron_lib import fixture
from neutron.plugins.ml2.drivers.agent import capabilities from neutron.plugins.ml2.drivers.agent import capabilities
from neutron.tests import base from neutron.tests import base
from neutron.tests import tools
class CapabilitiesTest(base.BaseTestCase): class CapabilitiesTest(base.BaseTestCase):
@@ -42,6 +42,6 @@ class CapabilitiesTest(base.BaseTestCase):
mock_callback = mock.Mock() mock_callback = mock.Mock()
mock_agent_type = mock.Mock() mock_agent_type = mock.Mock()
capabilities.register(mock_callback, mock_agent_type) capabilities.register(mock_callback, mock_agent_type)
args = tools.get_subscribe_args( self._mgr.subscribe.assert_called_with(
mock_callback, mock_agent_type, events.AFTER_INIT) mock_callback, mock_agent_type, events.AFTER_INIT,
self._mgr.subscribe.assert_called_with(*args) priority_group.PRIORITY_DEFAULT, False)

8
neutron/tests/unit/plugins/ml2/drivers/openvswitch/agent/test_ovs_capabilities.py
View File

@@ -14,12 +14,12 @@
from unittest import mock from unittest import mock
from neutron_lib.callbacks import events from neutron_lib.callbacks import events
from neutron_lib.callbacks import priority_group
from neutron_lib import fixture from neutron_lib import fixture
from neutron.plugins.ml2.drivers.openvswitch.agent import ovs_capabilities from neutron.plugins.ml2.drivers.openvswitch.agent import ovs_capabilities
from neutron.services.trunk.drivers.openvswitch.agent import driver from neutron.services.trunk.drivers.openvswitch.agent import driver
from neutron.tests import base from neutron.tests import base
from neutron.tests import tools
from neutron_lib import constants from neutron_lib import constants
@@ -33,6 +33,6 @@ class CapabilitiesTest(base.BaseTestCase):
def test_register(self): def test_register(self):
ovs_capabilities.register() ovs_capabilities.register()
args = tools.get_subscribe_args( self._mgr.subscribe.assert_called_with(
driver.init_handler, constants.AGENT_TYPE_OVS, events.AFTER_INIT) driver.init_handler, constants.AGENT_TYPE_OVS, events.AFTER_INIT,
self._mgr.subscribe.assert_called_with(*args) priority_group.PRIORITY_DEFAULT, False)

38
neutron/tests/unit/services/logapi/drivers/test_manager.py
View File

@@ -16,6 +16,7 @@
from unittest import mock from unittest import mock
from neutron_lib.callbacks import events from neutron_lib.callbacks import events
from neutron_lib.callbacks import priority_group
from neutron_lib import exceptions from neutron_lib import exceptions
from neutron_lib import fixture from neutron_lib import fixture
from neutron_lib.services.logapi import constants as log_const from neutron_lib.services.logapi import constants as log_const
@@ -23,7 +24,6 @@ from neutron_lib.services.logapi import constants as log_const
from neutron.services.logapi.common import exceptions as log_exc from neutron.services.logapi.common import exceptions as log_exc
from neutron.services.logapi.drivers import base as log_driver_base from neutron.services.logapi.drivers import base as log_driver_base
from neutron.services.logapi.drivers import manager as driver_mgr from neutron.services.logapi.drivers import manager as driver_mgr
from neutron.tests import tools
from neutron.tests.unit.services.logapi import base from neutron.tests.unit.services.logapi import base
@@ -162,28 +162,28 @@ class TestHandleResourceCallback(TestLogDriversManagerBase):
'fake_resource2', self.driver_manager.call) 'fake_resource2', self.driver_manager.call)
assert_calls = [ assert_calls = [
mock.call( mock.call(
*tools.get_subscribe_args( fake_resource_cb1.handle_event,
fake_resource_cb1.handle_event, 'fake_resource1', events.AFTER_CREATE,
'fake_resource1', events.AFTER_CREATE)), priority_group.PRIORITY_DEFAULT, False),
mock.call( mock.call(
*tools.get_subscribe_args( fake_resource_cb1.handle_event,
fake_resource_cb1.handle_event, 'fake_resource1', events.AFTER_UPDATE,
'fake_resource1', events.AFTER_UPDATE)), priority_group.PRIORITY_DEFAULT, False),
mock.call( mock.call(
*tools.get_subscribe_args( fake_resource_cb1.handle_event,
fake_resource_cb1.handle_event, 'fake_resource1', events.AFTER_DELETE,
'fake_resource1', events.AFTER_DELETE)), priority_group.PRIORITY_DEFAULT, False),
mock.call( mock.call(
*tools.get_subscribe_args( fake_resource_cb2.handle_event,
fake_resource_cb2.handle_event, 'fake_resource2', events.AFTER_CREATE,
'fake_resource2', events.AFTER_CREATE)), priority_group.PRIORITY_DEFAULT, False),
mock.call( mock.call(
*tools.get_subscribe_args( fake_resource_cb2.handle_event,
fake_resource_cb2.handle_event, 'fake_resource2', events.AFTER_UPDATE,
'fake_resource2', events.AFTER_UPDATE)), priority_group.PRIORITY_DEFAULT, False),
mock.call( mock.call(
*tools.get_subscribe_args( fake_resource_cb2.handle_event,
fake_resource_cb2.handle_event, 'fake_resource2', events.AFTER_DELETE,
'fake_resource2', events.AFTER_DELETE)), priority_group.PRIORITY_DEFAULT, False),
] ]
self._cb_mgr.subscribe.assert_has_calls(assert_calls) self._cb_mgr.subscribe.assert_has_calls(assert_calls)

38
neutron/tests/unit/services/trunk/rpc/test_backend.py
View File

@@ -14,13 +14,13 @@
from unittest import mock from unittest import mock
from neutron_lib.callbacks import events from neutron_lib.callbacks import events
from neutron_lib.callbacks import priority_group
from neutron_lib.callbacks import resources from neutron_lib.callbacks import resources
from neutron_lib import fixture from neutron_lib import fixture
from neutron.api.rpc.callbacks import resource_manager from neutron.api.rpc.callbacks import resource_manager
from neutron.services.trunk.rpc import backend from neutron.services.trunk.rpc import backend
from neutron.tests import base from neutron.tests import base
from neutron.tests import tools
class ServerSideRpcBackendTest(base.BaseTestCase): class ServerSideRpcBackendTest(base.BaseTestCase):
@@ -37,25 +37,29 @@ class ServerSideRpcBackendTest(base.BaseTestCase):
test_obj = backend.ServerSideRpcBackend() test_obj = backend.ServerSideRpcBackend()
calls = [mock.call( calls = [mock.call(
*tools.get_subscribe_args( test_obj.process_event,
test_obj.process_event, resources.TRUNK,
resources.TRUNK, events.AFTER_CREATE,
events.AFTER_CREATE)), priority_group.PRIORITY_DEFAULT,
False),
mock.call( mock.call(
*tools.get_subscribe_args( test_obj.process_event,
test_obj.process_event, resources.TRUNK,
resources.TRUNK, events.AFTER_DELETE,
events.AFTER_DELETE)), priority_group.PRIORITY_DEFAULT,
False),
mock.call( mock.call(
*tools.get_subscribe_args( test_obj.process_event,
test_obj.process_event, resources.SUBPORTS,
resources.SUBPORTS, events.AFTER_CREATE,
events.AFTER_CREATE)), priority_group.PRIORITY_DEFAULT,
False),
mock.call( mock.call(
*tools.get_subscribe_args( test_obj.process_event,
test_obj.process_event, resources.SUBPORTS,
resources.SUBPORTS, events.AFTER_DELETE,
events.AFTER_DELETE)) priority_group.PRIORITY_DEFAULT,
False),
] ]
self._mgr.subscribe.assert_has_calls(calls, any_order=True) self._mgr.subscribe.assert_has_calls(calls, any_order=True)