Merged regressiontests and modeltests into the test root.

5 files changed, 245 insertions, 0 deletions

diff --git a/tests/dispatch/__init__.py b/tests/dispatch/__init__.py
new file mode 100644
index 0000000000..679895bb5c
--- /dev/null
+++ b/tests/dispatch/__init__.py
@@ -0,0 +1,2 @@
+"""Unit-tests for the dispatch project
+"""
diff --git a/tests/dispatch/models.py b/tests/dispatch/models.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/tests/dispatch/models.py
diff --git a/tests/dispatch/tests/__init__.py b/tests/dispatch/tests/__init__.py
new file mode 100644
index 0000000000..b6d26217e1
--- /dev/null
+++ b/tests/dispatch/tests/__init__.py
@@ -0,0 +1,8 @@
+"""
+Unit-tests for the dispatch project
+"""
+
+from __future__ import absolute_import
+
+from .test_dispatcher import DispatcherTests, ReceiverTestCase
+from .test_saferef import SaferefTests
diff --git a/tests/dispatch/tests/test_dispatcher.py b/tests/dispatch/tests/test_dispatcher.py
new file mode 100644
index 0000000000..5f8f92acaf
--- /dev/null
+++ b/tests/dispatch/tests/test_dispatcher.py
@@ -0,0 +1,164 @@
+import gc
+import sys
+import time
+
+from django.dispatch import Signal, receiver
+from django.utils import unittest
+
+
+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"])
+
+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.assertTrue(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 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 testReceiverSingleSignal(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 testReceiverSignalList(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/test_saferef.py b/tests/dispatch/tests/test_saferef.py
new file mode 100644
index 0000000000..30eaddfe18
--- /dev/null
+++ b/tests/dispatch/tests/test_saferef.py
@@ -0,0 +1,71 @@
+from django.dispatch.saferef import safeRef
+from django.utils.six.moves import xrange
+from django.utils import unittest
+
+class Test1(object):
+    def x(self):
+        pass
+
+def test2(obj):
+    pass
+
+class Test2(object):
+    def __call__(self, obj):
+        pass
+
+class SaferefTests(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.assertTrue(safeRef(t.x) in self.ss)
+
+    def testValid(self):
+        """Test that the references are valid (return instance methods)"""
+        for s in self.ss:
+            self.assertTrue(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.assertTrue(safeRef(t.x) in sd)
+            else:
+                self.assertTrue(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