text: #    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 itertools
  from unittest import mock
  
  from neutron_lib.api.definitions import portbindings
  from neutron_lib.callbacks import registry
  from neutron_lib import constants as lib_consts
  from neutron_lib import context
  from neutron_lib import exceptions
  from neutron_lib.services.qos import base as qos_driver_base
  from neutron_lib.services.qos import constants as qos_consts
  from oslo_utils import uuidutils
  
  from neutron.api.rpc.callbacks.producer import registry as rpc_registry
  from neutron.objects import ports as ports_object
  from neutron.objects.qos import rule as rule_object
  from neutron.services.qos.drivers.linuxbridge import driver as lb_driver
  from neutron.services.qos.drivers import manager as driver_mgr
  from neutron.services.qos.drivers.openvswitch import driver as ovs_driver
  from neutron.services.qos.drivers.ovn import driver as ovn_driver
  from neutron.services.qos.drivers.sriov import driver as sriov_driver
  from neutron.tests.unit.services.qos import base
  
  
  class TestQosDriversManagerBase(base.BaseQosTestCase):
  
      def setUp(self):
          super(TestQosDriversManagerBase, self).setUp()
          self.config_parse()
          self.setup_coreplugin(load_plugins=False)
          self._loaded_qos_drivers = []
  
      def _delete_loaded_qos_drivers(self):
          registry._get_callback_manager().clear()
          for idx, _ in enumerate(self._loaded_qos_drivers):
              del self._loaded_qos_drivers[idx]
  
      def _create_manager_with_drivers(self, drivers_details):
          for name, driver_details in drivers_details.items():
  
              class QoSDriver(qos_driver_base.DriverBase):
                  @property
                  def is_loaded(self):
                      return driver_details['is_loaded']
  
              # the new ad-hoc driver will register on the QOS_PLUGIN registry
              self._loaded_qos_drivers.append(
                  QoSDriver(name,
                            driver_details.get('vif_types', []),
                            driver_details.get('vnic_types', []),
                            driver_details.get('rules', [])))
  
          return driver_mgr.QosServiceDriverManager()
  
  
  class TestQosDriversManagerMulti(TestQosDriversManagerBase):
      """Test calls happen to all drivers"""
      def test_driver_manager_empty_with_no_drivers(self):
          driver_manager = self._create_manager_with_drivers({})
          self.assertEqual(len(driver_manager._drivers), 0)
  
      def test_driver_manager_empty_with_no_loaded_drivers(self):
          driver_manager = self._create_manager_with_drivers(
              {'driver-A': {'is_loaded': False}})
          self.assertEqual(len(driver_manager._drivers), 0)
  
      def test_driver_manager_with_one_loaded_driver(self):
          driver_manager = self._create_manager_with_drivers(
              {'driver-A': {'is_loaded': True}})
          self.assertEqual(len(driver_manager._drivers), 1)
  
      def test_driver_manager_with_two_loaded_drivers(self):
          driver_manager = self._create_manager_with_drivers(
              {'driver-A': {'is_loaded': True},
               'driver-B': {'is_loaded': True}})
          self.assertEqual(len(driver_manager._drivers), 2)
  
  
  class TestQoSDriversRulesValidations(TestQosDriversManagerBase):
      """Test validation of rules for port"""
  
      def setUp(self):
          super(TestQoSDriversRulesValidations, self).setUp()
          self.ctxt = context.Context('fake_user', 'fake_tenant')
  
      def _get_port(self, vif_type, vnic_type):
          port_id = uuidutils.generate_uuid()
          port_binding = ports_object.PortBinding(
              self.ctxt, port_id=port_id, vif_type=vif_type, vnic_type=vnic_type)
          return ports_object.Port(
              self.ctxt, id=uuidutils.generate_uuid(), bindings=[port_binding])
  
      def _test_validate_rule_for_port(self, port, expected_result):
          driver_manager = self._create_manager_with_drivers({
              'driver-A': {
                  'is_loaded': True,
                  'rules': {
                      qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH: {
                          "min_kbps": {'type:values': None},
                          'direction': {
                              'type:values': lib_consts.VALID_DIRECTIONS}
                      }
                  },
                  'vif_types': [portbindings.VIF_TYPE_OVS],
                  'vnic_types': [portbindings.VNIC_NORMAL]
              }
          })
          rule = rule_object.QosMinimumBandwidthRule(
              self.ctxt, id=uuidutils.generate_uuid())
  
          is_rule_supported_mock = mock.Mock()
          if expected_result:
              is_rule_supported_mock.return_value = expected_result
          driver_manager._drivers[0].is_rule_supported = is_rule_supported_mock
  
          self.assertEqual(expected_result,
                           driver_manager.validate_rule_for_port(
                               mock.Mock(), rule, port))
          if expected_result:
              is_rule_supported_mock.assert_called_once_with(rule)
          else:
              is_rule_supported_mock.assert_not_called()
  
      def test_validate_rule_for_network(self):
          driver_manager = self._create_manager_with_drivers({
              'driver-A': {
                  'is_loaded': True,
                  'rules': {
                      qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH: {
                          "min_kbps": {'type:values': None},
                          'direction': {
                              'type:values': lib_consts.VALID_DIRECTIONS}
                      }
                  }
              }
          })
          rule = rule_object.QosMinimumBandwidthRule(
              self.ctxt, id=uuidutils.generate_uuid())
  
          is_rule_supported_mock = mock.Mock()
          is_rule_supported_mock.return_value = True
          driver_manager._drivers[0].is_rule_supported = is_rule_supported_mock
          self.assertTrue(driver_manager.validate_rule_for_network(
                           mock.Mock(), rule, mock.Mock()))
          is_rule_supported_mock.assert_called_once_with(rule)
  
      def test_validate_rule_for_port_rule_vif_type_supported(self):
          port = self._get_port(
              portbindings.VIF_TYPE_OVS, portbindings.VNIC_NORMAL)
          self._test_validate_rule_for_port(
              port, expected_result=True)
  
      def test_validate_rule_for_port_vif_type_not_supported(self):
          port = self._get_port(
              portbindings.VIF_TYPE_OTHER, portbindings.VNIC_NORMAL)
          self._test_validate_rule_for_port(
              port, expected_result=False)
  
      def test_validate_rule_for_port_unbound_vnic_type_supported(self):
          port = self._get_port(
              portbindings.VIF_TYPE_UNBOUND, portbindings.VNIC_NORMAL)
          self._test_validate_rule_for_port(
              port, expected_result=True)
  
      def test_validate_rule_for_port_unbound_vnic_type_not_supported(self):
          port = self._get_port(
              portbindings.VIF_TYPE_UNBOUND, portbindings.VNIC_BAREMETAL)
          self._test_validate_rule_for_port(
              port, expected_result=False)
  
  
  class TestQosDriversManagerRules(TestQosDriversManagerBase):
      """Test supported rules"""
  
      @mock.patch.object(rpc_registry, 'provide')
      def test_available_rules(self, *args):
          available_drivers = {'linuxbridge': lb_driver.SUPPORTED_RULES,
                               'ovs': ovs_driver.SUPPORTED_RULES,
                               'ovn': ovn_driver.SUPPORTED_RULES,
                               'sriov': sriov_driver.SUPPORTED_RULES}
          for drivers in itertools.combinations(available_drivers, 2):
              rules_0 = set(available_drivers[drivers[0]])
              rules_1 = set(available_drivers[drivers[1]])
              driver_manager = self._create_manager_with_drivers(
                  {drivers[0]: {'is_loaded': True, 'rules': rules_0},
                   drivers[1]: {'is_loaded': True, 'rules': rules_1}})
              self.assertEqual(
                  driver_manager.supported_rule_types(),
                  rules_0 & rules_1)
              self.assertEqual(
                  driver_manager.supported_rule_types(all_supported=True),
                  rules_0 | rules_1)
              self.assertEqual(
                  driver_manager.supported_rule_types(all_rules=True),
                  set(qos_consts.VALID_RULE_TYPES))
  
              self._delete_loaded_qos_drivers()
  
      def test__parse_parameter_values(self):
          range_parameter = {'type:range': [0, 10]}
          values_parameter = {'type:values': [1, 10, 100, 1000]}
          expected_parsed_range_parameter = {'start': 0, 'end': 10}
          expected_parsed_values_parameter = [1, 10, 100, 1000]
  
          parameter_values, parameter_type = (
              driver_mgr.QosServiceDriverManager._parse_parameter_values(
                  range_parameter))
          self.assertEqual(
              expected_parsed_range_parameter, parameter_values)
          self.assertEqual(
              lib_consts.VALUES_TYPE_RANGE, parameter_type)
  
          parameter_values, parameter_type = (
              driver_mgr.QosServiceDriverManager._parse_parameter_values(
                  values_parameter))
          self.assertEqual(
              expected_parsed_values_parameter, parameter_values)
          self.assertEqual(
              lib_consts.VALUES_TYPE_CHOICES, parameter_type)
  
      def test_supported_rule_type_details(self):
          driver_manager = self._create_manager_with_drivers({
              'driver-A': {
                  'is_loaded': True,
                  'rules': {
                      qos_consts.RULE_TYPE_BANDWIDTH_LIMIT: {
                          "max_kbps": {'type:range': [0, 1000]},
                          "max_burst_kbps": {'type:range': [0, 1000]}
                      }
                  }
              },
              'driver-B': {
                  'is_loaded': True,
                  'rules': {
                      qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH: {
                          "min_kbps": {'type:range': [0, 1000]},
                          'direction': {
                              'type:values': lib_consts.VALID_DIRECTIONS}
                      },
                      qos_consts.RULE_TYPE_DSCP_MARKING: {
                          "dscp_mark": {
                              'type:values': lib_consts.VALID_DSCP_MARKS}
                      }
                  }
              }
          })
          expected_rule_type_details = [{
              'name': 'driver-A',
              'supported_parameters': [{
                  'parameter_name': 'max_kbps',
                  'parameter_type': lib_consts.VALUES_TYPE_RANGE,
                  'parameter_values': {'start': 0, 'end': 1000}
              }, {
                  'parameter_name': 'max_burst_kbps',
                  'parameter_type': lib_consts.VALUES_TYPE_RANGE,
                  'parameter_values': {'start': 0, 'end': 1000}
              }]
          }]
          bandwidth_limit_details = driver_manager.supported_rule_type_details(
              qos_consts.RULE_TYPE_BANDWIDTH_LIMIT)
          self.assertEqual(
              len(expected_rule_type_details), len(bandwidth_limit_details))
          self.assertEqual(
              expected_rule_type_details[0]['name'],
              bandwidth_limit_details[0]['name'])
          self.assertEqual(
              len(expected_rule_type_details[0]['supported_parameters']),
              len(bandwidth_limit_details[0]['supported_parameters'])
          )
          for parameter in expected_rule_type_details[0]['supported_parameters']:
              self.assertIn(
                  parameter,
                  bandwidth_limit_details[0]['supported_parameters'])
  
      def test_supported_rule_type_details_no_drivers_loaded(self):
          driver_manager = self._create_manager_with_drivers({})
          self.assertEqual(
              [],
              driver_manager.supported_rule_type_details(
                  qos_consts.RULE_TYPE_BANDWIDTH_LIMIT))
  
  
  class TestQosDriversCalls(TestQosDriversManagerBase):
      """Test QoS driver calls"""
  
      def setUp(self):
          super(TestQosDriversCalls, self).setUp()
          self.driver_manager = self._create_manager_with_drivers(
              {'driver-A': {'is_loaded': True}})
  
      def test_implemented_call_methods(self):
          for method in qos_consts.QOS_CALL_METHODS:
              with mock.patch.object(qos_driver_base.DriverBase, method) as \
                      method_fnc:
                  context = mock.Mock()
                  policy = mock.Mock()
                  self.driver_manager.call(method, context, policy)
                  method_fnc.assert_called_once_with(context, policy)
  
      def test_not_implemented_call_methods(self):
          self.assertRaises(exceptions.DriverCallError, self.driver_manager.call,
                            'wrong_method', mock.Mock(), mock.Mock())
  