17563a802e
Neutron Manager is loaded at the very startup of the neutron server process and with it plugins are loaded and stored for lookup purposes as their references are widely used across the entire neutron codebase. Rather than holding these references directly in NeutronManager this patch refactors the code so that these references are held by a plugin directory. This allows subprojects and other parts of the Neutron codebase to use the directory in lieu of the manager. The result is a leaner, cleaner, and more decoupled code. Usage pattern [1,2] can be translated to [3,4] respectively. [1] manager.NeutronManager.get_service_plugins()[FOO] [2] manager.NeutronManager.get_plugin() [3] directory.get_plugin(FOO) [4] directory.get_plugin() The more entangled part is in the neutron unit tests, where the use of the manager can be simplified as mocking is typically replaced by a call to the directory add_plugin() method. This is safe as each test case gets its own copy of the plugin directory. That said, unit tests that look more like API tests and that rely on the entire plugin machinery, need some tweaking to avoid stumbling into plugin loading failures. Due to the massive use of the manager, deprecation warnings are considered impractical as they cause logs to bloat out of proportion. Follow-up patches that show how to adopt the directory in neutron subprojects are tagged with topic:plugin-directory. NeutronLibImpact Partially-implements: blueprint neutron-lib Change-Id: I7331e914234c5f0b7abe836604fdd7e4067551cf
286 lines
12 KiB
Python
286 lines
12 KiB
Python
# Copyright (c) 2015 Red Hat Inc.
|
|
# All Rights Reserved.
|
|
#
|
|
# 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.
|
|
|
|
import mock
|
|
|
|
from neutron.common import exceptions as n_exc
|
|
from neutron import context
|
|
from neutron.core_extensions import base as base_core
|
|
from neutron.core_extensions import qos as qos_core
|
|
from neutron.plugins.common import constants as plugin_constants
|
|
from neutron.services.qos import qos_consts
|
|
from neutron.tests import base
|
|
|
|
|
|
def _get_test_dbdata(qos_policy_id):
|
|
return {'id': None, 'qos_policy_binding': {'policy_id': qos_policy_id,
|
|
'network_id': 'fake_net_id'}}
|
|
|
|
|
|
class QosCoreResourceExtensionTestCase(base.BaseTestCase):
|
|
|
|
def setUp(self):
|
|
super(QosCoreResourceExtensionTestCase, self).setUp()
|
|
self.core_extension = qos_core.QosCoreResourceExtension()
|
|
policy_p = mock.patch('neutron.objects.qos.policy.QosPolicy')
|
|
self.policy_m = policy_p.start()
|
|
self.context = context.get_admin_context()
|
|
self.non_admin_context = context.Context('user_id', 'tenant_id')
|
|
|
|
def test_process_fields_no_qos_policy_id(self):
|
|
self.core_extension.process_fields(
|
|
self.context, base_core.PORT, {}, None)
|
|
self.assertFalse(self.policy_m.called)
|
|
|
|
def _mock_plugin_loaded(self, plugin_loaded):
|
|
plugins = {}
|
|
if plugin_loaded:
|
|
plugins[plugin_constants.QOS] = None
|
|
return mock.patch('neutron_lib.plugins.directory.get_plugins',
|
|
return_value=plugins)
|
|
|
|
def test_process_fields_no_qos_plugin_loaded(self):
|
|
with self._mock_plugin_loaded(False):
|
|
self.core_extension.process_fields(
|
|
self.context, base_core.PORT,
|
|
{qos_consts.QOS_POLICY_ID: None}, None)
|
|
self.assertFalse(self.policy_m.called)
|
|
|
|
def test_process_fields_port_new_policy(self):
|
|
with self._mock_plugin_loaded(True):
|
|
qos_policy_id = mock.Mock()
|
|
actual_port = {'id': mock.Mock(),
|
|
qos_consts.QOS_POLICY_ID: qos_policy_id}
|
|
qos_policy = mock.MagicMock()
|
|
self.policy_m.get_object = mock.Mock(return_value=qos_policy)
|
|
self.core_extension.process_fields(
|
|
self.context, base_core.PORT,
|
|
{qos_consts.QOS_POLICY_ID: qos_policy_id},
|
|
actual_port)
|
|
|
|
qos_policy.attach_port.assert_called_once_with(actual_port['id'])
|
|
|
|
def test_process_fields_port_updated_policy(self):
|
|
with self._mock_plugin_loaded(True):
|
|
qos_policy1_id = mock.Mock()
|
|
qos_policy2_id = mock.Mock()
|
|
port_id = mock.Mock()
|
|
actual_port = {'id': port_id,
|
|
qos_consts.QOS_POLICY_ID: qos_policy1_id}
|
|
old_qos_policy = mock.MagicMock()
|
|
self.policy_m.get_port_policy = mock.Mock(
|
|
return_value=old_qos_policy)
|
|
new_qos_policy = mock.MagicMock()
|
|
self.policy_m.get_object = mock.Mock(return_value=new_qos_policy)
|
|
self.core_extension.process_fields(
|
|
self.context, base_core.PORT,
|
|
{qos_consts.QOS_POLICY_ID: qos_policy2_id},
|
|
actual_port)
|
|
|
|
old_qos_policy.detach_port.assert_called_once_with(port_id)
|
|
new_qos_policy.attach_port.assert_called_once_with(port_id)
|
|
self.assertEqual(qos_policy2_id, actual_port['qos_policy_id'])
|
|
|
|
def test_process_resource_port_updated_no_policy(self):
|
|
with self._mock_plugin_loaded(True):
|
|
port_id = mock.Mock()
|
|
qos_policy_id = mock.Mock()
|
|
actual_port = {'id': port_id,
|
|
qos_consts.QOS_POLICY_ID: qos_policy_id}
|
|
old_qos_policy = mock.MagicMock()
|
|
self.policy_m.get_port_policy = mock.Mock(
|
|
return_value=old_qos_policy)
|
|
new_qos_policy = mock.MagicMock()
|
|
self.policy_m.get_object = mock.Mock(return_value=new_qos_policy)
|
|
self.core_extension.process_fields(
|
|
self.context, base_core.PORT,
|
|
{qos_consts.QOS_POLICY_ID: None},
|
|
actual_port)
|
|
|
|
old_qos_policy.detach_port.assert_called_once_with(port_id)
|
|
self.assertIsNone(actual_port['qos_policy_id'])
|
|
|
|
def _process_port_updated_policy(self, context, shared,
|
|
policy_tenant_id):
|
|
with self._mock_plugin_loaded(True):
|
|
port_id = mock.sentinel.port_id
|
|
qos_policy_id = mock.sentinel.policy_id
|
|
actual_port = {'id': port_id,
|
|
qos_consts.QOS_POLICY_ID: qos_policy_id}
|
|
old_qos_policy = mock.MagicMock()
|
|
old_qos_policy.shared = shared
|
|
old_qos_policy.tenant_id = policy_tenant_id
|
|
self.policy_m.get_port_policy = mock.Mock(
|
|
return_value=old_qos_policy)
|
|
self.core_extension.process_fields(
|
|
context, base_core.PORT,
|
|
{qos_consts.QOS_POLICY_ID: None},
|
|
actual_port)
|
|
|
|
old_qos_policy.detach_port.assert_called_once_with(port_id)
|
|
|
|
def test_process_resource_port_updated_remove_own_policy(self):
|
|
self._process_port_updated_policy(
|
|
context=self.non_admin_context,
|
|
shared=False,
|
|
policy_tenant_id=self.non_admin_context.tenant_id)
|
|
|
|
def test_process_resource_port_updated_admin_remove_provided_policy(self):
|
|
self._process_port_updated_policy(
|
|
context=self.context,
|
|
shared=False,
|
|
policy_tenant_id=self.non_admin_context.tenant_id)
|
|
|
|
def test_process_resource_port_updated_remove_shared_policy(self):
|
|
self._process_port_updated_policy(
|
|
context=self.non_admin_context,
|
|
shared=True,
|
|
policy_tenant_id=self.context.tenant_id)
|
|
|
|
def test_process_resource_port_updated_remove_provided_policy(self):
|
|
self.policy_m.is_accessible.return_value = False
|
|
self.assertRaises(n_exc.PolicyRemoveAuthorizationError,
|
|
self._process_port_updated_policy,
|
|
context=self.non_admin_context,
|
|
shared=False,
|
|
policy_tenant_id=self.context.tenant_id)
|
|
|
|
def test_process_resource_network_updated_no_policy(self):
|
|
with self._mock_plugin_loaded(True):
|
|
network_id = mock.Mock()
|
|
qos_policy_id = mock.Mock()
|
|
actual_network = {'id': network_id,
|
|
qos_consts.QOS_POLICY_ID: qos_policy_id}
|
|
old_qos_policy = mock.MagicMock()
|
|
self.policy_m.get_network_policy = mock.Mock(
|
|
return_value=old_qos_policy)
|
|
new_qos_policy = mock.MagicMock()
|
|
self.policy_m.get_object = mock.Mock(return_value=new_qos_policy)
|
|
self.core_extension.process_fields(
|
|
self.context, base_core.NETWORK,
|
|
{qos_consts.QOS_POLICY_ID: None},
|
|
actual_network)
|
|
|
|
old_qos_policy.detach_network.assert_called_once_with(network_id)
|
|
self.assertIsNone(actual_network['qos_policy_id'])
|
|
|
|
def test_process_fields_network_new_policy(self):
|
|
with self._mock_plugin_loaded(True):
|
|
qos_policy_id = mock.Mock()
|
|
actual_network = {'id': mock.Mock(),
|
|
qos_consts.QOS_POLICY_ID: qos_policy_id}
|
|
qos_policy = mock.MagicMock()
|
|
self.policy_m.get_object = mock.Mock(return_value=qos_policy)
|
|
self.core_extension.process_fields(
|
|
self.context, base_core.NETWORK,
|
|
{qos_consts.QOS_POLICY_ID: qos_policy_id}, actual_network)
|
|
|
|
qos_policy.attach_network.assert_called_once_with(
|
|
actual_network['id'])
|
|
|
|
def test_process_fields_network_updated_policy(self):
|
|
with self._mock_plugin_loaded(True):
|
|
qos_policy_id = mock.Mock()
|
|
network_id = mock.Mock()
|
|
actual_network = {'id': network_id,
|
|
qos_consts.QOS_POLICY_ID: qos_policy_id}
|
|
old_qos_policy = mock.MagicMock()
|
|
self.policy_m.get_network_policy = mock.Mock(
|
|
return_value=old_qos_policy)
|
|
new_qos_policy = mock.MagicMock()
|
|
self.policy_m.get_object = mock.Mock(return_value=new_qos_policy)
|
|
self.core_extension.process_fields(
|
|
self.context, base_core.NETWORK,
|
|
{qos_consts.QOS_POLICY_ID: qos_policy_id}, actual_network)
|
|
|
|
old_qos_policy.detach_network.assert_called_once_with(network_id)
|
|
new_qos_policy.attach_network.assert_called_once_with(network_id)
|
|
|
|
def _process_network_updated_policy(self, context, shared,
|
|
policy_tenant_id):
|
|
with self._mock_plugin_loaded(True):
|
|
qos_policy_id = mock.sentinel.policy_id
|
|
network_id = mock.sentinel.net_id
|
|
actual_network = {'id': network_id,
|
|
qos_consts.QOS_POLICY_ID: qos_policy_id}
|
|
old_qos_policy = mock.MagicMock()
|
|
old_qos_policy.shared = shared
|
|
old_qos_policy.tenant_id = policy_tenant_id
|
|
self.policy_m.get_network_policy.return_value = old_qos_policy
|
|
self.core_extension.process_fields(
|
|
context, base_core.NETWORK,
|
|
{qos_consts.QOS_POLICY_ID: None}, actual_network)
|
|
|
|
old_qos_policy.detach_network.assert_called_once_with(network_id)
|
|
|
|
def test_process_fields_network_updated_remove_shared_policy(self):
|
|
self._process_network_updated_policy(
|
|
context=self.non_admin_context,
|
|
shared=True,
|
|
policy_tenant_id=self.context.tenant_id)
|
|
|
|
def test_process_fields_network_updated_remove_own_policy(self):
|
|
self._process_network_updated_policy(
|
|
context=self.non_admin_context,
|
|
shared=True,
|
|
policy_tenant_id=self.non_admin_context.tenant_id)
|
|
|
|
def test_process_fields_network_updated_admin_remove_provided_policy(self):
|
|
self._process_network_updated_policy(
|
|
context=self.context,
|
|
shared=True,
|
|
policy_tenant_id=self.non_admin_context.tenant_id)
|
|
|
|
def test_process_fields_network_updated_remove_provided_policy(self):
|
|
self.policy_m.is_accessible.return_value = False
|
|
self.assertRaises(n_exc.PolicyRemoveAuthorizationError,
|
|
self._process_network_updated_policy,
|
|
context=self.non_admin_context,
|
|
shared=False,
|
|
policy_tenant_id=self.context.tenant_id)
|
|
|
|
def test_extract_fields_plugin_not_loaded(self):
|
|
with self._mock_plugin_loaded(False):
|
|
fields = self.core_extension.extract_fields(None, None)
|
|
self.assertEqual({}, fields)
|
|
|
|
def _test_extract_fields_for_port(self, qos_policy_id):
|
|
with self._mock_plugin_loaded(True):
|
|
fields = self.core_extension.extract_fields(
|
|
base_core.PORT, _get_test_dbdata(qos_policy_id))
|
|
self.assertEqual({qos_consts.QOS_POLICY_ID: qos_policy_id}, fields)
|
|
|
|
def test_extract_fields_no_port_policy(self):
|
|
self._test_extract_fields_for_port(None)
|
|
|
|
def test_extract_fields_port_policy_exists(self):
|
|
qos_policy_id = mock.Mock()
|
|
self._test_extract_fields_for_port(qos_policy_id)
|
|
|
|
def _test_extract_fields_for_network(self, qos_policy_id):
|
|
with self._mock_plugin_loaded(True):
|
|
fields = self.core_extension.extract_fields(
|
|
base_core.NETWORK, _get_test_dbdata(qos_policy_id))
|
|
self.assertEqual({qos_consts.QOS_POLICY_ID: qos_policy_id}, fields)
|
|
|
|
def test_extract_fields_no_network_policy(self):
|
|
self._test_extract_fields_for_network(None)
|
|
|
|
def test_extract_fields_network_policy_exists(self):
|
|
qos_policy_id = mock.Mock()
|
|
qos_policy = mock.Mock()
|
|
qos_policy.id = qos_policy_id
|
|
self._test_extract_fields_for_network(qos_policy_id)
|