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 ++++++++++++++++++++++++++++++++ tests/dispatch/tests/__init__.py | 0 tests/dispatch/tests/test_dispatcher.py | 191 -------------------------------- 3 files changed, 191 insertions(+), 191 deletions(-) create mode 100644 tests/dispatch/tests.py delete mode 100644 tests/dispatch/tests/__init__.py delete mode 100644 tests/dispatch/tests/test_dispatcher.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) diff --git a/tests/dispatch/tests/__init__.py b/tests/dispatch/tests/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/tests/dispatch/tests/test_dispatcher.py b/tests/dispatch/tests/test_dispatcher.py deleted file mode 100644 index deee6a2b78..0000000000 --- a/tests/dispatch/tests/test_dispatcher.py +++ /dev/null @@ -1,191 +0,0 @@ -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.3