neutron/tests/functional/agent/l2/extensions/test_ovs_agent_qos_extension.py

   1 # Copyright (c) 2015 Red Hat, Inc.
   2 # All Rights Reserved.
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15
  16 import copy
  17
  18 import mock
  19 from oslo_utils import uuidutils
  20
  21 from neutron.api.rpc.callbacks.consumer import registry as consumer_reg
  22 from neutron.api.rpc.callbacks import events
  23 from neutron.api.rpc.callbacks import resources
  24 from neutron.objects.qos import policy
  25 from neutron.objects.qos import rule
  26 from neutron.tests.common.agents import l2_extensions
  27 from neutron.tests.functional.agent.l2 import base
  28
  29
  30 TEST_POLICY_ID1 = "a2d72369-4246-4f19-bd3c-af51ec8d70cd"
  31 TEST_POLICY_ID2 = "46ebaec0-0570-43ac-82f6-60d2b03168c5"
  32 TEST_BW_LIMIT_RULE_1 = rule.QosBandwidthLimitRule(
  33         context=None,
  34         qos_policy_id=TEST_POLICY_ID1,
  35         id="5f126d84-551a-4dcf-bb01-0e9c0df0c793",
  36         max_kbps=1000,
  37         max_burst_kbps=10)
  38 TEST_BW_LIMIT_RULE_2 = rule.QosBandwidthLimitRule(
  39         context=None,
  40         qos_policy_id=TEST_POLICY_ID2,
  41         id="fa9128d9-44af-49b2-99bb-96548378ad42",
  42         max_kbps=900,
  43         max_burst_kbps=9)
  44
  45
  46 class OVSAgentQoSExtensionTestFramework(base.OVSAgentTestFramework):
  47     def setUp(self):
  48         super(OVSAgentQoSExtensionTestFramework, self).setUp()
  49         self.config.set_override('extensions', ['qos'], 'agent')
  50         self._set_pull_mock()
  51         self.set_test_qos_rules(TEST_POLICY_ID1, [TEST_BW_LIMIT_RULE_1])
  52         self.set_test_qos_rules(TEST_POLICY_ID2, [TEST_BW_LIMIT_RULE_2])
  53
  54     def _set_pull_mock(self):
  55
  56         self.qos_policies = {}
  57
  58         def _pull_mock(context, resource_type, resource_id):
  59             return self.qos_policies[resource_id]
  60
  61         self.pull = mock.patch(
  62             'neutron.api.rpc.handlers.resources_rpc.'
  63             'ResourcesPullRpcApi.pull').start()
  64         self.pull.side_effect = _pull_mock
  65
  66     def set_test_qos_rules(self, policy_id, policy_rules):
  67         """This function sets the policy test rules to be exposed."""
  68
  69         qos_policy = policy.QosPolicy(
  70             context=None,
  71             tenant_id=uuidutils.generate_uuid(),
  72             id=policy_id,
  73             name="Test Policy Name",
  74             description="This is a policy for testing purposes",
  75             shared=False,
  76             rules=policy_rules)
  77
  78         qos_policy.obj_reset_changes()
  79         self.qos_policies[policy_id] = qos_policy
  80
  81     def _create_test_port_dict(self, policy_id=None):
  82         port_dict = super(OVSAgentQoSExtensionTestFramework,
  83                           self)._create_test_port_dict()
  84         port_dict['qos_policy_id'] = policy_id
  85         port_dict['network_qos_policy_id'] = None
  86         return port_dict
  87
  88     def _get_device_details(self, port, network):
  89         dev = super(OVSAgentQoSExtensionTestFramework,
  90                     self)._get_device_details(port, network)
  91         dev['qos_policy_id'] = port['qos_policy_id']
  92         return dev
  93
  94     def _assert_bandwidth_limit_rule_is_set(self, port, rule):
  95         max_rate, burst = (
  96             self.agent.int_br.get_egress_bw_limit_for_port(port['vif_name']))
  97         self.assertEqual(max_rate, rule.max_kbps)
  98         self.assertEqual(burst, rule.max_burst_kbps)
  99
 100     def _assert_bandwidth_limit_rule_not_set(self, port):
 101         max_rate, burst = (
 102             self.agent.int_br.get_egress_bw_limit_for_port(port['vif_name']))
 103         self.assertIsNone(max_rate)
 104         self.assertIsNone(burst)
 105
 106     def wait_until_bandwidth_limit_rule_applied(self, port, rule):
 107         l2_extensions.wait_until_bandwidth_limit_rule_applied(
 108             self.agent.int_br, port['vif_name'], rule)
 109
 110     def _create_port_with_qos(self):
 111         port_dict = self._create_test_port_dict()
 112         port_dict['qos_policy_id'] = TEST_POLICY_ID1
 113         self.setup_agent_and_ports([port_dict])
 114         self.wait_until_ports_state(self.ports, up=True)
 115         self.wait_until_bandwidth_limit_rule_applied(port_dict,
 116                                                      TEST_BW_LIMIT_RULE_1)
 117         return port_dict
 118
 119
 120 class TestOVSAgentQosExtension(OVSAgentQoSExtensionTestFramework):
 121
 122     def test_port_creation_with_bandwidth_limit(self):
 123         """Make sure bandwidth limit rules are set in low level to ports."""
 124
 125         self.setup_agent_and_ports(
 126             port_dicts=self.create_test_ports(amount=1,
 127                                               policy_id=TEST_POLICY_ID1))
 128         self.wait_until_ports_state(self.ports, up=True)
 129
 130         for port in self.ports:
 131             self._assert_bandwidth_limit_rule_is_set(
 132                 port, TEST_BW_LIMIT_RULE_1)
 133
 134     def test_port_creation_with_different_bandwidth_limits(self):
 135         """Make sure different types of policies end on the right ports."""
 136
 137         port_dicts = self.create_test_ports(amount=3)
 138
 139         port_dicts[0]['qos_policy_id'] = TEST_POLICY_ID1
 140         port_dicts[1]['qos_policy_id'] = TEST_POLICY_ID2
 141
 142         self.setup_agent_and_ports(port_dicts)
 143         self.wait_until_ports_state(self.ports, up=True)
 144
 145         self._assert_bandwidth_limit_rule_is_set(self.ports[0],
 146                                                  TEST_BW_LIMIT_RULE_1)
 147
 148         self._assert_bandwidth_limit_rule_is_set(self.ports[1],
 149                                                  TEST_BW_LIMIT_RULE_2)
 150
 151         self._assert_bandwidth_limit_rule_not_set(self.ports[2])
 152
 153     def test_simple_port_policy_update(self):
 154         self.setup_agent_and_ports(
 155             port_dicts=self.create_test_ports(amount=1,
 156                                               policy_id=TEST_POLICY_ID1))
 157         self.wait_until_ports_state(self.ports, up=True)
 158         policy_copy = copy.deepcopy(self.qos_policies[TEST_POLICY_ID1])
 159         policy_copy.rules[0].max_kbps = 500
 160         policy_copy.rules[0].max_burst_kbps = 5
 161         consumer_reg.push(resources.QOS_POLICY, policy_copy, events.UPDATED)
 162         self.wait_until_bandwidth_limit_rule_applied(self.ports[0],
 163                                                      policy_copy.rules[0])
 164         self._assert_bandwidth_limit_rule_is_set(self.ports[0],
 165                                                  policy_copy.rules[0])
 166
 167     def test_port_qos_disassociation(self):
 168         """Test that qos_policy_id set to None will remove all qos rules from
 169            given port.
 170         """
 171         port_dict = self._create_port_with_qos()
 172
 173         port_dict['qos_policy_id'] = None
 174         self.agent.port_update(None, port=port_dict)
 175
 176         self.wait_until_bandwidth_limit_rule_applied(port_dict, None)
 177
 178     def test_port_qos_update_policy_id(self):
 179         """Test that change of qos policy id on given port refreshes all its
 180            rules.
 181         """
 182         port_dict = self._create_port_with_qos()
 183
 184         port_dict['qos_policy_id'] = TEST_POLICY_ID2
 185         self.agent.port_update(None, port=port_dict)
 186
 187         self.wait_until_bandwidth_limit_rule_applied(port_dict,
 188                                                      TEST_BW_LIMIT_RULE_2)
 189
 190     def test_policy_rule_delete(self):
 191         port_dict = self._create_port_with_qos()
 192
 193         policy_copy = copy.deepcopy(self.qos_policies[TEST_POLICY_ID1])
 194         policy_copy.rules = list()
 195         consumer_reg.push(resources.QOS_POLICY, policy_copy, events.UPDATED)
 196
 197         self.wait_until_bandwidth_limit_rule_applied(port_dict, None)