Refactor network fakes to sdk properties PART 5
Included resources: qos_policy qos_rule qos_rule_type Note: Parameters in unittests was modified for compatibility with sdk Change-Id: Iaa902d64ff1b29a07c28ed2100d437da506be475
This commit is contained in:
parent
30aa27b7f9
commit
f870548c7f
@ -73,7 +73,7 @@ ACTION_SHOW = 'get'
|
||||
|
||||
|
||||
def _get_columns(item):
|
||||
hidden_columns = ['location', 'tenant_id']
|
||||
hidden_columns = ['location', 'name', 'tenant_id']
|
||||
return utils.get_osc_show_columns_for_sdk_resource(
|
||||
item, {}, hidden_columns
|
||||
)
|
||||
@ -148,14 +148,6 @@ def _get_attrs(network_client, parsed_args, is_create=False):
|
||||
return attrs
|
||||
|
||||
|
||||
def _get_item_properties(item, fields):
|
||||
"""Return a tuple containing the item properties."""
|
||||
row = []
|
||||
for field in fields:
|
||||
row.append(item.get(field, ''))
|
||||
return tuple(row)
|
||||
|
||||
|
||||
def _rule_action_call(client, action, rule_type):
|
||||
rule_type = rule_type.replace('-', '_')
|
||||
func_name = f'{action}_qos_{rule_type}_rule'
|
||||
@ -357,10 +349,10 @@ class ListNetworkQosRule(command.Lister):
|
||||
qos = client.find_qos_policy(
|
||||
parsed_args.qos_policy, ignore_missing=False
|
||||
)
|
||||
data = qos.rules
|
||||
|
||||
return (
|
||||
column_headers,
|
||||
(_get_item_properties(s, columns) for s in data),
|
||||
(utils.get_dict_properties(s, columns) for s in qos.rules),
|
||||
)
|
||||
|
||||
|
||||
|
@ -97,7 +97,11 @@ class TestCase(testtools.TestCase):
|
||||
)
|
||||
|
||||
if parse_output:
|
||||
return json.loads(output)
|
||||
try:
|
||||
return json.loads(output)
|
||||
except json.JSONDecodeError:
|
||||
print(f'failed to decode: {output}')
|
||||
raise
|
||||
else:
|
||||
return output
|
||||
|
||||
|
@ -31,6 +31,20 @@ from openstack.network.v2 import network as _network
|
||||
from openstack.network.v2 import network_ip_availability as _ip_availability
|
||||
from openstack.network.v2 import network_segment_range as _segment_range
|
||||
from openstack.network.v2 import port as _port
|
||||
from openstack.network.v2 import (
|
||||
qos_bandwidth_limit_rule as _qos_bandwidth_limit_rule,
|
||||
)
|
||||
from openstack.network.v2 import (
|
||||
qos_dscp_marking_rule as _qos_dscp_marking_rule,
|
||||
)
|
||||
from openstack.network.v2 import (
|
||||
qos_minimum_bandwidth_rule as _qos_minimum_bandwidth_rule,
|
||||
)
|
||||
from openstack.network.v2 import (
|
||||
qos_minimum_packet_rate_rule as _qos_minimum_packet_rate_rule,
|
||||
)
|
||||
from openstack.network.v2 import qos_policy as _qos_policy
|
||||
from openstack.network.v2 import qos_rule_type as _qos_rule_type
|
||||
from openstack.network.v2 import rbac_policy as network_rbac
|
||||
from openstack.network.v2 import security_group as _security_group
|
||||
from openstack.network.v2 import segment as _segment
|
||||
@ -118,245 +132,6 @@ def create_one_extension(attrs=None):
|
||||
return extension
|
||||
|
||||
|
||||
class FakeNetworkQosPolicy:
|
||||
"""Fake one or more QoS policies."""
|
||||
|
||||
@staticmethod
|
||||
def create_one_qos_policy(attrs=None):
|
||||
"""Create a fake QoS policy.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:return:
|
||||
A FakeResource object with name, id, etc.
|
||||
"""
|
||||
attrs = attrs or {}
|
||||
qos_id = attrs.get('id') or 'qos-policy-id-' + uuid.uuid4().hex
|
||||
rule_attrs = {'qos_policy_id': qos_id}
|
||||
rules = [FakeNetworkQosRule.create_one_qos_rule(rule_attrs)]
|
||||
|
||||
# Set default attributes.
|
||||
qos_policy_attrs = {
|
||||
'name': 'qos-policy-name-' + uuid.uuid4().hex,
|
||||
'id': qos_id,
|
||||
'is_default': False,
|
||||
'project_id': 'project-id-' + uuid.uuid4().hex,
|
||||
'shared': False,
|
||||
'description': 'qos-policy-description-' + uuid.uuid4().hex,
|
||||
'rules': rules,
|
||||
'location': 'MUNCHMUNCHMUNCH',
|
||||
}
|
||||
|
||||
# Overwrite default attributes.
|
||||
qos_policy_attrs.update(attrs)
|
||||
|
||||
qos_policy = fakes.FakeResource(
|
||||
info=copy.deepcopy(qos_policy_attrs), loaded=True
|
||||
)
|
||||
|
||||
# Set attributes with special mapping in OpenStack SDK.
|
||||
qos_policy.is_shared = qos_policy_attrs['shared']
|
||||
|
||||
return qos_policy
|
||||
|
||||
@staticmethod
|
||||
def create_qos_policies(attrs=None, count=2):
|
||||
"""Create multiple fake QoS policies.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:param int count:
|
||||
The number of QoS policies to fake
|
||||
:return:
|
||||
A list of FakeResource objects faking the QoS policies
|
||||
"""
|
||||
qos_policies = []
|
||||
for i in range(0, count):
|
||||
qos_policies.append(
|
||||
FakeNetworkQosPolicy.create_one_qos_policy(attrs)
|
||||
)
|
||||
|
||||
return qos_policies
|
||||
|
||||
@staticmethod
|
||||
def get_qos_policies(qos_policies=None, count=2):
|
||||
"""Get an iterable MagicMock object with a list of faked QoS policies.
|
||||
|
||||
If qos policies list is provided, then initialize the Mock object
|
||||
with the list. Otherwise create one.
|
||||
|
||||
:param List qos_policies:
|
||||
A list of FakeResource objects faking qos policies
|
||||
:param int count:
|
||||
The number of QoS policies to fake
|
||||
:return:
|
||||
An iterable Mock object with side_effect set to a list of faked
|
||||
QoS policies
|
||||
"""
|
||||
if qos_policies is None:
|
||||
qos_policies = FakeNetworkQosPolicy.create_qos_policies(count)
|
||||
return mock.Mock(side_effect=qos_policies)
|
||||
|
||||
|
||||
class FakeNetworkSecGroup:
|
||||
"""Fake one security group."""
|
||||
|
||||
@staticmethod
|
||||
def create_one_security_group(attrs=None):
|
||||
"""Create a fake security group.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:return:
|
||||
A FakeResource object with name, id, etc.
|
||||
"""
|
||||
attrs = attrs or {}
|
||||
sg_id = attrs.get('id') or 'security-group-id-' + uuid.uuid4().hex
|
||||
|
||||
# Set default attributes.
|
||||
security_group_attrs = {
|
||||
'name': 'security-group-name-' + uuid.uuid4().hex,
|
||||
'id': sg_id,
|
||||
'project_id': 'project-id-' + uuid.uuid4().hex,
|
||||
'description': 'security-group-description-' + uuid.uuid4().hex,
|
||||
'location': 'MUNCHMUNCHMUNCH',
|
||||
}
|
||||
|
||||
security_group = fakes.FakeResource(
|
||||
info=copy.deepcopy(security_group_attrs), loaded=True
|
||||
)
|
||||
|
||||
return security_group
|
||||
|
||||
|
||||
class FakeNetworkQosRule:
|
||||
"""Fake one or more Network QoS rules."""
|
||||
|
||||
@staticmethod
|
||||
def create_one_qos_rule(attrs=None):
|
||||
"""Create a fake Network QoS rule.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:return:
|
||||
A FakeResource object with name, id, etc.
|
||||
"""
|
||||
attrs = attrs or {}
|
||||
|
||||
# Set default attributes.
|
||||
type = attrs.get('type') or choice(VALID_QOS_RULES)
|
||||
qos_rule_attrs = {
|
||||
'id': 'qos-rule-id-' + uuid.uuid4().hex,
|
||||
'qos_policy_id': 'qos-policy-id-' + uuid.uuid4().hex,
|
||||
'project_id': 'project-id-' + uuid.uuid4().hex,
|
||||
'type': type,
|
||||
'location': 'MUNCHMUNCHMUNCH',
|
||||
}
|
||||
if type == RULE_TYPE_BANDWIDTH_LIMIT:
|
||||
qos_rule_attrs['max_kbps'] = randint(1, 10000)
|
||||
qos_rule_attrs['max_burst_kbits'] = randint(1, 10000)
|
||||
qos_rule_attrs['direction'] = 'egress'
|
||||
elif type == RULE_TYPE_DSCP_MARKING:
|
||||
qos_rule_attrs['dscp_mark'] = choice(VALID_DSCP_MARKS)
|
||||
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)
|
||||
|
||||
qos_rule = fakes.FakeResource(
|
||||
info=copy.deepcopy(qos_rule_attrs), loaded=True
|
||||
)
|
||||
|
||||
return qos_rule
|
||||
|
||||
@staticmethod
|
||||
def create_qos_rules(attrs=None, count=2):
|
||||
"""Create multiple fake Network QoS rules.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:param int count:
|
||||
The number of Network QoS rule to fake
|
||||
:return:
|
||||
A list of FakeResource objects faking the Network QoS rules
|
||||
"""
|
||||
qos_rules = []
|
||||
for i in range(0, count):
|
||||
qos_rules.append(FakeNetworkQosRule.create_one_qos_rule(attrs))
|
||||
return qos_rules
|
||||
|
||||
@staticmethod
|
||||
def get_qos_rules(qos_rules=None, count=2):
|
||||
"""Get a list of faked Network QoS rules.
|
||||
|
||||
If Network QoS rules list is provided, then initialize the Mock
|
||||
object with the list. Otherwise create one.
|
||||
|
||||
:param List qos_rules:
|
||||
A list of FakeResource objects faking Network QoS rules
|
||||
:param int count:
|
||||
The number of QoS minimum bandwidth rules to fake
|
||||
:return:
|
||||
An iterable Mock object with side_effect set to a list of faked
|
||||
qos minimum bandwidth rules
|
||||
"""
|
||||
if qos_rules is None:
|
||||
qos_rules = FakeNetworkQosRule.create_qos_rules(count)
|
||||
return mock.Mock(side_effect=qos_rules)
|
||||
|
||||
|
||||
class FakeNetworkQosRuleType:
|
||||
"""Fake one or more Network QoS rule types."""
|
||||
|
||||
@staticmethod
|
||||
def create_one_qos_rule_type(attrs=None):
|
||||
"""Create a fake Network QoS rule type.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:return:
|
||||
A FakeResource object with name, id, etc.
|
||||
"""
|
||||
attrs = attrs or {}
|
||||
|
||||
# Set default attributes.
|
||||
qos_rule_type_attrs = {
|
||||
'type': 'rule-type-' + uuid.uuid4().hex,
|
||||
'location': 'MUNCHMUNCHMUNCH',
|
||||
}
|
||||
|
||||
# Overwrite default attributes.
|
||||
qos_rule_type_attrs.update(attrs)
|
||||
|
||||
return fakes.FakeResource(
|
||||
info=copy.deepcopy(qos_rule_type_attrs), loaded=True
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def create_qos_rule_types(attrs=None, count=2):
|
||||
"""Create multiple fake Network QoS rule types.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:param int count:
|
||||
The number of QoS rule types to fake
|
||||
:return:
|
||||
A list of FakeResource objects faking the QoS rule types
|
||||
"""
|
||||
qos_rule_types = []
|
||||
for i in range(0, count):
|
||||
qos_rule_types.append(
|
||||
FakeNetworkQosRuleType.create_one_qos_rule_type(attrs)
|
||||
)
|
||||
|
||||
return qos_rule_types
|
||||
|
||||
|
||||
class FakeRouter:
|
||||
"""Fake one or more routers."""
|
||||
|
||||
@ -1926,6 +1701,207 @@ def get_service_profile(flavor_profile=None, count=2):
|
||||
return mock.Mock(side_effect=flavor_profile)
|
||||
|
||||
|
||||
def create_one_qos_policy(attrs=None):
|
||||
"""Create a fake QoS policy.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:return:
|
||||
A QoSPolicy object with name, id, etc.
|
||||
"""
|
||||
attrs = attrs or {}
|
||||
qos_id = attrs.get('id') or 'qos-policy-id-' + uuid.uuid4().hex
|
||||
rules = []
|
||||
|
||||
# Set default attributes.
|
||||
qos_policy_attrs = {
|
||||
'name': 'qos-policy-name-' + uuid.uuid4().hex,
|
||||
'id': qos_id,
|
||||
'is_default': False,
|
||||
'project_id': 'project-id-' + uuid.uuid4().hex,
|
||||
'shared': False,
|
||||
'description': 'qos-policy-description-' + uuid.uuid4().hex,
|
||||
'rules': rules,
|
||||
'location': 'MUNCHMUNCHMUNCH',
|
||||
}
|
||||
|
||||
# Overwrite default attributes.
|
||||
qos_policy_attrs.update(attrs)
|
||||
|
||||
qos_policy = _qos_policy.QoSPolicy(**qos_policy_attrs)
|
||||
|
||||
return qos_policy
|
||||
|
||||
|
||||
def create_qos_policies(attrs=None, count=2):
|
||||
"""Create multiple fake QoS policies.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:param int count:
|
||||
The number of QoS policies to fake
|
||||
:return:
|
||||
A list of QoSPolicy objects faking the QoS policies
|
||||
"""
|
||||
qos_policies = []
|
||||
for i in range(0, count):
|
||||
qos_policies.append(create_one_qos_policy(attrs))
|
||||
|
||||
return qos_policies
|
||||
|
||||
|
||||
def get_qos_policies(qos_policies=None, count=2):
|
||||
"""Get an iterable MagicMock object with a list of faked QoS policies.
|
||||
|
||||
If qos policies list is provided, then initialize the Mock object
|
||||
with the list. Otherwise create one.
|
||||
|
||||
:param List qos_policies:
|
||||
A list of QoSPolicy objects faking qos policies
|
||||
:param int count:
|
||||
The number of QoS policies to fake
|
||||
:return:
|
||||
An iterable Mock object with side_effect set to a list of faked
|
||||
QoS policies
|
||||
"""
|
||||
if qos_policies is None:
|
||||
qos_policies = create_qos_policies(count)
|
||||
return mock.Mock(side_effect=qos_policies)
|
||||
|
||||
|
||||
def create_one_qos_rule(attrs=None):
|
||||
"""Create a fake Network QoS rule.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:return:
|
||||
A QoSRule object with id, type, etc.
|
||||
"""
|
||||
attrs = attrs or {}
|
||||
|
||||
# Set default attributes.
|
||||
type = attrs.get('type') or choice(VALID_QOS_RULES)
|
||||
qos_rule_attrs = {
|
||||
'id': 'qos-rule-id-' + uuid.uuid4().hex,
|
||||
'qos_policy_id': 'qos-policy-id-' + uuid.uuid4().hex,
|
||||
'project_id': 'project-id-' + uuid.uuid4().hex,
|
||||
'type': type,
|
||||
'location': 'MUNCHMUNCHMUNCH',
|
||||
}
|
||||
|
||||
if type == RULE_TYPE_BANDWIDTH_LIMIT:
|
||||
qos_rule_attrs['max_kbps'] = randint(1, 10000)
|
||||
qos_rule_attrs['max_burst_kbps'] = randint(1, 10000)
|
||||
qos_rule_attrs['direction'] = 'egress'
|
||||
|
||||
qos_rule_attrs.update(attrs)
|
||||
qos_rule = _qos_bandwidth_limit_rule.QoSBandwidthLimitRule(
|
||||
**qos_rule_attrs
|
||||
)
|
||||
|
||||
elif type == RULE_TYPE_DSCP_MARKING:
|
||||
qos_rule_attrs['dscp_mark'] = choice(VALID_DSCP_MARKS)
|
||||
|
||||
qos_rule_attrs.update(attrs)
|
||||
qos_rule = _qos_dscp_marking_rule.QoSDSCPMarkingRule(**qos_rule_attrs)
|
||||
|
||||
elif type == RULE_TYPE_MINIMUM_BANDWIDTH:
|
||||
qos_rule_attrs['min_kbps'] = randint(1, 10000)
|
||||
qos_rule_attrs['direction'] = 'egress'
|
||||
|
||||
qos_rule_attrs.update(attrs)
|
||||
|
||||
qos_rule = _qos_minimum_bandwidth_rule.QoSMinimumBandwidthRule(
|
||||
**qos_rule_attrs
|
||||
)
|
||||
else: # type == RULE_TYPE_MINIMUM_PACKET_RATE:
|
||||
qos_rule_attrs['min_kpps'] = randint(1, 10000)
|
||||
qos_rule_attrs['direction'] = 'egress'
|
||||
|
||||
qos_rule_attrs.update(attrs)
|
||||
|
||||
qos_rule = _qos_minimum_packet_rate_rule.QoSMinimumPacketRateRule(
|
||||
**qos_rule_attrs
|
||||
)
|
||||
|
||||
return qos_rule
|
||||
|
||||
|
||||
def create_qos_rules(attrs=None, count=2):
|
||||
"""Create multiple fake Network QoS rules.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:param int count:
|
||||
The number of Network QoS rule to fake
|
||||
:return:
|
||||
A list of QoS Rules objects faking the Network QoS rules
|
||||
"""
|
||||
qos_rules = []
|
||||
for i in range(0, count):
|
||||
qos_rules.append(create_one_qos_rule(attrs))
|
||||
return qos_rules
|
||||
|
||||
|
||||
def get_qos_rules(qos_rules=None, count=2):
|
||||
"""Get a list of faked Network QoS rules.
|
||||
|
||||
If Network QoS rules list is provided, then initialize the Mock
|
||||
object with the list. Otherwise create one.
|
||||
|
||||
:param List qos_rules:
|
||||
A list of FakeResource objects faking Network QoS rules
|
||||
:param int count:
|
||||
The number of QoS minimum bandwidth rules to fake
|
||||
:return:
|
||||
An iterable Mock object with side_effect set to a list of faked
|
||||
qos minimum bandwidth rules
|
||||
"""
|
||||
if qos_rules is None:
|
||||
qos_rules = create_qos_rules(count)
|
||||
return mock.Mock(side_effect=qos_rules)
|
||||
|
||||
|
||||
def create_one_qos_rule_type(attrs=None):
|
||||
"""Create a fake Network QoS rule type.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:return:
|
||||
A QoSRuleType object with name, id, etc.
|
||||
"""
|
||||
attrs = attrs or {}
|
||||
|
||||
# Set default attributes.
|
||||
qos_rule_type_attrs = {
|
||||
'type': 'rule-type-' + uuid.uuid4().hex,
|
||||
'location': 'MUNCHMUNCHMUNCH',
|
||||
}
|
||||
|
||||
# Overwrite default attributes.
|
||||
qos_rule_type_attrs.update(attrs)
|
||||
qos_rule_type = _qos_rule_type.QoSRuleType(**qos_rule_type_attrs)
|
||||
|
||||
return qos_rule_type
|
||||
|
||||
|
||||
def create_qos_rule_types(attrs=None, count=2):
|
||||
"""Create multiple fake Network QoS rule types.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:param int count:
|
||||
The number of QoS rule types to fake
|
||||
:return:
|
||||
A list of QoSRuleType objects faking the QoS rule types
|
||||
"""
|
||||
qos_rule_types = []
|
||||
for i in range(0, count):
|
||||
qos_rule_types.append(create_one_qos_rule_type(attrs))
|
||||
|
||||
return qos_rule_types
|
||||
|
||||
|
||||
def create_one_local_ip(attrs=None):
|
||||
"""Create a fake local ip.
|
||||
|
||||
|
@ -234,10 +234,8 @@ class TestCreateFloatingIPNetwork(TestFloatingIPNetwork):
|
||||
self.assertEqual(self.data, data)
|
||||
|
||||
def test_create_floating_ip_with_qos(self):
|
||||
qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=qos_policy
|
||||
)
|
||||
qos_policy = network_fakes.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy.return_value = qos_policy
|
||||
arglist = [
|
||||
'--qos-policy',
|
||||
qos_policy.id,
|
||||
@ -893,10 +891,8 @@ class TestSetFloatingIP(TestFloatingIPNetwork):
|
||||
)
|
||||
|
||||
def test_qos_policy_option(self):
|
||||
qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=qos_policy
|
||||
)
|
||||
qos_policy = network_fakes.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy.return_value = qos_policy
|
||||
arglist = [
|
||||
"--qos-policy",
|
||||
qos_policy.id,
|
||||
@ -922,10 +918,8 @@ class TestSetFloatingIP(TestFloatingIPNetwork):
|
||||
)
|
||||
|
||||
def test_port_and_qos_policy_option(self):
|
||||
qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=qos_policy
|
||||
)
|
||||
qos_policy = network_fakes.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy.return_value = qos_policy
|
||||
arglist = [
|
||||
"--qos-policy",
|
||||
qos_policy.id,
|
||||
|
@ -47,7 +47,7 @@ class TestCreateNetworkIdentityV3(TestNetwork):
|
||||
'availability_zone_hints': ["nova"],
|
||||
}
|
||||
)
|
||||
qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy(
|
||||
qos_policy = network_fakes.create_one_qos_policy(
|
||||
attrs={'id': _network.qos_policy_id}
|
||||
)
|
||||
|
||||
@ -622,7 +622,7 @@ class TestListNetwork(TestNetwork):
|
||||
self.network_client.networks = mock.Mock(return_value=self._network)
|
||||
|
||||
self._agent = network_fakes.create_one_network_agent()
|
||||
self.network_client.get_agent = mock.Mock(return_value=self._agent)
|
||||
self.network_client.get_agent.return_value = self._agent
|
||||
|
||||
self.network_client.dhcp_agent_hosting_networks = mock.Mock(
|
||||
return_value=self._network
|
||||
@ -967,7 +967,7 @@ class TestListNetwork(TestNetwork):
|
||||
class TestSetNetwork(TestNetwork):
|
||||
# The network to set.
|
||||
_network = network_fakes.create_one_network({'tags': ['green', 'red']})
|
||||
qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy(
|
||||
qos_policy = network_fakes.create_one_qos_policy(
|
||||
attrs={'id': _network.qos_policy_id}
|
||||
)
|
||||
|
||||
@ -1266,7 +1266,7 @@ class TestShowNetwork(TestNetwork):
|
||||
class TestUnsetNetwork(TestNetwork):
|
||||
# The network to set.
|
||||
_network = network_fakes.create_one_network({'tags': ['green', 'red']})
|
||||
qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy(
|
||||
qos_policy = network_fakes.create_one_qos_policy(
|
||||
attrs={'id': _network.qos_policy_id}
|
||||
)
|
||||
|
||||
|
@ -35,11 +35,10 @@ class TestCreateNetworkQosPolicy(TestQosPolicy):
|
||||
project = identity_fakes_v3.FakeProject.create_one_project()
|
||||
|
||||
# The new qos policy created.
|
||||
new_qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy(
|
||||
attrs={
|
||||
'project_id': project.id,
|
||||
}
|
||||
new_qos_policy = network_fakes.create_one_qos_policy(
|
||||
attrs={'project_id': project.id}
|
||||
)
|
||||
|
||||
columns = (
|
||||
'description',
|
||||
'id',
|
||||
@ -48,6 +47,7 @@ class TestCreateNetworkQosPolicy(TestQosPolicy):
|
||||
'project_id',
|
||||
'rules',
|
||||
'shared',
|
||||
'tags',
|
||||
)
|
||||
|
||||
data = (
|
||||
@ -57,7 +57,8 @@ class TestCreateNetworkQosPolicy(TestQosPolicy):
|
||||
new_qos_policy.name,
|
||||
new_qos_policy.project_id,
|
||||
new_qos_policy.rules,
|
||||
new_qos_policy.shared,
|
||||
new_qos_policy.is_shared,
|
||||
new_qos_policy.tags,
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
@ -158,17 +159,13 @@ class TestCreateNetworkQosPolicy(TestQosPolicy):
|
||||
|
||||
class TestDeleteNetworkQosPolicy(TestQosPolicy):
|
||||
# The address scope to delete.
|
||||
_qos_policies = network_fakes.FakeNetworkQosPolicy.create_qos_policies(
|
||||
count=2
|
||||
)
|
||||
_qos_policies = network_fakes.create_qos_policies(count=2)
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.network_client.delete_qos_policy = mock.Mock(return_value=None)
|
||||
self.network_client.find_qos_policy = (
|
||||
network_fakes.FakeNetworkQosPolicy.get_qos_policies(
|
||||
qos_policies=self._qos_policies
|
||||
)
|
||||
self.network_client.find_qos_policy = network_fakes.get_qos_policies(
|
||||
qos_policies=self._qos_policies
|
||||
)
|
||||
|
||||
# Get the command object to test
|
||||
@ -245,9 +242,7 @@ class TestDeleteNetworkQosPolicy(TestQosPolicy):
|
||||
|
||||
class TestListNetworkQosPolicy(TestQosPolicy):
|
||||
# The QoS policies to list up.
|
||||
qos_policies = network_fakes.FakeNetworkQosPolicy.create_qos_policies(
|
||||
count=3
|
||||
)
|
||||
qos_policies = network_fakes.create_qos_policies(count=3)
|
||||
columns = (
|
||||
'ID',
|
||||
'Name',
|
||||
@ -261,7 +256,7 @@ class TestListNetworkQosPolicy(TestQosPolicy):
|
||||
(
|
||||
qos_policy.id,
|
||||
qos_policy.name,
|
||||
qos_policy.shared,
|
||||
qos_policy.is_shared,
|
||||
qos_policy.is_default,
|
||||
qos_policy.project_id,
|
||||
)
|
||||
@ -345,7 +340,7 @@ class TestListNetworkQosPolicy(TestQosPolicy):
|
||||
|
||||
class TestSetNetworkQosPolicy(TestQosPolicy):
|
||||
# The QoS policy to set.
|
||||
_qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
_qos_policy = network_fakes.create_one_qos_policy()
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
@ -428,7 +423,7 @@ class TestSetNetworkQosPolicy(TestQosPolicy):
|
||||
|
||||
class TestShowNetworkQosPolicy(TestQosPolicy):
|
||||
# The QoS policy to show.
|
||||
_qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
_qos_policy = network_fakes.create_one_qos_policy()
|
||||
columns = (
|
||||
'description',
|
||||
'id',
|
||||
@ -437,6 +432,7 @@ class TestShowNetworkQosPolicy(TestQosPolicy):
|
||||
'project_id',
|
||||
'rules',
|
||||
'shared',
|
||||
'tags',
|
||||
)
|
||||
data = (
|
||||
_qos_policy.description,
|
||||
@ -445,7 +441,8 @@ class TestShowNetworkQosPolicy(TestQosPolicy):
|
||||
_qos_policy.name,
|
||||
_qos_policy.project_id,
|
||||
network_qos_policy.RulesColumn(_qos_policy.rules),
|
||||
_qos_policy.shared,
|
||||
_qos_policy.is_shared,
|
||||
_qos_policy.tags,
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
|
@ -14,6 +14,7 @@
|
||||
# under the License.
|
||||
|
||||
from unittest import mock
|
||||
import uuid
|
||||
|
||||
from osc_lib import exceptions
|
||||
|
||||
@ -54,9 +55,7 @@ DSCP_VALID_MARKS = [
|
||||
class TestNetworkQosRule(network_fakes.TestNetworkV2):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.qos_policy = (
|
||||
network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
)
|
||||
self.qos_policy = network_fakes.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=self.qos_policy
|
||||
)
|
||||
@ -72,15 +71,12 @@ class TestCreateNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': RULE_TYPE_MINIMUM_BANDWIDTH,
|
||||
}
|
||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
||||
attrs
|
||||
)
|
||||
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.columns = (
|
||||
'direction',
|
||||
'id',
|
||||
'min_kbps',
|
||||
'project_id',
|
||||
'qos_policy_id',
|
||||
'type',
|
||||
)
|
||||
|
||||
@ -89,7 +85,6 @@ class TestCreateNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
|
||||
self.new_rule.id,
|
||||
self.new_rule.min_kbps,
|
||||
self.new_rule.project_id,
|
||||
self.new_rule.qos_policy_id,
|
||||
self.new_rule.type,
|
||||
)
|
||||
self.network_client.create_qos_minimum_bandwidth_rule = mock.Mock(
|
||||
@ -179,15 +174,12 @@ class TestCreateNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
|
||||
'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.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.columns = (
|
||||
'direction',
|
||||
'id',
|
||||
'min_kpps',
|
||||
'project_id',
|
||||
'qos_policy_id',
|
||||
'type',
|
||||
)
|
||||
|
||||
@ -196,7 +188,6 @@ class TestCreateNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
|
||||
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_client.create_qos_minimum_packet_rate_rule = mock.Mock(
|
||||
@ -286,14 +277,11 @@ class TestCreateNetworkQosRuleDSCPMarking(TestNetworkQosRule):
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': RULE_TYPE_DSCP_MARKING,
|
||||
}
|
||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
||||
attrs
|
||||
)
|
||||
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.columns = (
|
||||
'dscp_mark',
|
||||
'id',
|
||||
'project_id',
|
||||
'qos_policy_id',
|
||||
'type',
|
||||
)
|
||||
|
||||
@ -301,7 +289,6 @@ class TestCreateNetworkQosRuleDSCPMarking(TestNetworkQosRule):
|
||||
self.new_rule.dscp_mark,
|
||||
self.new_rule.id,
|
||||
self.new_rule.project_id,
|
||||
self.new_rule.qos_policy_id,
|
||||
self.new_rule.type,
|
||||
)
|
||||
self.network_client.create_qos_dscp_marking_rule = mock.Mock(
|
||||
@ -384,26 +371,22 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
||||
}
|
||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
||||
attrs
|
||||
)
|
||||
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.columns = (
|
||||
'direction',
|
||||
'id',
|
||||
'max_burst_kbits',
|
||||
'max_burst_kbps',
|
||||
'max_kbps',
|
||||
'project_id',
|
||||
'qos_policy_id',
|
||||
'type',
|
||||
)
|
||||
|
||||
self.data = (
|
||||
self.new_rule.direction,
|
||||
self.new_rule.id,
|
||||
self.new_rule.max_burst_kbits,
|
||||
self.new_rule.max_burst_kbps,
|
||||
self.new_rule.max_kbps,
|
||||
self.new_rule.project_id,
|
||||
self.new_rule.qos_policy_id,
|
||||
self.new_rule.type,
|
||||
)
|
||||
self.network_client.create_qos_bandwidth_limit_rule = mock.Mock(
|
||||
@ -443,20 +426,19 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
|
||||
('qos_policy', self.new_rule.qos_policy_id),
|
||||
]
|
||||
|
||||
rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
||||
rule = network_fakes.create_one_qos_rule(
|
||||
{
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
||||
}
|
||||
)
|
||||
rule.max_burst_kbits = 0
|
||||
rule.max_burst_kbps = 0
|
||||
expected_data = (
|
||||
rule.direction,
|
||||
rule.id,
|
||||
rule.max_burst_kbits,
|
||||
rule.max_burst_kbps,
|
||||
rule.max_kbps,
|
||||
rule.project_id,
|
||||
rule.qos_policy_id,
|
||||
rule.type,
|
||||
)
|
||||
|
||||
@ -485,7 +467,7 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
|
||||
'--max-kbps',
|
||||
str(self.new_rule.max_kbps),
|
||||
'--max-burst-kbits',
|
||||
str(self.new_rule.max_burst_kbits),
|
||||
str(self.new_rule.max_burst_kbps),
|
||||
'--egress',
|
||||
self.new_rule.qos_policy_id,
|
||||
]
|
||||
@ -493,7 +475,7 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
|
||||
verifylist = [
|
||||
('type', RULE_TYPE_BANDWIDTH_LIMIT),
|
||||
('max_kbps', self.new_rule.max_kbps),
|
||||
('max_burst_kbits', self.new_rule.max_burst_kbits),
|
||||
('max_burst_kbits', self.new_rule.max_burst_kbps),
|
||||
('egress', True),
|
||||
('qos_policy', self.new_rule.qos_policy_id),
|
||||
]
|
||||
@ -505,7 +487,7 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
|
||||
self.qos_policy.id,
|
||||
**{
|
||||
'max_kbps': self.new_rule.max_kbps,
|
||||
'max_burst_kbps': self.new_rule.max_burst_kbits,
|
||||
'max_burst_kbps': self.new_rule.max_burst_kbps,
|
||||
'direction': self.new_rule.direction,
|
||||
},
|
||||
)
|
||||
@ -545,17 +527,13 @@ class TestDeleteNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': RULE_TYPE_MINIMUM_BANDWIDTH,
|
||||
}
|
||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
||||
attrs
|
||||
)
|
||||
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.qos_policy.rules = [self.new_rule]
|
||||
self.network_client.delete_qos_minimum_bandwidth_rule = mock.Mock(
|
||||
return_value=None
|
||||
)
|
||||
self.network_client.find_qos_minimum_bandwidth_rule = (
|
||||
network_fakes.FakeNetworkQosRule.get_qos_rules(
|
||||
qos_rules=self.new_rule
|
||||
)
|
||||
network_fakes.get_qos_rules(qos_rules=self.new_rule)
|
||||
)
|
||||
|
||||
# Get the command object to test
|
||||
@ -612,17 +590,13 @@ class TestDeleteNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
|
||||
'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.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.qos_policy.rules = [self.new_rule]
|
||||
self.network_client.delete_qos_minimum_packet_rate_rule = mock.Mock(
|
||||
return_value=None
|
||||
)
|
||||
self.network_client.find_qos_minimum_packet_rate_rule = (
|
||||
network_fakes.FakeNetworkQosRule.get_qos_rules(
|
||||
qos_rules=self.new_rule
|
||||
)
|
||||
network_fakes.get_qos_rules(qos_rules=self.new_rule)
|
||||
)
|
||||
|
||||
# Get the command object to test
|
||||
@ -679,17 +653,13 @@ class TestDeleteNetworkQosRuleDSCPMarking(TestNetworkQosRule):
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': RULE_TYPE_DSCP_MARKING,
|
||||
}
|
||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
||||
attrs
|
||||
)
|
||||
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.qos_policy.rules = [self.new_rule]
|
||||
self.network_client.delete_qos_dscp_marking_rule = mock.Mock(
|
||||
return_value=None
|
||||
)
|
||||
self.network_client.find_qos_dscp_marking_rule = (
|
||||
network_fakes.FakeNetworkQosRule.get_qos_rules(
|
||||
qos_rules=self.new_rule
|
||||
)
|
||||
network_fakes.get_qos_rules(qos_rules=self.new_rule)
|
||||
)
|
||||
|
||||
# Get the command object to test
|
||||
@ -746,17 +716,13 @@ class TestDeleteNetworkQosRuleBandwidthLimit(TestNetworkQosRule):
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
||||
}
|
||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
||||
attrs
|
||||
)
|
||||
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.qos_policy.rules = [self.new_rule]
|
||||
self.network_client.delete_qos_bandwidth_limit_rule = mock.Mock(
|
||||
return_value=None
|
||||
)
|
||||
self.network_client.find_qos_bandwidth_limit_rule = (
|
||||
network_fakes.FakeNetworkQosRule.get_qos_rules(
|
||||
qos_rules=self.new_rule
|
||||
)
|
||||
network_fakes.get_qos_rules(qos_rules=self.new_rule)
|
||||
)
|
||||
|
||||
# Get the command object to test
|
||||
@ -813,9 +779,7 @@ class TestSetNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': RULE_TYPE_MINIMUM_BANDWIDTH,
|
||||
}
|
||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
||||
attrs=attrs
|
||||
)
|
||||
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.qos_policy.rules = [self.new_rule]
|
||||
self.network_client.update_qos_minimum_bandwidth_rule = mock.Mock(
|
||||
return_value=None
|
||||
@ -917,9 +881,7 @@ class TestSetNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
|
||||
'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.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.qos_policy.rules = [self.new_rule]
|
||||
self.network_client.update_qos_minimum_packet_rate_rule = mock.Mock(
|
||||
return_value=None
|
||||
@ -1021,9 +983,7 @@ class TestSetNetworkQosRuleDSCPMarking(TestNetworkQosRule):
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': RULE_TYPE_DSCP_MARKING,
|
||||
}
|
||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
||||
attrs=attrs
|
||||
)
|
||||
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.qos_policy.rules = [self.new_rule]
|
||||
self.network_client.update_qos_dscp_marking_rule = mock.Mock(
|
||||
return_value=None
|
||||
@ -1124,21 +1084,14 @@ class TestSetNetworkQosRuleBandwidthLimit(TestNetworkQosRule):
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
||||
}
|
||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
||||
attrs=attrs
|
||||
)
|
||||
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.qos_policy.rules = [self.new_rule]
|
||||
self.network_client.update_qos_bandwidth_limit_rule = mock.Mock(
|
||||
return_value=None
|
||||
)
|
||||
self.network_client.find_qos_bandwidth_limit_rule = mock.Mock(
|
||||
return_value=self.new_rule
|
||||
)
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=self.qos_policy
|
||||
self.network_client.update_qos_bandwidth_limit_rule.return_value = None
|
||||
self.network_client.find_qos_bandwidth_limit_rule.return_value = (
|
||||
self.new_rule
|
||||
)
|
||||
self.network_client.find_qos_policy.return_value = self.qos_policy
|
||||
|
||||
# Get the command object to test
|
||||
self.cmd = network_qos_rule.SetNetworkQosRule(self.app, None)
|
||||
|
||||
def test_set_nothing(self):
|
||||
@ -1203,23 +1156,23 @@ class TestSetNetworkQosRuleBandwidthLimit(TestNetworkQosRule):
|
||||
self._set_max_burst_kbits(max_burst_kbits=0)
|
||||
|
||||
def _reset_max_burst_kbits(self, max_burst_kbits):
|
||||
self.new_rule.max_burst_kbits = max_burst_kbits
|
||||
self.new_rule.max_burst_kbps = max_burst_kbits
|
||||
|
||||
def _set_max_burst_kbits(self, max_burst_kbits=None):
|
||||
if max_burst_kbits:
|
||||
self.addCleanup(
|
||||
self._reset_max_burst_kbits, self.new_rule.max_burst_kbits
|
||||
self._reset_max_burst_kbits, self.new_rule.max_burst_kbps
|
||||
)
|
||||
self.new_rule.max_burst_kbits = max_burst_kbits
|
||||
self.new_rule.max_burst_kbps = max_burst_kbits
|
||||
|
||||
arglist = [
|
||||
'--max-burst-kbits',
|
||||
str(self.new_rule.max_burst_kbits),
|
||||
str(self.new_rule.max_burst_kbps),
|
||||
self.new_rule.qos_policy_id,
|
||||
self.new_rule.id,
|
||||
]
|
||||
verifylist = [
|
||||
('max_burst_kbits', self.new_rule.max_burst_kbits),
|
||||
('max_burst_kbits', self.new_rule.max_burst_kbps),
|
||||
('qos_policy', self.new_rule.qos_policy_id),
|
||||
('id', self.new_rule.id),
|
||||
]
|
||||
@ -1228,7 +1181,7 @@ class TestSetNetworkQosRuleBandwidthLimit(TestNetworkQosRule):
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
|
||||
attrs = {
|
||||
'max_burst_kbps': self.new_rule.max_burst_kbits,
|
||||
'max_burst_kbps': self.new_rule.max_burst_kbps,
|
||||
}
|
||||
self.network_client.update_qos_bandwidth_limit_rule.assert_called_with(
|
||||
self.new_rule, self.qos_policy.id, **attrs
|
||||
@ -1297,43 +1250,36 @@ class TestSetNetworkQosRuleBandwidthLimit(TestNetworkQosRule):
|
||||
class TestListNetworkQosRule(TestNetworkQosRule):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
attrs = {
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'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)
|
||||
)
|
||||
attrs['type'] = RULE_TYPE_BANDWIDTH_LIMIT
|
||||
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,
|
||||
{
|
||||
'max_kbps': 1024,
|
||||
'max_burst_kbps': 1024,
|
||||
'direction': 'egress',
|
||||
'id': 'qos-rule-id-' + uuid.uuid4().hex,
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': 'bandwidth_limit',
|
||||
},
|
||||
{
|
||||
'dscp_mark': 0,
|
||||
'id': 'qos-rule-id-' + uuid.uuid4().hex,
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': 'dscp_marking',
|
||||
},
|
||||
{
|
||||
'min_kbps': 1024,
|
||||
'direction': 'egress',
|
||||
'id': 'qos-rule-id-' + uuid.uuid4().hex,
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': 'minimum_bandwidth',
|
||||
},
|
||||
{
|
||||
'min_kpps': 2800,
|
||||
'direction': 'egress',
|
||||
'id': 'qos-rule-id-' + uuid.uuid4().hex,
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': 'minimum_packet_rate',
|
||||
},
|
||||
]
|
||||
self.network_client.find_qos_minimum_bandwidth_rule = mock.Mock(
|
||||
return_value=self.new_rule_min_bw
|
||||
)
|
||||
self.network_client.find_qos_minimum_packet_rate_rule = mock.Mock(
|
||||
return_value=self.new_rule_min_pps
|
||||
)
|
||||
self.network_client.find_qos_dscp_marking_rule = mock.Mock(
|
||||
return_value=self.new_rule_dscp_mark
|
||||
)
|
||||
self.network_client.find_qos_bandwidth_limit_rule = mock.Mock(
|
||||
return_value=self.new_rule_max_bw
|
||||
)
|
||||
self.columns = (
|
||||
'ID',
|
||||
'QoS Policy ID',
|
||||
@ -1349,20 +1295,17 @@ class TestListNetworkQosRule(TestNetworkQosRule):
|
||||
for index in range(len(self.qos_policy.rules)):
|
||||
self.data.append(
|
||||
(
|
||||
self.qos_policy.rules[index].id,
|
||||
self.qos_policy.rules[index].qos_policy_id,
|
||||
self.qos_policy.rules[index].type,
|
||||
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', ''),
|
||||
self.qos_policy.rules[index]['id'],
|
||||
self.qos_policy.id,
|
||||
self.qos_policy.rules[index]['type'],
|
||||
self.qos_policy.rules[index].get('max_kbps', ''),
|
||||
self.qos_policy.rules[index].get('max_burst_kbps', ''),
|
||||
self.qos_policy.rules[index].get('min_kbps', ''),
|
||||
self.qos_policy.rules[index].get('min_kpps', ''),
|
||||
self.qos_policy.rules[index].get('dscp_mark', ''),
|
||||
self.qos_policy.rules[index].get('direction', ''),
|
||||
)
|
||||
)
|
||||
# Get the command object to test
|
||||
self.cmd = network_qos_rule.ListNetworkQosRule(self.app, None)
|
||||
|
||||
def test_qos_rule_list(self):
|
||||
@ -1391,16 +1334,13 @@ class TestShowNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': RULE_TYPE_MINIMUM_BANDWIDTH,
|
||||
}
|
||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
||||
attrs
|
||||
)
|
||||
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.qos_policy.rules = [self.new_rule]
|
||||
self.columns = (
|
||||
'direction',
|
||||
'id',
|
||||
'min_kbps',
|
||||
'project_id',
|
||||
'qos_policy_id',
|
||||
'type',
|
||||
)
|
||||
self.data = (
|
||||
@ -1408,7 +1348,6 @@ class TestShowNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
|
||||
self.new_rule.id,
|
||||
self.new_rule.min_kbps,
|
||||
self.new_rule.project_id,
|
||||
self.new_rule.qos_policy_id,
|
||||
self.new_rule.type,
|
||||
)
|
||||
|
||||
@ -1459,16 +1398,13 @@ class TestShowNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
|
||||
'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.new_rule = network_fakes.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 = (
|
||||
@ -1476,7 +1412,6 @@ class TestShowNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
|
||||
self.new_rule.id,
|
||||
self.new_rule.min_kpps,
|
||||
self.new_rule.project_id,
|
||||
self.new_rule.qos_policy_id,
|
||||
self.new_rule.type,
|
||||
)
|
||||
|
||||
@ -1527,22 +1462,18 @@ class TestShowNetworkQosDSCPMarking(TestNetworkQosRule):
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': RULE_TYPE_DSCP_MARKING,
|
||||
}
|
||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
||||
attrs
|
||||
)
|
||||
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.qos_policy.rules = [self.new_rule]
|
||||
self.columns = (
|
||||
'dscp_mark',
|
||||
'id',
|
||||
'project_id',
|
||||
'qos_policy_id',
|
||||
'type',
|
||||
)
|
||||
self.data = (
|
||||
self.new_rule.dscp_mark,
|
||||
self.new_rule.id,
|
||||
self.new_rule.project_id,
|
||||
self.new_rule.qos_policy_id,
|
||||
self.new_rule.type,
|
||||
)
|
||||
|
||||
@ -1593,26 +1524,22 @@ class TestShowNetworkQosBandwidthLimit(TestNetworkQosRule):
|
||||
'qos_policy_id': self.qos_policy.id,
|
||||
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
||||
}
|
||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
||||
attrs
|
||||
)
|
||||
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||
self.qos_policy.rules = [self.new_rule]
|
||||
self.columns = (
|
||||
'direction',
|
||||
'id',
|
||||
'max_burst_kbits',
|
||||
'max_burst_kbps',
|
||||
'max_kbps',
|
||||
'project_id',
|
||||
'qos_policy_id',
|
||||
'type',
|
||||
)
|
||||
self.data = (
|
||||
self.new_rule.direction,
|
||||
self.new_rule.id,
|
||||
self.new_rule.max_burst_kbits,
|
||||
self.new_rule.max_burst_kbps,
|
||||
self.new_rule.max_kbps,
|
||||
self.new_rule.project_id,
|
||||
self.new_rule.qos_policy_id,
|
||||
self.new_rule.type,
|
||||
)
|
||||
|
||||
|
@ -28,9 +28,8 @@ class TestNetworkQosRuleType(network_fakes.TestNetworkV2):
|
||||
class TestShowNetworkQosRuleType(TestNetworkQosRuleType):
|
||||
attrs = {'drivers': [{'name': 'driver 1', 'supported_parameters': []}]}
|
||||
# The QoS policies to show.
|
||||
qos_rule_type = (
|
||||
network_fakes.FakeNetworkQosRuleType.create_one_qos_rule_type(attrs)
|
||||
)
|
||||
qos_rule_type = network_fakes.create_one_qos_rule_type(attrs)
|
||||
columns = ('drivers', 'rule_type_name')
|
||||
columns = ('drivers', 'rule_type_name')
|
||||
data = [qos_rule_type.drivers, qos_rule_type.type]
|
||||
|
||||
@ -76,9 +75,8 @@ class TestShowNetworkQosRuleType(TestNetworkQosRuleType):
|
||||
|
||||
class TestListNetworkQosRuleType(TestNetworkQosRuleType):
|
||||
# The QoS policies to list up.
|
||||
qos_rule_types = (
|
||||
network_fakes.FakeNetworkQosRuleType.create_qos_rule_types(count=3)
|
||||
)
|
||||
qos_rule_types = network_fakes.create_qos_rule_types(count=3)
|
||||
|
||||
columns = ('Type',)
|
||||
data = []
|
||||
for qos_rule_type in qos_rule_types:
|
||||
|
@ -34,8 +34,8 @@ class TestNetworkRBAC(network_fakes.TestNetworkV2):
|
||||
@ddt.ddt
|
||||
class TestCreateNetworkRBAC(TestNetworkRBAC):
|
||||
network_object = network_fakes.create_one_network()
|
||||
qos_object = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
sg_object = network_fakes.FakeNetworkSecGroup.create_one_security_group()
|
||||
qos_object = network_fakes.create_one_qos_policy()
|
||||
sg_object = network_fakes.FakeSecurityGroup.create_one_security_group()
|
||||
as_object = network_fakes.create_one_address_scope()
|
||||
snp_object = network_fakes.FakeSubnetPool.create_one_subnet_pool()
|
||||
ag_object = network_fakes.create_one_address_group()
|
||||
|
@ -582,7 +582,7 @@ class TestCreatePort(TestPort):
|
||||
self.assertCountEqual(self.data, data)
|
||||
|
||||
def test_create_port_with_qos(self):
|
||||
qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
qos_policy = network_fakes.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=qos_policy
|
||||
)
|
||||
@ -2371,7 +2371,7 @@ class TestSetPort(TestPort):
|
||||
)
|
||||
|
||||
def test_set_port_with_qos(self):
|
||||
qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
qos_policy = network_fakes.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=qos_policy
|
||||
)
|
||||
|
@ -555,9 +555,7 @@ class TestCreateRouter(TestRouter):
|
||||
def test_create_with_qos_policy(self):
|
||||
_network = network_fakes.create_one_network()
|
||||
self.network_client.find_network = mock.Mock(return_value=_network)
|
||||
_qos_policy = (
|
||||
network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
)
|
||||
_qos_policy = network_fakes.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=_qos_policy
|
||||
)
|
||||
@ -590,9 +588,7 @@ class TestCreateRouter(TestRouter):
|
||||
self.assertCountEqual(self.data, data)
|
||||
|
||||
def test_create_with_qos_policy_no_external_gateway(self):
|
||||
_qos_policy = (
|
||||
network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
)
|
||||
_qos_policy = network_fakes.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=_qos_policy
|
||||
)
|
||||
@ -1668,7 +1664,7 @@ class TestSetRouter(TestRouter):
|
||||
self._test_set_tags(with_tags=False)
|
||||
|
||||
def test_set_gateway_ip_qos(self):
|
||||
qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
qos_policy = network_fakes.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=qos_policy
|
||||
)
|
||||
@ -1725,7 +1721,7 @@ class TestSetRouter(TestRouter):
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_set_unset_gateway_ip_qos(self):
|
||||
qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
qos_policy = network_fakes.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=qos_policy
|
||||
)
|
||||
@ -1753,7 +1749,7 @@ class TestSetRouter(TestRouter):
|
||||
)
|
||||
|
||||
def test_set_gateway_ip_qos_no_gateway(self):
|
||||
qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
qos_policy = network_fakes.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=qos_policy
|
||||
)
|
||||
@ -1775,7 +1771,7 @@ class TestSetRouter(TestRouter):
|
||||
)
|
||||
|
||||
def test_unset_gateway_ip_qos_no_gateway(self):
|
||||
qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
qos_policy = network_fakes.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=qos_policy
|
||||
)
|
||||
@ -1932,9 +1928,7 @@ class TestUnsetRouter(TestRouter):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.fake_network = network_fakes.create_one_network()
|
||||
self.fake_qos_policy = (
|
||||
network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
)
|
||||
self.fake_qos_policy = network_fakes.create_one_qos_policy()
|
||||
self._testrouter = network_fakes.FakeRouter.create_one_router(
|
||||
{
|
||||
'routes': [
|
||||
@ -2102,7 +2096,7 @@ class TestUnsetRouter(TestRouter):
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_unset_gateway_ip_qos_no_network(self):
|
||||
qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
qos_policy = network_fakes.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=qos_policy
|
||||
)
|
||||
@ -2122,7 +2116,7 @@ class TestUnsetRouter(TestRouter):
|
||||
)
|
||||
|
||||
def test_unset_gateway_ip_qos_no_qos(self):
|
||||
qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
||||
qos_policy = network_fakes.create_one_qos_policy()
|
||||
self.network_client.find_qos_policy = mock.Mock(
|
||||
return_value=qos_policy
|
||||
)
|
||||
|
Loading…
x
Reference in New Issue
Block a user