neutron/tests/unit/services/qos/test_qos_plugin.py

   1 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   2 #    not use this file except in compliance with the License. You may obtain
   3 #    a copy of the License at
   4 #
   5 #         http://www.apache.org/licenses/LICENSE-2.0
   6 #
   7 #    Unless required by applicable law or agreed to in writing, software
   8 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   9 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  10 #    License for the specific language governing permissions and limitations
  11 #    under the License.
  12
  13 import mock
  14 from oslo_config import cfg
  15
  16 from neutron.common import exceptions as n_exc
  17 from neutron import context
  18 from neutron import manager
  19 from neutron.objects import base as base_object
  20 from neutron.objects.qos import policy as policy_object
  21 from neutron.objects.qos import rule as rule_object
  22 from neutron.plugins.common import constants
  23 from neutron.tests.unit.services.qos import base
  24
  25
  26 DB_PLUGIN_KLASS = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'
  27
  28
  29 class TestQosPlugin(base.BaseQosTestCase):
  30
  31     def setUp(self):
  32         super(TestQosPlugin, self).setUp()
  33         self.setup_coreplugin()
  34
  35         mock.patch('neutron.db.api.create_object').start()
  36         mock.patch('neutron.db.api.update_object').start()
  37         mock.patch('neutron.db.api.delete_object').start()
  38         mock.patch('neutron.db.api.get_object').start()
  39         mock.patch(
  40             'neutron.objects.qos.policy.QosPolicy.obj_load_attr').start()
  41
  42         cfg.CONF.set_override("core_plugin", DB_PLUGIN_KLASS)
  43         cfg.CONF.set_override("service_plugins", ["qos"])
  44
  45         mgr = manager.NeutronManager.get_instance()
  46         self.qos_plugin = mgr.get_service_plugins().get(
  47             constants.QOS)
  48
  49         self.qos_plugin.notification_driver_manager = mock.Mock()
  50
  51         self.ctxt = context.Context('fake_user', 'fake_tenant')
  52         self.policy_data = {
  53             'policy': {'id': 7777777,
  54                        'tenant_id': 888888,
  55                        'name': 'test-policy',
  56                        'description': 'Test policy description',
  57                        'shared': True}}
  58
  59         self.rule_data = {
  60             'bandwidth_limit_rule': {'id': 7777777,
  61                                      'max_kbps': 100,
  62                                      'max_burst_kbps': 150}}
  63
  64         self.policy = policy_object.QosPolicy(
  65             self.ctxt, **self.policy_data['policy'])
  66
  67         self.rule = rule_object.QosBandwidthLimitRule(
  68             self.ctxt, **self.rule_data['bandwidth_limit_rule'])
  69
  70     def _validate_notif_driver_params(self, method_name):
  71         method = getattr(self.qos_plugin.notification_driver_manager,
  72                          method_name)
  73         self.assertTrue(method.called)
  74         self.assertIsInstance(
  75             method.call_args[0][1], policy_object.QosPolicy)
  76
  77     def test_add_policy(self):
  78         self.qos_plugin.create_policy(self.ctxt, self.policy_data)
  79         self._validate_notif_driver_params('create_policy')
  80
  81     def test_update_policy(self):
  82         fields = base_object.get_updatable_fields(
  83             policy_object.QosPolicy, self.policy_data['policy'])
  84         self.qos_plugin.update_policy(
  85             self.ctxt, self.policy.id, {'policy': fields})
  86         self._validate_notif_driver_params('update_policy')
  87
  88     @mock.patch('neutron.db.api.get_object', return_value=None)
  89     def test_delete_policy(self, *mocks):
  90         self.qos_plugin.delete_policy(self.ctxt, self.policy.id)
  91         self._validate_notif_driver_params('delete_policy')
  92
  93     def test_create_policy_rule(self):
  94         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
  95                         return_value=self.policy):
  96             self.qos_plugin.create_policy_bandwidth_limit_rule(
  97                 self.ctxt, self.policy.id, self.rule_data)
  98             self._validate_notif_driver_params('update_policy')
  99
 100     def test_update_policy_rule(self):
 101         _policy = policy_object.QosPolicy(
 102             self.ctxt, **self.policy_data['policy'])
 103         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
 104                         return_value=_policy):
 105             setattr(_policy, "rules", [self.rule])
 106             self.qos_plugin.update_policy_bandwidth_limit_rule(
 107                 self.ctxt, self.rule.id, self.policy.id, self.rule_data)
 108             self._validate_notif_driver_params('update_policy')
 109
 110     def test_update_policy_rule_bad_policy(self):
 111         _policy = policy_object.QosPolicy(
 112             self.ctxt, **self.policy_data['policy'])
 113         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
 114                         return_value=_policy):
 115             setattr(_policy, "rules", [])
 116             self.assertRaises(
 117                 n_exc.QosRuleNotFound,
 118                 self.qos_plugin.update_policy_bandwidth_limit_rule,
 119                 self.ctxt, self.rule.id, self.policy.id,
 120                 self.rule_data)
 121
 122     def test_delete_policy_rule(self):
 123         _policy = policy_object.QosPolicy(
 124             self.ctxt, **self.policy_data['policy'])
 125         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
 126                         return_value=_policy):
 127             setattr(_policy, "rules", [self.rule])
 128             self.qos_plugin.delete_policy_bandwidth_limit_rule(
 129                         self.ctxt, self.rule.id, _policy.id)
 130             self._validate_notif_driver_params('update_policy')
 131
 132     def test_delete_policy_rule_bad_policy(self):
 133         _policy = policy_object.QosPolicy(
 134             self.ctxt, **self.policy_data['policy'])
 135         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
 136                         return_value=_policy):
 137             setattr(_policy, "rules", [])
 138             self.assertRaises(
 139                 n_exc.QosRuleNotFound,
 140                 self.qos_plugin.delete_policy_bandwidth_limit_rule,
 141                 self.ctxt, self.rule.id, _policy.id)
 142
 143     def test_get_policy_bandwidth_limit_rules_for_policy(self):
 144         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
 145                         return_value=self.policy):
 146             with mock.patch('neutron.objects.qos.rule.'
 147                             'QosBandwidthLimitRule.'
 148                             'get_objects') as get_object_mock:
 149                 self.qos_plugin.get_policy_bandwidth_limit_rules(
 150                     self.ctxt, self.policy.id)
 151                 get_object_mock.assert_called_once_with(
 152                     self.ctxt, qos_policy_id=self.policy.id)
 153
 154     def test_get_policy_bandwidth_limit_rules_for_policy_with_filters(self):
 155         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
 156                         return_value=self.policy):
 157             with mock.patch('neutron.objects.qos.rule.'
 158                             'QosBandwidthLimitRule.'
 159                             'get_objects') as get_object_mock:
 160
 161                 filters = {'filter': 'filter_id'}
 162                 self.qos_plugin.get_policy_bandwidth_limit_rules(
 163                     self.ctxt, self.policy.id, filters=filters)
 164                 get_object_mock.assert_called_once_with(
 165                     self.ctxt, qos_policy_id=self.policy.id,
 166                     filter='filter_id')
 167
 168     def test_get_policy_for_nonexistent_policy(self):
 169         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
 170                         return_value=None):
 171             self.assertRaises(
 172                 n_exc.QosPolicyNotFound,
 173                 self.qos_plugin.get_policy,
 174                 self.ctxt, self.policy.id)
 175
 176     def test_get_policy_bandwidth_limit_rule_for_nonexistent_policy(self):
 177         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
 178                         return_value=None):
 179             self.assertRaises(
 180                 n_exc.QosPolicyNotFound,
 181                 self.qos_plugin.get_policy_bandwidth_limit_rule,
 182                 self.ctxt, self.rule.id, self.policy.id)
 183
 184     def test_get_policy_bandwidth_limit_rules_for_nonexistent_policy(self):
 185         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
 186                         return_value=None):
 187             self.assertRaises(
 188                 n_exc.QosPolicyNotFound,
 189                 self.qos_plugin.get_policy_bandwidth_limit_rules,
 190                 self.ctxt, self.policy.id)
 191
 192     def test_create_policy_rule_for_nonexistent_policy(self):
 193         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
 194                         return_value=None):
 195             self.assertRaises(
 196                 n_exc.QosPolicyNotFound,
 197                 self.qos_plugin.create_policy_bandwidth_limit_rule,
 198                 self.ctxt, self.policy.id, self.rule_data)
 199
 200     def test_update_policy_rule_for_nonexistent_policy(self):
 201         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
 202                         return_value=None):
 203             self.assertRaises(
 204                 n_exc.QosPolicyNotFound,
 205                 self.qos_plugin.update_policy_bandwidth_limit_rule,
 206                 self.ctxt, self.rule.id, self.policy.id, self.rule_data)
 207
 208     def test_delete_policy_rule_for_nonexistent_policy(self):
 209         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
 210                         return_value=None):
 211             self.assertRaises(
 212                 n_exc.QosPolicyNotFound,
 213                 self.qos_plugin.delete_policy_bandwidth_limit_rule,
 214                 self.ctxt, self.rule.id, self.policy.id)