signal.py
''' @author: Yoriz ''' import inspect import weakref class Ysignal(object): '''WeakRef Signal/Slots''' def __init__(self): '''Initialise attributes to store observers''' self._functions = weakref.WeakSet() self._methods = weakref.WeakKeyDictionary() def emit_slot(self, slot, *args, **kwargs): '''emit a signal to the passed in slot only''' slot(*args, **kwargs) def emit(self, *args, **kwargs): '''emit a signal to all slots''' self._emit_functions(*args, **kwargs) self._emit_methods(*args, **kwargs) def _emit_functions(self, *args, **kwargs): '''Emits a signal to any Function slots''' for func in tuple(self._functions): func(*args, **kwargs) def _emit_methods(self, *args, **kwargs): '''Emits a signal to any Method slots''' for obj, funcs in self._methods.items(): for func in tuple(funcs): method = getattr(obj, func.__name__) method(*args, **kwargs) def bind(self, slot): '''Add a slot to the list of listeners''' if inspect.ismethod(slot): self._bind_method(slot) else: self._bind_function(slot) def _bind_function(self, slot): '''Add a Function slot''' self._functions.add(slot) def _bind_method(self, slot): '''Add a Method slot''' try: self._methods[slot.__self__].add(slot.__func__) except KeyError: self._methods[slot.__self__] = set() self._bind_method(slot) def unbind(self, slot): '''Remove slot from the list of listeners''' if inspect.ismethod(slot): self._unbind_method(slot) else: self._unbind_function(slot) def _unbind_function(self, slot): '''Remove a Function slot''' try: self._functions.remove(slot) except (ValueError, KeyError): pass def _unbind_method(self, slot): '''Remove a Method slot''' try: self._methods[slot.__self__].remove(slot.__func__) except (ValueError, KeyError): pass def unbind_all(self): '''Remove all slots''' self._functions.clear() self._methods.clear()test_ysignal.py
''' @author: Yoriz ''' import unittest import ysignal class TestYsignal(unittest.TestCase): def setUp(self): self.signal = ysignal.Ysignal() self.attr1 = None self.attr2 = None self.attr3 = 0 def tearDown(self): pass def set_attr1(self, value): self.attr1 = value def method_for_test(self): pass def test_connect_function(self): def test_function(): pass func = test_function self.signal.bind(func) self.assertSetEqual(set((func,)), self.signal._functions) def test_connect_method(self): self.signal.bind(self.set_attr1) self.assertIn(self, self.signal._methods) funcs = self.signal._methods.get(self, set()) self.assertIn(self.set_attr1.__func__, funcs) def test_emit_function(self): value = 'EmitFunction' def test_function(value): self.attr1 = value self.signal.bind(test_function) self.signal.emit(value=value) self.assertEqual(self.attr1, value) def test_emit_method(self): value = 'EmitMethod' self.signal.bind(self.set_attr1) self.signal.emit(value=value) self.assertEqual(self.attr1, value) def test_emit_function_and_method(self): value = 'EmitBothTypes' def test_function(value): self.attr2 = value self.signal.bind(test_function) self.signal.bind(self.set_attr1) self.signal.emit(value=value) self.assertEqual(self.attr1, value) self.assertEqual(self.attr2, value) def test_emit_slot_function(self): value = 'EmitSlotFunction' def test_function(value): self.attr1 = value self.signal.emit_slot(test_function, value=value) self.assertEqual(self.attr1, value) def test_emit_slot_method(self): value = 'EmitMethod' self.signal.emit_slot(self.set_attr1, value=value) self.assertEqual(self.attr1, value) def test_disconnect_slot_function(self): def test_function(): pass def test_function2(): pass func = test_function func2 = test_function2 self.signal.bind(func) self.signal.bind(func2) self.assertSetEqual(set((func, func2)), self.signal._functions) self.signal.unbind(func2) self.assertSetEqual(set((func,)), self.signal._functions) def test_disconnect_slot_method(self): self.signal.bind(self.set_attr1) self.signal.bind(self.method_for_test) self.assertIn(self, self.signal._methods) funcs = self.signal._methods.get(self, []) self.assertSetEqual(set((self.set_attr1.__func__, self.method_for_test.__func__)), funcs) self.signal.unbind(self.method_for_test) funcs = self.signal._methods.get(self, set()) self.assertSetEqual(set((self.set_attr1.__func__,)), funcs) def test_disconnect_all(self): def test_function(): pass def test_function2(): pass func = test_function func2 = test_function2 self.signal.bind(func) self.signal.bind(func2) self.signal.bind(self.set_attr1) self.signal.bind(self.method_for_test) self.signal.unbind_all() self.assertSetEqual(set(), self.signal._functions) funcs = self.signal._methods.get(self, set()) self.assertSetEqual(set(), funcs) if __name__ == '__main__': unittest.main()