← Back to branch summary

~ubuntu-branches/ubuntu/precise/ubuntuone-client/precise-201112142106

« back to all changes in this revision

Viewing changes to tests/platform/linux/test_filesystem_notifications.py

  • Committer: Bazaar Package Importer
  • Author(s): Rodney Dawes
  • Date: 2011-08-09 12:47:56 UTC
  • mfrom: (1.1.53 upstream)
  • Revision ID: james.westby@ubuntu.com-20110809124756-7nzilp3oix0a1yl9
Tags: 1.7.1-0ubuntu1
* New upstream release.
* debian/*:
  - Removed obsolete pycompat file
  - Removed ubuntuone-client-gnome deps and binary packaging, as it was
    moved out to separate project upstream.
  - Updated copyright to remove obsolete file reference
* debian/patches:
  - Removed patches which have been included upstream.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/platform/linux/test_filesystem_notifications.py
1
1
#
2
 
# Author: Facundo Batista <facundo@canonical.com>
3
 
#
4
 
# Author: Guillermo Gonzalez <guillermo.gonzalez@canonical.com>
5
 
#
6
 
# Copyright 2009 Canonical Ltd.
 
2
# Authors: Facundo Batista <facundo@canonical.com>
 
3
#          Guillermo Gonzalez <guillermo.gonzalez@canonical.com>
 
4
#
 
5
# Copyright 2009-2011 Canonical Ltd.
7
6
#
8
7
# This program is free software: you can redistribute it and/or modify it
9
8
# under the terms of the GNU General Public License version 3, as published
21
20
 
22
21
import logging
23
22
import os
24
 
import unittest
25
23
 
26
24
from twisted.internet import defer
27
25
 
34
32
from ubuntuone.syncdaemon import volume_manager
35
33
from ubuntuone.syncdaemon.tritcask import Tritcask
36
34
 
 
35
# We normally access to private attribs in tests
 
36
# pylint: disable=W0212
 
37
 
37
38
 
38
39
class BaseFSMonitorTestCase(testcase.BaseTwistedTestCase):
39
40
    """Test the structures where we have the path/watch."""
40
41
 
41
42
    timeout = 3
42
43
 
 
44
    @defer.inlineCallbacks
43
45
    def setUp(self):
44
46
        """Set up."""
45
 
        testcase.BaseTwistedTestCase.setUp(self)
 
47
        yield super(BaseFSMonitorTestCase, self).setUp()
46
48
        fsmdir = self.mktemp('fsmdir')
47
49
        partials_dir = self.mktemp('partials_dir')
48
50
        self.root_dir = self.mktemp('root_dir')
70
72
        self.log_handler.setLevel(logging.DEBUG)
71
73
        self.monitor.log.addHandler(self.log_handler)
72
74
 
 
75
    @defer.inlineCallbacks
73
76
    def tearDown(self):
74
77
        """Clean up the tests."""
75
78
        self.monitor.shutdown()
76
 
        self.rmtree(self.tmpdir)
77
79
        self.monitor.log.removeHandler(self.log_handler)
78
 
        testcase.BaseTwistedTestCase.tearDown(self)
 
80
        yield super(BaseFSMonitorTestCase, self).tearDown()
79
81
 
80
82
 
81
83
class WatchManagerTests(BaseFSMonitorTestCase):
89
91
        self.monitor.add_watch(path)
90
92
 
91
93
        # we have the watch, remove the dir, the watch should be gone
92
 
        self.assertTrue(self.monitor.has_watch(path))
 
94
        self.assertIn(path, self.monitor._general_watchs)
93
95
        os.rmdir(path)
94
96
 
95
97
        def check(_):
96
98
            """Check state after the event."""
97
 
            self.assertFalse(self.monitor.has_watch(path))
 
99
            self.assertNotIn(path, self.monitor._general_watchs)
98
100
 
99
101
        self.deferred.addCallback(check)
100
102
        return self.deferred
109
111
 
110
112
        # we have the watch, rename the dir, new name should have the watch,
111
113
        # old name should not
112
 
        self.assertTrue(self.monitor.has_watch(path1))
 
114
        self.assertIn(path1, self.monitor._general_watchs)
113
115
        os.rename(path1, path2)
114
116
 
115
117
        def check(_):
116
118
            """Check state after the event."""
117
 
            self.assertTrue(self.monitor.has_watch(path2))
118
 
            self.assertFalse(self.monitor.has_watch(path1))
 
119
            self.assertIn(path2, self.monitor._general_watchs)
 
120
            self.assertNotIn(path1, self.monitor._general_watchs)
119
121
 
120
122
        self.deferred.addCallback(check)
121
123
        return self.deferred
131
133
        self.monitor.add_watch(path1)
132
134
 
133
135
        # we have the watch, move it outside watched structure, no more watches
134
 
        self.assertTrue(self.monitor.has_watch(path1))
 
136
        self.assertIn(path1, self.monitor._general_watchs)
135
137
        os.rename(path1, path2)
136
138
 
137
139
        def check(_):
138
140
            """Check state after the event."""
139
 
            self.assertFalse(self.monitor.has_watch(path1))
140
 
            self.assertFalse(self.monitor.has_watch(path2))
 
141
            self.assertNotIn(path1, self.monitor._general_watchs)
 
142
            self.assertNotIn(path2, self.monitor._general_watchs)
141
143
 
142
144
        self.deferred.addCallback(check)
143
145
        return self.deferred
208
210
            return to_check
209
211
 
210
212
 
211
 
class BaseTwisted(BaseFSMonitorTestCase):
212
 
    """Base class for twisted tests."""
213
 
 
214
 
    # this timeout must be bigger than the one used in event_queue
215
 
    timeout = 2
216
 
 
217
 
    def setUp(self):
218
 
        """Setup the test."""
219
 
        BaseFSMonitorTestCase.setUp(self)
220
 
 
221
 
        # create the deferred for the tests
222
 
        self._deferred = defer.Deferred()
223
 
 
224
 
    def finished_ok(self):
225
 
        """Called to indicate that the tests finished ok."""
226
 
        self._deferred.callback(True)
227
 
 
228
 
    def finished_error(self, msg):
229
 
        """Called to indicate that the tests finished badly."""
230
 
        self._deferred.errback(Exception(msg))
231
 
 
232
 
    def failUnlessEqual(self, first, second, msg=''):
233
 
        """Fail the test if C{first} and C{second} are not equal.
234
 
 
235
 
        @param msg: A string describing the failure that's included in the
236
 
            exception.
237
 
 
238
 
        """
239
 
        if not first == second:
240
 
            if msg is None:
241
 
                msg = ''
242
 
            if len(msg) > 0:
243
 
                msg += '\n'
244
 
            exception = self.failureException(
245
 
                '%snot equal:\na = %s\nb = %s\n'
246
 
                % (msg, repr(first), repr(second)))
247
 
            self.finished_error(exception)
248
 
            raise exception
249
 
        return first
250
 
    assertEqual = assertEquals = failUnlessEquals = failUnlessEqual
251
 
 
252
 
 
253
213
class WatchTests(BaseFSMonitorTestCase):
254
214
    """Test the EQ API to add and remove watchs."""
255
215
 
263
223
 
264
224
    def test_add_general_watch(self):
265
225
        """Test that general watchs can be added."""
266
 
        # we should have what we asked for
267
226
        self.monitor.add_watch(self.root_dir)
268
227
 
269
 
        # check only added dir in watchs
270
 
        # pylint: disable-msg=W0212
271
 
        self.assertTrue(self.root_dir in self.monitor._general_watchs)
272
 
        self.assertTrue("not-added-dir" not in self.monitor._general_watchs)
 
228
        # check only added dir in watchs, and logs
 
229
        self.assertIn(self.root_dir, self.monitor._general_watchs)
 
230
        self.assertNotIn("not-added-dir", self.monitor._general_watchs)
 
231
        self.assertTrue(self.log_handler.check_debug(
 
232
                                "Adding general inotify watch", self.root_dir))
273
233
 
274
234
        # nothing in the udf ancestors watch
275
235
        self.assertEqual(self.monitor._ancestors_watchs, {})
276
236
 
 
237
    def test_add_general_watch_twice(self):
 
238
        """Test that general watchs can be added."""
 
239
        self.monitor.add_watch(self.root_dir)
 
240
        self.assertTrue(self.log_handler.check_debug(
 
241
                                "Adding general inotify watch", self.root_dir))
 
242
        self.assertIn(self.root_dir, self.monitor._general_watchs)
 
243
 
 
244
        # go again
 
245
        self.monitor.add_watch(self.root_dir)
 
246
        self.assertTrue(self.log_handler.check_debug("Watch already there for",
 
247
                                                     self.root_dir))
 
248
        self.assertIn(self.root_dir, self.monitor._general_watchs)
 
249
 
277
250
    @defer.inlineCallbacks
278
251
    def test_add_watch_on_udf_ancestor(self):
279
252
        """Test that ancestors watchs can be added."""
284
257
        self.monitor.add_watch(path_ancestor)
285
258
 
286
259
        # check only added dir in watchs
287
 
        # pylint: disable-msg=W0212
288
260
        self.assertTrue(path_ancestor in self.monitor._ancestors_watchs)
289
261
        self.assertTrue("not-added-dir" not in self.monitor._ancestors_watchs)
290
262
 
299
271
        yield self._create_udf(path_udf)
300
272
        self.monitor.add_watch(path_udf)
301
273
 
302
 
        # pylint: disable-msg=W0212
303
274
        self.assertTrue(path_udf in self.monitor._general_watchs)
304
275
        self.assertEqual(self.monitor._ancestors_watchs, {})
305
276
 
313
284
        os.mkdir(path_ancestor)
314
285
        self.monitor.add_watch(path_ancestor)
315
286
 
316
 
        # pylint: disable-msg=W0212
317
287
        self.assertTrue(path_ancestor in self.monitor._general_watchs)
318
288
        self.assertEqual(self.monitor._ancestors_watchs, {})
319
289
 
326
296
        """
327
297
        not_existing_dir = "not-added-dir"
328
298
        self.monitor.add_watch(not_existing_dir)
329
 
        self.assertTrue(self.monitor.has_watch(not_existing_dir))
 
299
        self.assertIn(not_existing_dir, self.monitor._general_watchs)
330
300
        self.monitor.rm_watch(not_existing_dir)
331
 
        self.assertFalse(self.monitor.has_watch(not_existing_dir))
 
301
        self.assertNotIn(not_existing_dir, self.monitor._general_watchs)
332
302
 
333
303
    @defer.inlineCallbacks
334
304
    def test_rm_watch_wrong(self):
351
321
        self.monitor.add_watch(self.root_dir)
352
322
        self.monitor.rm_watch(self.root_dir)
353
323
 
354
 
        # pylint: disable-msg=W0212
355
324
        self.assertEqual(self.monitor._general_watchs, {})
356
325
        self.assertEqual(self.monitor._ancestors_watchs, {})
357
326
 
366
335
 
367
336
        # remove what we added
368
337
        self.monitor.rm_watch(path_ancestor)
369
 
        # pylint: disable-msg=W0212
370
338
        self.assertEqual(self.monitor._general_watchs, {})
371
339
        self.assertEqual(self.monitor._ancestors_watchs, {})
372
 
 
373
 
    def test_has_watch_general(self):
374
 
        """Test that a general path is watched."""
375
 
        self.assertFalse(self.monitor.has_watch(self.root_dir))
376
 
 
377
 
        # add
378
 
        self.monitor.add_watch(self.root_dir)
379
 
        self.assertTrue(self.monitor.has_watch(self.root_dir))
380
 
 
381
 
        # remove
382
 
        self.monitor.rm_watch(self.root_dir)
383
 
        self.assertFalse(self.monitor.has_watch(self.root_dir))
384
 
 
385
 
    @defer.inlineCallbacks
386
 
    def test_has_watch_ancestor(self):
387
 
        """Test that an ancestor path is watched."""
388
 
        path_udf = os.path.join(self.home_dir, "path/to/UDF")
389
 
        yield self._create_udf(path_udf)
390
 
        path_ancestor = os.path.join(self.home_dir, "path")
391
 
 
392
 
        self.assertFalse(self.monitor.has_watch(path_ancestor))
393
 
 
394
 
        # add
395
 
        # create the udf and add the watch
396
 
        self.monitor.add_watch(path_ancestor)
397
 
        self.assertTrue(self.monitor.has_watch(path_ancestor))
398
 
 
399
 
        # remove
400
 
        self.monitor.rm_watch(path_ancestor)
401
 
        self.assertFalse(self.monitor.has_watch(path_ancestor))
402
 
 
403
 
 
404
 
def test_suite():
405
 
    """Collect the tests."""
406
 
    import sys
407
 
    if sys.platform == 'linux2':
408
 
        tests = unittest.TestLoader().loadTestsFromName(__name__)
409
 
    else:
410
 
        tests = []
411
 
    return tests
412
 
 
413
 
if __name__ == "__main__":
414
 
    unittest.main()