From 136a3ffe21988d49e443867d129cc01fb62b34cd Mon Sep 17 00:00:00 2001 From: Tim Graham Date: Mon, 7 Jul 2014 19:21:47 -0400 Subject: Removed unnecessary directory in dispatch tests. --- tests/dispatch/tests.py | 191 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 191 insertions(+) create mode 100644 tests/dispatch/tests.py (limited to 'tests/dispatch/tests.py') diff --git a/tests/dispatch/tests.py b/tests/dispatch/tests.py new file mode 100644 index 0000000000..deee6a2b78 --- /dev/null +++ b/tests/dispatch/tests.py @@ -0,0 +1,191 @@ +import gc +import sys +import time +import unittest +import weakref +from types import TracebackType + +from django.dispatch import Signal, receiver + + +if sys.platform.startswith('java'): + def garbage_collect(): + # Some JVM GCs will execute finalizers in a different thread, meaning + # we need to wait for that to complete before we go on looking for the + # effects of that. + gc.collect() + time.sleep(0.1) +elif hasattr(sys, "pypy_version_info"): + def garbage_collect(): + # Collecting weakreferences can take two collections on PyPy. + gc.collect() + gc.collect() +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"]) +b_signal = Signal(providing_args=["val"]) +c_signal = Signal(providing_args=["val"]) +d_signal = Signal(providing_args=["val"], use_caching=True) + + +class DispatcherTests(unittest.TestCase): + """Test suite for dispatcher (barely started)""" + + def _testIsClean(self, signal): + """Assert that everything has been cleaned up automatically""" + # Note that dead weakref cleanup happens as side effect of using + # the signal's receivers through the signals API. So, first do a + # call to an API method to force cleanup. + self.assertFalse(signal.has_listeners()) + self.assertEqual(signal.receivers, []) + + def test_exact(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 test_ignored_sender(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 test_garbage_collected(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 test_cached_garbaged_collected(self): + """ + Make sure signal caching sender receivers don't prevent garbage + collection of senders. + """ + class sender: + pass + wref = weakref.ref(sender) + d_signal.connect(receiver_1_arg) + d_signal.send(sender, val='garbage') + del sender + garbage_collect() + try: + self.assertIsNone(wref()) + finally: + # Disconnect after reference check since it flushes the tested cache. + d_signal.disconnect(receiver_1_arg) + + def test_multiple_registration(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 test_uid_registration(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 test_robust(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.assertIsInstance(err, ValueError) + self.assertEqual(err.args, ('this',)) + self.assertTrue(hasattr(err, '__traceback__')) + self.assertTrue(isinstance(err.__traceback__, TracebackType)) + a_signal.disconnect(fails) + self._testIsClean(a_signal) + + def test_disconnection(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 test_has_listeners(self): + self.assertFalse(a_signal.has_listeners()) + self.assertFalse(a_signal.has_listeners(sender=object())) + receiver_1 = Callable() + a_signal.connect(receiver_1) + self.assertTrue(a_signal.has_listeners()) + self.assertTrue(a_signal.has_listeners(sender=object())) + a_signal.disconnect(receiver_1) + self.assertFalse(a_signal.has_listeners()) + self.assertFalse(a_signal.has_listeners(sender=object())) + + +class ReceiverTestCase(unittest.TestCase): + """ + Test suite for receiver. + + """ + def test_receiver_single_signal(self): + @receiver(a_signal) + def f(val, **kwargs): + self.state = val + self.state = False + a_signal.send(sender=self, val=True) + self.assertTrue(self.state) + + def test_receiver_signal_list(self): + @receiver([a_signal, b_signal, c_signal]) + def f(val, **kwargs): + self.state.append(val) + self.state = [] + a_signal.send(sender=self, val='a') + c_signal.send(sender=self, val='c') + b_signal.send(sender=self, val='b') + self.assertIn('a', self.state) + self.assertIn('b', self.state) + self.assertIn('c', self.state) -- cgit v1.2.1