diff --git a/whitebox_neutron_tempest_plugin/tests/scenario/test_qos.py b/whitebox_neutron_tempest_plugin/tests/scenario/test_qos.py index 45dc4d8..0cd13fd 100644 --- a/whitebox_neutron_tempest_plugin/tests/scenario/test_qos.py +++ b/whitebox_neutron_tempest_plugin/tests/scenario/test_qos.py @@ -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):