openstack
/
neutron
Code Issues Proposed changes

Merge "QoS: fix get bandwidth limit rules to filter them per policy"

This commit is contained in:
Jenkins 2015-08-25 22:04:36 +00:00 committed by Gerrit Code Review
parent 79ee9ad7e9 6063e70927
commit dea4f9fac1
3 changed files with 52 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
neutron/services/qos/qos_plugin.py
View File

@ -23,6 +23,7 @@ from neutron.objects.qos import policy as policy_object
from neutron.objects.qos import rule as rule_object
from neutron.objects.qos import rule_type as rule_type_object
from neutron.services.qos.notification_drivers import manager as driver_mgr
from neutron.services.qos import qos_consts
LOG = logging.getLogger(__name__)
@ -151,6 +152,8 @@ class QoSPlugin(qos.QoSPluginBase):
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
self._get_policy_obj(context, policy_id)
filters = filters or dict()
filters[qos_consts.QOS_POLICY_ID] = policy_id
return rule_object.QosBandwidthLimitRule.get_objects(context,
**filters)

24
neutron/tests/api/test_qos.py
View File

@ -419,3 +419,27 @@ class QosBandwidthLimitRuleTestJSON(base.BaseAdminNetworkTest):
exceptions.Forbidden,
self.client.create_bandwidth_limit_rule,
'policy', 1, 2)
@test.attr(type='smoke')
@test.idempotent_id('ce0bd0c2-54d9-4e29-85f1-cfb36ac3ebe2')
def test_get_rules_by_policy(self):
policy1 = self.create_qos_policy(name='test-policy1',
description='test policy1',
shared=False)
rule1 = self.create_qos_bandwidth_limit_rule(policy_id=policy1['id'],
max_kbps=200,
max_burst_kbps=1337)
policy2 = self.create_qos_policy(name='test-policy2',
description='test policy2',
shared=False)
rule2 = self.create_qos_bandwidth_limit_rule(policy_id=policy2['id'],
max_kbps=5000,
max_burst_kbps=2523)
# Test 'list rules'
rules = self.admin_client.list_bandwidth_limit_rules(policy1['id'])
rules = rules['bandwidth_limit_rules']
rules_ids = [r['id'] for r in rules]
self.assertIn(rule1['id'], rules_ids)
self.assertNotIn(rule2['id'], rules_ids)

25
neutron/tests/unit/services/qos/test_qos_plugin.py
View File

@ -111,6 +111,31 @@ class TestQosPlugin(base.BaseQosTestCase):
self.ctxt, self.rule.id, self.policy.id)
self._validate_notif_driver_params('update_policy')
def test_get_policy_bandwidth_limit_rules_for_policy(self):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
return_value=self.policy):
with mock.patch('neutron.objects.qos.rule.'
'QosBandwidthLimitRule.'
'get_objects') as get_object_mock:
self.qos_plugin.get_policy_bandwidth_limit_rules(
self.ctxt, self.policy.id)
get_object_mock.assert_called_once_with(
self.ctxt, qos_policy_id=self.policy.id)
def test_get_policy_bandwidth_limit_rules_for_policy_with_filters(self):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
return_value=self.policy):
with mock.patch('neutron.objects.qos.rule.'
'QosBandwidthLimitRule.'
'get_objects') as get_object_mock:
filters = {'filter': 'filter_id'}
self.qos_plugin.get_policy_bandwidth_limit_rules(
self.ctxt, self.policy.id, filters=filters)
get_object_mock.assert_called_once_with(
self.ctxt, qos_policy_id=self.policy.id,
filter='filter_id')
def test_get_policy_for_nonexistent_policy(self):
with mock.patch('neutron.objects.qos.policy.QosPolicy.get_by_id',
return_value=None):