Merge "Refactor network fakes to sdk properties PART 5"
This commit is contained in:
commit
0874e2d33f
openstackclient
@ -73,7 +73,7 @@ ACTION_SHOW = 'get'
|
|||||||
|
|
||||||
|
|
||||||
def _get_columns(item):
|
def _get_columns(item):
|
||||||
hidden_columns = ['location', 'tenant_id']
|
hidden_columns = ['location', 'name', 'tenant_id']
|
||||||
return utils.get_osc_show_columns_for_sdk_resource(
|
return utils.get_osc_show_columns_for_sdk_resource(
|
||||||
item, {}, hidden_columns
|
item, {}, hidden_columns
|
||||||
)
|
)
|
||||||
@ -148,14 +148,6 @@ def _get_attrs(network_client, parsed_args, is_create=False):
|
|||||||
return attrs
|
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):
|
def _rule_action_call(client, action, rule_type):
|
||||||
rule_type = rule_type.replace('-', '_')
|
rule_type = rule_type.replace('-', '_')
|
||||||
func_name = f'{action}_qos_{rule_type}_rule'
|
func_name = f'{action}_qos_{rule_type}_rule'
|
||||||
@ -357,10 +349,10 @@ class ListNetworkQosRule(command.Lister):
|
|||||||
qos = client.find_qos_policy(
|
qos = client.find_qos_policy(
|
||||||
parsed_args.qos_policy, ignore_missing=False
|
parsed_args.qos_policy, ignore_missing=False
|
||||||
)
|
)
|
||||||
data = qos.rules
|
|
||||||
return (
|
return (
|
||||||
column_headers,
|
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:
|
if parse_output:
|
||||||
return json.loads(output)
|
try:
|
||||||
|
return json.loads(output)
|
||||||
|
except json.JSONDecodeError:
|
||||||
|
print(f'failed to decode: {output}')
|
||||||
|
raise
|
||||||
else:
|
else:
|
||||||
return output
|
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_ip_availability as _ip_availability
|
||||||
from openstack.network.v2 import network_segment_range as _segment_range
|
from openstack.network.v2 import network_segment_range as _segment_range
|
||||||
from openstack.network.v2 import port as _port
|
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 rbac_policy as network_rbac
|
||||||
from openstack.network.v2 import security_group as _security_group
|
from openstack.network.v2 import security_group as _security_group
|
||||||
from openstack.network.v2 import segment as _segment
|
from openstack.network.v2 import segment as _segment
|
||||||
@ -118,245 +132,6 @@ def create_one_extension(attrs=None):
|
|||||||
return extension
|
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:
|
class FakeRouter:
|
||||||
"""Fake one or more routers."""
|
"""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)
|
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):
|
def create_one_local_ip(attrs=None):
|
||||||
"""Create a fake local ip.
|
"""Create a fake local ip.
|
||||||
|
|
||||||
|
@ -234,10 +234,8 @@ class TestCreateFloatingIPNetwork(TestFloatingIPNetwork):
|
|||||||
self.assertEqual(self.data, data)
|
self.assertEqual(self.data, data)
|
||||||
|
|
||||||
def test_create_floating_ip_with_qos(self):
|
def test_create_floating_ip_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(
|
self.network_client.find_qos_policy.return_value = qos_policy
|
||||||
return_value=qos_policy
|
|
||||||
)
|
|
||||||
arglist = [
|
arglist = [
|
||||||
'--qos-policy',
|
'--qos-policy',
|
||||||
qos_policy.id,
|
qos_policy.id,
|
||||||
@ -893,10 +891,8 @@ class TestSetFloatingIP(TestFloatingIPNetwork):
|
|||||||
)
|
)
|
||||||
|
|
||||||
def test_qos_policy_option(self):
|
def test_qos_policy_option(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(
|
self.network_client.find_qos_policy.return_value = qos_policy
|
||||||
return_value=qos_policy
|
|
||||||
)
|
|
||||||
arglist = [
|
arglist = [
|
||||||
"--qos-policy",
|
"--qos-policy",
|
||||||
qos_policy.id,
|
qos_policy.id,
|
||||||
@ -922,10 +918,8 @@ class TestSetFloatingIP(TestFloatingIPNetwork):
|
|||||||
)
|
)
|
||||||
|
|
||||||
def test_port_and_qos_policy_option(self):
|
def test_port_and_qos_policy_option(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(
|
self.network_client.find_qos_policy.return_value = qos_policy
|
||||||
return_value=qos_policy
|
|
||||||
)
|
|
||||||
arglist = [
|
arglist = [
|
||||||
"--qos-policy",
|
"--qos-policy",
|
||||||
qos_policy.id,
|
qos_policy.id,
|
||||||
|
@ -47,7 +47,7 @@ class TestCreateNetworkIdentityV3(TestNetwork):
|
|||||||
'availability_zone_hints': ["nova"],
|
'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}
|
attrs={'id': _network.qos_policy_id}
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -622,7 +622,7 @@ class TestListNetwork(TestNetwork):
|
|||||||
self.network_client.networks = mock.Mock(return_value=self._network)
|
self.network_client.networks = mock.Mock(return_value=self._network)
|
||||||
|
|
||||||
self._agent = network_fakes.create_one_network_agent()
|
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(
|
self.network_client.dhcp_agent_hosting_networks = mock.Mock(
|
||||||
return_value=self._network
|
return_value=self._network
|
||||||
@ -967,7 +967,7 @@ class TestListNetwork(TestNetwork):
|
|||||||
class TestSetNetwork(TestNetwork):
|
class TestSetNetwork(TestNetwork):
|
||||||
# The network to set.
|
# The network to set.
|
||||||
_network = network_fakes.create_one_network({'tags': ['green', 'red']})
|
_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}
|
attrs={'id': _network.qos_policy_id}
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -1266,7 +1266,7 @@ class TestShowNetwork(TestNetwork):
|
|||||||
class TestUnsetNetwork(TestNetwork):
|
class TestUnsetNetwork(TestNetwork):
|
||||||
# The network to set.
|
# The network to set.
|
||||||
_network = network_fakes.create_one_network({'tags': ['green', 'red']})
|
_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}
|
attrs={'id': _network.qos_policy_id}
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -35,11 +35,10 @@ class TestCreateNetworkQosPolicy(TestQosPolicy):
|
|||||||
project = identity_fakes_v3.FakeProject.create_one_project()
|
project = identity_fakes_v3.FakeProject.create_one_project()
|
||||||
|
|
||||||
# The new qos policy created.
|
# The new qos policy created.
|
||||||
new_qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy(
|
new_qos_policy = network_fakes.create_one_qos_policy(
|
||||||
attrs={
|
attrs={'project_id': project.id}
|
||||||
'project_id': project.id,
|
|
||||||
}
|
|
||||||
)
|
)
|
||||||
|
|
||||||
columns = (
|
columns = (
|
||||||
'description',
|
'description',
|
||||||
'id',
|
'id',
|
||||||
@ -48,6 +47,7 @@ class TestCreateNetworkQosPolicy(TestQosPolicy):
|
|||||||
'project_id',
|
'project_id',
|
||||||
'rules',
|
'rules',
|
||||||
'shared',
|
'shared',
|
||||||
|
'tags',
|
||||||
)
|
)
|
||||||
|
|
||||||
data = (
|
data = (
|
||||||
@ -57,7 +57,8 @@ class TestCreateNetworkQosPolicy(TestQosPolicy):
|
|||||||
new_qos_policy.name,
|
new_qos_policy.name,
|
||||||
new_qos_policy.project_id,
|
new_qos_policy.project_id,
|
||||||
new_qos_policy.rules,
|
new_qos_policy.rules,
|
||||||
new_qos_policy.shared,
|
new_qos_policy.is_shared,
|
||||||
|
new_qos_policy.tags,
|
||||||
)
|
)
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
@ -158,17 +159,13 @@ class TestCreateNetworkQosPolicy(TestQosPolicy):
|
|||||||
|
|
||||||
class TestDeleteNetworkQosPolicy(TestQosPolicy):
|
class TestDeleteNetworkQosPolicy(TestQosPolicy):
|
||||||
# The address scope to delete.
|
# The address scope to delete.
|
||||||
_qos_policies = network_fakes.FakeNetworkQosPolicy.create_qos_policies(
|
_qos_policies = network_fakes.create_qos_policies(count=2)
|
||||||
count=2
|
|
||||||
)
|
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
super().setUp()
|
super().setUp()
|
||||||
self.network_client.delete_qos_policy = mock.Mock(return_value=None)
|
self.network_client.delete_qos_policy = mock.Mock(return_value=None)
|
||||||
self.network_client.find_qos_policy = (
|
self.network_client.find_qos_policy = network_fakes.get_qos_policies(
|
||||||
network_fakes.FakeNetworkQosPolicy.get_qos_policies(
|
qos_policies=self._qos_policies
|
||||||
qos_policies=self._qos_policies
|
|
||||||
)
|
|
||||||
)
|
)
|
||||||
|
|
||||||
# Get the command object to test
|
# Get the command object to test
|
||||||
@ -245,9 +242,7 @@ class TestDeleteNetworkQosPolicy(TestQosPolicy):
|
|||||||
|
|
||||||
class TestListNetworkQosPolicy(TestQosPolicy):
|
class TestListNetworkQosPolicy(TestQosPolicy):
|
||||||
# The QoS policies to list up.
|
# The QoS policies to list up.
|
||||||
qos_policies = network_fakes.FakeNetworkQosPolicy.create_qos_policies(
|
qos_policies = network_fakes.create_qos_policies(count=3)
|
||||||
count=3
|
|
||||||
)
|
|
||||||
columns = (
|
columns = (
|
||||||
'ID',
|
'ID',
|
||||||
'Name',
|
'Name',
|
||||||
@ -261,7 +256,7 @@ class TestListNetworkQosPolicy(TestQosPolicy):
|
|||||||
(
|
(
|
||||||
qos_policy.id,
|
qos_policy.id,
|
||||||
qos_policy.name,
|
qos_policy.name,
|
||||||
qos_policy.shared,
|
qos_policy.is_shared,
|
||||||
qos_policy.is_default,
|
qos_policy.is_default,
|
||||||
qos_policy.project_id,
|
qos_policy.project_id,
|
||||||
)
|
)
|
||||||
@ -345,7 +340,7 @@ class TestListNetworkQosPolicy(TestQosPolicy):
|
|||||||
|
|
||||||
class TestSetNetworkQosPolicy(TestQosPolicy):
|
class TestSetNetworkQosPolicy(TestQosPolicy):
|
||||||
# The QoS policy to set.
|
# 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):
|
def setUp(self):
|
||||||
super().setUp()
|
super().setUp()
|
||||||
@ -428,7 +423,7 @@ class TestSetNetworkQosPolicy(TestQosPolicy):
|
|||||||
|
|
||||||
class TestShowNetworkQosPolicy(TestQosPolicy):
|
class TestShowNetworkQosPolicy(TestQosPolicy):
|
||||||
# The QoS policy to show.
|
# The QoS policy to show.
|
||||||
_qos_policy = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
_qos_policy = network_fakes.create_one_qos_policy()
|
||||||
columns = (
|
columns = (
|
||||||
'description',
|
'description',
|
||||||
'id',
|
'id',
|
||||||
@ -437,6 +432,7 @@ class TestShowNetworkQosPolicy(TestQosPolicy):
|
|||||||
'project_id',
|
'project_id',
|
||||||
'rules',
|
'rules',
|
||||||
'shared',
|
'shared',
|
||||||
|
'tags',
|
||||||
)
|
)
|
||||||
data = (
|
data = (
|
||||||
_qos_policy.description,
|
_qos_policy.description,
|
||||||
@ -445,7 +441,8 @@ class TestShowNetworkQosPolicy(TestQosPolicy):
|
|||||||
_qos_policy.name,
|
_qos_policy.name,
|
||||||
_qos_policy.project_id,
|
_qos_policy.project_id,
|
||||||
network_qos_policy.RulesColumn(_qos_policy.rules),
|
network_qos_policy.RulesColumn(_qos_policy.rules),
|
||||||
_qos_policy.shared,
|
_qos_policy.is_shared,
|
||||||
|
_qos_policy.tags,
|
||||||
)
|
)
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
|
@ -14,6 +14,7 @@
|
|||||||
# under the License.
|
# under the License.
|
||||||
|
|
||||||
from unittest import mock
|
from unittest import mock
|
||||||
|
import uuid
|
||||||
|
|
||||||
from osc_lib import exceptions
|
from osc_lib import exceptions
|
||||||
|
|
||||||
@ -54,9 +55,7 @@ DSCP_VALID_MARKS = [
|
|||||||
class TestNetworkQosRule(network_fakes.TestNetworkV2):
|
class TestNetworkQosRule(network_fakes.TestNetworkV2):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
super().setUp()
|
super().setUp()
|
||||||
self.qos_policy = (
|
self.qos_policy = network_fakes.create_one_qos_policy()
|
||||||
network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
|
||||||
)
|
|
||||||
self.network_client.find_qos_policy = mock.Mock(
|
self.network_client.find_qos_policy = mock.Mock(
|
||||||
return_value=self.qos_policy
|
return_value=self.qos_policy
|
||||||
)
|
)
|
||||||
@ -72,15 +71,12 @@ class TestCreateNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_MINIMUM_BANDWIDTH,
|
'type': RULE_TYPE_MINIMUM_BANDWIDTH,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs
|
|
||||||
)
|
|
||||||
self.columns = (
|
self.columns = (
|
||||||
'direction',
|
'direction',
|
||||||
'id',
|
'id',
|
||||||
'min_kbps',
|
'min_kbps',
|
||||||
'project_id',
|
'project_id',
|
||||||
'qos_policy_id',
|
|
||||||
'type',
|
'type',
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -89,7 +85,6 @@ class TestCreateNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
|
|||||||
self.new_rule.id,
|
self.new_rule.id,
|
||||||
self.new_rule.min_kbps,
|
self.new_rule.min_kbps,
|
||||||
self.new_rule.project_id,
|
self.new_rule.project_id,
|
||||||
self.new_rule.qos_policy_id,
|
|
||||||
self.new_rule.type,
|
self.new_rule.type,
|
||||||
)
|
)
|
||||||
self.network_client.create_qos_minimum_bandwidth_rule = mock.Mock(
|
self.network_client.create_qos_minimum_bandwidth_rule = mock.Mock(
|
||||||
@ -179,15 +174,12 @@ class TestCreateNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_MINIMUM_PACKET_RATE,
|
'type': RULE_TYPE_MINIMUM_PACKET_RATE,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs
|
|
||||||
)
|
|
||||||
self.columns = (
|
self.columns = (
|
||||||
'direction',
|
'direction',
|
||||||
'id',
|
'id',
|
||||||
'min_kpps',
|
'min_kpps',
|
||||||
'project_id',
|
'project_id',
|
||||||
'qos_policy_id',
|
|
||||||
'type',
|
'type',
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -196,7 +188,6 @@ class TestCreateNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
|
|||||||
self.new_rule.id,
|
self.new_rule.id,
|
||||||
self.new_rule.min_kpps,
|
self.new_rule.min_kpps,
|
||||||
self.new_rule.project_id,
|
self.new_rule.project_id,
|
||||||
self.new_rule.qos_policy_id,
|
|
||||||
self.new_rule.type,
|
self.new_rule.type,
|
||||||
)
|
)
|
||||||
self.network_client.create_qos_minimum_packet_rate_rule = mock.Mock(
|
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,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_DSCP_MARKING,
|
'type': RULE_TYPE_DSCP_MARKING,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs
|
|
||||||
)
|
|
||||||
self.columns = (
|
self.columns = (
|
||||||
'dscp_mark',
|
'dscp_mark',
|
||||||
'id',
|
'id',
|
||||||
'project_id',
|
'project_id',
|
||||||
'qos_policy_id',
|
|
||||||
'type',
|
'type',
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -301,7 +289,6 @@ class TestCreateNetworkQosRuleDSCPMarking(TestNetworkQosRule):
|
|||||||
self.new_rule.dscp_mark,
|
self.new_rule.dscp_mark,
|
||||||
self.new_rule.id,
|
self.new_rule.id,
|
||||||
self.new_rule.project_id,
|
self.new_rule.project_id,
|
||||||
self.new_rule.qos_policy_id,
|
|
||||||
self.new_rule.type,
|
self.new_rule.type,
|
||||||
)
|
)
|
||||||
self.network_client.create_qos_dscp_marking_rule = mock.Mock(
|
self.network_client.create_qos_dscp_marking_rule = mock.Mock(
|
||||||
@ -384,26 +371,22 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs
|
|
||||||
)
|
|
||||||
self.columns = (
|
self.columns = (
|
||||||
'direction',
|
'direction',
|
||||||
'id',
|
'id',
|
||||||
'max_burst_kbits',
|
'max_burst_kbps',
|
||||||
'max_kbps',
|
'max_kbps',
|
||||||
'project_id',
|
'project_id',
|
||||||
'qos_policy_id',
|
|
||||||
'type',
|
'type',
|
||||||
)
|
)
|
||||||
|
|
||||||
self.data = (
|
self.data = (
|
||||||
self.new_rule.direction,
|
self.new_rule.direction,
|
||||||
self.new_rule.id,
|
self.new_rule.id,
|
||||||
self.new_rule.max_burst_kbits,
|
self.new_rule.max_burst_kbps,
|
||||||
self.new_rule.max_kbps,
|
self.new_rule.max_kbps,
|
||||||
self.new_rule.project_id,
|
self.new_rule.project_id,
|
||||||
self.new_rule.qos_policy_id,
|
|
||||||
self.new_rule.type,
|
self.new_rule.type,
|
||||||
)
|
)
|
||||||
self.network_client.create_qos_bandwidth_limit_rule = mock.Mock(
|
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),
|
('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,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
rule.max_burst_kbits = 0
|
rule.max_burst_kbps = 0
|
||||||
expected_data = (
|
expected_data = (
|
||||||
rule.direction,
|
rule.direction,
|
||||||
rule.id,
|
rule.id,
|
||||||
rule.max_burst_kbits,
|
rule.max_burst_kbps,
|
||||||
rule.max_kbps,
|
rule.max_kbps,
|
||||||
rule.project_id,
|
rule.project_id,
|
||||||
rule.qos_policy_id,
|
|
||||||
rule.type,
|
rule.type,
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -485,7 +467,7 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
|
|||||||
'--max-kbps',
|
'--max-kbps',
|
||||||
str(self.new_rule.max_kbps),
|
str(self.new_rule.max_kbps),
|
||||||
'--max-burst-kbits',
|
'--max-burst-kbits',
|
||||||
str(self.new_rule.max_burst_kbits),
|
str(self.new_rule.max_burst_kbps),
|
||||||
'--egress',
|
'--egress',
|
||||||
self.new_rule.qos_policy_id,
|
self.new_rule.qos_policy_id,
|
||||||
]
|
]
|
||||||
@ -493,7 +475,7 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
|
|||||||
verifylist = [
|
verifylist = [
|
||||||
('type', RULE_TYPE_BANDWIDTH_LIMIT),
|
('type', RULE_TYPE_BANDWIDTH_LIMIT),
|
||||||
('max_kbps', self.new_rule.max_kbps),
|
('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),
|
('egress', True),
|
||||||
('qos_policy', self.new_rule.qos_policy_id),
|
('qos_policy', self.new_rule.qos_policy_id),
|
||||||
]
|
]
|
||||||
@ -505,7 +487,7 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
|
|||||||
self.qos_policy.id,
|
self.qos_policy.id,
|
||||||
**{
|
**{
|
||||||
'max_kbps': self.new_rule.max_kbps,
|
'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,
|
'direction': self.new_rule.direction,
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
@ -545,17 +527,13 @@ class TestDeleteNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_MINIMUM_BANDWIDTH,
|
'type': RULE_TYPE_MINIMUM_BANDWIDTH,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs
|
|
||||||
)
|
|
||||||
self.qos_policy.rules = [self.new_rule]
|
self.qos_policy.rules = [self.new_rule]
|
||||||
self.network_client.delete_qos_minimum_bandwidth_rule = mock.Mock(
|
self.network_client.delete_qos_minimum_bandwidth_rule = mock.Mock(
|
||||||
return_value=None
|
return_value=None
|
||||||
)
|
)
|
||||||
self.network_client.find_qos_minimum_bandwidth_rule = (
|
self.network_client.find_qos_minimum_bandwidth_rule = (
|
||||||
network_fakes.FakeNetworkQosRule.get_qos_rules(
|
network_fakes.get_qos_rules(qos_rules=self.new_rule)
|
||||||
qos_rules=self.new_rule
|
|
||||||
)
|
|
||||||
)
|
)
|
||||||
|
|
||||||
# Get the command object to test
|
# Get the command object to test
|
||||||
@ -612,17 +590,13 @@ class TestDeleteNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_MINIMUM_PACKET_RATE,
|
'type': RULE_TYPE_MINIMUM_PACKET_RATE,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs
|
|
||||||
)
|
|
||||||
self.qos_policy.rules = [self.new_rule]
|
self.qos_policy.rules = [self.new_rule]
|
||||||
self.network_client.delete_qos_minimum_packet_rate_rule = mock.Mock(
|
self.network_client.delete_qos_minimum_packet_rate_rule = mock.Mock(
|
||||||
return_value=None
|
return_value=None
|
||||||
)
|
)
|
||||||
self.network_client.find_qos_minimum_packet_rate_rule = (
|
self.network_client.find_qos_minimum_packet_rate_rule = (
|
||||||
network_fakes.FakeNetworkQosRule.get_qos_rules(
|
network_fakes.get_qos_rules(qos_rules=self.new_rule)
|
||||||
qos_rules=self.new_rule
|
|
||||||
)
|
|
||||||
)
|
)
|
||||||
|
|
||||||
# Get the command object to test
|
# Get the command object to test
|
||||||
@ -679,17 +653,13 @@ class TestDeleteNetworkQosRuleDSCPMarking(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_DSCP_MARKING,
|
'type': RULE_TYPE_DSCP_MARKING,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs
|
|
||||||
)
|
|
||||||
self.qos_policy.rules = [self.new_rule]
|
self.qos_policy.rules = [self.new_rule]
|
||||||
self.network_client.delete_qos_dscp_marking_rule = mock.Mock(
|
self.network_client.delete_qos_dscp_marking_rule = mock.Mock(
|
||||||
return_value=None
|
return_value=None
|
||||||
)
|
)
|
||||||
self.network_client.find_qos_dscp_marking_rule = (
|
self.network_client.find_qos_dscp_marking_rule = (
|
||||||
network_fakes.FakeNetworkQosRule.get_qos_rules(
|
network_fakes.get_qos_rules(qos_rules=self.new_rule)
|
||||||
qos_rules=self.new_rule
|
|
||||||
)
|
|
||||||
)
|
)
|
||||||
|
|
||||||
# Get the command object to test
|
# Get the command object to test
|
||||||
@ -746,17 +716,13 @@ class TestDeleteNetworkQosRuleBandwidthLimit(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs
|
|
||||||
)
|
|
||||||
self.qos_policy.rules = [self.new_rule]
|
self.qos_policy.rules = [self.new_rule]
|
||||||
self.network_client.delete_qos_bandwidth_limit_rule = mock.Mock(
|
self.network_client.delete_qos_bandwidth_limit_rule = mock.Mock(
|
||||||
return_value=None
|
return_value=None
|
||||||
)
|
)
|
||||||
self.network_client.find_qos_bandwidth_limit_rule = (
|
self.network_client.find_qos_bandwidth_limit_rule = (
|
||||||
network_fakes.FakeNetworkQosRule.get_qos_rules(
|
network_fakes.get_qos_rules(qos_rules=self.new_rule)
|
||||||
qos_rules=self.new_rule
|
|
||||||
)
|
|
||||||
)
|
)
|
||||||
|
|
||||||
# Get the command object to test
|
# Get the command object to test
|
||||||
@ -813,9 +779,7 @@ class TestSetNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_MINIMUM_BANDWIDTH,
|
'type': RULE_TYPE_MINIMUM_BANDWIDTH,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs=attrs
|
|
||||||
)
|
|
||||||
self.qos_policy.rules = [self.new_rule]
|
self.qos_policy.rules = [self.new_rule]
|
||||||
self.network_client.update_qos_minimum_bandwidth_rule = mock.Mock(
|
self.network_client.update_qos_minimum_bandwidth_rule = mock.Mock(
|
||||||
return_value=None
|
return_value=None
|
||||||
@ -917,9 +881,7 @@ class TestSetNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_MINIMUM_PACKET_RATE,
|
'type': RULE_TYPE_MINIMUM_PACKET_RATE,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs=attrs
|
|
||||||
)
|
|
||||||
self.qos_policy.rules = [self.new_rule]
|
self.qos_policy.rules = [self.new_rule]
|
||||||
self.network_client.update_qos_minimum_packet_rate_rule = mock.Mock(
|
self.network_client.update_qos_minimum_packet_rate_rule = mock.Mock(
|
||||||
return_value=None
|
return_value=None
|
||||||
@ -1021,9 +983,7 @@ class TestSetNetworkQosRuleDSCPMarking(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_DSCP_MARKING,
|
'type': RULE_TYPE_DSCP_MARKING,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs=attrs
|
|
||||||
)
|
|
||||||
self.qos_policy.rules = [self.new_rule]
|
self.qos_policy.rules = [self.new_rule]
|
||||||
self.network_client.update_qos_dscp_marking_rule = mock.Mock(
|
self.network_client.update_qos_dscp_marking_rule = mock.Mock(
|
||||||
return_value=None
|
return_value=None
|
||||||
@ -1124,21 +1084,14 @@ class TestSetNetworkQosRuleBandwidthLimit(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs=attrs
|
|
||||||
)
|
|
||||||
self.qos_policy.rules = [self.new_rule]
|
self.qos_policy.rules = [self.new_rule]
|
||||||
self.network_client.update_qos_bandwidth_limit_rule = mock.Mock(
|
self.network_client.update_qos_bandwidth_limit_rule.return_value = None
|
||||||
return_value=None
|
self.network_client.find_qos_bandwidth_limit_rule.return_value = (
|
||||||
)
|
self.new_rule
|
||||||
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.find_qos_policy.return_value = self.qos_policy
|
||||||
|
|
||||||
# Get the command object to test
|
|
||||||
self.cmd = network_qos_rule.SetNetworkQosRule(self.app, None)
|
self.cmd = network_qos_rule.SetNetworkQosRule(self.app, None)
|
||||||
|
|
||||||
def test_set_nothing(self):
|
def test_set_nothing(self):
|
||||||
@ -1203,23 +1156,23 @@ class TestSetNetworkQosRuleBandwidthLimit(TestNetworkQosRule):
|
|||||||
self._set_max_burst_kbits(max_burst_kbits=0)
|
self._set_max_burst_kbits(max_burst_kbits=0)
|
||||||
|
|
||||||
def _reset_max_burst_kbits(self, max_burst_kbits):
|
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):
|
def _set_max_burst_kbits(self, max_burst_kbits=None):
|
||||||
if max_burst_kbits:
|
if max_burst_kbits:
|
||||||
self.addCleanup(
|
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 = [
|
arglist = [
|
||||||
'--max-burst-kbits',
|
'--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.qos_policy_id,
|
||||||
self.new_rule.id,
|
self.new_rule.id,
|
||||||
]
|
]
|
||||||
verifylist = [
|
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),
|
('qos_policy', self.new_rule.qos_policy_id),
|
||||||
('id', self.new_rule.id),
|
('id', self.new_rule.id),
|
||||||
]
|
]
|
||||||
@ -1228,7 +1181,7 @@ class TestSetNetworkQosRuleBandwidthLimit(TestNetworkQosRule):
|
|||||||
result = self.cmd.take_action(parsed_args)
|
result = self.cmd.take_action(parsed_args)
|
||||||
|
|
||||||
attrs = {
|
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.network_client.update_qos_bandwidth_limit_rule.assert_called_with(
|
||||||
self.new_rule, self.qos_policy.id, **attrs
|
self.new_rule, self.qos_policy.id, **attrs
|
||||||
@ -1297,43 +1250,36 @@ class TestSetNetworkQosRuleBandwidthLimit(TestNetworkQosRule):
|
|||||||
class TestListNetworkQosRule(TestNetworkQosRule):
|
class TestListNetworkQosRule(TestNetworkQosRule):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
super().setUp()
|
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.qos_policy.rules = [
|
||||||
self.new_rule_min_bw,
|
{
|
||||||
self.new_rule_min_pps,
|
'max_kbps': 1024,
|
||||||
self.new_rule_dscp_mark,
|
'max_burst_kbps': 1024,
|
||||||
self.new_rule_max_bw,
|
'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 = (
|
self.columns = (
|
||||||
'ID',
|
'ID',
|
||||||
'QoS Policy ID',
|
'QoS Policy ID',
|
||||||
@ -1349,20 +1295,17 @@ class TestListNetworkQosRule(TestNetworkQosRule):
|
|||||||
for index in range(len(self.qos_policy.rules)):
|
for index in range(len(self.qos_policy.rules)):
|
||||||
self.data.append(
|
self.data.append(
|
||||||
(
|
(
|
||||||
self.qos_policy.rules[index].id,
|
self.qos_policy.rules[index]['id'],
|
||||||
self.qos_policy.rules[index].qos_policy_id,
|
self.qos_policy.id,
|
||||||
self.qos_policy.rules[index].type,
|
self.qos_policy.rules[index]['type'],
|
||||||
getattr(self.qos_policy.rules[index], 'max_kbps', ''),
|
self.qos_policy.rules[index].get('max_kbps', ''),
|
||||||
getattr(
|
self.qos_policy.rules[index].get('max_burst_kbps', ''),
|
||||||
self.qos_policy.rules[index], 'max_burst_kbps', ''
|
self.qos_policy.rules[index].get('min_kbps', ''),
|
||||||
),
|
self.qos_policy.rules[index].get('min_kpps', ''),
|
||||||
getattr(self.qos_policy.rules[index], 'min_kbps', ''),
|
self.qos_policy.rules[index].get('dscp_mark', ''),
|
||||||
getattr(self.qos_policy.rules[index], 'min_kpps', ''),
|
self.qos_policy.rules[index].get('direction', ''),
|
||||||
getattr(self.qos_policy.rules[index], 'dscp_mark', ''),
|
|
||||||
getattr(self.qos_policy.rules[index], 'direction', ''),
|
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
# Get the command object to test
|
|
||||||
self.cmd = network_qos_rule.ListNetworkQosRule(self.app, None)
|
self.cmd = network_qos_rule.ListNetworkQosRule(self.app, None)
|
||||||
|
|
||||||
def test_qos_rule_list(self):
|
def test_qos_rule_list(self):
|
||||||
@ -1391,16 +1334,13 @@ class TestShowNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_MINIMUM_BANDWIDTH,
|
'type': RULE_TYPE_MINIMUM_BANDWIDTH,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs
|
|
||||||
)
|
|
||||||
self.qos_policy.rules = [self.new_rule]
|
self.qos_policy.rules = [self.new_rule]
|
||||||
self.columns = (
|
self.columns = (
|
||||||
'direction',
|
'direction',
|
||||||
'id',
|
'id',
|
||||||
'min_kbps',
|
'min_kbps',
|
||||||
'project_id',
|
'project_id',
|
||||||
'qos_policy_id',
|
|
||||||
'type',
|
'type',
|
||||||
)
|
)
|
||||||
self.data = (
|
self.data = (
|
||||||
@ -1408,7 +1348,6 @@ class TestShowNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
|
|||||||
self.new_rule.id,
|
self.new_rule.id,
|
||||||
self.new_rule.min_kbps,
|
self.new_rule.min_kbps,
|
||||||
self.new_rule.project_id,
|
self.new_rule.project_id,
|
||||||
self.new_rule.qos_policy_id,
|
|
||||||
self.new_rule.type,
|
self.new_rule.type,
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -1459,16 +1398,13 @@ class TestShowNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_MINIMUM_PACKET_RATE,
|
'type': RULE_TYPE_MINIMUM_PACKET_RATE,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs
|
|
||||||
)
|
|
||||||
self.qos_policy.rules = [self.new_rule]
|
self.qos_policy.rules = [self.new_rule]
|
||||||
self.columns = (
|
self.columns = (
|
||||||
'direction',
|
'direction',
|
||||||
'id',
|
'id',
|
||||||
'min_kpps',
|
'min_kpps',
|
||||||
'project_id',
|
'project_id',
|
||||||
'qos_policy_id',
|
|
||||||
'type',
|
'type',
|
||||||
)
|
)
|
||||||
self.data = (
|
self.data = (
|
||||||
@ -1476,7 +1412,6 @@ class TestShowNetworkQosRuleMinimumPacketRate(TestNetworkQosRule):
|
|||||||
self.new_rule.id,
|
self.new_rule.id,
|
||||||
self.new_rule.min_kpps,
|
self.new_rule.min_kpps,
|
||||||
self.new_rule.project_id,
|
self.new_rule.project_id,
|
||||||
self.new_rule.qos_policy_id,
|
|
||||||
self.new_rule.type,
|
self.new_rule.type,
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -1527,22 +1462,18 @@ class TestShowNetworkQosDSCPMarking(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_DSCP_MARKING,
|
'type': RULE_TYPE_DSCP_MARKING,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs
|
|
||||||
)
|
|
||||||
self.qos_policy.rules = [self.new_rule]
|
self.qos_policy.rules = [self.new_rule]
|
||||||
self.columns = (
|
self.columns = (
|
||||||
'dscp_mark',
|
'dscp_mark',
|
||||||
'id',
|
'id',
|
||||||
'project_id',
|
'project_id',
|
||||||
'qos_policy_id',
|
|
||||||
'type',
|
'type',
|
||||||
)
|
)
|
||||||
self.data = (
|
self.data = (
|
||||||
self.new_rule.dscp_mark,
|
self.new_rule.dscp_mark,
|
||||||
self.new_rule.id,
|
self.new_rule.id,
|
||||||
self.new_rule.project_id,
|
self.new_rule.project_id,
|
||||||
self.new_rule.qos_policy_id,
|
|
||||||
self.new_rule.type,
|
self.new_rule.type,
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -1593,26 +1524,22 @@ class TestShowNetworkQosBandwidthLimit(TestNetworkQosRule):
|
|||||||
'qos_policy_id': self.qos_policy.id,
|
'qos_policy_id': self.qos_policy.id,
|
||||||
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
'type': RULE_TYPE_BANDWIDTH_LIMIT,
|
||||||
}
|
}
|
||||||
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
|
self.new_rule = network_fakes.create_one_qos_rule(attrs)
|
||||||
attrs
|
|
||||||
)
|
|
||||||
self.qos_policy.rules = [self.new_rule]
|
self.qos_policy.rules = [self.new_rule]
|
||||||
self.columns = (
|
self.columns = (
|
||||||
'direction',
|
'direction',
|
||||||
'id',
|
'id',
|
||||||
'max_burst_kbits',
|
'max_burst_kbps',
|
||||||
'max_kbps',
|
'max_kbps',
|
||||||
'project_id',
|
'project_id',
|
||||||
'qos_policy_id',
|
|
||||||
'type',
|
'type',
|
||||||
)
|
)
|
||||||
self.data = (
|
self.data = (
|
||||||
self.new_rule.direction,
|
self.new_rule.direction,
|
||||||
self.new_rule.id,
|
self.new_rule.id,
|
||||||
self.new_rule.max_burst_kbits,
|
self.new_rule.max_burst_kbps,
|
||||||
self.new_rule.max_kbps,
|
self.new_rule.max_kbps,
|
||||||
self.new_rule.project_id,
|
self.new_rule.project_id,
|
||||||
self.new_rule.qos_policy_id,
|
|
||||||
self.new_rule.type,
|
self.new_rule.type,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -28,9 +28,8 @@ class TestNetworkQosRuleType(network_fakes.TestNetworkV2):
|
|||||||
class TestShowNetworkQosRuleType(TestNetworkQosRuleType):
|
class TestShowNetworkQosRuleType(TestNetworkQosRuleType):
|
||||||
attrs = {'drivers': [{'name': 'driver 1', 'supported_parameters': []}]}
|
attrs = {'drivers': [{'name': 'driver 1', 'supported_parameters': []}]}
|
||||||
# The QoS policies to show.
|
# The QoS policies to show.
|
||||||
qos_rule_type = (
|
qos_rule_type = network_fakes.create_one_qos_rule_type(attrs)
|
||||||
network_fakes.FakeNetworkQosRuleType.create_one_qos_rule_type(attrs)
|
columns = ('drivers', 'rule_type_name')
|
||||||
)
|
|
||||||
columns = ('drivers', 'rule_type_name')
|
columns = ('drivers', 'rule_type_name')
|
||||||
data = [qos_rule_type.drivers, qos_rule_type.type]
|
data = [qos_rule_type.drivers, qos_rule_type.type]
|
||||||
|
|
||||||
@ -76,9 +75,8 @@ class TestShowNetworkQosRuleType(TestNetworkQosRuleType):
|
|||||||
|
|
||||||
class TestListNetworkQosRuleType(TestNetworkQosRuleType):
|
class TestListNetworkQosRuleType(TestNetworkQosRuleType):
|
||||||
# The QoS policies to list up.
|
# The QoS policies to list up.
|
||||||
qos_rule_types = (
|
qos_rule_types = network_fakes.create_qos_rule_types(count=3)
|
||||||
network_fakes.FakeNetworkQosRuleType.create_qos_rule_types(count=3)
|
|
||||||
)
|
|
||||||
columns = ('Type',)
|
columns = ('Type',)
|
||||||
data = []
|
data = []
|
||||||
for qos_rule_type in qos_rule_types:
|
for qos_rule_type in qos_rule_types:
|
||||||
|
@ -34,8 +34,8 @@ class TestNetworkRBAC(network_fakes.TestNetworkV2):
|
|||||||
@ddt.ddt
|
@ddt.ddt
|
||||||
class TestCreateNetworkRBAC(TestNetworkRBAC):
|
class TestCreateNetworkRBAC(TestNetworkRBAC):
|
||||||
network_object = network_fakes.create_one_network()
|
network_object = network_fakes.create_one_network()
|
||||||
qos_object = network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
qos_object = network_fakes.create_one_qos_policy()
|
||||||
sg_object = network_fakes.FakeNetworkSecGroup.create_one_security_group()
|
sg_object = network_fakes.FakeSecurityGroup.create_one_security_group()
|
||||||
as_object = network_fakes.create_one_address_scope()
|
as_object = network_fakes.create_one_address_scope()
|
||||||
snp_object = network_fakes.FakeSubnetPool.create_one_subnet_pool()
|
snp_object = network_fakes.FakeSubnetPool.create_one_subnet_pool()
|
||||||
ag_object = network_fakes.create_one_address_group()
|
ag_object = network_fakes.create_one_address_group()
|
||||||
|
@ -582,7 +582,7 @@ class TestCreatePort(TestPort):
|
|||||||
self.assertCountEqual(self.data, data)
|
self.assertCountEqual(self.data, data)
|
||||||
|
|
||||||
def test_create_port_with_qos(self):
|
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(
|
self.network_client.find_qos_policy = mock.Mock(
|
||||||
return_value=qos_policy
|
return_value=qos_policy
|
||||||
)
|
)
|
||||||
@ -2371,7 +2371,7 @@ class TestSetPort(TestPort):
|
|||||||
)
|
)
|
||||||
|
|
||||||
def test_set_port_with_qos(self):
|
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(
|
self.network_client.find_qos_policy = mock.Mock(
|
||||||
return_value=qos_policy
|
return_value=qos_policy
|
||||||
)
|
)
|
||||||
|
@ -555,9 +555,7 @@ class TestCreateRouter(TestRouter):
|
|||||||
def test_create_with_qos_policy(self):
|
def test_create_with_qos_policy(self):
|
||||||
_network = network_fakes.create_one_network()
|
_network = network_fakes.create_one_network()
|
||||||
self.network_client.find_network = mock.Mock(return_value=_network)
|
self.network_client.find_network = mock.Mock(return_value=_network)
|
||||||
_qos_policy = (
|
_qos_policy = network_fakes.create_one_qos_policy()
|
||||||
network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
|
||||||
)
|
|
||||||
self.network_client.find_qos_policy = mock.Mock(
|
self.network_client.find_qos_policy = mock.Mock(
|
||||||
return_value=_qos_policy
|
return_value=_qos_policy
|
||||||
)
|
)
|
||||||
@ -590,9 +588,7 @@ class TestCreateRouter(TestRouter):
|
|||||||
self.assertCountEqual(self.data, data)
|
self.assertCountEqual(self.data, data)
|
||||||
|
|
||||||
def test_create_with_qos_policy_no_external_gateway(self):
|
def test_create_with_qos_policy_no_external_gateway(self):
|
||||||
_qos_policy = (
|
_qos_policy = network_fakes.create_one_qos_policy()
|
||||||
network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
|
||||||
)
|
|
||||||
self.network_client.find_qos_policy = mock.Mock(
|
self.network_client.find_qos_policy = mock.Mock(
|
||||||
return_value=_qos_policy
|
return_value=_qos_policy
|
||||||
)
|
)
|
||||||
@ -1668,7 +1664,7 @@ class TestSetRouter(TestRouter):
|
|||||||
self._test_set_tags(with_tags=False)
|
self._test_set_tags(with_tags=False)
|
||||||
|
|
||||||
def test_set_gateway_ip_qos(self):
|
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(
|
self.network_client.find_qos_policy = mock.Mock(
|
||||||
return_value=qos_policy
|
return_value=qos_policy
|
||||||
)
|
)
|
||||||
@ -1725,7 +1721,7 @@ class TestSetRouter(TestRouter):
|
|||||||
self.assertIsNone(result)
|
self.assertIsNone(result)
|
||||||
|
|
||||||
def test_set_unset_gateway_ip_qos(self):
|
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(
|
self.network_client.find_qos_policy = mock.Mock(
|
||||||
return_value=qos_policy
|
return_value=qos_policy
|
||||||
)
|
)
|
||||||
@ -1753,7 +1749,7 @@ class TestSetRouter(TestRouter):
|
|||||||
)
|
)
|
||||||
|
|
||||||
def test_set_gateway_ip_qos_no_gateway(self):
|
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(
|
self.network_client.find_qos_policy = mock.Mock(
|
||||||
return_value=qos_policy
|
return_value=qos_policy
|
||||||
)
|
)
|
||||||
@ -1775,7 +1771,7 @@ class TestSetRouter(TestRouter):
|
|||||||
)
|
)
|
||||||
|
|
||||||
def test_unset_gateway_ip_qos_no_gateway(self):
|
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(
|
self.network_client.find_qos_policy = mock.Mock(
|
||||||
return_value=qos_policy
|
return_value=qos_policy
|
||||||
)
|
)
|
||||||
@ -1932,9 +1928,7 @@ class TestUnsetRouter(TestRouter):
|
|||||||
def setUp(self):
|
def setUp(self):
|
||||||
super().setUp()
|
super().setUp()
|
||||||
self.fake_network = network_fakes.create_one_network()
|
self.fake_network = network_fakes.create_one_network()
|
||||||
self.fake_qos_policy = (
|
self.fake_qos_policy = network_fakes.create_one_qos_policy()
|
||||||
network_fakes.FakeNetworkQosPolicy.create_one_qos_policy()
|
|
||||||
)
|
|
||||||
self._testrouter = network_fakes.FakeRouter.create_one_router(
|
self._testrouter = network_fakes.FakeRouter.create_one_router(
|
||||||
{
|
{
|
||||||
'routes': [
|
'routes': [
|
||||||
@ -2102,7 +2096,7 @@ class TestUnsetRouter(TestRouter):
|
|||||||
self.assertIsNone(result)
|
self.assertIsNone(result)
|
||||||
|
|
||||||
def test_unset_gateway_ip_qos_no_network(self):
|
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(
|
self.network_client.find_qos_policy = mock.Mock(
|
||||||
return_value=qos_policy
|
return_value=qos_policy
|
||||||
)
|
)
|
||||||
@ -2122,7 +2116,7 @@ class TestUnsetRouter(TestRouter):
|
|||||||
)
|
)
|
||||||
|
|
||||||
def test_unset_gateway_ip_qos_no_qos(self):
|
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(
|
self.network_client.find_qos_policy = mock.Mock(
|
||||||
return_value=qos_policy
|
return_value=qos_policy
|
||||||
)
|
)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user