x/whitebox-neutron-tempest-plugin
Code Issues Proposed changes

Adapt _validate_qos_rules_nbdb to flat and vlan networks

Browse Source 
With recent neutron changes in how BW limits are applied when to vlan
and flat networks, the validation of the QoS configuration on the OVN
NB DB needs to be adapted.

Change-Id: I34db362f161085b1c45bd14eb9eabbd1ddafd070
This commit is contained in:
Eduardo Olivares 2025-04-03 15:40:00 +02:00
parent 2ad8456264
commit 45ff7d4f93
1 changed files with 66 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
whitebox_neutron_tempest_plugin/tests/scenario/test_qos.py
View File

@ -1255,6 +1255,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.
@ -1266,43 +1303,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):