Filter the ports by VNIC type

In ``check_baremetal_ports_dhcp_options``, the ports need to be
retrieved based on the VNIC type. The VNIC type is stored in
"ml2_port_bindings" table, not "ports" table.  Because the ML2Plugin
``get_ports`` method can filter only by "ports" database register
fields, it is needed to retrieve those ports using a custom database
query and then use the plugin method to retrieve the ports, converted
to dictionaries and the extensions processed.

Closes-Bug: #1979643
Change-Id: Iab227d9de75e4b1b927043ce27bdc346066478e8
(cherry picked from commit b497ad1)

Author: ralonsoh

neutron/objects/ports.py

@@ -684,13 +684,12 @@ def get_ports_by_binding_type_and_host(cls, context,
         return [cls._load_object(context, db_obj) for db_obj in query.all()]
 
     @classmethod
-    def get_ports_by_vnic_type_and_host(
-            cls, context, vnic_type, host):
+    def get_ports_by_vnic_type_and_host(cls, context, vnic_type, host=None):
         query = context.session.query(models_v2.Port).join(
             ml2_models.PortBinding)
-        query = query.filter(
-            ml2_models.PortBinding.vnic_type == vnic_type,
-            ml2_models.PortBinding.host == host)
+        query = query.filter(ml2_models.PortBinding.vnic_type == vnic_type)
+        if host:
+            query = query.filter(ml2_models.PortBinding.host == host)
         return [cls._load_object(context, db_obj) for db_obj in query.all()]
 
     @classmethod

neutron/plugins/ml2/drivers/ovn/mech_driver/ovsdb/maintenance.py

@@ -41,6 +41,7 @@
 from neutron.db import ovn_revision_numbers_db as revision_numbers_db
 from neutron.db import segments_db
 from neutron.objects import router as router_obj
+from neutron.objects import ports as ports_obj
 from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb import ovn_db_sync
 
 
@@ -907,9 +908,10 @@ def check_baremetal_ports_dhcp_options(self):
             return
 
         context = n_context.get_admin_context()
+        ports = ports_obj.Port.get_ports_by_vnic_type_and_host(
+            context, portbindings.VNIC_BAREMETAL)
         ports = self._ovn_client._plugin.get_ports(
-            context,
-            filters={portbindings.VNIC_TYPE: portbindings.VNIC_BAREMETAL})
+            context, filters={'id': [p.id for p in ports]})
         if not ports:
             raise periodics.NeverAgain()
 

neutron/tests/unit/common/ovn/test_utils.py

@@ -416,19 +416,19 @@ def test_get_lsp_dhcp_opts_dhcp_disabled_for_baremetal(self):
         # Assert no options were passed
         self.assertEqual({}, options)
 
-   def test_get_lsp_dhcp_opts_for_domain_search(self):
-        opt = {'opt_name': 'domain-search',
-               'opt_value': 'openstack.org,ovn.org',
-               'ip_version': 4}
-        port = {portbindings.VNIC_TYPE: portbindings.VNIC_NORMAL,
-                edo_ext.EXTRADHCPOPTS: [opt]}
-
-        dhcp_disabled, options = utils.get_lsp_dhcp_opts(port, 4)
-        self.assertFalse(dhcp_disabled)
-        # Assert option got translated to "domain_search_list" and
-        # the value is a string (double-quoted)
-        expected_options = {'domain_search_list': '"openstack.org,ovn.org"'}
-        self.assertEqual(expected_options, options)
+    def test_get_lsp_dhcp_opts_for_domain_search(self):
+         opt = {'opt_name': 'domain-search',
+                'opt_value': 'openstack.org,ovn.org',
+                'ip_version': 4}
+         port = {portbindings.VNIC_TYPE: portbindings.VNIC_NORMAL,
+                 edo_ext.EXTRADHCPOPTS: [opt]}
+
+         dhcp_disabled, options = utils.get_lsp_dhcp_opts(port, 4)
+         self.assertFalse(dhcp_disabled)
+         # Assert option got translated to "domain_search_list" and
+         # the value is a string (double-quoted)
+         expected_options = {'domain_search_list': '"openstack.org,ovn.org"'}
+         self.assertEqual(expected_options, options)
 
 class TestGetDhcpDnsServers(base.BaseTestCase):