Merge "Adapt _validate_qos_rules_nbdb to flat and vlan networks"

This commit is contained in:
Zuul 2025-04-08 11:13:02 +00:00 committed by Gerrit Code Review
commit e5585cdaeb

View File

@ -1266,6 +1266,43 @@ class QosTestOvn(base.BaseTempestTestCaseOvn, QosBaseTest):
return policy_id
def _validate_bw_limit_nbdb_qos(self, qos_settings):
for line in qos_settings.splitlines():
if line.startswith('bandwidth'):
bandwidth_settings = line
break
self.assertTrue(
'{burst=%d, rate=%d}' % (self.MAX_BURST_KBPS, self.MAX_KBPS)
in bandwidth_settings,
'Bandwidth options are not set as expected')
LOG.debug('BW limit options found')
def _validate_bw_limit_nbdb_lsp(self, lsp_settings):
for line in lsp_settings.splitlines():
if line.startswith('options'):
bandwidth_settings = line
break
self.assertTrue(
'qos_burst="%d", qos_max_rate="%d"' % (
self.MAX_BURST_KBPS, self.MAX_KBPS)
in bandwidth_settings,
'Bandwidth options are not set as expected')
LOG.debug('BW limit options found')
def _validate_dscp_nbdb_qos(self, qos_settings):
for line in qos_settings.splitlines():
if line.startswith('action'):
dscp_settings = line
break
self.assertTrue(
'{dscp=%s}' % (self.DSCP_MARK_OPTION,)
in dscp_settings,
'DSCP options are not set as expected')
LOG.debug('DSCP options found')
def _validate_qos_rules_nbdb(
self, port_id=None, fip_id=None, expected_empty=False):
# Validates QoS bw and dscp rules with constant values in OVN NBDB.
@ -1277,43 +1314,42 @@ class QosTestOvn(base.BaseTempestTestCaseOvn, QosBaseTest):
self.assertTrue(port_id or fip_id,
'At least one of the input params is required')
cmds = []
if port_id:
cmds.append(r'{} find qos match="inport\ \=\=\ \"{}\""'.format(
self.nbctl, port_id))
if fip_id:
cmds.append(
r'%s find qos external_ids={"neutron\:fip_id"="%s"}' % (
self.nbctl, fip_id))
for cmd in cmds:
policy_settings = self.run_on_master_controller(
cmd).rstrip()
cmd = r'{} find qos match="inport\ \=\=\ \"{}\""'.format(
self.nbctl, port_id)
qos_settings = self.run_on_master_controller(cmd).rstrip()
if expected_empty:
self.assertFalse(policy_settings,
self.assertFalse(qos_settings,
'QoS is not supposed to be applied on this '
'port in OVN NBDB')
LOG.debug('Success: no QoS policies found, as expected')
else:
for line in policy_settings.splitlines():
if line.startswith('action'):
dscp_settings = line
if line.startswith('bandwidth'):
bandwidth_settings = line
network_id = self.os_admin.network_client.show_port(
port_id)['port']['network_id']
network_type = self.os_admin.network_client.show_network(
network_id)['network']['provider:network_type']
if network_type not in ('vlan', 'flat'):
self._validate_dscp_nbdb_qos(qos_settings)
self._validate_bw_limit_nbdb_qos(qos_settings)
else:
self._validate_dscp_nbdb_qos(qos_settings)
cmd = '{} list logical_switch_port {}'.format(self.nbctl,
port_id)
lsp_settings = self.run_on_master_controller(cmd).rstrip()
self._validate_bw_limit_nbdb_lsp(lsp_settings)
self.assertTrue(
'{burst=%d, rate=%d}' % (self.MAX_BURST_KBPS,
self.MAX_KBPS)
in bandwidth_settings,
'Bandwidth options are not set as expected')
LOG.debug('BW limit options found')
self.assertTrue(
'{dscp=%s}' % (self.DSCP_MARK_OPTION,)
in dscp_settings,
'DSCP options are not set as expected')
LOG.debug('DSCP options found')
if fip_id:
cmd = r'%s find qos external_ids={"neutron\:fip_id"="%s"}' % (
self.nbctl, fip_id)
qos_settings = self.run_on_master_controller(cmd).rstrip()
if expected_empty:
self.assertFalse(qos_settings,
'QoS is not supposed to be applied on this '
'FIP in OVN NBDB')
LOG.debug('Success: no QoS policies found, as expected')
else:
self._validate_dscp_nbdb_qos(qos_settings)
self._validate_bw_limit_nbdb_qos(qos_settings)
@decorators.idempotent_id('08b74ece-d7f2-4a80-9a1e-5fb7ec928a9b')
def test_attach_qos_port_to_vm_with_another_port(self):