Add object loader

Change-Id: I921a3ba37ca49eec29fb74c0a131308791b1ef7c
This commit is contained in:
Federico Ressi 2019-01-10 14:27:50 +01:00
parent b9dfc8c04c
commit 492905d71b
3 changed files with 245 additions and 0 deletions

View File

@ -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

View File

@ -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()

View File

@ -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)