openstack/horizon
Code Issues Proposed changes

Merge "SDK for Neutron QoS"

Browse Source
This commit is contained in:
Zuul
2025-07-02 23:56:06 +00:00
committed by Gerrit Code Review
parent 8b5c5e2975 bfcec735ae
commit b605bccd97
5 changed files with 207 additions and 235 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
openstack_dashboard/api/neutron.py
View File

@@ -2355,14 +2355,12 @@ def list_extensions(request):
:param request: django request object
"""
neutron_api = neutronclient(request)
neutron_api = networkclient(request)
try:
extensions_list = neutron_api.list_extensions()
extensions_list = neutron_api.extensions()
except exceptions.ServiceCatalogException:
return {}
if 'extensions' in extensions_list:
return tuple(extensions_list['extensions'])
return ()
return tuple(extensions_list)
@profiler.trace
@@ -2534,30 +2532,27 @@ def policy_create(request, **kwargs):
:param shared: boolean (true or false)
:return: QoSPolicy object
"""
body = {'policy': kwargs}
policy = neutronclient(request).create_qos_policy(body=body).get('policy')
return QoSPolicy(policy)
policy = networkclient(request).create_qos_policy(**kwargs)
return QoSPolicy(policy.to_dict())
def policy_list(request, **kwargs):
"""List of QoS Policies."""
policies = neutronclient(request).list_qos_policies(
**kwargs).get('policies')
return [QoSPolicy(p) for p in policies]
policies = networkclient(request).qos_policies(**kwargs)
return [QoSPolicy(p.to_dict()) for p in policies]
@profiler.trace
def policy_get(request, policy_id, **kwargs):
"""Get QoS policy for a given policy id."""
policy = neutronclient(request).show_qos_policy(
policy_id, **kwargs).get('policy')
return QoSPolicy(policy)
policy = networkclient(request).get_qos_policy(policy_id)
return QoSPolicy(policy.to_dict())
@profiler.trace
def policy_delete(request, policy_id):
"""Delete QoS policy for a given policy id."""
neutronclient(request).delete_qos_policy(policy_id)
networkclient(request).delete_qos_policy(policy_id)
class DSCPMarkingRule(NeutronAPIDictWrapper):
@@ -2573,28 +2568,25 @@ def dscp_marking_rule_create(request, policy_id, **kwargs):
:param dscp_mark: integer
:return: A dscp_mark_rule object.
"""
body = {'dscp_marking_rule': kwargs}
rule = 'dscp_marking_rule'
dscp_marking_rule = neutronclient(request)\
.create_dscp_marking_rule(policy_id, body).get(rule)
return DSCPMarkingRule(dscp_marking_rule)
dscp_marking_rule = networkclient(request).create_qos_dscp_marking_rule(
policy_id, **kwargs)
return DSCPMarkingRule(dscp_marking_rule.to_dict())
@profiler.trace
def dscp_marking_rule_update(request, policy_id, rule_id, **kwargs):
"""Update a DSCP Marking Limit Rule."""
body = {'dscp_marking_rule': kwargs}
ruleType = 'dscp_marking_rule'
dscpmarking_update = neutronclient(request)\
.update_dscp_marking_rule(rule_id, policy_id, body).get(ruleType)
return DSCPMarkingRule(dscpmarking_update)
dscpmarking_update = networkclient(request).update_qos_dscp_marking_rule(
rule_id, policy_id, **kwargs)
return DSCPMarkingRule(dscpmarking_update.to_dict())
def dscp_marking_rule_delete(request, policy_id, rule_id):
"""Deletes a DSCP Marking Rule."""
neutronclient(request).delete_dscp_marking_rule(rule_id, policy_id)
networkclient(request).delete_qos_dscp_marking_rule(
rule_id, policy_id)
class MinimumBandwidthRule(NeutronAPIDictWrapper):
@@ -2611,11 +2603,10 @@ def minimum_bandwidth_rule_create(request, policy_id, **kwargs):
:param direction: string (egress or ingress)
:return: A minimum_bandwidth_rule object.
"""
body = {'minimum_bandwidth_rule': kwargs}
rule = 'minimum_bandwidth_rule'
minimum_bandwidth_rule = neutronclient(request)\
.create_minimum_bandwidth_rule(policy_id, body).get(rule)
return MinimumBandwidthRule(minimum_bandwidth_rule)
minimum_bandwidth_rule = networkclient(
request).create_qos_minimum_bandwidth_rule(
policy_id, **kwargs)
return MinimumBandwidthRule(minimum_bandwidth_rule.to_dict())
@profiler.trace
@@ -2628,18 +2619,16 @@ def minimum_bandwidth_rule_update(request, policy_id, rule_id, **kwargs):
:param direction: string (egress or ingress)
:return: A minimum_bandwidth_rule object.
"""
body = {'minimum_bandwidth_rule': kwargs}
ruleType = 'minimum_bandwidth_rule'
minbandwidth_update = neutronclient(request)\
.update_minimum_bandwidth_rule(rule_id, policy_id, body)\
.get(ruleType)
return MinimumBandwidthRule(minbandwidth_update)
minbandwidth_update = networkclient(
request).update_qos_minimum_bandwidth_rule(
rule_id, policy_id, **kwargs)
return MinimumBandwidthRule(minbandwidth_update.to_dict())
def minimum_bandwidth_rule_delete(request, policy_id, rule_id):
"""Deletes a Minimum Bandwidth Rule."""
neutronclient(request).delete_minimum_bandwidth_rule(rule_id, policy_id)
networkclient(request).delete_qos_minimum_bandwidth_rule(
rule_id, policy_id)
class BandwidthLimitRule(NeutronAPIDictWrapper):
@@ -2657,11 +2646,9 @@ def bandwidth_limit_rule_create(request, policy_id, **kwargs):
:param direction: string (egress or ingress)
:return: A bandwidth_limit_rule object.
"""
body = {'bandwidth_limit_rule': kwargs}
rule = 'bandwidth_limit_rule'
bandwidth_limit_rule = neutronclient(request)\
.create_bandwidth_limit_rule(policy_id, body).get(rule)
return BandwidthLimitRule(bandwidth_limit_rule)
bandwidth_limit_rule = networkclient(
request).create_qos_bandwidth_limit_rule(policy_id, **kwargs)
return BandwidthLimitRule(bandwidth_limit_rule.to_dict())
@profiler.trace
@@ -2675,18 +2662,15 @@ def bandwidth_limit_rule_update(request, policy_id, rule_id, **kwargs):
:param direction: string (egress or ingress)
:return: A bandwidth_limit_rule object.
"""
body = {'bandwidth_limit_rule': kwargs}
ruleType = 'bandwidth_limit_rule'
bandwidthlimit_update = neutronclient(request)\
.update_bandwidth_limit_rule(rule_id, policy_id, body)\
.get(ruleType)
return BandwidthLimitRule(bandwidthlimit_update)
bandwidthlimit_update = networkclient(
request).update_qos_bandwidth_limit_rule(rule_id, policy_id, **kwargs)
return BandwidthLimitRule(bandwidthlimit_update.to_dict())
@profiler.trace
def bandwidth_limit_rule_delete(request, policy_id, rule_id):
"""Deletes a Bandwidth Limit Rule."""
neutronclient(request).delete_bandwidth_limit_rule(rule_id, policy_id)
networkclient(request).delete_qos_bandwidth_limit_rule(rule_id, policy_id)
class MinimumPacketRateRule(NeutronAPIDictWrapper):
@@ -2703,11 +2687,10 @@ def minimum_packet_rate_rule_create(request, policy_id, **kwargs):
:param direction: string (egress or ingress)
:return: A minimum_packet_rate_rule object.
"""
body = {'minimum_packet_rate_rule': kwargs}
rule = 'minimum_packet_rate_rule'
minimum_packet_rate_rule = neutronclient(request)\
.create_minimum_packet_rate_rule(policy_id, body).get(rule)
return MinimumPacketRateRule(minimum_packet_rate_rule)
minimum_packet_rate_rule = networkclient(
request).create_qos_minimum_packet_rate_rule(
policy_id, **kwargs)
return MinimumPacketRateRule(minimum_packet_rate_rule.to_dict())
@profiler.trace
@@ -2720,17 +2703,16 @@ def minimum_packet_rate_rule_update(request, policy_id, rule_id, **kwargs):
:param direction: string (egress or ingress)
:return: A minimum_packet_rate_rule object.
"""
body = {'minimum_packet_rate_rule': kwargs}
ruleType = 'minimum_packet_rate_rule'
minpacketrate_update = neutronclient(request)\
.update_minimum_packet_rate_rule(rule_id, policy_id, body)\
.get(ruleType)
minpacketrate_update = networkclient(
request).update_qos_minimum_packet_rate_rule(
rule_id, policy_id, **kwargs)
return MinimumPacketRateRule(minpacketrate_update)
def minimum_packet_rate_rule_delete(request, policy_id, rule_id):
"""Deletes a Minimum Packet Rate Rule."""
neutronclient(request).delete_minimum_packet_rate_rule(rule_id, policy_id)
networkclient(request).delete_qos_minimum_packet_rate_rule(
rule_id, policy_id)
@profiler.trace

2
openstack_dashboard/test/selenium/ui/test_actions.py
View File

@@ -36,7 +36,7 @@ def test_delete_multiple_instance_rows(live_server, driver, dashboard_data,
mocked_i_l_d.return_value = [dashboard_data.images.list()]
mocked_f_l.return_value = dashboard_data.flavors.list()
mocked_l_e.return_value = {
'extensions': dashboard_data.api_extensions.list()}
'extensions': dashboard_data.api_extensions_sdk}
mocked_t_a_l.return_value = dashboard_data.limits['absolute']
novaclient = mock_novaclient.return_value
novaclient.versions.get_current.return_value = "2.0"

52
openstack_dashboard/test/test_data/neutron_data.py
View File

@@ -15,12 +15,19 @@
import copy
from openstack.network.v2 import agent as sdk_agent
from openstack.network.v2 import extension
from openstack.network.v2 import floating_ip as sdk_fip
from openstack.network.v2 import network as sdk_net
from openstack.network.v2 import network_ip_availability as \
sdk_ip_availability
from openstack.network.v2 import port as sdk_port
from openstack.network.v2 import port_forwarding as sdk_pf
from openstack.network.v2 import qos_bandwidth_limit_rule
from openstack.network.v2 import qos_dscp_marking_rule
from openstack.network.v2 import qos_minimum_bandwidth_rule
from openstack.network.v2 import qos_minimum_packet_rate_rule
from openstack.network.v2 import qos_policy
from openstack.network.v2 import router as sdk_router
from openstack.network.v2 import security_group as sdk_sec_group
from openstack.network.v2 import security_group_rule as sdk_sec_group_rule
@@ -72,13 +79,7 @@ def data(TEST):
TEST.api_vips = utils.TestDataContainer()
TEST.api_members = utils.TestDataContainer()
TEST.api_monitors = utils.TestDataContainer()
TEST.api_extensions = utils.TestDataContainer()
TEST.api_rbac_policies = utils.TestDataContainer()
TEST.api_qos_policies = utils.TestDataContainer()
TEST.api_dscp_mark_rule = utils.TestDataContainer()
TEST.api_bandwidth_limit_rule = utils.TestDataContainer()
TEST.api_minimum_bandwidth_rule = utils.TestDataContainer()
TEST.api_minimum_packet_rate_rule = utils.TestDataContainer()
# Data returned by SDK:
TEST.api_networks_sdk = list()
@@ -96,6 +97,12 @@ def data(TEST):
TEST.api_floating_ips_sdk = list()
TEST.api_security_groups_sdk = list()
TEST.api_security_group_rules_sdk = list()
TEST.api_qos_policies_sdk = list()
TEST.api_dscp_mark_rule_sdk = list()
TEST.api_bandwidth_limit_rule_sdk = list()
TEST.api_minimum_bandwidth_rule_sdk = list()
TEST.api_minimum_packet_rate_rule_sdk = list()
TEST.api_extensions_sdk = list()
# 1st network.
network_dict = {'is_admin_state_up': True,
@@ -1000,13 +1007,13 @@ def data(TEST):
"alias": "security-groups-shared-filtering",
"description": "Support filtering security groups on "
"the shared field"}
TEST.api_extensions.add(extension_1)
TEST.api_extensions.add(extension_2)
TEST.api_extensions.add(extension_3)
TEST.api_extensions.add(extension_4)
TEST.api_extensions.add(extension_5)
TEST.api_extensions.add(extension_6)
TEST.api_extensions.add(extension_7)
TEST.api_extensions_sdk.append(extension.Extension(**extension_1))
TEST.api_extensions_sdk.append(extension.Extension(**extension_2))
TEST.api_extensions_sdk.append(extension.Extension(**extension_3))
TEST.api_extensions_sdk.append(extension.Extension(**extension_4))
TEST.api_extensions_sdk.append(extension.Extension(**extension_5))
TEST.api_extensions_sdk.append(extension.Extension(**extension_6))
TEST.api_extensions_sdk.append(extension.Extension(**extension_7))
# 1st agent.
agent_dict = {"binary": "neutron-openvswitch-agent",
@@ -1102,12 +1109,12 @@ def data(TEST):
policy_dict = {'id': 'a21dcd22-7189-cccc-aa32-22adafaf16a7',
'name': 'policy 1',
'tenant_id': '1'}
TEST.api_qos_policies.add(policy_dict)
TEST.api_qos_policies_sdk.append(qos_policy.QoSPolicy(**policy_dict))
TEST.qos_policies.add(neutron.QoSPolicy(policy_dict))
policy_dict1 = {'id': 'a21dcd22-7189-ssss-aa32-22adafaf16a7',
'name': 'policy 2',
'tenant_id': '1'}
TEST.api_qos_policies.add(policy_dict1)
TEST.api_qos_policies_sdk.append(qos_policy.QoSPolicy(**policy_dict1))
TEST.qos_policies.add(neutron.QoSPolicy(policy_dict1))
# qos rule - dscp mark
@@ -1115,7 +1122,8 @@ def data(TEST):
"id": "5f126d84-551a-4dcf-bb01-0e9c0df0c794",
"dscp_mark": 26
}
TEST.api_dscp_mark_rule.add(dscp_mark_rule_dict)
TEST.api_dscp_mark_rule_sdk.append(
qos_dscp_marking_rule.QoSDSCPMarkingRule(**dscp_mark_rule_dict))
TEST.dscp_mark_rule.add(neutron.DSCPMarkingRule(dscp_mark_rule_dict))
# qos rule - bandwidth limit
@@ -1125,7 +1133,9 @@ def data(TEST):
"max_burst_kbps": 0,
"direction": "egress"
}
TEST.api_bandwidth_limit_rule.add(bandwidth_limit_rule)
TEST.api_bandwidth_limit_rule_sdk.append(
qos_bandwidth_limit_rule.QoSBandwidthLimitRule(
**bandwidth_limit_rule))
TEST.bandwidth_limit_rule.add(neutron.BandwidthLimitRule(
bandwidth_limit_rule))
@@ -1135,7 +1145,9 @@ def data(TEST):
"min_kbps": 10000,
"direction": "egress"
}
TEST.api_minimum_bandwidth_rule.add(minimum_bandwidth_rule)
TEST.api_minimum_bandwidth_rule_sdk.append(
qos_minimum_bandwidth_rule.QoSMinimumBandwidthRule(
**minimum_bandwidth_rule))
TEST.minimum_bandwidth_rule.add(neutron.MinimumBandwidthRule(
minimum_bandwidth_rule))
@@ -1145,7 +1157,9 @@ def data(TEST):
"min_kpps": 10000,
"direction": "egress"
}
TEST.api_minimum_packet_rate_rule.add(minimum_packet_rate_rule)
TEST.api_minimum_packet_rate_rule_sdk.append(
qos_minimum_packet_rate_rule.QoSMinimumPacketRateRule(
**minimum_packet_rate_rule))
TEST.minimum_packet_rate_rule.add(neutron.MinimumPacketRateRule(
minimum_packet_rate_rule))

6
openstack_dashboard/test/unit/api/rest/test_neutron.py
View File

@@ -223,10 +223,12 @@ class NeutronExtensionsTestCase(test.RestAPITestCase):
@mock.patch.object(api.neutron, 'list_extensions')
def test_list_extensions(self, mock_list_extensions):
request = self.mock_rest_request(**{'GET': {}})
mock_list_extensions.return_value = self.api_extensions.list()
mock_list_extensions.return_value = self.api_extensions_sdk
response = neutron.Extensions().get(request)
self.assertStatusCode(response, 200)
self.assertItemsCollectionEqual(response, self.api_extensions.list())
self.assertItemsCollectionEqual(
response,
[ext.to_dict() for ext in self.api_extensions_sdk])
mock_list_extensions.assert_called_once_with(request)

276
openstack_dashboard/test/unit/api/test_neutron.py
View File

@@ -1562,19 +1562,18 @@ class NeutronApiTests(test.APIMockTestCase):
networkclient.remove_interface_from_router.assert_called_once_with(
router=router_id, port_id=fake_port, subnet_id=None)
# Mocking neutronclient() does not work because api.neutron.list_extensions
# is decorated with memoized_with_request, so we need to mock
# neutronclient.v2_0.client directly.
@mock.patch('neutronclient.v2_0.client.Client.list_extensions')
def test_is_extension_supported(self, mock_list_extensions):
extensions = self.api_extensions.list()
mock_list_extensions.return_value = {'extensions': extensions}
@mock.patch.object(api.neutron, 'networkclient')
def test_is_extension_supported(self, mock_networkclient):
extensions = self.api_extensions_sdk
networkclient = mock_networkclient.return_value
networkclient.extensions.return_value = extensions
self.assertTrue(
api.neutron.is_extension_supported(self.request, 'quotas'))
self.assertFalse(
api.neutron.is_extension_supported(self.request, 'doesntexist'))
mock_list_extensions.assert_called_once_with()
networkclient.extensions.assert_called_once_with()
@mock.patch.object(api.neutron, 'networkclient')
def test_router_static_route_list(self, mock_networkclient):
@@ -1775,74 +1774,65 @@ class NeutronApiTests(test.APIMockTestCase):
expected_calls.append(mock.call(id=tuple(port_ids[i:i + 4])))
network_client.ports.assert_has_calls(expected_calls)
@mock.patch.object(api.neutron, 'neutronclient')
def test_qos_policies_list(self, mock_neutronclient):
@mock.patch.object(api.neutron, 'networkclient')
def test_qos_policies_list(self, mock_networkclient):
exp_policies = self.qos_policies.list()
api_qos_policies = {'policies': self.api_qos_policies.list()}
api_qos_policies = self.api_qos_policies_sdk
neutronclient = mock_neutronclient.return_value
neutronclient.list_qos_policies.return_value = api_qos_policies
networkclient = mock_networkclient.return_value
networkclient.qos_policies.return_value = api_qos_policies
ret_val = api.neutron.policy_list(self.request)
self.assertEqual(len(ret_val), len(exp_policies))
self.assertIsInstance(ret_val[0], api.neutron.QoSPolicy)
self.assertEqual(exp_policies[0].name, ret_val[0].name)
neutronclient.list_qos_policies.assert_called_once_with()
networkclient.qos_policies.assert_called_once_with()
@mock.patch.object(api.neutron, 'neutronclient')
def test_qos_policy_create(self, mock_neutronclient):
qos_policy = self.api_qos_policies.first()
post_data = {'policy': {'name': qos_policy['name']}}
@mock.patch.object(api.neutron, 'networkclient')
def test_qos_policy_create(self, mock_networkclient):
qos_policy = self.api_qos_policies_sdk[0]
post_data = {'name': qos_policy['name']}
neutronclient = mock_neutronclient.return_value
neutronclient.create_qos_policy.return_value = {'policy': qos_policy}
networkclient = mock_networkclient.return_value
networkclient.create_qos_policy.return_value = qos_policy
ret_val = api.neutron.policy_create(self.request,
name=qos_policy['name'])
self.assertIsInstance(ret_val, api.neutron.QoSPolicy)
self.assertEqual(qos_policy['name'], ret_val.name)
neutronclient.create_qos_policy.assert_called_once_with(body=post_data)
networkclient.create_qos_policy.assert_called_once_with(**post_data)
@mock.patch.object(api.neutron, 'neutronclient')
def test_dscp_mark_rule_create(self, mock_neutronclient):
qos_policy = self.api_qos_policies.first()
dscp_mark_rule = self.api_dscp_mark_rule.first()
post_data = {'dscp_marking_rule': {
"dscp_mark": dscp_mark_rule["dscp_mark"]}
}
@mock.patch.object(api.neutron, 'networkclient')
def test_dscp_mark_rule_create(self, mock_networkclient):
qos_policy = self.api_qos_policies_sdk[0]
dscp_mark_rule = self.api_dscp_mark_rule_sdk[0]
post_data = {'dscp_mark': dscp_mark_rule['dscp_mark']}
neutronclient = mock_neutronclient.return_value
nclient = mock_networkclient.return_value
neutronclient.create_dscp_marking_rule.return_value = {
'dscp_marking_rule': dscp_mark_rule
}
nclient.create_qos_dscp_marking_rule.return_value = dscp_mark_rule
ret_val = api.neutron.dscp_marking_rule_create(
self.request,
policy_id=qos_policy['id'],
dscp_mark=dscp_mark_rule['dscp_mark'])
policy_id=qos_policy['id'], dscp_mark=dscp_mark_rule['dscp_mark'])
self.assertIsInstance(ret_val, api.neutron.DSCPMarkingRule)
self.assertEqual(dscp_mark_rule['dscp_mark'], ret_val.dscp_mark)
neutronclient.create_dscp_marking_rule.assert_called_once_with(
qos_policy['id'], post_data)
self.assertEqual(dscp_mark_rule.dscp_mark, ret_val.dscp_mark)
nclient.create_qos_dscp_marking_rule.assert_called_once_with(
qos_policy['id'], **post_data)
@mock.patch.object(api.neutron, 'neutronclient')
def test_dscp_mark_rule_update(self, mock_neutronclient):
qos_policy = self.api_qos_policies.first()
dscp_mark_rule = self.api_dscp_mark_rule.first()
@mock.patch.object(api.neutron, 'networkclient')
def test_dscp_mark_rule_update(self, mock_networkclient):
qos_policy = self.api_qos_policies_sdk[0]
dscp_mark_rule = self.api_dscp_mark_rule_sdk[0]
dscp_mark_rule["dscp_mark"] = 28
post_data = {'dscp_marking_rule': {
"dscp_mark": 28}
}
post_data = {"dscp_mark": 28}
neutronclient = mock_neutronclient.return_value
nclient = mock_networkclient.return_value
neutronclient.update_dscp_marking_rule.return_value = {
'dscp_marking_rule': dscp_mark_rule
}
nclient.update_qos_dscp_marking_rule.return_value = dscp_mark_rule
ret_val = api.neutron.dscp_marking_rule_update(
self.request,
@@ -1852,39 +1842,34 @@ class NeutronApiTests(test.APIMockTestCase):
self.assertIsInstance(ret_val, api.neutron.DSCPMarkingRule)
self.assertEqual(
dscp_mark_rule['dscp_mark'], ret_val.dscp_mark)
neutronclient.update_dscp_marking_rule.assert_called_once_with(
dscp_mark_rule['id'], qos_policy['id'], post_data)
dscp_mark_rule.dscp_mark, ret_val.dscp_mark)
nclient.update_qos_dscp_marking_rule.assert_called_once_with(
dscp_mark_rule['id'], qos_policy['id'], **post_data)
@mock.patch.object(api.neutron, 'neutronclient')
def test_dscp_mark_rule_delete(self, mock_neutronclient):
qos_policy = self.api_qos_policies.first()
dscp_mark_rule = self.api_dscp_mark_rule.first()
@mock.patch.object(api.neutron, 'networkclient')
def test_dscp_mark_rule_delete(self, mock_networkclient):
qos_policy = self.api_qos_policies_sdk[0]
dscp_mark_rule = self.api_dscp_mark_rule_sdk[0]
neutronclient = mock_neutronclient.return_value
networkclient = mock_networkclient.return_value
neutronclient.delete_dscp_marking_rule.return_value = None
networkclient.delete_dscp_marking_rule.return_value = None
api.neutron.dscp_marking_rule_delete(
self.request, qos_policy['id'], dscp_mark_rule['id'])
neutronclient.delete_dscp_marking_rule.assert_called_once_with(
networkclient.delete_qos_dscp_marking_rule.assert_called_once_with(
dscp_mark_rule['id'], qos_policy['id'])
@mock.patch.object(api.neutron, 'neutronclient')
def test_bandwidth_limit_rule_create(self, mock_neutronclient):
qos_policy = self.api_qos_policies.first()
bwd_limit_rule = self.api_bandwidth_limit_rule.first()
post_data = {
'bandwidth_limit_rule': {
"max_kbps": bwd_limit_rule["max_kbps"]
}
}
@mock.patch.object(api.neutron, 'networkclient')
def test_bandwidth_limit_rule_create(self, mock_networkclient):
qos_policy = self.api_qos_policies_sdk[0]
bwd_limit_rule = self.api_bandwidth_limit_rule_sdk[0]
post_data = {"max_kbps": bwd_limit_rule["max_kbps"]}
neutronclient = mock_neutronclient.return_value
nclient = mock_networkclient.return_value
neutronclient.create_bandwidth_limit_rule.return_value = {
'bandwidth_limit_rule': bwd_limit_rule}
nclient.create_qos_bandwidth_limit_rule.return_value = bwd_limit_rule
ret_val = api.neutron.bandwidth_limit_rule_create(
self.request,
@@ -1893,24 +1878,19 @@ class NeutronApiTests(test.APIMockTestCase):
self.assertIsInstance(ret_val, api.neutron.BandwidthLimitRule)
self.assertEqual(bwd_limit_rule["max_kbps"], ret_val.max_kbps)
neutronclient.create_bandwidth_limit_rule.assert_called_once_with(
qos_policy['id'], post_data)
nclient.create_qos_bandwidth_limit_rule.assert_called_once_with(
qos_policy['id'], **post_data)
@mock.patch.object(api.neutron, 'neutronclient')
def test_bandwidth_limit_rule_update(self, mock_neutronclient):
qos_policy = self.api_qos_policies.first()
bwd_limit_rule = self.api_bandwidth_limit_rule.first()
@mock.patch.object(api.neutron, 'networkclient')
def test_bandwidth_limit_rule_update(self, mock_networkclient):
qos_policy = self.api_qos_policies_sdk[0]
bwd_limit_rule = self.api_bandwidth_limit_rule_sdk[0]
bwd_limit_rule["max_kbps"] = 20000
post_data = {
"bandwidth_limit_rule": {
"max_kbps": 20000
}
}
post_data = {"max_kbps": 20000}
neutronclient = mock_neutronclient.return_value
nclient = mock_networkclient.return_value
neutronclient.update_bandwidth_limit_rule.return_value = {
'bandwidth_limit_rule': bwd_limit_rule}
nclient.update_qos_bandwidth_limit_rule.return_value = bwd_limit_rule
ret_val = api.neutron.bandwidth_limit_rule_update(
self.request,
@@ -1920,35 +1900,33 @@ class NeutronApiTests(test.APIMockTestCase):
self.assertIsInstance(ret_val, api.neutron.BandwidthLimitRule)
self.assertEqual(bwd_limit_rule["max_kbps"], ret_val.max_kbps)
neutronclient.update_bandwidth_limit_rule.assert_called_once_with(
bwd_limit_rule['id'], qos_policy['id'], post_data)
nclient.update_qos_bandwidth_limit_rule.assert_called_once_with(
bwd_limit_rule['id'], qos_policy['id'], **post_data)
@mock.patch.object(api.neutron, 'neutronclient')
def test_bandwidth_limit_rule_delete(self, mock_neutronclient):
qos_policy = self.api_qos_policies.first()
bandwidth_limit_rule = self.api_bandwidth_limit_rule.first()
@mock.patch.object(api.neutron, 'networkclient')
def test_bandwidth_limit_rule_delete(self, mock_networkclient):
qos_policy = self.api_qos_policies_sdk[0]
bandwidth_limit_rule = self.api_bandwidth_limit_rule_sdk[0]
neutronclient = mock_neutronclient.return_value
nclient = mock_networkclient.return_value
neutronclient.delete_bandwidth_limit_rule.return_value = None
nclient.delete_qos_bandwidth_limit_rule.return_value = None
api.neutron.bandwidth_limit_rule_delete(
self.request, qos_policy['id'], bandwidth_limit_rule['id'])
neutronclient.delete_bandwidth_limit_rule.assert_called_once_with(
nclient.delete_qos_bandwidth_limit_rule.assert_called_once_with(
bandwidth_limit_rule['id'], qos_policy['id'])
@mock.patch.object(api.neutron, 'neutronclient')
def test_minimum_bandwidth_rule_create(self, mock_neutronclient):
qos_policy = self.api_qos_policies.first()
min_bwd_rule = self.api_minimum_bandwidth_rule.first()
post_data = {'minimum_bandwidth_rule': {
"min_kbps": min_bwd_rule["min_kbps"]}}
@mock.patch.object(api.neutron, 'networkclient')
def test_minimum_bandwidth_rule_create(self, mock_networkclient):
qos_policy = self.api_qos_policies_sdk[0]
min_bwd_rule = self.api_minimum_bandwidth_rule_sdk[0]
post_data = {"min_kbps": min_bwd_rule["min_kbps"]}
neutronclient = mock_neutronclient.return_value
nclient = mock_networkclient.return_value
neutronclient.create_minimum_bandwidth_rule.return_value = {
'minimum_bandwidth_rule': min_bwd_rule}
nclient.create_qos_minimum_bandwidth_rule.return_value = min_bwd_rule
ret_val = api.neutron.minimum_bandwidth_rule_create(
self.request,
@@ -1957,21 +1935,19 @@ class NeutronApiTests(test.APIMockTestCase):
self.assertIsInstance(ret_val, api.neutron.MinimumBandwidthRule)
self.assertEqual(min_bwd_rule["min_kbps"], ret_val.min_kbps)
neutronclient.create_minimum_bandwidth_rule.assert_called_once_with(
qos_policy['id'], post_data)
nclient.create_qos_minimum_bandwidth_rule.assert_called_once_with(
qos_policy['id'], **post_data)
@mock.patch.object(api.neutron, 'neutronclient')
def test_minimum_bandwidth_rule_update(self, mock_neutronclient):
qos_policy = self.api_qos_policies.first()
min_bwd_rule = self.api_minimum_bandwidth_rule.first()
@mock.patch.object(api.neutron, 'networkclient')
def test_minimum_bandwidth_rule_update(self, mock_networkclient):
qos_policy = self.api_qos_policies_sdk[0]
min_bwd_rule = self.api_minimum_bandwidth_rule_sdk[0]
min_bwd_rule['min_kbps'] = 20000
post_data = {'minimum_bandwidth_rule': {
"min_kbps": 20000}}
post_data = {"min_kbps": 20000}
neutronclient = mock_neutronclient.return_value
nclient = mock_networkclient.return_value
neutronclient.update_minimum_bandwidth_rule.return_value = {
'minimum_bandwidth_rule': min_bwd_rule}
nclient.update_qos_minimum_bandwidth_rule.return_value = min_bwd_rule
ret_val = api.neutron.minimum_bandwidth_rule_update(
self.request,
@@ -1981,35 +1957,34 @@ class NeutronApiTests(test.APIMockTestCase):
self.assertIsInstance(ret_val, api.neutron.MinimumBandwidthRule)
self.assertEqual(min_bwd_rule["min_kbps"], ret_val.min_kbps)
neutronclient.update_minimum_bandwidth_rule.assert_called_once_with(
min_bwd_rule['id'], qos_policy['id'], post_data)
nclient.update_qos_minimum_bandwidth_rule.assert_called_once_with(
min_bwd_rule['id'], qos_policy['id'], **post_data)
@mock.patch.object(api.neutron, 'neutronclient')
def test_minimum_bandwidth_rule_delete(self, mock_neutronclient):
qos_policy = self.api_qos_policies.first()
min_bwd_rule = self.api_minimum_bandwidth_rule.first()
@mock.patch.object(api.neutron, 'networkclient')
def test_minimum_bandwidth_rule_delete(self, mock_networkclient):
qos_policy = self.api_qos_policies_sdk[0]
min_bwd_rule = self.api_minimum_bandwidth_rule_sdk[0]
neutronclient = mock_neutronclient.return_value
nclient = mock_networkclient.return_value
neutronclient.delete_minimum_bandwidth_rule.return_value = None
nclient.delete_qos_minimum_bandwidth_rule.return_value = None
api.neutron.minimum_bandwidth_rule_delete(
self.request, qos_policy['id'], min_bwd_rule['id'])
neutronclient.delete_minimum_bandwidth_rule.assert_called_once_with(
nclient.delete_qos_minimum_bandwidth_rule.assert_called_once_with(
min_bwd_rule['id'], qos_policy['id'])
@mock.patch.object(api.neutron, 'neutronclient')
def test_minimum_packer_rate_rule_create(self, mock_neutronclient):
qos_policy = self.api_qos_policies.first()
min_pckt_rt_rule = self.api_minimum_packet_rate_rule.first()
post_data = {'minimum_packet_rate_rule': {
"min_kpps": min_pckt_rt_rule["min_kpps"]}}
@mock.patch.object(api.neutron, 'networkclient')
def test_minimum_packer_rate_rule_create(self, mock_networkclient):
qos_policy = self.api_qos_policies_sdk[0]
min_pckt_rt_rule = self.api_minimum_packet_rate_rule_sdk[0]
post_data = {"min_kpps": min_pckt_rt_rule["min_kpps"]}
neutronclient = mock_neutronclient.return_value
nclient = mock_networkclient.return_value
neutronclient.create_minimum_packet_rate_rule.return_value = {
'minimum_packet_rate_rule': min_pckt_rt_rule}
nclient.create_qos_minimum_packet_rate_rule.return_value = \
min_pckt_rt_rule
ret_val = api.neutron.minimum_packet_rate_rule_create(
self.request,
@@ -2018,21 +1993,20 @@ class NeutronApiTests(test.APIMockTestCase):
self.assertIsInstance(ret_val, api.neutron.MinimumPacketRateRule)
self.assertEqual(min_pckt_rt_rule['min_kpps'], ret_val.min_kpps)
neutronclient.create_minimum_packet_rate_rule.assert_called_once_with(
qos_policy['id'], post_data)
nclient.create_qos_minimum_packet_rate_rule.assert_called_once_with(
qos_policy['id'], **post_data)
@mock.patch.object(api.neutron, 'neutronclient')
def test_minimum_packer_rate_rule_update(self, mock_neutronclient):
qos_policy = self.api_qos_policies.first()
min_pckt_rt_rule = self.api_minimum_packet_rate_rule.first()
@mock.patch.object(api.neutron, 'networkclient')
def test_minimum_packer_rate_rule_update(self, mock_networkclient):
qos_policy = self.api_qos_policies_sdk[0]
min_pckt_rt_rule = self.api_minimum_packet_rate_rule_sdk[0]
min_pckt_rt_rule['min_kpps'] = 11000
post_data = {'minimum_packet_rate_rule': {
"min_kpps": 11000}}
post_data = {"min_kpps": 11000}
neutronclient = mock_neutronclient.return_value
nclient = mock_networkclient.return_value
neutronclient.update_minimum_packet_rate_rule.return_value = {
'minimum_packet_rate_rule': min_pckt_rt_rule}
nclient.update_qos_minimum_packet_rate_rule.return_value = \
min_pckt_rt_rule
ret_val = api.neutron.minimum_packet_rate_rule_update(
self.request,
@@ -2042,22 +2016,22 @@ class NeutronApiTests(test.APIMockTestCase):
self.assertIsInstance(ret_val, api.neutron.MinimumPacketRateRule)
self.assertEqual(min_pckt_rt_rule["min_kpps"], ret_val.min_kpps)
neutronclient.update_minimum_packet_rate_rule.assert_called_once_with(
min_pckt_rt_rule['id'], qos_policy['id'], post_data)
nclient.update_qos_minimum_packet_rate_rule.assert_called_once_with(
min_pckt_rt_rule['id'], qos_policy['id'], **post_data)
@mock.patch.object(api.neutron, 'neutronclient')
def test_minimum_packet_rate_rule_delete(self, mock_neutronclient):
qos_policy = self.api_qos_policies.first()
min_pckt_rt_rule = self.api_minimum_packet_rate_rule.first()
@mock.patch.object(api.neutron, 'networkclient')
def test_minimum_packet_rate_rule_delete(self, mock_networkclient):
qos_policy = self.api_qos_policies_sdk[0]
min_pckt_rt_rule = self.api_minimum_packet_rate_rule_sdk[0]
neutronclient = mock_neutronclient.return_value
nclient = mock_networkclient.return_value
neutronclient.delete_minimum_packet_rate_rule.return_value = None
nclient.delete_qos_minimum_packet_rate_rule.return_value = None
api.neutron.minimum_packet_rate_rule_delete(
self.request, qos_policy['id'], min_pckt_rt_rule['id'])
neutronclient.delete_minimum_packet_rate_rule.assert_called_once_with(
nclient.delete_qos_minimum_packet_rate_rule.assert_called_once_with(
min_pckt_rt_rule['id'], qos_policy['id'])