From 45ff7d4f9308d73ed993315df9133d09b0709040 Mon Sep 17 00:00:00 2001
From: Eduardo Olivares <eolivare@redhat.com>
Date: Thu, 3 Apr 2025 15:40:00 +0200
Subject: [PATCH] Adapt _validate_qos_rules_nbdb to flat and vlan networks

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
---
 .../tests/scenario/test_qos.py                | 96 +++++++++++++------
 1 file changed, 66 insertions(+), 30 deletions(-)

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):