Merge "consume load_class_by_alias_or_classname from neutron-lib"

neutron/agent/common/utils.py

@@ -15,11 +15,11 @@
 
 import os
 
+from neutron_lib.utils import runtime
 from oslo_config import cfg
 from oslo_log import log as logging
 from oslo_utils import timeutils
 
-from neutron.common import utils as neutron_utils
 from neutron.conf.agent import common as config
 from neutron.conf.agent.database import agents_db
 
@@ -48,7 +48,7 @@ def load_interface_driver(conf):
     """
 
     try:
-        loaded_class = neutron_utils.load_class_by_alias_or_classname(
+        loaded_class = runtime.load_class_by_alias_or_classname(
                 INTERFACE_NAMESPACE, conf.interface_driver)
         return loaded_class(conf)
     except ImportError:

neutron/agent/firewall.py

@@ -20,8 +20,7 @@ import six
 
 from neutron_lib.api.definitions import port_security as psec
 from neutron_lib import constants as n_const
-
-from neutron.common import utils
+from neutron_lib.utils import runtime
 
 
 INGRESS_DIRECTION = n_const.INGRESS_DIRECTION
@@ -49,7 +48,7 @@ def port_sec_enabled(port):
 
 
 def load_firewall_driver_class(driver):
-    return utils.load_class_by_alias_or_classname(
+    return runtime.load_class_by_alias_or_classname(
         'neutron.agent.firewall_drivers', driver)
 
 

neutron/common/utils.py

@@ -42,9 +42,7 @@ from oslo_db import exception as db_exc
 from oslo_log import log as logging
 from oslo_utils import excutils
 from oslo_utils import fileutils
-from oslo_utils import importutils
 import six
-from stevedore import driver
 
 import neutron
 from neutron._i18n import _
@@ -283,37 +281,6 @@ class DelayedStringRenderer(object):
         return str(self.function(*self.args, **self.kwargs))
 
 
-def load_class_by_alias_or_classname(namespace, name):
-    """Load class using stevedore alias or the class name
-
-    :param namespace: namespace where the alias is defined
-    :param name: alias or class name of the class to be loaded
-    :returns: class if calls can be loaded
-    :raises ImportError if class cannot be loaded
-    """
-
-    if not name:
-        LOG.error("Alias or class name is not set")
-        raise ImportError(_("Class not found."))
-    try:
-        # Try to resolve class by alias
-        mgr = driver.DriverManager(
-            namespace, name, warn_on_missing_entrypoint=False)
-        class_to_load = mgr.driver
-    except RuntimeError:
-        e1_info = sys.exc_info()
-        # Fallback to class name
-        try:
-            class_to_load = importutils.import_class(name)
-        except (ImportError, ValueError):
-            LOG.error("Error loading class by alias",
-                      exc_info=e1_info)
-            LOG.error("Error loading class by class name",
-                      exc_info=True)
-            raise ImportError(_("Class not found."))
-    return class_to_load
-
-
 def _hex_format(port, mask=0):
 
     def hex_str(num):

neutron/manager.py

@@ -27,7 +27,6 @@ from osprofiler import profiler
 import six
 
 from neutron._i18n import _
-from neutron.common import utils
 from neutron.plugins.common import constants
 
 
@@ -156,8 +155,8 @@ class NeutronManager(object):
         """
 
         try:
-            return utils.load_class_by_alias_or_classname(namespace,
-                    plugin_provider)
+            return runtime.load_class_by_alias_or_classname(namespace,
+                                                            plugin_provider)
         except ImportError:
             with excutils.save_and_reraise_exception():
                 LOG.error("Plugin '%s' not found.", plugin_provider)

neutron/services/metering/drivers/utils.py

@@ -13,10 +13,9 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from neutron_lib.utils import runtime
 from oslo_log import log as logging
 
-from neutron.common import utils
-
 
 LOG = logging.getLogger(__name__)
 
@@ -32,7 +31,7 @@ def load_metering_driver(plugin, conf):
     """
 
     try:
-        loaded_class = utils.load_class_by_alias_or_classname(
+        loaded_class = runtime.load_class_by_alias_or_classname(
                 METERING_NAMESPACE, conf.driver)
         return loaded_class(plugin, conf)
     except ImportError:

neutron/tests/unit/agent/l2/extensions/test_qos.py

@@ -25,6 +25,7 @@ from neutron.api.rpc.callbacks.consumer import registry
 from neutron.api.rpc.callbacks import events
 from neutron.api.rpc.callbacks import resources
 from neutron.api.rpc.handlers import resources_rpc
+from neutron import manager
 from neutron.objects.qos import policy
 from neutron.objects.qos import rule
 from neutron.plugins.ml2.drivers.openvswitch.agent import (
@@ -229,10 +230,10 @@ class QosExtensionBaseTestCase(base.BaseTestCase):
         self.qos_ext.consume_api(self.agent_api)
 
         # Don't rely on used driver
-        mock.patch(
-            'neutron.manager.NeutronManager.load_class_for_provider',
-            return_value=lambda: mock.Mock(spec=qos_linux.QosLinuxAgentDriver)
-        ).start()
+        mock.patch.object(
+            manager.NeutronManager, 'load_class_for_provider',
+            return_value=lambda: mock.Mock(
+                spec=qos_linux.QosLinuxAgentDriver)).start()
 
 
 class QosExtensionRpcTestCase(QosExtensionBaseTestCase):