"""This module implements dependency injection mechanisms.""" __author__ = "Martin Sivak " __all__ = ["DI_ENABLE", "di_enable", "inject", "usesclassinject"] from functools import wraps, partial from types import FunctionType DI_ENABLE = True def di_enable(method): """This decorator enables DI mechanisms in an environment where DI is disabled by default. Can be used only on methods or simple functions. """ @wraps(method) def caller(*args, **kwargs): """The replacement method doing the DI enablement. """ global DI_ENABLE old = DI_ENABLE DI_ENABLE = True ret = method(*args, **kwargs) DI_ENABLE = old return ret return caller class DiRegistry(object): """This class is the internal core of the DI engine. It records the injected objects, handles the execution and cleanup tasks associated with the DI mechanisms. """ def __init__(self, obj): self._obj = obj self._used_objects = {} self._obj._di_ = self._used_objects def __get__(self, obj, objtype): """Support instance methods.""" return partial(self.__call__, obj) def register(self, *args, **kwargs): """Add registered injections to the instance of DiRegistry """ self._used_objects.update(kwargs) for used_object in args: if hasattr(used_object, "__name__"): self._used_objects[used_object.__name__] = used_object elif isinstance(used_object, basestring): pass # it is already global, so this is just an annotation else: raise ValueError("%s is not a string or object with __name__" % used_object) def __call__(self, *args, **kwargs): if not issubclass(type(self._obj), FunctionType): # call constructor or callable class # (which use @usesclassinject if needed) return self._obj(*args, **kwargs) else: return di_call(self._used_objects, self._obj, *args, **kwargs) def func_globals(func): """Helper method that allows access to globals depending on the Python version. """ if hasattr(func, "func_globals"): return func.func_globals # Python 2 else: return func.__globals__ # Python 3 def di_call(di_dict, method, *args, **kwargs): """This method is the core of dependency injection framework. It modifies methods global namespace to define all the injected variables, executed the method under test and then restores the global namespace back. This variant is used on plain functions. The modified global namespace is discarded after the method finishes so all new global variables and changes to scalars will be lost. """ # modify the globals new_globals = func_globals(method).copy() new_globals.update(di_dict) # create new func with modified globals new_method = FunctionType(method.func_code, new_globals, method.func_name, method.func_defaults, method.func_closure) # execute the method and return it's ret value return new_method(*args, **kwargs) def inject(*args, **kwargs): """Decorator that registers all the injections we want to pass into a unit possibly under test. It can be used to decorate class, method or simple function, but if it is a decorated class, it's methods has to be decorated with @usesinject to use the DI mechanism. """ def inject_decorate(obj): """The actual decorator generated by @inject.""" if not DI_ENABLE: return obj if not isinstance(obj, DiRegistry): obj = DiRegistry(obj) obj.register(*args, **kwargs) return obj return inject_decorate def usesclassinject(method): """This decorator marks a method inside of @inject decorated class as a method that should use the dependency injection mechanisms. """ if not DI_ENABLE: return method @wraps(method) def call(*args, **kwargs): """The replacement method acting as a proxy to @inject decorated class and it's DI mechanisms.""" self = args[0] return di_call(self._di_, method, *args, **kwargs) return call ### Unittests are defined below this point import unittest class BareFuncTestCase(unittest.TestCase): @inject(injected_func = str.lower) def method(self, arg): return injected_func(arg) @inject(injected_func = method) def method2(self, arg): return injected_func(self, arg) def test_bare_inject(self): """Tests the injection to plain methods.""" self.assertEqual("a", self.method("A")) def test_double_inject(self): """Tests the injection to two plain methods.""" self.assertEqual("a", self.method2("A")) def test_inject_global_tainting(self): """Tests whether the global namespace is clean after the injection is done.""" global injected_func injected_func = None self.method("A") self.assertEqual(None, injected_func) @inject(injected_func = str.lower) class Test(object): """Test fixture for class injection.""" @usesclassinject def method(self, arg): return injected_func(arg) @inject(injected_func = str.lower) class TestInit(object): """Test fixture for injection to __init__.""" @usesclassinject def __init__(self, arg): self.value = injected_func(arg) @inject(injected_func = str.lower) class TestCallable(object): """Test fixture for callable classes.""" @usesclassinject def __call__(self, arg): return injected_func(arg) class TestCallableSingle(object): """Test fixture for callable classes with simple method injection.""" @inject(injected_func = str.lower) def __call__(self, arg): return injected_func(arg) class ClassDITestCase(unittest.TestCase): def test_class_inject(self): """Test injection to instance method.""" obj = Test() self.assertEqual("a", obj.method("A")) def test_class_init_inject(self): """Test injection to class constructor.""" obj = TestInit("A") self.assertEqual("a", obj.value) def test_callable_class(self): """Test class injection to callable class.""" obj = TestCallable() self.assertEqual("a", obj("A")) def test_callable_class_single(self): """Test method injection to callable class.""" obj = TestCallableSingle() self.assertEqual("a", obj("A")) if __name__ == "__main__": unittest.main()