Add object loader
Change-Id: I921a3ba37ca49eec29fb74c0a131308791b1ef7c
This commit is contained in:
parent
b9dfc8c04c
commit
492905d71b
|
@ -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
|
||||
|
|
|
@ -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()
|
|
@ -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__, '<non-existing>'])
|
||||
self.assertRaises(ImportError, tobiko.load_object, object_id)
|
||||
|
||||
def test_load_object_with_non_existing_member(self):
|
||||
object_id = '.'.join([SomeClass.__module__,
|
||||
SomeClass.__name__,
|
||||
'<non-existing>'])
|
||||
self.assertRaises(AttributeError, tobiko.load_object, object_id)
|
Loading…
Reference in New Issue