diff options
Diffstat (limited to 'parts/django/tests/regressiontests/dispatch')
5 files changed, 210 insertions, 0 deletions
diff --git a/parts/django/tests/regressiontests/dispatch/__init__.py b/parts/django/tests/regressiontests/dispatch/__init__.py new file mode 100644 index 0000000..679895b --- /dev/null +++ b/parts/django/tests/regressiontests/dispatch/__init__.py @@ -0,0 +1,2 @@ +"""Unit-tests for the dispatch project +""" diff --git a/parts/django/tests/regressiontests/dispatch/models.py b/parts/django/tests/regressiontests/dispatch/models.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/parts/django/tests/regressiontests/dispatch/models.py diff --git a/parts/django/tests/regressiontests/dispatch/tests/__init__.py b/parts/django/tests/regressiontests/dispatch/tests/__init__.py new file mode 100644 index 0000000..150bb01 --- /dev/null +++ b/parts/django/tests/regressiontests/dispatch/tests/__init__.py @@ -0,0 +1,6 @@ +""" +Unit-tests for the dispatch project +""" + +from test_saferef import * +from test_dispatcher import * diff --git a/parts/django/tests/regressiontests/dispatch/tests/test_dispatcher.py b/parts/django/tests/regressiontests/dispatch/tests/test_dispatcher.py new file mode 100644 index 0000000..ad3a05f --- /dev/null +++ b/parts/django/tests/regressiontests/dispatch/tests/test_dispatcher.py @@ -0,0 +1,123 @@ +from django.dispatch import Signal +import unittest +import sys +import gc +import django.utils.copycompat as copy + +if sys.platform.startswith('java'): + def garbage_collect(): + """Run the garbage collector and wait a bit to let it do his work""" + import time + gc.collect() + time.sleep(0.1) +else: + def garbage_collect(): + gc.collect() + +def receiver_1_arg(val, **kwargs): + return val + +class Callable(object): + def __call__(self, val, **kwargs): + return val + + def a(self, val, **kwargs): + return val + +a_signal = Signal(providing_args=["val"]) + +class DispatcherTests(unittest.TestCase): + """Test suite for dispatcher (barely started)""" + + def _testIsClean(self, signal): + """Assert that everything has been cleaned up automatically""" + self.assertEqual(signal.receivers, []) + + # force cleanup just in case + signal.receivers = [] + + def testExact(self): + a_signal.connect(receiver_1_arg, sender=self) + expected = [(receiver_1_arg,"test")] + result = a_signal.send(sender=self, val="test") + self.assertEqual(result, expected) + a_signal.disconnect(receiver_1_arg, sender=self) + self._testIsClean(a_signal) + + def testIgnoredSender(self): + a_signal.connect(receiver_1_arg) + expected = [(receiver_1_arg,"test")] + result = a_signal.send(sender=self, val="test") + self.assertEqual(result, expected) + a_signal.disconnect(receiver_1_arg) + self._testIsClean(a_signal) + + def testGarbageCollected(self): + a = Callable() + a_signal.connect(a.a, sender=self) + expected = [] + del a + garbage_collect() + result = a_signal.send(sender=self, val="test") + self.assertEqual(result, expected) + self._testIsClean(a_signal) + + def testMultipleRegistration(self): + a = Callable() + a_signal.connect(a) + a_signal.connect(a) + a_signal.connect(a) + a_signal.connect(a) + a_signal.connect(a) + a_signal.connect(a) + result = a_signal.send(sender=self, val="test") + self.assertEqual(len(result), 1) + self.assertEqual(len(a_signal.receivers), 1) + del a + del result + garbage_collect() + self._testIsClean(a_signal) + + def testUidRegistration(self): + def uid_based_receiver_1(**kwargs): + pass + + def uid_based_receiver_2(**kwargs): + pass + + a_signal.connect(uid_based_receiver_1, dispatch_uid = "uid") + a_signal.connect(uid_based_receiver_2, dispatch_uid = "uid") + self.assertEqual(len(a_signal.receivers), 1) + a_signal.disconnect(dispatch_uid = "uid") + self._testIsClean(a_signal) + + def testRobust(self): + """Test the sendRobust function""" + def fails(val, **kwargs): + raise ValueError('this') + a_signal.connect(fails) + result = a_signal.send_robust(sender=self, val="test") + err = result[0][1] + self.assert_(isinstance(err, ValueError)) + self.assertEqual(err.args, ('this',)) + a_signal.disconnect(fails) + self._testIsClean(a_signal) + + def testDisconnection(self): + receiver_1 = Callable() + receiver_2 = Callable() + receiver_3 = Callable() + a_signal.connect(receiver_1) + a_signal.connect(receiver_2) + a_signal.connect(receiver_3) + a_signal.disconnect(receiver_1) + del receiver_2 + garbage_collect() + a_signal.disconnect(receiver_3) + self._testIsClean(a_signal) + +def getSuite(): + return unittest.makeSuite(DispatcherTests,'test') + +if __name__ == "__main__": + unittest.main() diff --git a/parts/django/tests/regressiontests/dispatch/tests/test_saferef.py b/parts/django/tests/regressiontests/dispatch/tests/test_saferef.py new file mode 100644 index 0000000..c0ec879 --- /dev/null +++ b/parts/django/tests/regressiontests/dispatch/tests/test_saferef.py @@ -0,0 +1,79 @@ +from django.dispatch.saferef import * + +import unittest + +class Test1(object): + def x(self): + pass + +def test2(obj): + pass + +class Test2(object): + def __call__(self, obj): + pass + +class Tester(unittest.TestCase): + def setUp(self): + ts = [] + ss = [] + for x in xrange(5000): + t = Test1() + ts.append(t) + s = safeRef(t.x, self._closure) + ss.append(s) + ts.append(test2) + ss.append(safeRef(test2, self._closure)) + for x in xrange(30): + t = Test2() + ts.append(t) + s = safeRef(t, self._closure) + ss.append(s) + self.ts = ts + self.ss = ss + self.closureCount = 0 + + def tearDown(self): + del self.ts + del self.ss + + def testIn(self): + """Test the "in" operator for safe references (cmp)""" + for t in self.ts[:50]: + self.assert_(safeRef(t.x) in self.ss) + + def testValid(self): + """Test that the references are valid (return instance methods)""" + for s in self.ss: + self.assert_(s()) + + def testShortCircuit (self): + """Test that creation short-circuits to reuse existing references""" + sd = {} + for s in self.ss: + sd[s] = 1 + for t in self.ts: + if hasattr(t, 'x'): + self.assert_(sd.has_key(safeRef(t.x))) + self.assert_(safeRef(t.x) in sd) + else: + self.assert_(sd.has_key(safeRef(t))) + self.assert_(safeRef(t) in sd) + + def testRepresentation (self): + """Test that the reference object's representation works + + XXX Doesn't currently check the results, just that no error + is raised + """ + repr(self.ss[-1]) + + def _closure(self, ref): + """Dumb utility mechanism to increment deletion counter""" + self.closureCount +=1 + +def getSuite(): + return unittest.makeSuite(Tester,'test') + +if __name__ == "__main__": + unittest.main() |