Fix load_db_migration_drivers

Migration plugins registered under the ``neutron.ovn.db_migration``
entry point are plain callables (functions), not subclasses of
``BaseOvnDbSynchronizer``. The shared ``_load_drivers`` helper applies
an ``issubclass`` guard that is only valid for class-based sync plugins,
causing a ``TypeError`` when loading function-based migration plugins and
ultimately a ``NoMatches`` exception that aborts the OVS-to-OVN database
migration step.

Fix this by giving ``load_db_migration_drivers`` its own
``EnabledExtensionManager`` construction that skips the ``issubclass``
check and simply filters by driver name.

Closes-Bug: #2160210

Assisted-By: Claude Sonnet 4.6
Change-Id: I5c9b689558a9553ff6968868b43bf3d9239c92ad
Signed-off-by: Michal Nasiadka <mnasiadka@gmail.com>

neutron/cmd/ovn/neutron_ovn_db_sync_util.py

@@ -13,6 +13,7 @@
 #    under the License.
 
 import copy
+import inspect
 
 from neutron_lib.ovn import constants as ovn_const
 from neutron_lib.ovn import db_sync as db_sync_base
@@ -113,7 +114,9 @@ def setup_conf():
 def _load_drivers(entry_point, driver_name=None):
 
     def load_driver(ext):
-        if not issubclass(ext.plugin, db_sync_base.BaseOvnDbSynchronizer):
+        if (inspect.isclass(ext.plugin) and
+                not issubclass(ext.plugin,
+                               db_sync_base.BaseOvnDbSynchronizer)):
             LOG.error("Extension '%s' is not an instance of "
                       "%s and will not be loaded",
                       ext.name, db_sync_base.BaseOvnDbSynchronizer)

neutron/tests/unit/cmd/ovn/test_neutron_ovn_db_sync_util.py

@@ -12,6 +12,8 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from unittest import mock
+
 from oslo_config import cfg
 
 from neutron.cmd.ovn import neutron_ovn_db_sync_util as util
@@ -29,3 +31,51 @@ class TestNeutronOVNDBSyncUtil(base.BaseTestCase):
         # value. Validate that the setup code overrides them. LP: #1882020
         self.assertFalse(cfg.CONF.notify_nova_on_port_status_changes)
         self.assertFalse(cfg.CONF.notify_nova_on_port_data_changes)
+
+    def test_load_db_migration_drivers_calls_callable_plugin(self):
+        migration_fn = mock.Mock()
+        fake_ext = mock.Mock()
+        fake_ext.name = 'neutron'
+        fake_ext.plugin = migration_fn
+
+        with mock.patch(
+                'stevedore.enabled.EnabledExtensionManager') as mock_mgr_cls:
+            mock_mgr = mock.Mock()
+            mock_mgr_cls.return_value = mock_mgr
+            mock_mgr.__iter__ = mock.Mock(return_value=iter([fake_ext]))
+
+            util.load_db_migration_drivers()
+
+        mock_mgr_cls.assert_called_once_with(
+            'neutron.ovn.db_migration',
+            check_func=mock.ANY,
+            invoke_on_load=False)
+
+        # Verify the check_func accepts a callable (non-class) plugin without
+        # raising TypeError from issubclass()
+        check_func = mock_mgr_cls.call_args[1]['check_func']
+        self.assertTrue(check_func(fake_ext))
+
+    def test_load_db_migration_drivers_filters_by_name(self):
+        fake_ext_a = mock.Mock()
+        fake_ext_a.name = 'neutron'
+        fake_ext_b = mock.Mock()
+        fake_ext_b.name = 'other'
+
+        with mock.patch(
+                'stevedore.enabled.EnabledExtensionManager') as mock_mgr_cls:
+            util.load_db_migration_drivers(driver_name='neutron')
+
+        check_func = mock_mgr_cls.call_args[1]['check_func']
+        self.assertTrue(check_func(fake_ext_a))
+        self.assertFalse(check_func(fake_ext_b))
+
+    def test_migrate_neutron_dbs_to_ovn_calls_plugin(self):
+        migration_fn = mock.Mock()
+        fake_ext = mock.Mock()
+        fake_ext.name = 'neutron'
+        fake_ext.plugin = migration_fn
+
+        util.migrate_neutron_dbs_to_ovn(fake_ext)
+
+        migration_fn.assert_called_once_with()