← Back to branch summary

~ubuntu-branches/ubuntu/jaunty/python-django/jaunty

« back to all changes in this revision

Viewing changes to tests/regressiontests/dispatch/tests/test_dispatcher.py

  • Committer: Bazaar Package Importer
  • Author(s): Scott James Remnant, Eddy Mulyono
  • Date: 2008-09-16 12:18:47 UTC
  • mfrom: (1.1.5 upstream) (4.1.1 lenny)
  • Revision ID: james.westby@ubuntu.com-20080916121847-mg225rg5mnsdqzr0
Tags: 1.0-1ubuntu1
https://launchpad.net/bugs/264191
* Merge from Debian (LP: #264191), remaining changes:
  - Run test suite on build.

[Eddy Mulyono]
* Update patch to workaround network test case failures.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/regressiontests/dispatch/tests/test_dispatcher.py
1
 
from django.dispatch.dispatcher import *
2
 
from django.dispatch import dispatcher, robust
 
1
from django.dispatch import Signal
3
2
import unittest
4
3
import copy
5
 
 
6
 
def x(a):
7
 
    return a
8
 
 
9
 
class Dummy(object):
10
 
    pass
 
4
import sys
 
5
import gc
 
6
 
 
7
if sys.platform.startswith('java'):
 
8
    def garbage_collect():
 
9
        """Run the garbage collector and wait a bit to let it do his work"""
 
10
        import time
 
11
        gc.collect()
 
12
        time.sleep(0.1)
 
13
else:
 
14
    def garbage_collect():
 
15
        gc.collect()
 
16
 
 
17
def receiver_1_arg(val, **kwargs):
 
18
    return val
11
19
 
12
20
class Callable(object):
13
 
    def __call__(self, a):
14
 
        return a
 
21
    def __call__(self, val, **kwargs):
 
22
        return val
15
23
    
16
 
    def a(self, a):
17
 
        return a
 
24
    def a(self, val, **kwargs):
 
25
        return val
 
26
 
 
27
a_signal = Signal(providing_args=["val"])
18
28
 
19
29
class DispatcherTests(unittest.TestCase):
20
30
    """Test suite for dispatcher (barely started)"""
21
 
    
22
 
    def setUp(self):
23
 
        # track the initial state, since it's possible that others have bleed receivers in
24
 
        self.sendersBack = copy.copy(dispatcher.sendersBack)
25
 
        self.connections = copy.copy(dispatcher.connections)
26
 
        self.senders = copy.copy(dispatcher.senders)
27
 
    
28
 
    def _testIsClean(self):
 
31
 
 
32
    def _testIsClean(self, signal):
29
33
        """Assert that everything has been cleaned up automatically"""
30
 
        self.assertEqual(dispatcher.sendersBack, self.sendersBack)
31
 
        self.assertEqual(dispatcher.connections, self.connections)
32
 
        self.assertEqual(dispatcher.senders, self.senders)
 
34
        self.assertEqual(signal.receivers, [])
 
35
 
 
36
        # force cleanup just in case
 
37
        signal.receivers = []
33
38
    
34
39
    def testExact(self):
35
 
        a = Dummy()
36
 
        signal = 'this'
37
 
        connect(x, signal, a)
38
 
        expected = [(x,a)]
39
 
        result = send('this',a, a=a)
40
 
        self.assertEqual(result, expected)
41
 
        disconnect(x, signal, a)
42
 
        self.assertEqual(list(getAllReceivers(a,signal)), [])
43
 
        self._testIsClean()
44
 
    
45
 
    def testAnonymousSend(self):
46
 
        a = Dummy()
47
 
        signal = 'this'
48
 
        connect(x, signal)
49
 
        expected = [(x,a)]
50
 
        result = send(signal,None, a=a)
51
 
        self.assertEqual(result, expected)
52
 
        disconnect(x, signal)
53
 
        self.assertEqual(list(getAllReceivers(None,signal)), [])
54
 
        self._testIsClean()
55
 
    
56
 
    def testAnyRegistration(self):
57
 
        a = Dummy()
58
 
        signal = 'this'
59
 
        connect(x, signal, Any)
60
 
        expected = [(x,a)]
61
 
        result = send('this',object(), a=a)
62
 
        self.assertEqual(result, expected)
63
 
        disconnect(x, signal, Any)
64
 
        expected = []
65
 
        result = send('this',object(), a=a)
66
 
        self.assertEqual(result, expected)
67
 
        self.assertEqual(list(getAllReceivers(Any,signal)), [])
68
 
        
69
 
        self._testIsClean()
70
 
    
71
 
    def testAnyRegistration2(self):
72
 
        a = Dummy()
73
 
        signal = 'this'
74
 
        connect(x, Any, a)
75
 
        expected = [(x,a)]
76
 
        result = send('this',a, a=a)
77
 
        self.assertEqual(result, expected)
78
 
        disconnect(x, Any, a)
79
 
        self.assertEqual(list(getAllReceivers(a,Any)), [])
80
 
        self._testIsClean()
 
40
        a_signal.connect(receiver_1_arg, sender=self)
 
41
        expected = [(receiver_1_arg,"test")]
 
42
        result = a_signal.send(sender=self, val="test")
 
43
        self.assertEqual(result, expected)
 
44
        a_signal.disconnect(receiver_1_arg, sender=self)
 
45
        self._testIsClean(a_signal)
 
46
 
 
47
    def testIgnoredSender(self):
 
48
        a_signal.connect(receiver_1_arg)
 
49
        expected = [(receiver_1_arg,"test")]
 
50
        result = a_signal.send(sender=self, val="test")
 
51
        self.assertEqual(result, expected)
 
52
        a_signal.disconnect(receiver_1_arg)
 
53
        self._testIsClean(a_signal)
81
54
    
82
55
    def testGarbageCollected(self):
83
56
        a = Callable()
84
 
        b = Dummy()
85
 
        signal = 'this'
86
 
        connect(a.a, signal, b)
87
 
        expected = []
88
 
        del a
89
 
        result = send('this',b, a=b)
90
 
        self.assertEqual(result, expected)
91
 
        self.assertEqual(list(getAllReceivers(b,signal)), [])
92
 
        self._testIsClean()
93
 
    
94
 
    def testGarbageCollectedObj(self):
95
 
        class x:
96
 
            def __call__(self, a):
97
 
                return a
98
 
        a = Callable()
99
 
        b = Dummy()
100
 
        signal = 'this'
101
 
        connect(a, signal, b)
102
 
        expected = []
103
 
        del a
104
 
        result = send('this',b, a=b)
105
 
        self.assertEqual(result, expected)
106
 
        self.assertEqual(list(getAllReceivers(b,signal)), [])
107
 
        self._testIsClean()
108
 
 
 
57
        a_signal.connect(a.a, sender=self)
 
58
        expected = []
 
59
        del a
 
60
        garbage_collect()
 
61
        result = a_signal.send(sender=self, val="test")
 
62
        self.assertEqual(result, expected)
 
63
        self._testIsClean(a_signal)
109
64
    
110
65
    def testMultipleRegistration(self):
111
66
        a = Callable()
112
 
        b = Dummy()
113
 
        signal = 'this'
114
 
        connect(a, signal, b)
115
 
        connect(a, signal, b)
116
 
        connect(a, signal, b)
117
 
        connect(a, signal, b)
118
 
        connect(a, signal, b)
119
 
        connect(a, signal, b)
120
 
        result = send('this',b, a=b)
 
67
        a_signal.connect(a)
 
68
        a_signal.connect(a)
 
69
        a_signal.connect(a)
 
70
        a_signal.connect(a)
 
71
        a_signal.connect(a)
 
72
        a_signal.connect(a)
 
73
        result = a_signal.send(sender=self, val="test")
121
74
        self.assertEqual(len(result), 1)
122
 
        self.assertEqual(len(list(getAllReceivers(b,signal))), 1)
 
75
        self.assertEqual(len(a_signal.receivers), 1)
123
76
        del a
124
 
        del b
125
77
        del result
126
 
        self._testIsClean()
 
78
        garbage_collect()
 
79
        self._testIsClean(a_signal)
 
80
 
 
81
    def testUidRegistration(self):
 
82
        def uid_based_receiver_1(**kwargs):
 
83
            pass
 
84
 
 
85
        def uid_based_receiver_2(**kwargs):
 
86
            pass
 
87
 
 
88
        a_signal.connect(uid_based_receiver_1, dispatch_uid = "uid")
 
89
        a_signal.connect(uid_based_receiver_2, dispatch_uid = "uid")
 
90
        self.assertEqual(len(a_signal.receivers), 1)
 
91
        a_signal.disconnect(dispatch_uid = "uid")
 
92
        self._testIsClean(a_signal)
127
93
    
128
94
    def testRobust(self):
129
95
        """Test the sendRobust function"""
130
 
        def fails():
 
96
        def fails(val, **kwargs):
131
97
            raise ValueError('this')
132
 
        a = object()
133
 
        signal = 'this'
134
 
        connect(fails, Any, a)
135
 
        result = robust.sendRobust('this',a, a=a)
 
98
        a_signal.connect(fails)
 
99
        result = a_signal.send_robust(sender=self, val="test")
136
100
        err = result[0][1]
137
101
        self.assert_(isinstance(err, ValueError))
138
102
        self.assertEqual(err.args, ('this',))
 
103
        a_signal.disconnect(fails)
 
104
        self._testIsClean(a_signal)
139
105
 
140
106
def getSuite():
141
107
    return unittest.makeSuite(DispatcherTests,'test')
142
108
 
143
109
if __name__ == "__main__":
144
 
    unittest.main ()
 
110
    unittest.main()