diff --git a/tobiko/__init__.py b/tobiko/__init__.py index 155443454..62557ca46 100644 --- a/tobiko/__init__.py +++ b/tobiko/__init__.py @@ -14,6 +14,10 @@ from __future__ import absolute_import from tobiko.common.managers import fixture as fixture_manager from tobiko.common.managers import testcase as testcase_manager +from tobiko.common.managers import loader as loader_manager + + +load_object = loader_manager.load_object Fixture = fixture_manager.Fixture diff --git a/tobiko/common/managers/loader.py b/tobiko/common/managers/loader.py new file mode 100644 index 000000000..7959326da --- /dev/null +++ b/tobiko/common/managers/loader.py @@ -0,0 +1,147 @@ +# Copyright 2019 Red Hat +# +# 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 __future__ import absolute_import + +import importlib +import inspect +import weakref +import sys + + +_REF_TO_NONE = object() + + +def load_object(object_id, manager=None, new_loader=None, cached=True): + manager = manager or LOADERS + loader = manager.get_loader(object_id=object_id, new_loader=new_loader) + return loader.load(manager=manager, cached=cached) + + +class ObjectLoader(object): + """Previously loaded object meta-data""" + + # Weak reference to target object + _ref = None + + # Flag that tells if referenced object is an imported module + _is_module = None + + def __init__(self, object_id): + # Object ID + self._id = object_id + if '.' in object_id: + # Extract object name and parent id from object id + parent_id, name = object_id.rsplit('.', 1) + self._name = name + self._parent_id = parent_id + else: + # Root objects have no parent + self._name = object_id + self._parent_id = None + + @property + def is_module(self): + return self._is_module + + @property + def id(self): + return self._id + + def __repr__(self): + return '{cls!s}({id!r})'.format(cls=type(self).__name__, id=self._id) + + def get(self): + if self._is_module: + return sys.modules.get(self._id) + + ref = self._ref + if ref is _REF_TO_NONE: + return None + + if callable(ref): + obj = ref() + if obj is not None: + return obj + + msg = "Object {!r} not cached".format(self._id) + raise RuntimeError(msg) + + def load(self, manager, cached=True): + if cached: + try: + return self.get() + except RuntimeError: + pass + + obj = None + parent_id = self._parent_id + if parent_id: + parent_loader = manager.get_loader(object_id=parent_id, + new_loader=type(self)) + parent = parent_loader.load(manager=manager, cached=cached) + name = self._name + try: + obj = getattr(parent, name) + except AttributeError: + if not parent_loader.is_module: + # Child cannot be a module if parent isn't a module + raise + else: + if obj is None: + # Cannot create weak reference to None + self._ref = _REF_TO_NONE + elif inspect.ismodule(obj): + # Cannot create weak reference to Module + self._is_module = True + else: + self._ref = weakref.ref(obj) + return obj + + if obj is None: + obj = importlib.import_module(self._id) + self._is_module = True + return obj + + +class LoaderManager(object): + + def __init__(self): + # Dictionary used to cache object references + self._loaders = {} + + new_loader = ObjectLoader + + def get_loader(self, object_id, new_loader=None): + """Get existing ObjectInfo or create new one + + It implements singleton pattern by caching previously created + ObjectInfo instances to OBJECT_INFOS for later retrieval + """ + try: + return self._loaders[object_id] + except KeyError: + pass + + new_loader = new_loader or self.new_loader + loader = new_loader(object_id=object_id) + if not isinstance(loader, ObjectLoader): + msg = "{!r} is not instance of class ObjectLoader".format( + loader) + raise TypeError(msg) + + self._loaders[object_id] = loader + return loader + + +LOADERS = LoaderManager() diff --git a/tobiko/tests/test_loader.py b/tobiko/tests/test_loader.py new file mode 100644 index 000000000..f8d1ef0ef --- /dev/null +++ b/tobiko/tests/test_loader.py @@ -0,0 +1,94 @@ +# Copyright 2019 Red Hat +# +# 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 __future__ import absolute_import + +import sys + +import tobiko +from tobiko.common.managers import loader +from tobiko.tests.unit import TobikoUnitTest + + +SOME_NONE = None + + +class SomeClass(object): + + def some_method(self): + pass + + +class TestLoader(TobikoUnitTest): + + def setUp(self): + super(TestLoader, self).setUp() + self.manager = loader.LoaderManager() + self.patch('tobiko.common.managers.loader.LOADERS', self.manager) + + def test_load_object_with_none(self): + object_id = '.'.join([__name__, 'SOME_NONE']) + obj = tobiko.load_object(object_id) + self.assertIsNone(obj) + + _loader = self.manager.get_loader(object_id) + self.assertEqual(_loader.id, object_id) + self.assertFalse(_loader.is_module) + self.assertIs(_loader.get(), obj) + self.assertIs(_loader, self.manager.get_loader(object_id)) + + def test_load_object_with_module(self): + object_id = __name__ + obj = tobiko.load_object(object_id) + self.assertIs(sys.modules[object_id], obj) + + _loader = self.manager.get_loader(object_id) + self.assertEqual(_loader.id, object_id) + self.assertTrue(_loader.is_module) + self.assertIs(_loader.get(), obj) + self.assertIs(_loader, self.manager.get_loader(object_id)) + + def test_load_object_with_class(self): + object_id = '.'.join([SomeClass.__module__, + SomeClass.__name__]) + obj = tobiko.load_object(object_id) + self.assertIs(SomeClass, obj) + + _loader = self.manager.get_loader(object_id) + self.assertEqual(_loader.id, object_id) + self.assertFalse(_loader.is_module) + self.assertIs(_loader.get(), obj) + self.assertIs(_loader, self.manager.get_loader(object_id)) + + def test_load_object_with_class_method(self): + object_id = '.'.join([SomeClass.__module__, + SomeClass.__name__, + SomeClass.some_method.__name__]) + obj = tobiko.load_object(object_id) + self.assertEqual(SomeClass.some_method, obj) + + _loader = self.manager.get_loader(object_id) + self.assertEqual(_loader.id, object_id) + self.assertFalse(_loader.is_module) + self.assertIs(_loader.get(), obj) + self.assertIs(_loader, self.manager.get_loader(object_id)) + + def test_load_object_with_non_existing(self): + object_id = '.'.join([SomeClass.__module__, '']) + self.assertRaises(ImportError, tobiko.load_object, object_id) + + def test_load_object_with_non_existing_member(self): + object_id = '.'.join([SomeClass.__module__, + SomeClass.__name__, + '']) + self.assertRaises(AttributeError, tobiko.load_object, object_id)