← Back to branch summary

~rodrigo-moya/ubuntuone-client/contact-default-icon

« back to all changes in this revision

Viewing changes to tests/syncdaemon/test_eq_inotify.py

  • Committer: Tarmac
  • Author(s): facundo at com
  • Date: 2010-06-01 16:42:43 UTC
  • mfrom: (526.1.1 fix-watch-when-dir-moved)
  • Revision ID: dobey@wayofthemonkey.com-20100601164243-r93sfevwcmpikwqt
Fix internal watch structures when directory is moved.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/syncdaemon/test_eq_inotify.py
187
187
        self.assertFalse(self.eq.inotify_has_watch(path_ancestor))
188
188
 
189
189
 
 
190
class WatchManagerTests(BaseEQTestCase):
 
191
    """Test the structures where we have the path/watch."""
 
192
 
 
193
    timeout = 3
 
194
 
 
195
    def setUp(self):
 
196
        """Set up."""
 
197
        BaseEQTestCase.setUp(self)
 
198
        self.deferred = deferred = defer.Deferred()
 
199
 
 
200
        class HitMe(object):
 
201
            # class-closure, cannot use self, pylint: disable-msg=E0213
 
202
            def handle_default(*a):
 
203
                    deferred.callback(True)
 
204
 
 
205
        self.eq.subscribe(HitMe())
 
206
 
 
207
    def test_watch_updated_when_deleting_dir(self):
 
208
        """Internal data structures are fixed when deleting the dir."""
 
209
        path = os.path.join(self.root_dir, "path")
 
210
        os.mkdir(path)
 
211
        self.eq.inotify_add_watch(self.root_dir)
 
212
        self.eq.inotify_add_watch(path)
 
213
 
 
214
        # we have the watch, remove the dir, the watch should be gone
 
215
        self.assertTrue(self.eq.inotify_has_watch(path))
 
216
        os.rmdir(path)
 
217
 
 
218
        def check(_):
 
219
            """Check state after the event."""
 
220
            self.assertFalse(self.eq.inotify_has_watch(path))
 
221
 
 
222
        self.deferred.addCallback(check)
 
223
        return self.deferred
 
224
 
 
225
    def test_watch_updated_when_renaming_dir(self):
 
226
        """Internal data structures are fixed when renaming the dir."""
 
227
        path1 = os.path.join(self.root_dir, "path1")
 
228
        path2 = os.path.join(self.root_dir, "path2")
 
229
        os.mkdir(path1)
 
230
        self.eq.inotify_add_watch(self.root_dir)
 
231
        self.eq.inotify_add_watch(path1)
 
232
 
 
233
        # we have the watch, rename the dir, new name should have the watch,
 
234
        # old name should not
 
235
        self.assertTrue(self.eq.inotify_has_watch(path1))
 
236
        os.rename(path1, path2)
 
237
 
 
238
        def check(_):
 
239
            """Check state after the event."""
 
240
            self.assertTrue(self.eq.inotify_has_watch(path2))
 
241
            self.assertFalse(self.eq.inotify_has_watch(path1))
 
242
 
 
243
        self.deferred.addCallback(check)
 
244
        return self.deferred
 
245
 
 
246
    def test_watch_updated_when_movingout_dir(self):
 
247
        """Internal data structures are fixed when moving out the dir."""
 
248
        notu1 = os.path.join(self.root_dir, "notu1")
 
249
        path1 = os.path.join(self.root_dir, "path1")
 
250
        path2 = os.path.join(notu1, "path2")
 
251
        os.mkdir(notu1)
 
252
        os.mkdir(path1)
 
253
        self.eq.inotify_add_watch(self.root_dir)
 
254
        self.eq.inotify_add_watch(path1)
 
255
 
 
256
        # we have the watch, move it outside watched structure, no more watches
 
257
        self.assertTrue(self.eq.inotify_has_watch(path1))
 
258
        os.rename(path1, path2)
 
259
 
 
260
        def check(_):
 
261
            """Check state after the event."""
 
262
            self.assertFalse(self.eq.inotify_has_watch(path1))
 
263
            self.assertFalse(self.eq.inotify_has_watch(path2))
 
264
 
 
265
        self.deferred.addCallback(check)
 
266
        return self.deferred
 
267
 
 
268
    def test_fix_path_not_there(self):
 
269
        """Try to fix path but it's not there."""
 
270
        self.eq._general_watchs = {}
 
271
        self.eq._ancestors_watchs = {}
 
272
        self.eq.inotify_watch_fix("not-there", "new-one")
 
273
        self.assertTrue(self.log_handler.check_warning("Tried to fix",
 
274
                                                       "not-there"))
 
275
 
 
276
    def test_fix_path_general(self):
 
277
        """Fixing path in general watches."""
 
278
        self.eq._general_watchs = {'/path1/foo': 1, '/other': 2}
 
279
        self.eq._ancestors_watchs = {'/foo': 3}
 
280
        self.eq.inotify_watch_fix('/path1/foo', '/path1/new')
 
281
        self.assertEqual(self.eq._general_watchs, {'/path1/new': 1, '/other': 2})
 
282
        self.assertEqual(self.eq._ancestors_watchs, {'/foo': 3})
 
283
 
 
284
    def test_fix_path_ancestors(self):
 
285
        """Fixing path in ancestors watches."""
 
286
        self.eq._general_watchs = {'/bar': 3}
 
287
        self.eq._ancestors_watchs = {'/oth': 1, '/other': 2}
 
288
        self.eq.inotify_watch_fix('/oth', '/baz')
 
289
        self.assertEqual(self.eq._general_watchs, {'/bar': 3})
 
290
        self.assertEqual(self.eq._ancestors_watchs, {'/baz': 1, '/other': 2})
 
291
 
 
292
 
190
293
class DynamicHitMe(object):
191
294
    """Helper class to test a sequence of signals."""
192
295