← Back to branch summary

~facundo/ubuntuone-client/support-ssl-error

« back to all changes in this revision

Viewing changes to tests/syncdaemon/test_main.py

  • Committer: Tarmac
  • Author(s): Natalia B. Bidart
  • Date: 2011-08-10 19:08:43 UTC
  • mfrom: (1099.1.2 clean-env-aq)
  • Revision ID: tarmac-20110810190843-80bere8l3k3471x6
- Fake ExternalInterface when testing the main module (LP: #823982).
- Do a proper clean on test_action_queue tests (LP: #823895).

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/syncdaemon/test_main.py
57
57
        """Something! :)"""
58
58
 
59
59
 
 
60
class FakedExternalInterface(object):
 
61
    """Do nothing."""
 
62
 
 
63
    def __init__(self, *a, **kw):
 
64
        self._called = []
 
65
        self.connect = lambda *a, **kw: self._called.append(('connect', a, kw))
 
66
        self.shutdown = lambda *a, **kw: None
 
67
 
 
68
 
60
69
class MainTests(BaseTwistedTestCase):
61
70
    """ Basic tests to check main.Main """
62
71
 
71
80
        # perform the extra setup steps according to the platform
72
81
        setup_main_test(self)
73
82
 
 
83
        self.patch(main_mod, 'ExternalInterface', FakedExternalInterface)
 
84
 
74
85
        self.handler = MementoHandler()
75
86
        self.handler.setLevel(logging.DEBUG)
76
87
        self._logger = logging.getLogger('ubuntuone.SyncDaemon')
78
89
        self.addCleanup(self._logger.removeHandler, self.handler)
79
90
 
80
91
    def build_main(self, **kwargs):
81
 
        """
82
 
        Build and return a Main object, using reasonable defaults for
83
 
        the tests, plus whatever extra kwargs are passed in.
 
92
        """Build and return a Main object.
 
93
 
 
94
        Use reasonable defaults for the tests, plus whatever extra kwargs are
 
95
        passed in.
 
96
 
84
97
        """
85
98
        # get the params using the platform code to ensure they are correct
86
 
        params =  get_main_params(self, _get_main_common_params(self))
 
99
        params = get_main_params(self, _get_main_common_params(self))
87
100
        params.update(kwargs)
88
101
        m = main_mod.Main(**params)
 
102
        self.addCleanup(m.shutdown)
89
103
        m.local_rescan = lambda *_: m.event_q.push('SYS_LOCAL_RESCAN_DONE')
90
104
        return m
91
105
 
92
106
    def test_main_initialization(self):
93
107
        """test that creating a Main instance works as expected."""
94
108
        main = self.build_main()
95
 
        main.shutdown()
 
109
        self.assertIsInstance(main, main_mod.Main)
96
110
 
97
111
    def test_main_start(self):
98
112
        """Test that Main.start works."""
99
113
        main = self.build_main()
100
114
        main.start()
101
 
        main.shutdown()
102
115
 
103
116
    def test_main_restarts_on_critical_error(self):
104
117
        """Test that Main restarts when syncdaemon gets into UNKNOWN_ERROR."""
108
121
        main.start()
109
122
        main.event_q.push('SYS_UNKNOWN_ERROR')
110
123
        self.assertTrue(self.restarted)
111
 
        main.shutdown()
112
124
 
113
125
    def test_shutdown_pushes_sys_quit(self):
114
126
        """When shutting down, the SYS_QUIT event is pushed."""
115
 
        main = self.build_main()
 
127
        params = get_main_params(self, _get_main_common_params(self))
 
128
        main = main_mod.Main(**params)
116
129
        events = []
117
130
        self.patch(main.event_q, 'push',
118
131
                   lambda *a, **kw: events.append((a, kw)))
147
160
        main.start()
148
161
        main.event_q.push('SYS_NET_CONNECTED')
149
162
        main.event_q.push('SYS_USER_CONNECT', access_token='')
150
 
        d0.addCallback(lambda _: main.shutdown())
151
163
        return d0
152
164
 
153
165
    def test_create_dirs_already_exists_dirs(self):
154
166
        """test that creating a Main instance works as expected."""
155
167
        link = os.path.join(self.root, 'Shared With Me')
156
 
        self.assertFalse(path_exists(link))
 
168
        self.assertFalse(is_link(link))
157
169
        self.assertTrue(path_exists(self.shares))
158
170
        self.assertTrue(path_exists(self.root))
159
171
        main = self.build_main()
160
 
        self.assertTrue(path_exists(main.shares_dir_link))
161
172
        # check that the shares link is actually a link
162
173
        self.assertTrue(is_link(main.shares_dir_link))
163
174
        self.assertEquals(link, main.shares_dir_link)
164
 
        main.shutdown()
165
175
 
166
176
    def test_create_dirs_already_exists_symlink_too(self):
167
177
        """test that creating a Main instance works as expected."""
168
178
        link = os.path.join(self.root, 'Shared With Me')
169
179
        make_link(self.shares, link)
170
 
        self.assertTrue(path_exists(link))
171
180
        self.assertTrue(is_link(link))
172
181
        self.assertTrue(path_exists(self.shares))
173
182
        self.assertTrue(path_exists(self.root))
174
183
        main = self.build_main()
175
184
        # check that the shares link is actually a link
176
185
        self.assertTrue(is_link(main.shares_dir_link))
177
 
        main.shutdown()
178
186
 
179
187
    def test_create_dirs_already_exists_but_not_symlink(self):
180
188
        """test that creating a Main instance works as expected."""
188
196
        # check that the shares link is actually a link
189
197
        self.assertEquals(main.shares_dir_link, link)
190
198
        self.assertFalse(is_link(main.shares_dir_link))
191
 
        main.shutdown()
192
199
 
193
200
    def test_create_dirs_none_exists(self):
194
201
        """test that creating a Main instance works as expected."""
195
 
        link = os.path.join(self.root, 'Shared With Me')
196
202
        # remove the existing dirs
197
203
        remove_dir(self.root)
198
204
        remove_dir(self.shares)
199
205
        main = self.build_main()
200
206
        # check that the shares link is actually a link
201
 
        self.assertTrue(path_exists(link))
202
207
        self.assertTrue(is_link(main.shares_dir_link))
203
208
        self.assertTrue(path_exists(self.shares))
204
209
        self.assertTrue(path_exists(self.root))
205
 
        main.shutdown()
206
210
 
207
211
    def test_connect_if_autoconnect_is_enabled(self):
208
212
        """If autoconnect option is enabled, connect the syncdaemon."""
211
215
        user_config.set_autoconnect(True)
212
216
        self.addCleanup(user_config.set_autoconnect, orig)
213
217
 
214
 
        self.connect_called = False
215
 
        self.patch(main_mod.ubuntuone.platform.ExternalInterface, 'connect',
216
 
                   lambda *a, **kw: setattr(self, 'connect_called', (a, kw)))
217
 
 
218
218
        main = self.build_main()
219
 
        self.addCleanup(main.shutdown)
220
 
 
221
 
        self.assertEqual(self.connect_called, ((main.external,),
222
 
                                               {'autoconnecting': True}))
 
219
        expected = [('connect', (), {'autoconnecting': True})]
 
220
        self.assertEqual(main.external._called, expected)
223
221
 
224
222
    def test_dont_connect_if_autoconnect_is_disabled(self):
225
223
        """If autoconnect option is disabled, do not connect the syncdaemon."""
228
226
        user_config.set_autoconnect(False)
229
227
        self.addCleanup(user_config.set_autoconnect, orig)
230
228
 
231
 
        self.connect_called = False
232
 
        self.patch(main_mod.ubuntuone.platform.ExternalInterface, 'connect',
233
 
                   lambda *a, **kw: setattr(self, 'connect_called', True))
234
 
 
235
229
        main = self.build_main()
236
 
        self.addCleanup(main.shutdown)
237
 
 
238
 
        self.assertFalse(self.connect_called)
 
230
        self.assertEqual(main.external._called, [])
239
231
 
240
232
    def _get_listeners(self, main):
241
233
        """Return the subscribed objects."""
247
239
 
248
240
    def test_event_logger_starts_if_available(self):
249
241
        """The event logger is started if available."""
250
 
        self.patch(main_mod.status_listener,
 
242
        self.patch(main_mod.event_logging,
251
243
                   "get_listener", lambda *a: FakeListener())
252
244
        main = self.build_main()
253
 
        self.addCleanup(main.shutdown)
254
245
        self.assertIn(main.eventlog_listener, self._get_listeners(main))
255
246
 
256
247
    def test_event_logger_not_started_if_not_available(self):
257
248
        """The event logger is not started if it's not available."""
258
249
        self.patch(main_mod.event_logging, "get_listener", lambda *a: None)
259
250
        main = self.build_main()
260
 
        self.addCleanup(main.shutdown)
261
251
        self.assertNotIn(main.eventlog_listener, self._get_listeners(main))
262
252
 
263
253
    def test_status_listener_is_installed(self):
265
255
        self.patch(main_mod.status_listener,
266
256
                   "get_listener", lambda *a: FakeListener())
267
257
        main = self.build_main()
268
 
        self.addCleanup(main.shutdown)
269
258
        self.assertIn(main.status_listener, self._get_listeners(main))
270
259
 
271
260
    def test_status_listener_not_installed_when_disabled(self):
272
261
        """The status listener is not started if it's not available."""
273
262
        self.patch(main_mod.status_listener, "get_listener", lambda *a: None)
274
263
        main = self.build_main()
275
 
        self.addCleanup(main.shutdown)
276
264
        self.assertNotIn(main.status_listener, self._get_listeners(main))
277
265
 
278
266
    def test_get_rootdir(self):
279
267
        """The get_rootdir returns the root dir."""
280
 
        expected = '~/Ubuntu Test One'
 
268
        expected = os.path.expanduser(os.path.join('~', 'Ubuntu Test One'))
281
269
        main = self.build_main(root_dir=expected)
282
 
        self.addCleanup(main.shutdown)
283
 
 
284
270
        self.assertEqual(main.get_rootdir(), expected)
285
271
 
286
272
    def test_get_sharesdir(self):
287
273
        """The get_sharesdir returns the shares dir."""
288
 
        expected = '~/Share it to Me'
 
274
        expected = os.path.expanduser(os.path.join('~', 'Share it to Me'))
289
275
        main = self.build_main(shares_dir=expected)
290
 
        self.addCleanup(main.shutdown)
291
 
 
292
276
        self.assertEqual(main.get_sharesdir(), expected)
293
277
 
294
278
    def test_get_sharesdirlink(self):
295
279
        """The get_sharesdirlink returns the shares dir link."""
296
280
        expected = 'Share it to Me'
297
281
        main = self.build_main(shares_symlink_name=expected)
298
 
        self.addCleanup(main.shutdown)
299
 
 
300
282
        self.assertEqual(main.get_sharesdir_link(),
301
283
                         os.path.join(main.get_rootdir(), expected))
302
284
 
303
285
    def test_version_is_logged(self):
304
286
        """Test that the client version is logged."""
305
 
        main = self.build_main()
306
 
        self.addCleanup(main.shutdown)
 
287
        self.build_main()
307
288
        self.assertTrue(self.handler.check_info("client version", VERSION))