openstack/neutron
Code Issues Proposed changes

use qos DriverBase from neutron-lib

Browse Source 
neutron-lib contains the QoS DriverBase class. This patch removes
DriverBase and its associated UTs from neutron and uses them from
neutron-lib.

This patch also switches test_ovs_bridge to use mock.patch.object rather
than mock.patch as py27 was incurring sporadic failures running this
test module.

NeutronLibImpact

Change-Id: Ic8027b73f82d691bd3c465061a4c8d8301288b07
This commit is contained in:
Boden R
2017-07-25 14:22:38 -06:00
parent 25db0f0266
commit d37e974e46
7 changed files with 13 additions and 265 deletions
Download Patch File Download Diff File Expand all files Collapse all files

162
neutron/services/qos/drivers/base.py
View File

@ -1,162 +0,0 @@
# Copyright 2016 Hewlett Packard Enterprise Development Company, LP
# Copyright 2016 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.
from neutron_lib.api import validators as lib_validators
from neutron_lib.callbacks import events
from neutron_lib.callbacks import registry
from oslo_log import log as logging
from neutron.services.qos import qos_consts
LOG = logging.getLogger(__name__)
@registry.has_registry_receivers
class DriverBase(object):
def __init__(self, name, vif_types, vnic_types,
supported_rules,
requires_rpc_notifications=False):
"""Instantiate a qos driver.
:param name: driver name.
:param vif_types: list of interfaces (VIFs) supported.
:param vnic_types: list of vnic types supported.
:param supported_rules: dict of supported rules.
:param requires_rpc_notifications: indicates if this driver
expects rpc push notifications to be sent from the driver.
"""
self.name = name
self.vif_types = vif_types
self.vnic_types = vnic_types
self.supported_rules = supported_rules
self.requires_rpc_notifications = requires_rpc_notifications
@registry.receives(qos_consts.QOS_PLUGIN, [events.AFTER_INIT])
def _register(self, resource, event, trigger, **kwargs):
if self.is_loaded:
# trigger is the QosServiceDriverManager
trigger.register_driver(self)
def is_loaded(self):
"""True if the driver is active for the Neutron Server.
Implement this property to determine if your driver is actively
configured for this Neutron Server deployment.
"""
return True
def is_vif_type_compatible(self, vif_type):
"""True if the driver is compatible with the VIF type."""
return vif_type in self.vif_types
def is_vnic_compatible(self, vnic_type):
"""True if the driver is compatible with the specific VNIC type."""
return vnic_type in self.vnic_types
def is_rule_supported(self, rule):
supported_parameters = self.supported_rules.get(rule.rule_type)
if not supported_parameters:
LOG.debug("Rule type %(rule_type)s is not supported by "
"%(driver_name)s",
{'rule_type': rule.rule_type,
'driver_name': self.name})
return False
for parameter, validators in supported_parameters.items():
parameter_value = rule.get(parameter)
for validator_type, validator_data in validators.items():
validator_function = lib_validators.get_validator(
validator_type)
validate_result = validator_function(parameter_value,
validator_data)
# NOTE(slaweq): validator functions returns None if data is
# valid or string with reason why data is not valid
if validate_result:
LOG.debug("Parameter %(parameter)s=%(value)s in "
"rule type %(rule_type)s is not "
"supported by %(driver_name)s. "
"Validate result: %(validate_result)s",
{'parameter': parameter,
'value': parameter_value,
'rule_type': rule.rule_type,
'driver_name': self.name,
'validate_result': validate_result})
return False
return True
def create_policy(self, context, policy):
"""Create policy invocation.
This method can be implemented by the specific driver subclass
to update the backend where necessary with the specific policy
information.
:param context: current running context information
:param policy: a QoSPolicy object being created, which will have no
rules.
"""
def create_policy_precommit(self, context, policy):
"""Create policy precommit.
This method can be implemented by the specific driver subclass
to handle the precommit event of a policy that is being created.
:param context: current running context information
:param policy: a QoSPolicy object being created, which will have no
rules.
"""
def update_policy(self, context, policy):
"""Update policy invocation.
This method can be implemented by the specific driver subclass
to update the backend where necessary.
:param context: current running context information
:param policy: a QoSPolicy object being updated.
"""
def update_policy_precommit(self, context, policy):
"""Update policy precommit.
This method can be implemented by the specific driver subclass
to handle update precommit event of a policy that is being updated.
:param context: current running context information
:param policy: a QoSPolicy object being updated.
"""
def delete_policy(self, context, policy):
"""Delete policy invocation.
This method can be implemented by the specific driver subclass
to delete the backend policy where necessary.
:param context: current running context information
:param policy: a QoSPolicy object being deleted
"""
def delete_policy_precommit(self, context, policy):
"""Delete policy precommit.
This method can be implemented by the specific driver subclass
to handle delete precommit event of a policy that is being deleted.
:param context: current running context information
:param policy: a QoSPolicy object being deleted
"""

2
neutron/services/qos/drivers/linuxbridge/driver.py
View File

@ -14,10 +14,10 @@
# under the License.
from neutron_lib.api.definitions import portbindings
from neutron_lib.services.qos import base
from oslo_log import log as logging
from neutron.common import constants
from neutron.services.qos.drivers import base
from neutron.services.qos import qos_consts
LOG = logging.getLogger(__name__)

2
neutron/services/qos/drivers/openvswitch/driver.py
View File

@ -14,10 +14,10 @@
# under the License.
from neutron_lib.api.definitions import portbindings
from neutron_lib.services.qos import base
from oslo_log import log as logging
from neutron.common import constants
from neutron.services.qos.drivers import base
from neutron.services.qos import qos_consts
LOG = logging.getLogger(__name__)

2
neutron/services/qos/drivers/sriov/driver.py
View File

@ -14,10 +14,10 @@
# under the License.
from neutron_lib.api.definitions import portbindings
from neutron_lib.services.qos import base
from oslo_log import log as logging
from neutron.common import constants
from neutron.services.qos.drivers import base
from neutron.services.qos import qos_consts
LOG = logging.getLogger(__name__)

17
neutron/tests/unit/plugins/ml2/drivers/openvswitch/agent/openflow/native/test_ovs_bridge.py
View File

@ -15,18 +15,19 @@
import mock
from neutron.agent.common import ovs_lib
from neutron.plugins.ml2.drivers.openvswitch.agent.openflow.native \
import ofswitch
from neutron.tests.unit.plugins.ml2.drivers.openvswitch.agent \
import ovs_test_base
class OVSAgentBridgeTestCase(ovs_test_base.OVSRyuTestBase):
def test__get_dp(self):
mock.patch(
'neutron.agent.common.ovs_lib.OVSBridge.get_datapath_id',
return_value="3e9").start()
mock.patch(
"neutron.plugins.ml2.drivers.openvswitch.agent.openflow.native."
"ofswitch.OpenFlowSwitchMixin._get_dp_by_dpid",
mock.patch.object(
ovs_lib.OVSBridge, 'get_datapath_id', return_value="3e9").start()
mock.patch.object(
ofswitch.OpenFlowSwitchMixin, "_get_dp_by_dpid",
side_effect=RuntimeError).start()
br = self.br_int_cls('br-int')
br._cached_dpid = int("3e9", 16)
@ -40,8 +41,8 @@ class OVSAgentBridgeTestCase(ovs_test_base.OVSRyuTestBase):
if tb == 'Bridge':
return []
mock.patch('neutron.agent.common.ovs_lib.OVSBridge.db_get_val',
side_effect=_mock_db_get_val).start()
mock.patch.object(ovs_lib.OVSBridge, 'db_get_val',
side_effect=_mock_db_get_val).start()
br = self.br_int_cls('br-int')
# make sure that in case of any misconfiguration when no datapath is
# found a proper exception, not a TypeError is raised

91
neutron/tests/unit/services/qos/drivers/test_base.py
View File

@ -1,91 +0,0 @@
# Copyright 2017 OVH SAS
# 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.
from neutron_lib.api.definitions import portbindings
from neutron_lib import context
from oslo_utils import uuidutils
from neutron.common import constants
from neutron.objects.qos import rule as rule_object
from neutron.services.qos.drivers import base as qos_base_driver
from neutron.services.qos import qos_consts
from neutron.tests import base
SUPPORTED_RULES = {
qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH: {
"min_kbps": {'type:values': None},
'direction': {'type:values': [constants.EGRESS_DIRECTION]}
}
}
class FakeDriver(qos_base_driver.DriverBase):
@staticmethod
def create():
return FakeDriver(
name='fake_driver',
vif_types=[portbindings.VIF_TYPE_OVS],
vnic_types=[portbindings.VNIC_NORMAL],
supported_rules=SUPPORTED_RULES,
requires_rpc_notifications=False)
class TestDriverBase(base.BaseTestCase):
def setUp(self):
super(TestDriverBase, self).setUp()
self.driver = FakeDriver.create()
self.rule_data = {
'minimum_bandwidth_rule': {
'id': uuidutils.generate_uuid(),
'min_kbps': 100,
'direction': constants.EGRESS_DIRECTION
},
'dscp_marking_rule': {
'id': uuidutils.generate_uuid(),
'dscp_mark': 16
}
}
ctxt = context.Context('fake_user', 'fake_tenant')
self.minimum_bandwidth_rule = rule_object.QosMinimumBandwidthRule(
ctxt, **self.rule_data['minimum_bandwidth_rule'])
self.dscp_rule = rule_object.QosDscpMarkingRule(
ctxt, **self.rule_data['dscp_marking_rule'])
def test_is_vif_type_compatible(self):
self.assertFalse(
self.driver.is_vif_type_compatible(portbindings.VIF_TYPE_OTHER))
self.assertTrue(
self.driver.is_vif_type_compatible(portbindings.VIF_TYPE_OVS))
def test_is_vnic_compatible(self):
self.assertFalse(
self.driver.is_vnic_compatible(portbindings.VNIC_BAREMETAL))
self.assertTrue(
self.driver.is_vnic_compatible(portbindings.VNIC_NORMAL))
def test_is_rule_supported(self):
# Rule which is in SUPPORTED_RULES should be supported
self.assertTrue(
self.driver.is_rule_supported(self.minimum_bandwidth_rule))
# Rule which is not in SUPPORTED_RULES should not be supported
self.assertFalse(self.driver.is_rule_supported(self.dscp_rule))
# Rule which is in SUPPORTED_RULES but got not supported parameter
# should not be supported
self.minimum_bandwidth_rule.direction = constants.INGRESS_DIRECTION
self.assertFalse(
self.driver.is_rule_supported(self.minimum_bandwidth_rule))

2
neutron/tests/unit/services/qos/drivers/test_manager.py
View File

@ -13,13 +13,13 @@
import mock
from neutron_lib.api.definitions import portbindings
from neutron_lib import context
from neutron_lib.services.qos import base as qos_driver_base
from oslo_utils import uuidutils
from neutron.common import constants
from neutron.common import exceptions
from neutron.objects import ports as ports_object
from neutron.objects.qos import rule as rule_object
from neutron.services.qos.drivers import base as qos_driver_base
from neutron.services.qos.drivers import manager as driver_mgr
from neutron.services.qos import qos_consts
from neutron.tests.unit.services.qos import base