boulder-oracle-sprint: Merged to trunk [4692].

git-svn-id: http://code.djangoproject.com/svn/django/branches/boulder-oracle-sprint@4695 bcc190cf-cafb-0310-a4f2-bffc1f526a37

6 files changed, 264 insertions, 0 deletions

diff --git a/tests/regressiontests/dispatch/__init__.py b/tests/regressiontests/dispatch/__init__.py
new file mode 100644
index 0000000000..679895bb5c
--- /dev/null
+++ b/tests/regressiontests/dispatch/__init__.py
@@ -0,0 +1,2 @@
+"""Unit-tests for the dispatch project
+"""
diff --git a/tests/regressiontests/dispatch/models.py b/tests/regressiontests/dispatch/models.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/tests/regressiontests/dispatch/models.py
diff --git a/tests/regressiontests/dispatch/tests/__init__.py b/tests/regressiontests/dispatch/tests/__init__.py
new file mode 100644
index 0000000000..0fdefe48a7
--- /dev/null
+++ b/tests/regressiontests/dispatch/tests/__init__.py
@@ -0,0 +1,7 @@
+"""
+Unit-tests for the dispatch project
+"""
+
+from test_dispatcher import *
+from test_robustapply import *
+from test_saferef import *
diff --git a/tests/regressiontests/dispatch/tests/test_dispatcher.py b/tests/regressiontests/dispatch/tests/test_dispatcher.py
new file mode 100644
index 0000000000..0bc99a1505
--- /dev/null
+++ b/tests/regressiontests/dispatch/tests/test_dispatcher.py
@@ -0,0 +1,144 @@
+from django.dispatch.dispatcher import *
+from django.dispatch import dispatcher, robust
+import unittest
+import copy
+
+def x(a):
+    return a
+
+class Dummy(object):
+    pass
+
+class Callable(object):
+    def __call__(self, a):
+        return a
+    
+    def a(self, a):
+        return a
+
+class DispatcherTests(unittest.TestCase):
+    """Test suite for dispatcher (barely started)"""
+    
+    def setUp(self):
+        # track the initial state, since it's possible that others have bleed receivers in
+        self.sendersBack = copy.copy(dispatcher.sendersBack)
+        self.connections = copy.copy(dispatcher.connections)
+        self.senders = copy.copy(dispatcher.senders)
+    
+    def _testIsClean(self):
+        """Assert that everything has been cleaned up automatically"""
+        self.assertEqual(dispatcher.sendersBack, self.sendersBack)
+        self.assertEqual(dispatcher.connections, self.connections)
+        self.assertEqual(dispatcher.senders, self.senders)
+    
+    def testExact(self):
+        a = Dummy()
+        signal = 'this'
+        connect(x, signal, a)
+        expected = [(x,a)]
+        result = send('this',a, a=a)
+        self.assertEqual(result, expected)
+        disconnect(x, signal, a)
+        self.assertEqual(list(getAllReceivers(a,signal)), [])
+        self._testIsClean()
+    
+    def testAnonymousSend(self):
+        a = Dummy()
+        signal = 'this'
+        connect(x, signal)
+        expected = [(x,a)]
+        result = send(signal,None, a=a)
+        self.assertEqual(result, expected)
+        disconnect(x, signal)
+        self.assertEqual(list(getAllReceivers(None,signal)), [])
+        self._testIsClean()
+    
+    def testAnyRegistration(self):
+        a = Dummy()
+        signal = 'this'
+        connect(x, signal, Any)
+        expected = [(x,a)]
+        result = send('this',object(), a=a)
+        self.assertEqual(result, expected)
+        disconnect(x, signal, Any)
+        expected = []
+        result = send('this',object(), a=a)
+        self.assertEqual(result, expected)
+        self.assertEqual(list(getAllReceivers(Any,signal)), [])
+        
+        self._testIsClean()
+    
+    def testAnyRegistration2(self):
+        a = Dummy()
+        signal = 'this'
+        connect(x, Any, a)
+        expected = [(x,a)]
+        result = send('this',a, a=a)
+        self.assertEqual(result, expected)
+        disconnect(x, Any, a)
+        self.assertEqual(list(getAllReceivers(a,Any)), [])
+        self._testIsClean()
+    
+    def testGarbageCollected(self):
+        a = Callable()
+        b = Dummy()
+        signal = 'this'
+        connect(a.a, signal, b)
+        expected = []
+        del a
+        result = send('this',b, a=b)
+        self.assertEqual(result, expected)
+        self.assertEqual(list(getAllReceivers(b,signal)), [])
+        self._testIsClean()
+    
+    def testGarbageCollectedObj(self):
+        class x:
+            def __call__(self, a):
+                return a
+        a = Callable()
+        b = Dummy()
+        signal = 'this'
+        connect(a, signal, b)
+        expected = []
+        del a
+        result = send('this',b, a=b)
+        self.assertEqual(result, expected)
+        self.assertEqual(list(getAllReceivers(b,signal)), [])
+        self._testIsClean()
+
+    
+    def testMultipleRegistration(self):
+        a = Callable()
+        b = Dummy()
+        signal = 'this'
+        connect(a, signal, b)
+        connect(a, signal, b)
+        connect(a, signal, b)
+        connect(a, signal, b)
+        connect(a, signal, b)
+        connect(a, signal, b)
+        result = send('this',b, a=b)
+        self.assertEqual(len(result), 1)
+        self.assertEqual(len(list(getAllReceivers(b,signal))), 1)
+        del a
+        del b
+        del result
+        self._testIsClean()
+    
+    def testRobust(self):
+        """Test the sendRobust function"""
+        def fails():
+            raise ValueError('this')
+        a = object()
+        signal = 'this'
+        connect(fails, Any, a)
+        result = robust.sendRobust('this',a, a=a)
+        err = result[0][1]
+        self.assert_(isinstance(err, ValueError))
+        self.assertEqual(err.args, ('this',))
+
+def getSuite():
+    return unittest.makeSuite(DispatcherTests,'test')
+
+if __name__ == "__main__":
+    unittest.main ()
diff --git a/tests/regressiontests/dispatch/tests/test_robustapply.py b/tests/regressiontests/dispatch/tests/test_robustapply.py
new file mode 100644
index 0000000000..499450eec4
--- /dev/null
+++ b/tests/regressiontests/dispatch/tests/test_robustapply.py
@@ -0,0 +1,34 @@
+from django.dispatch.robustapply import *
+
+import unittest
+
+def noArgument():
+    pass
+
+def oneArgument(blah):
+    pass
+
+def twoArgument(blah, other):
+    pass
+
+class TestCases(unittest.TestCase):
+    def test01(self):
+        robustApply(noArgument)
+    
+    def test02(self):
+        self.assertRaises(TypeError, robustApply, noArgument, "this")
+    
+    def test03(self):
+        self.assertRaises(TypeError, robustApply, oneArgument)
+    
+    def test04(self):
+        """Raise error on duplication of a particular argument"""
+        self.assertRaises(TypeError, robustApply, oneArgument, "this", blah = "that")
+
+def getSuite():
+    return unittest.makeSuite(TestCases,'test')
+
+
+if __name__ == "__main__":
+    unittest.main()
+    
diff --git a/tests/regressiontests/dispatch/tests/test_saferef.py b/tests/regressiontests/dispatch/tests/test_saferef.py
new file mode 100644
index 0000000000..233a2023e0
--- /dev/null
+++ b/tests/regressiontests/dispatch/tests/test_saferef.py
@@ -0,0 +1,77 @@
+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)))
+            else:
+                self.assert_(sd.has_key(safeRef(t)))
+    
+    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()