« back to all changes in this revision
Viewing changes to spectlib/net/test/test_connectionmanager.py
- browse files at revision 51
- compare with another revision
- download diff
- download tarball
- view history from revision 51
- Committer: nekohayo
- Date: 2007-02-04 18:04:12 UTC
- Revision ID: svn-v3-trunk0:cc143b5f-0d1a-0410-b6b4-bdbe827cf625:trunk:70
Killed the net/ folder, system-wide install works better now. Also added network connection status icons for
Wout, removed some prints from watch.py.
Wout, removed some prints from watch.py.
- files added:
- data/icons/notifier/big/network-offline.png
- files removed:
- spectlib/net
- spectlib/net/test
- files modified:
- data/doc/ChangeLog
added
removed
spectlib/net/test/test_connectionmanager.py
1
import unittest
2
from spectlib.net.connectionmanager import (NMListener, FallbackListener,
3
CallbackRunner, get_net_listener)
4
import dbus
5
import urllib2
6
7
class MockNetworkManager(object):
8
def __init__(self):
9
self.status = 0
10
self.callbacks = {'DeviceNoLongerActive' : [],
11
'DeviceNowActive' : []}
12
13
def state(self):
14
return self.status
15
16
def setConnected(self):
17
self.status = 3
18
for callback in self.callbacks['DeviceNowActive'] :
19
callback(dbus.String('Mock device'))
20
21
def setDisconnected(self):
22
self.status = 4
23
for callback in self.callbacks['DeviceNoLongerActive'] :
24
callback(dbus.String('Mock device'))
25
26
def connect_to_signal(self, signal_name, handler_function,
27
dbus_interface=None):
28
if signal_name not in self.callbacks :
29
self.callbacks[signal_name] = []
30
self.callbacks[signal_name].append(handler_function)
31
32
def get_object(self, *args, **kwargs):
33
return self
34
35
def Interface(self, *args, **kwargs):
36
return self
37
38
39
class MockFailNetworkManager(MockNetworkManager):
40
excepitonMessage='The name org.freedesktop.NetworkManager was not provided\
41
by any .service files'
42
def state(self):
43
raise dbus.DBusException(self.excepitonMessage)
44
45
class LogingCallback(object):
46
def __init__(self):
47
self.log = []
48
49
def __call__(self, *args, **kwargs):
50
self.log.append((args, kwargs))
51
52
class TestCallbackRunner(unittest.TestCase):
53
def setUp(self):
54
self.cbRunner = CallbackRunner()
55
56
def test_addingCallback(self):
57
"""
58
Add a callback and trigger it
59
"""
60
callback = LogingCallback()
61
self.cbRunner.add_callback(callback)
62
self.cbRunner._run_callbacks()
63
self.assertEqual([(tuple() , { })], callback.log)
64
65
def test_callbackArgs(self):
66
"""
67
Add a callback, with some args, and trigger it
68
"""
69
callback = LogingCallback()
70
self.cbRunner.add_callback(callback, 'foo', 'bar')
71
self.cbRunner._run_callbacks()
72
self.assertEqual([(('foo', 'bar') , { })], callback.log)
73
74
def test_callbackKWArgs(self):
75
"""
76
Add a callback, with some KWArgs, and trigger it
77
"""
78
callback = LogingCallback()
79
self.cbRunner.add_callback(callback, foo='bar')
80
self.cbRunner._run_callbacks()
81
self.assertEqual([(tuple() , {'foo' : 'bar'})], callback.log)
82
83
def test_callbackArgsAndKWargs(self):
84
"""
85
Add a callback, with both Args and KWArgs, and trigger it.
86
"""
87
callback = LogingCallback()
88
self.cbRunner.add_callback(callback, 'empathy', foo='bar')
89
self.cbRunner._run_callbacks()
90
self.assertEqual([(('empathy', ) , {'foo' : 'bar'})], callback.log)
91
92
def test_multipleCallbacks(self):
93
"""
94
Add two callbacks. Each should be run with its specified arguments
95
"""
96
cb1 = LogingCallback()
97
cb2 = LogingCallback()
98
self.cbRunner.add_callback(cb1, 'cb1 called')
99
self.cbRunner.add_callback(cb2, 'cb2 called')
100
self.cbRunner._run_callbacks()
101
self.assertEqual([(('cb1 called',), {})], cb1.log)
102
self.assertEqual([(('cb2 called',), {})], cb2.log)
103
104
def test_registerCallbackMultipleTimes(self):
105
"""
106
Registering a callback twice should update the args and kwargs, but
107
NOT cause it to be called twice
108
"""
109
callback = LogingCallback()
110
self.cbRunner.add_callback(callback, 'foo')
111
self.cbRunner.add_callback(callback, 'bar')
112
self.cbRunner._run_callbacks()
113
self.assertEqual([(('bar', ) , {})], callback.log)
114
115
def test_callbackOnlyRunOnce(self):
116
"""
117
Callbacks should be removed after they are run. Each call to
118
add_callback should trigger at most one callback
119
"""
120
callback = LogingCallback()
121
self.cbRunner.add_callback(callback, 'foo')
122
self.cbRunner._run_callbacks()
123
self.cbRunner._run_callbacks()
124
self.assertEqual([(('foo', ) , {})], callback.log)
125
126
127
class TestNMConnectionListener(unittest.TestCase) :
128
def setUp(self):
129
self._oldDBUSInterface = dbus.Interface
130
dbus.Interface = MockNetworkManager.Interface
131
self.mockNM = MockNetworkManager()
132
self.nmListener = NMListener(self.mockNM)
133
134
def tearDown(self) :
135
dbus.Interface = self._oldDBUSInterface
136
137
def test_connected(self):
138
self.mockNM.setConnected()
139
self.assertTrue(self.nmListener.connected())
140
141
def test_unconnected(self):
142
self.mockNM.setDisconnected()
143
self.assertFalse(self.nmListener.connected())
144
145
def test_noNetworkManager(self):
146
self.nmListener.nmIface = MockFailNetworkManager()
147
self.assertFalse(self.nmListener.has_networkmanager())
148
149
def test_hasNetworkManager(self):
150
self.assertTrue(self.nmListener.has_networkmanager())
151
152
def test_cachesNetworkStatus(self):
153
"""
154
The listener should cache the connection status, and should not
155
use DBUS each call to connected()
156
"""
157
self.mockNM.setConnected()
158
#Ensure the status is cached
159
self.nmListener.connected()
160
#Now, ensure we can't get status from DBUS
161
self.nmListener.nmIface = None
162
self.assertTrue(self.nmListener.connected())
163
164
def test_callbackOnStateChange(self):
165
"""
166
All the callbacks should be run on change of state from disconnected
167
-> connected
168
"""
169
self.mockNM.setDisconnected()
170
cb = LogingCallback()
171
self.nmListener.add_callback(cb)
172
self.mockNM.setConnected()
173
self.assertEqual([(tuple(), dict())], cb.log)
174
175
176
def mockFailingUrlOpen(url) :
177
raise IOError('foo')
178
179
def mockWorkingUrlOpen(url) :
180
class MockUrl(object) :
181
def close(self) :
182
pass
183
return MockUrl()
184
185
186
class TestFallbackConnectionListener(unittest.TestCase) :
187
def setUp(self):
188
self.fbListener = FallbackListener()
189
self.realUrlOpen = urllib2.urlopen
190
191
def tearDown(self):
192
urllib2.urlopen = self.realUrlOpen
193
194
def test_connected(self):
195
urllib2.urlopen = mockWorkingUrlOpen
196
self.assertTrue(self.fbListener.connected())
197
198
def test_disconnected(self):
199
urllib2.urlopen = mockFailingUrlOpen
200
self.assertFalse(self.fbListener.connected())
201
202
203
if __name__ == '__main__':
204
unittest.main()
Loggerhead is a web-based interface for Breezy
Version: 2.0.1