Merge "Add test_ovn_dbs.py"

whitebox_neutron_tempest_plugin/tests/scenario/base.py

@@ -59,6 +59,8 @@ class BaseTempestWhiteboxTestCase(base.BaseTempestTestCase):
         sriov_agents = [
             agent for agent in agents if 'sriov' in agent['binary']]
         cls.has_sriov_support = True if sriov_agents else False
+        if WB_CONF.openstack_type == 'devstack':
+            cls.master_node_client = cls.get_node_client('localhost')
 
     @classmethod
     def run_on_master_controller(cls, cmd):

whitebox_neutron_tempest_plugin/tests/scenario/test_ovn_dbs.py

@@ -0,0 +1,75 @@
+# Copyright 2024 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from oslo_log import log
+from tempest import config
+from tempest.lib import decorators
+
+from whitebox_neutron_tempest_plugin.tests.scenario import base as wb_base
+
+CONF = config.CONF
+LOG = log.getLogger(__name__)
+
+
+class OvnDbsMonitoringTest(wb_base.BaseTempestTestCaseOvn):
+    credentials = ['primary', 'admin']
+
+    def start_monitoring(self):
+        self.monitoring_file = self.master_node_client.exec_command(
+            'mktemp').rstrip()
+        self.process = self.master_node_client.open_session()
+        cmd = '{} OVN_Southbound Chassis external_ids > {}'.format(
+            self.sbmonitorcmd, self.monitoring_file)
+        LOG.debug('Executing command: {}'.format(cmd))
+        self.process.exec_command(cmd)
+        self.addCleanup(self.stop_process)
+
+    def stop_process(self):
+        if self.process:
+            self.process.close()
+            self.process = None
+
+    def check_monitoring(self):
+        output = self.master_node_client.exec_command(
+            'sudo cat ' + self.monitoring_file).splitlines()
+        self.assertEqual(1, len(output))
+
+    @decorators.idempotent_id('e0c93863-a4a7-4b16-8f45-793965658cbd')
+    def test_create_server_ovn_dbs_monitoring(self):
+        """Check no changes are performed on the external_ids value from the
+        Chassis table on the OVN SBDB due to the creation of a network, a
+        subnet and a VM instance
+        This test covers the watchlist bug rhbz#1962178
+        Scenario:
+        1. The monitoring of the external_ids value from the SBDB Chassis table
+           is activated
+        2. A network, a subnet and a VM instance are created
+        3. The moniroting initiated at 1 is stopped and the result is
+           validated: no changes are expected, which means only one row has
+           been printed during the test
+        """
+        self.start_monitoring()
+        network = self.create_network()
+        self.create_subnet(network=network)
+        keypair = self.create_keypair()
+        server_kwargs = {
+            'flavor_ref': CONF.compute.flavor_ref,
+            'image_ref': CONF.compute.image_ref,
+            'key_name': keypair['name'],
+            'networks': [{'uuid': network['id']}]
+        }
+        self.create_server(**server_kwargs)
+        self.stop_process()
+        self.check_monitoring()