openstack/python-openstackclient
Code Issues Proposed changes

Merge "Add support for CRUD operations for QoS minimum packet rate rule"

Browse Source
This commit is contained in:
Zuul 2022-07-20 14:13:42 +00:00 committed by Gerrit Code Review
commit b3ae3f5552
5 changed files with 436 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
openstackclient/network/v2/network_qos_rule.py
View File

@ -25,16 +25,20 @@ from openstackclient.network import common
RULE_TYPE_BANDWIDTH_LIMIT = 'bandwidth-limit'
RULE_TYPE_DSCP_MARKING = 'dscp-marking'
RULE_TYPE_MINIMUM_BANDWIDTH = 'minimum-bandwidth'
RULE_TYPE_MINIMUM_PACKET_RATE = 'minimum-packet-rate'
MANDATORY_PARAMETERS = {
RULE_TYPE_MINIMUM_BANDWIDTH: {'min_kbps', 'direction'},
RULE_TYPE_MINIMUM_PACKET_RATE: {'min_kpps', 'direction'},
RULE_TYPE_DSCP_MARKING: {'dscp_mark'},
RULE_TYPE_BANDWIDTH_LIMIT: {'max_kbps'}}
OPTIONAL_PARAMETERS = {
RULE_TYPE_MINIMUM_BANDWIDTH: set(),
RULE_TYPE_MINIMUM_PACKET_RATE: set(),
RULE_TYPE_DSCP_MARKING: set(),
RULE_TYPE_BANDWIDTH_LIMIT: {'direction', 'max_burst_kbps'}}
DIRECTION_EGRESS = 'egress'
DIRECTION_INGRESS = 'ingress'
DIRECTION_ANY = 'any'
DSCP_VALID_MARKS = [0, 8, 10, 12, 14, 16, 18, 20, 22, 24, 26, 28, 30, 32,
34, 36, 38, 40, 46, 48, 56]
@ -98,10 +102,20 @@ def _get_attrs(network_client, parsed_args, is_create=False):
attrs['dscp_mark'] = parsed_args.dscp_mark
if parsed_args.min_kbps is not None:
attrs['min_kbps'] = parsed_args.min_kbps
if parsed_args.min_kpps is not None:
attrs['min_kpps'] = parsed_args.min_kpps
if parsed_args.ingress:
attrs['direction'] = 'ingress'
attrs['direction'] = DIRECTION_INGRESS
if parsed_args.egress:
attrs['direction'] = 'egress'
attrs['direction'] = DIRECTION_EGRESS
if parsed_args.any:
if rule_type == RULE_TYPE_MINIMUM_PACKET_RATE:
attrs['direction'] = DIRECTION_ANY
else:
msg = (_('Direction "any" can only be used with '
'%(rule_type_min_pps)s rule type') %
{'rule_type_min_pps': RULE_TYPE_MINIMUM_PACKET_RATE})
raise exceptions.CommandError(msg)
_check_type_parameters(attrs, rule_type, is_create)
return attrs
@ -160,6 +174,13 @@ def _add_rule_arguments(parser):
type=int,
help=_('Minimum guaranteed bandwidth in kbps')
)
parser.add_argument(
'--min-kpps',
dest='min_kpps',
metavar='<min-kpps>',
type=int,
help=_('Minimum guaranteed packet rate in kpps')
)
direction_group = parser.add_mutually_exclusive_group()
direction_group.add_argument(
'--ingress',
@ -171,6 +192,12 @@ def _add_rule_arguments(parser):
action='store_true',
help=_("Egress traffic direction from the project point of view")
)
direction_group.add_argument(
'--any',
action='store_true',
help=_("Any traffic direction from the project point of view. Can be "
"used only with minimum packet rate rule.")
)
class CreateNetworkQosRule(command.ShowOne,
@ -190,6 +217,7 @@ class CreateNetworkQosRule(command.ShowOne,
metavar='<type>',
required=True,
choices=[RULE_TYPE_MINIMUM_BANDWIDTH,
RULE_TYPE_MINIMUM_PACKET_RATE,
RULE_TYPE_DSCP_MARKING,
RULE_TYPE_BANDWIDTH_LIMIT],
help=(_('QoS rule type (%s)') %
@ -200,10 +228,10 @@ class CreateNetworkQosRule(command.ShowOne,
def take_action(self, parsed_args):
network_client = self.app.client_manager.network
try:
attrs = _get_attrs(network_client, parsed_args, is_create=True)
attrs.update(
self._parse_extra_properties(parsed_args.extra_properties))
try:
obj = _rule_action_call(
network_client, ACTION_CREATE, parsed_args.type)(
attrs.pop('qos_policy_id'), **attrs)
@ -270,6 +298,7 @@ class ListNetworkQosRule(command.Lister):
'max_kbps',
'max_burst_kbps',
'min_kbps',
'min_kpps',
'dscp_mark',
'direction',
)
@ -280,6 +309,7 @@ class ListNetworkQosRule(command.Lister):
'Max Kbps',
'Max Burst Kbits',
'Min Kbps',
'Min Kpps',
'DSCP mark',
'Direction',
)

67
openstackclient/tests/functional/network/v2/test_network_qos_rule.py
View File

@ -85,6 +85,73 @@ class NetworkQosRuleTestsMinimumBandwidth(common.NetworkTests):
self.assertEqual(7500, cmd_output['min_kbps'])
class NetworkQosRuleTestsMinimumPacketRate(common.NetworkTests):
"""Functional tests for QoS minimum packet rate rule"""
def setUp(self):
super(NetworkQosRuleTestsMinimumPacketRate, self).setUp()
# Nothing in this class works with Nova Network
if not self.haz_network:
self.skipTest("No Network service present")
self.QOS_POLICY_NAME = 'qos_policy_%s' % uuid.uuid4().hex
self.openstack(
'network qos policy create %s' % self.QOS_POLICY_NAME
)
self.addCleanup(self.openstack,
'network qos policy delete %s' % self.QOS_POLICY_NAME)
cmd_output = json.loads(self.openstack(
'network qos rule create -f json '
'--type minimum-packet-rate '
'--min-kpps 2800 '
'--egress %s' %
self.QOS_POLICY_NAME
))
self.RULE_ID = cmd_output['id']
self.addCleanup(self.openstack,
'network qos rule delete %s %s' %
(self.QOS_POLICY_NAME, self.RULE_ID))
self.assertTrue(self.RULE_ID)
def test_qos_rule_create_delete(self):
# This is to check the output of qos rule delete
policy_name = uuid.uuid4().hex
self.openstack('network qos policy create -f json %s' % policy_name)
self.addCleanup(self.openstack,
'network qos policy delete %s' % policy_name)
rule = json.loads(self.openstack(
'network qos rule create -f json '
'--type minimum-packet-rate '
'--min-kpps 2800 '
'--egress %s' % policy_name
))
raw_output = self.openstack(
'network qos rule delete %s %s' %
(policy_name, rule['id']))
self.assertEqual('', raw_output)
def test_qos_rule_list(self):
cmd_output = json.loads(self.openstack(
'network qos rule list -f json %s' % self.QOS_POLICY_NAME))
self.assertIn(self.RULE_ID, [rule['ID'] for rule in cmd_output])
def test_qos_rule_show(self):
cmd_output = json.loads(self.openstack(
'network qos rule show -f json %s %s' %
(self.QOS_POLICY_NAME, self.RULE_ID)))
self.assertEqual(self.RULE_ID, cmd_output['id'])
def test_qos_rule_set(self):
self.openstack('network qos rule set --min-kpps 7500 --any %s %s' %
(self.QOS_POLICY_NAME, self.RULE_ID))
cmd_output = json.loads(self.openstack(
'network qos rule show -f json %s %s' %
(self.QOS_POLICY_NAME, self.RULE_ID)))
self.assertEqual(7500, cmd_output['min_kpps'])
self.assertEqual('any', cmd_output['direction'])
class NetworkQosRuleTestsDSCPMarking(common.NetworkTests):
"""Functional tests for QoS DSCP marking rule"""

7
openstackclient/tests/unit/network/v2/fakes.py
View File

@ -58,9 +58,11 @@ QUOTA = {
RULE_TYPE_BANDWIDTH_LIMIT = 'bandwidth-limit'
RULE_TYPE_DSCP_MARKING = 'dscp-marking'
RULE_TYPE_MINIMUM_BANDWIDTH = 'minimum-bandwidth'
RULE_TYPE_MINIMUM_PACKET_RATE = 'minimum-packet-rate'
VALID_QOS_RULES = [RULE_TYPE_BANDWIDTH_LIMIT,
RULE_TYPE_DSCP_MARKING,
RULE_TYPE_MINIMUM_BANDWIDTH]
RULE_TYPE_MINIMUM_BANDWIDTH,
RULE_TYPE_MINIMUM_PACKET_RATE]
VALID_DSCP_MARKS = [0, 8, 10, 12, 14, 16, 18, 20, 22, 24, 26, 28, 30, 32,
34, 36, 38, 40, 46, 48, 56]
@ -274,6 +276,9 @@ class FakeNetworkQosRule(object):
elif type == RULE_TYPE_MINIMUM_BANDWIDTH:
qos_rule_attrs['min_kbps'] = randint(1, 10000)
qos_rule_attrs['direction'] = 'egress'
elif type == RULE_TYPE_MINIMUM_PACKET_RATE:
qos_rule_attrs['min_kpps'] = randint(1, 10000)
qos_rule_attrs['direction'] = 'egress'
# Overwrite default attributes.
qos_rule_attrs.update(attrs)

327
openstackclient/tests/unit/network/v2/test_network_qos_rule.py
View File

@ -25,6 +25,7 @@ from openstackclient.tests.unit import utils as tests_utils
RULE_TYPE_BANDWIDTH_LIMIT = 'bandwidth-limit'
RULE_TYPE_DSCP_MARKING = 'dscp-marking'
RULE_TYPE_MINIMUM_BANDWIDTH = 'minimum-bandwidth'
RULE_TYPE_MINIMUM_PACKET_RATE = 'minimum-packet-rate'
DSCP_VALID_MARKS = [0, 8, 10, 12, 14, 16, 18, 20, 22, 24, 26, 28, 30, 32,
34, 36, 38, 40, 46, 48, 56]
@ -126,8 +127,101 @@ class TestCreateNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
try:
self.cmd.take_action(parsed_args)
except exceptions.CommandError as e:
msg = ('"Create" rule command for type "minimum-bandwidth" '
'requires arguments: direction, min_kbps')
msg = ('Failed to create Network QoS rule: "Create" rule command '
'for type "minimum-bandwidth" requires arguments: '
'direction, min_kbps')
self.assertEqual(msg, str(e))
class TestCreateNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
def test_check_type_parameters(self):
pass
def setUp(self):
super(TestCreateNetworkQosRuleMinimumPacketRate, self).setUp()
attrs = {'qos_policy_id': self.qos_policy.id,
'type': RULE_TYPE_MINIMUM_PACKET_RATE}
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
attrs)
self.columns = (
'direction',
'id',
'min_kpps',
'project_id',
'qos_policy_id',
'type'
)
self.data = (
self.new_rule.direction,
self.new_rule.id,
self.new_rule.min_kpps,
self.new_rule.project_id,
self.new_rule.qos_policy_id,
self.new_rule.type,
)
self.network.create_qos_minimum_packet_rate_rule = mock.Mock(
return_value=self.new_rule)
# Get the command object to test
self.cmd = network_qos_rule.CreateNetworkQosRule(self.app,
self.namespace)
def test_create_no_options(self):
arglist = []
verifylist = []
# Missing required args should bail here
self.assertRaises(tests_utils.ParserException, self.check_parser,
self.cmd, arglist, verifylist)
def test_create_default_options(self):
arglist = [
'--type', RULE_TYPE_MINIMUM_PACKET_RATE,
'--min-kpps', str(self.new_rule.min_kpps),
'--egress',
self.new_rule.qos_policy_id,
]
verifylist = [
('type', RULE_TYPE_MINIMUM_PACKET_RATE),
('min_kpps', self.new_rule.min_kpps),
('egress', True),
('qos_policy', self.new_rule.qos_policy_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = (self.cmd.take_action(parsed_args))
self.network.create_qos_minimum_packet_rate_rule.\
assert_called_once_with(
self.qos_policy.id,
**{'min_kpps': self.new_rule.min_kpps,
'direction': self.new_rule.direction})
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
def test_create_wrong_options(self):
arglist = [
'--type', RULE_TYPE_MINIMUM_PACKET_RATE,
'--min-kbps', '10000',
self.new_rule.qos_policy_id,
]
verifylist = [
('type', RULE_TYPE_MINIMUM_PACKET_RATE),
('min_kbps', 10000),
('qos_policy', self.new_rule.qos_policy_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
try:
self.cmd.take_action(parsed_args)
except exceptions.CommandError as e:
msg = ('Failed to create Network QoS rule: "Create" rule command '
'for type "minimum-packet-rate" requires arguments: '
'direction, min_kpps')
self.assertEqual(msg, str(e))
@ -212,8 +306,8 @@ class TestCreateNetworkQosRuleDSCPMarking(TestNetworkQosRule):
try:
self.cmd.take_action(parsed_args)
except exceptions.CommandError as e:
msg = ('"Create" rule command for type "dscp-marking" '
'requires arguments: dscp_mark')
msg = ('Failed to create Network QoS rule: "Create" rule command '
'for type "dscp-marking" requires arguments: dscp_mark')
self.assertEqual(msg, str(e))
@ -351,8 +445,8 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
try:
self.cmd.take_action(parsed_args)
except exceptions.CommandError as e:
msg = ('"Create" rule command for type "bandwidth-limit" '
'requires arguments: max_kbps')
msg = ('Failed to create Network QoS rule: "Create" rule command '
'for type "bandwidth-limit" requires arguments: max_kbps')
self.assertEqual(msg, str(e))
@ -415,6 +509,65 @@ class TestDeleteNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
self.assertEqual(msg, str(e))
class TestDeleteNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
def setUp(self):
super(TestDeleteNetworkQosRuleMinimumPacketRate, self).setUp()
attrs = {'qos_policy_id': self.qos_policy.id,
'type': RULE_TYPE_MINIMUM_PACKET_RATE}
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
attrs)
self.qos_policy.rules = [self.new_rule]
self.network.delete_qos_minimum_packet_rate_rule = mock.Mock(
return_value=None)
self.network.find_qos_minimum_packet_rate_rule = (
network_fakes.FakeNetworkQosRule.get_qos_rules(
qos_rules=self.new_rule)
)
# Get the command object to test
self.cmd = network_qos_rule.DeleteNetworkQosRule(self.app,
self.namespace)
def test_qos_policy_delete(self):
arglist = [
self.new_rule.qos_policy_id,
self.new_rule.id,
]
verifylist = [
('qos_policy', self.new_rule.qos_policy_id),
('id', self.new_rule.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.network.find_qos_policy.assert_called_once_with(
self.qos_policy.id, ignore_missing=False)
self.network.delete_qos_minimum_packet_rate_rule.\
assert_called_once_with(self.new_rule.id, self.qos_policy.id)
self.assertIsNone(result)
def test_qos_policy_delete_error(self):
arglist = [
self.new_rule.qos_policy_id,
self.new_rule.id,
]
verifylist = [
('qos_policy', self.new_rule.qos_policy_id),
('id', self.new_rule.id),
]
self.network.delete_qos_minimum_packet_rate_rule.side_effect = \
Exception('Error message')
try:
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
except exceptions.CommandError as e:
msg = ('Failed to delete Network QoS rule ID "%(rule)s": %(e)s' %
{'rule': self.new_rule.id, 'e': 'Error message'})
self.assertEqual(msg, str(e))
class TestDeleteNetworkQosRuleDSCPMarking(TestNetworkQosRule):
def setUp(self):
@ -627,6 +780,100 @@ class TestSetNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
self.assertEqual(msg, str(e))
class TestSetNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
def setUp(self):
super(TestSetNetworkQosRuleMinimumPacketRate, self).setUp()
attrs = {'qos_policy_id': self.qos_policy.id,
'type': RULE_TYPE_MINIMUM_PACKET_RATE}
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
attrs=attrs)
self.qos_policy.rules = [self.new_rule]
self.network.update_qos_minimum_packet_rate_rule = mock.Mock(
return_value=None)
self.network.find_qos_minimum_packet_rate_rule = mock.Mock(
return_value=self.new_rule)
self.network.find_qos_policy = mock.Mock(
return_value=self.qos_policy)
# Get the command object to test
self.cmd = (network_qos_rule.SetNetworkQosRule(self.app,
self.namespace))
def test_set_nothing(self):
arglist = [
self.new_rule.qos_policy_id,
self.new_rule.id,
]
verifylist = [
('qos_policy', self.new_rule.qos_policy_id),
('id', self.new_rule.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.network.update_qos_minimum_packet_rate_rule.assert_called_with(
self.new_rule, self.qos_policy.id)
self.assertIsNone(result)
def test_set_min_kpps(self):
self._set_min_kpps()
def test_set_min_kpps_to_zero(self):
self._set_min_kpps(min_kpps=0)
def _set_min_kpps(self, min_kpps=None):
if min_kpps:
previous_min_kpps = self.new_rule.min_kpps
self.new_rule.min_kpps = min_kpps
arglist = [
'--min-kpps', str(self.new_rule.min_kpps),
self.new_rule.qos_policy_id,
self.new_rule.id,
]
verifylist = [
('min_kpps', self.new_rule.min_kpps),
('qos_policy', self.new_rule.qos_policy_id),
('id', self.new_rule.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
attrs = {
'min_kpps': self.new_rule.min_kpps,
}
self.network.update_qos_minimum_packet_rate_rule.assert_called_with(
self.new_rule, self.qos_policy.id, **attrs)
self.assertIsNone(result)
if min_kpps:
self.new_rule.min_kpps = previous_min_kpps
def test_set_wrong_options(self):
arglist = [
'--min-kbps', str(10000),
self.new_rule.qos_policy_id,
self.new_rule.id,
]
verifylist = [
('min_kbps', 10000),
('qos_policy', self.new_rule.qos_policy_id),
('id', self.new_rule.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
try:
self.cmd.take_action(parsed_args)
except exceptions.CommandError as e:
msg = ('Failed to set Network QoS rule ID "%(rule)s": Rule type '
'"minimum-packet-rate" only requires arguments: direction, '
'min_kpps' % {'rule': self.new_rule.id})
self.assertEqual(msg, str(e))
class TestSetNetworkQosRuleDSCPMarking(TestNetworkQosRule):
def setUp(self):
@ -893,6 +1140,9 @@ class TestListNetworkQosRule(TestNetworkQosRule):
'type': RULE_TYPE_MINIMUM_BANDWIDTH}
self.new_rule_min_bw = (network_fakes.FakeNetworkQosRule.
create_one_qos_rule(attrs=attrs))
attrs['type'] = RULE_TYPE_MINIMUM_PACKET_RATE
self.new_rule_min_pps = (network_fakes.FakeNetworkQosRule.
create_one_qos_rule(attrs=attrs))
attrs['type'] = RULE_TYPE_DSCP_MARKING
self.new_rule_dscp_mark = (network_fakes.FakeNetworkQosRule.
create_one_qos_rule(attrs=attrs))
@ -900,10 +1150,13 @@ class TestListNetworkQosRule(TestNetworkQosRule):
self.new_rule_max_bw = (network_fakes.FakeNetworkQosRule.
create_one_qos_rule(attrs=attrs))
self.qos_policy.rules = [self.new_rule_min_bw,
self.new_rule_min_pps,
self.new_rule_dscp_mark,
self.new_rule_max_bw]
self.network.find_qos_minimum_bandwidth_rule = mock.Mock(
return_value=self.new_rule_min_bw)
self.network.find_qos_minimum_packet_rate_rule = mock.Mock(
return_value=self.new_rule_min_pps)
self.network.find_qos_dscp_marking_rule = mock.Mock(
return_value=self.new_rule_dscp_mark)
self.network.find_qos_bandwidth_limit_rule = mock.Mock(
@ -915,6 +1168,7 @@ class TestListNetworkQosRule(TestNetworkQosRule):
'Max Kbps',
'Max Burst Kbits',
'Min Kbps',
'Min Kpps',
'DSCP mark',
'Direction',
)
@ -927,6 +1181,7 @@ class TestListNetworkQosRule(TestNetworkQosRule):
getattr(self.qos_policy.rules[index], 'max_kbps', ''),
getattr(self.qos_policy.rules[index], 'max_burst_kbps', ''),
getattr(self.qos_policy.rules[index], 'min_kbps', ''),
getattr(self.qos_policy.rules[index], 'min_kpps', ''),
getattr(self.qos_policy.rules[index], 'dscp_mark', ''),
getattr(self.qos_policy.rules[index], 'direction', ''),
))
@ -1014,6 +1269,66 @@ class TestShowNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
self.assertEqual(list(self.data), list(data))
class TestShowNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
def setUp(self):
super(TestShowNetworkQosRuleMinimumPacketRate, self).setUp()
attrs = {'qos_policy_id': self.qos_policy.id,
'type': RULE_TYPE_MINIMUM_PACKET_RATE}
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
attrs)
self.qos_policy.rules = [self.new_rule]
self.columns = (
'direction',
'id',
'min_kpps',
'project_id',
'qos_policy_id',
'type'
)
self.data = (
self.new_rule.direction,
self.new_rule.id,
self.new_rule.min_kpps,
self.new_rule.project_id,
self.new_rule.qos_policy_id,
self.new_rule.type,
)
self.network.get_qos_minimum_packet_rate_rule = mock.Mock(
return_value=self.new_rule)
# Get the command object to test
self.cmd = network_qos_rule.ShowNetworkQosRule(self.app,
self.namespace)
def test_show_no_options(self):
arglist = []
verifylist = []
# Missing required args should bail here
self.assertRaises(tests_utils.ParserException, self.check_parser,
self.cmd, arglist, verifylist)
def test_show_all_options(self):
arglist = [
self.new_rule.qos_policy_id,
self.new_rule.id,
]
verifylist = [
('qos_policy', self.new_rule.qos_policy_id),
('id', self.new_rule.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.network.get_qos_minimum_packet_rate_rule.assert_called_once_with(
self.new_rule.id, self.qos_policy.id)
self.assertEqual(self.columns, columns)
self.assertEqual(list(self.data), list(data))
class TestShowNetworkQosDSCPMarking(TestNetworkQosRule):
def setUp(self):

7
releasenotes/notes/qos-min-pps-rule-bfe22cea1966c4a0.yaml Normal file
View File

@ -0,0 +1,7 @@
---
features:
- |
Add support for QoS minimum packet rate rule to following commands:
``network qos rule create``, ``network qos rule delete``,
``network qos rule list``, ``network qos rule show`` and
``network qos rule set``