Merge lp:~diegosarmentero/ubuntuone-client/darwin4-fsevents into lp:ubuntuone-client

[Diego Sarmentero]

- Adding filesystem notifications feature for darwin (LP: #1013323)

Now you can execute:
./run-mac-tests tests/platform/filesystem_notifications/

[Alejandro J. Cura (alecu)]

The branch looks mostly fine.

A few questions, though:

 * Why is test_file_moved_from_partial skipped? I think this is a critical operation that should be tested.

 * Why is there a test that still uses time.sleep? As mentioned in previous branches we should not be using sleep at all in tests.

And some fixes needed, too:

It looks wrong to import fsevents in /platform/filesystem_notifications/__init__.py, so please, move each platform specific definition of EVENT_CODES to ubuntuone.platform.filesystem_notifications.windows and .darwin

---

Lines 827 and 828 are duplicated.

---

Please move the logic in these few lines to a function, add some unit tests for it, and use it from both places:

837 + index = 0
838 + if not self._path.endswith(os.path.sep):
839 + index = 1
840 + path = path[len(self._path) + index:]

review: Needs Information

[Manuel de la Peña (mandel)]

The diff is misleading because it should depend on lp:~mandel/ubuntuone-client/clean-fsevents. Diego can you do a resummit with the correct dependency so that we are not scared regading the size :)

[Diego Sarmentero (diegosarmentero)]

Answering alecu:
* Why is test_file_moved_from_partial skipped? I think this is a critical operation that should be tested.
This is marked for fix in

There is a bug for this, in which i'm working on:
https://bugs.launchpad.net/ubuntuone-client/+bug/1012706

The rest of the things has been fixed.

[Manuel de la Peña (mandel)]

Looks like there is a thread somewhere that is not killed in the tests. I'm running the tests via u1trial and once they are done the process does not stop.

review: Needs Fixing

[lp:~diegosarmentero/ubuntuone-client/darwin4-fsevents]

1296. By Diego Sarmentero on 2012-07-11

removing time.sleep

[Diego Sarmentero (diegosarmentero)]

> Looks like there is a thread somewhere that is not killed in the tests. I'm
> running the tests via u1trial and once they are done the process does not
> stop.

Fixed!

[Alejandro J. Cura (alecu)]

The tearDown in TestWatchManager should not be calling self.patch.
In the same tearDown, the deferred returned by self.manager.stop() should be yielded on.

---
typos:

"""Test that we can get a Watch using is wd.""" ->
"""Test that we can get a Watch using its wd."""

---

This line should not be part of the tests; it should be part of the stopping of the watcher thread:

self.manager.manager.observer.join()

That part of the code (stopping the watcher thread) is particularly dirty: I just found that there's a __del__ in WatchManager that we missed on the review of the previous branch, and that should not be there, since the use of __del__ is discouraged in python, and much more so when it's being used to work around a thread not being correctly finished.

I think that the stop method should be returning a deferred that's fired when the join is done; but please, let's discuss together a better solution to this on IRC or Mumble.
---

[Alejandro J. Cura (alecu)]

review: Needs Fixing

[lp:~diegosarmentero/ubuntuone-client/darwin4-fsevents]

1297. By Diego Sarmentero on 2012-07-12

adding join on stop

1298. By Diego Sarmentero on 2012-07-12

moving patch to setUp

1299. By Diego Sarmentero on 2012-07-12

reverting

[Manuel de la Peña (mandel)]

Looks goo to me knowing that we will merge this tests later.

review: Approve

[Alejandro J. Cura (alecu)]

I still would like better handling of the watches being stopped, so for that I'm creating a different bug: #1024102

Otherwise the code here looks ok.

review: Approve

=== modified file 'tests/platform/filesystem_notifications/test_darwin.py'
--- tests/platform/filesystem_notifications/test_darwin.py	2012-07-10 18:43:56 +0000
+++ tests/platform/filesystem_notifications/test_darwin.py	2012-07-12 13:53:19 +0000
@@ -32,7 +32,9 @@
 import os
 import tempfile
 import thread
+import itertools
 
+import fsevents
 from twisted.internet import defer
 
 from contrib.testing.testcase import BaseTwistedTestCase
@@ -41,15 +43,20 @@
     darwin as filesystem_notifications,
 )
 from ubuntuone.platform.filesystem_notifications.common import (
+    NotifyProcessor,
     Watch,
     WatchManager,
 )
 from ubuntuone.platform.filesystem_notifications.pyinotify_agnostic import (
     EventsCodes,
     ProcessEvent,
+    IN_CLOSE_WRITE,
     IN_CREATE,
     IN_DELETE,
+    IN_OPEN,
 )
+from tests.platform.filesystem_notifications import BaseFSMonitorTestCase
+
 
 # A reverse mapping for the tests
 REVERSE_MACOS_ACTIONS = {}
@@ -135,6 +142,7 @@
         yield super(TestWatch, self).setUp()
         self.basedir = self.mktemp('test_root')
         self.mask = None
+        self.stream = None
         self.memento = MementoHandler()
         self.memento.setLevel(logging.DEBUG)
         self.raw_events = []
@@ -747,3 +755,632 @@
         watch._subdirs.add(path)
         watch._update_subdirs(path, REVERSE_MACOS_ACTIONS[IN_DELETE])
         self.assertTrue(path not in watch._subdirs)
+
+
+class TestWatchManager(BaseTwistedTestCase):
+    """Test the watch manager."""
+
+    @defer.inlineCallbacks
+    def setUp(self):
+        """Set each of the tests."""
+        yield super(TestWatchManager, self).setUp()
+        self.parent_path = '/Users/username/'
+        self.path = self.parent_path + 'path'
+        self.watch = Watch(1, self.path, None)
+        self.manager = WatchManager(None)
+        self.manager._wdm = {1: self.watch}
+
+    @defer.inlineCallbacks
+    def test_stop(self):
+        """Test that the different watches are stopped."""
+        self.was_called = False
+
+        def fake_stop_watching(watch):
+            """Fake stop watch."""
+            self.was_called = True
+            return defer.succeed(True)
+
+        self.patch(Watch, "stop_watching", fake_stop_watching)
+        self.patch(self.manager.manager.observer, "unschedule", lambda x: None)
+        yield self.manager.stop()
+        self.assertTrue(self.was_called, 'The watch stop should be called.')
+
+    def test_stop_multiple(self):
+        """The watches should became watching=False and the observer stopped."""
+        self.patch(self.manager.manager.observer, "unschedule", lambda x: None)
+        second_path = self.parent_path + "second_path"
+        second_watch = Watch(2, second_path, None)
+        second_watch._watching = True
+        self.manager._wdm[2] = second_watch
+        self.manager.stop()
+        self.assertFalse(second_watch.platform_watch.watching)
+        self.assertEqual(second_watch._subdirs, set())
+        # Give time to the thread to be finished.
+        self.manager.manager.observer.join()
+        self.assertFalse(self.manager.manager.observer.is_alive())
+
+    def test_get_present_watch(self):
+        """Test that we can get a Watch using is wd."""
+        self.assertEqual(self.watch, self.manager.get_watch(1))
+
+    def test_get_missing_watch(self):
+        """Test that we get an error when trying to get a missing wd."""
+        self.assertRaises(KeyError, self.manager.get_watch, (1,))
+
+    @defer.inlineCallbacks
+    def test_delete_present_watch(self):
+        """Test that we can remove a present watch."""
+        self.was_called = False
+
+        def stop_watching():
+            """Fake stop watch."""
+            self.was_called = True
+            return defer.succeed(True)
+
+        def fake_unschedule(s):
+            """Fake function that should receive a Stream object."""
+            self.stream = s
+
+        self.patch(self.manager.manager.observer, "unschedule",
+            fake_unschedule)
+
+        self.watch.stop_watching = stop_watching
+        yield self.manager.del_watch(1)
+        self.assertIsInstance(self.stream, fsevents.Stream)
+        self.assertRaises(KeyError, self.manager.get_watch, (1,))
+
+    def test_add_single_watch(self):
+        """Test the addition of a new single watch."""
+        self.was_called = False
+
+        def fake_start_watching(*args):
+            """Fake start watch."""
+            self.was_called = True
+
+        self.patch(Watch, "start_watching", fake_start_watching)
+        self.manager._wdm = {}
+
+        mask = 'bit_mask'
+        self.manager.add_watch(self.path, mask)
+        self.assertEqual(1, len(self.manager._wdm))
+        self.assertTrue(self.was_called, 'The watch start was not called.')
+        self.assertEqual(self.path + os.path.sep, self.manager._wdm[0].path)
+
+    def test_get_watch_present_wd(self):
+        """Test that the correct path is returned."""
+        self.assertEqual(self.path + os.path.sep, self.manager.get_path(1))
+
+    def test_get_watch_missing_wd(self):
+        """Test that the correct path is returned."""
+        self.manager._wdm = {}
+        self.assertEqual(None, self.manager.get_path(1))
+
+    def test_get_wd_exact_path(self):
+        """Test the wd is returned when there is a watch for the path."""
+        self.assertEqual(1, self.manager.get_wd(self.path))
+
+    def test_get_wd_child_path(self):
+        """Test the wd is returned when we have a child path."""
+        child = os.path.join(self.path, 'child')
+        self.assertEqual(1, self.manager.get_wd(child))
+
+    def test_get_wd_unwatched(self):
+        """A watch on an unwatched path returns None."""
+        self.assertEqual(None, self.manager.get_wd(self.parent_path))
+
+    def test_rm_present_wd(self):
+        """Test the removal of a present watch."""
+        self.patch(self.watch, "stop_watching", lambda: None)
+        self.patch(self.manager.manager.observer, "unschedule", lambda x: None)
+        self.manager.rm_watch(1)
+        self.assertEqual(None, self.manager._wdm.get(1))
+
+    def test_rm_root_path(self):
+        """Test the removal of a root path."""
+        events = []
+
+        def fake_processor(event):
+            """Memorize the processed events."""
+            events.append(event.pathname)
+
+        self.watch._processor = fake_processor
+        self.manager.rm_path(self.path)
+        self.assertEqual(self.watch, self.manager._wdm.get(1))
+        self.watch._watching = True
+        event = FakeFileEvent(256, None, os.path.join(self.path, 'test'))
+        self.watch.platform_watch._process_events(event)
+        self.assertEqual(0, len(events))
+
+    def test_rm_child_path(self):
+        """Test the removal of a child path."""
+        events = []
+
+        def fake_processor(event):
+            """Memorize the processed events."""
+            events.append(event.pathname)
+
+        self.patch(filesystem_notifications.reactor, 'callFromThread',
+            lambda x, e: x(e))
+
+        self.watch._processor = fake_processor
+        child = os.path.join(self.path, 'child')
+        self.manager.rm_path(child)
+        self.assertEqual(self.watch, self.manager._wdm[1])
+        # assert that the correct event is ignored
+        self.watch.platform_watch.watching = True
+        event = FakeFileEvent(256, None, os.path.join('child', 'test'))
+        self.watch.platform_watch._process_events(event)
+        self.assertEqual(0, len(events))
+        # assert that other events are not ignored
+        event2 = FakeFileEvent(256, None, 'test')
+        self.watch.platform_watch._process_events(event2)
+        self.assertEqual(1, len(events))
+
+
+class TestWatchManagerAddWatches(BaseTwistedTestCase):
+    """Test the watch manager."""
+    timeout = 5
+
+    def test_add_watch_twice(self):
+        """Adding a watch twice succeeds when the watch is running."""
+        self.patch(Watch, "start_watching", lambda self: None)
+        manager = WatchManager(None)
+        # no need to stop watching because start_watching is fake
+
+        path = '/Users/username/path'
+        mask = 'fake bit mask'
+        d1 = manager.add_watch(path, mask)
+        d2 = manager.add_watch(path, mask)
+
+        self.assertTrue(d1.result, "Should not be called yet.")
+        self.assertFalse(d2.result, "Should not be called yet.")
+
+
+class FakeEvent(object):
+    """Fake event."""
+
+    def __init__(self, wd=0, dir=True, name=None, path=None, pathname=None,
+                 cookie=None):
+        """Create fake event."""
+        self.dir = dir
+        self.wd = wd
+        self.name = name
+        self.path = path
+        self.pathname = pathname
+        self.cookie = cookie
+
+
+class FakeLog(object):
+    """A fake log that is used by the general processor."""
+
+    def __init__(self):
+        """Create the fake."""
+        self.called_methods = []
+
+    def info(self, *args):
+        """Fake the info call."""
+        self.called_methods.append(('info', args))
+
+
+class FakeGeneralProcessor(object):
+    """Fake implementation of the general processor."""
+
+    def __init__(self):
+        """Create the fake."""
+        self.called_methods = []
+        self.paths_to_return = []
+        self.log = FakeLog()
+        self.share_id = None
+        self.ignore = False
+
+    def rm_from_mute_filter(self, event, paths):
+        """Fake rm_from_mute_filter."""
+        self.called_methods.append(('rm_from_mute_filter', event, paths))
+
+    def add_to_mute_filter(self, event, paths):
+        """Fake add_to_move_filter."""
+        self.called_methods.append(('add_to_mute_filter', event, paths))
+
+    def is_ignored(self, path):
+        """Fake is_ignored."""
+        self.called_methods.append(('is_ignored', path))
+        return self.ignore
+
+    def push_event(self, event):
+        """Fake push event."""
+        self.called_methods.append(('push_event', event))
+
+    def eq_push(self, event, path=None, path_to=None, path_from=None):
+        """Fake event to push event."""
+        self.called_methods.append(('eq_push', event, path, path_to,
+                                    path_from))
+
+    def get_paths_starting_with(self, fullpath, include_base=False):
+        """Fake get_paths_starting_with."""
+        self.called_methods.append(('get_paths_starting_with', fullpath,
+                                    include_base))
+        return self.paths_to_return
+
+    def get_path_share_id(self, path):
+        """Fake get_path_share_id."""
+        self.called_methods.append(('get_path_share_id', path))
+        return self.share_id
+
+    def rm_watch(self, path):
+        """Fake the remove watch."""
+        self.called_methods.append(('rm_watch', path))
+
+    def freeze_begin(self, path):
+        """Fake freeze_begin"""
+        self.called_methods.append(('freeze_begin', path))
+
+    def freeze_rollback(self):
+        """Fake rollback."""
+        self.called_methods.append(('freeze_rollback',))
+
+    def freeze_commit(self, path):
+        """Fake freeze commit."""
+        self.called_methods.append(('freeze_commit', path))
+
+
+class TestNotifyProcessor(BaseTwistedTestCase):
+    """Test the notify processor."""
+
+    @defer.inlineCallbacks
+    def setUp(self):
+        """set up the diffeent tests."""
+        yield super(TestNotifyProcessor, self).setUp()
+        self.processor = NotifyProcessor(None)
+        self.general = FakeGeneralProcessor()
+        self.processor.general_processor = self.general
+
+    def test_rm_from_mute_filter(self):
+        """Test that we remove the event from the mute filter."""
+        event = 'event'
+        paths = 'paths'
+        self.processor.rm_from_mute_filter(event, paths)
+        self.assertEqual(1, len(self.general.called_methods))
+        self.assertEqual('rm_from_mute_filter',
+                         self.general.called_methods[0][0])
+        self.assertEqual(event, self.general.called_methods[0][1])
+        self.assertEqual(paths, self.general.called_methods[0][2])
+
+    def test_add_to_mute_filter(self):
+        """Test that we add the path to the mute filter."""
+        event = 'event'
+        paths = 'paths'
+        self.processor.add_to_mute_filter(event, paths)
+        self.assertEqual(1, len(self.general.called_methods))
+        self.assertEqual('add_to_mute_filter',
+                         self.general.called_methods[0][0])
+        self.assertEqual(event, self.general.called_methods[0][1])
+        self.assertEqual(paths, self.general.called_methods[0][2])
+
+    def test_is_ignored(self):
+        """Test that we do ensure that the path is ignored."""
+        path = 'path'
+        self.processor.is_ignored(path)
+        self.assertEqual(1, len(self.general.called_methods))
+        self.assertEqual('is_ignored',
+                         self.general.called_methods[0][0])
+        self.assertEqual(path, self.general.called_methods[0][1])
+
+    def test_release_held_event(self):
+        """Test that we do release the held event."""
+        event = 'event'
+        # set the held event to assert that is pushed
+        self.processor.held_event = event
+        self.processor.release_held_event()
+        self.assertEqual('push_event',
+                         self.general.called_methods[0][0])
+        self.assertEqual(event, self.general.called_methods[0][1])
+
+    def test_process_IN_MODIFY_dir(self):
+        """Test that the modify works as exepcted with dirs."""
+        event = FakeEvent(dir=True)
+        self.processor.process_IN_MODIFY(event)
+        # no method should be called
+        self.assertEqual(0, len(self.general.called_methods))
+
+    def test_process_IN_MODIFY_file(self):
+        """Test that the modify works as expected with files."""
+        event = FakeEvent(dir=False, wd=0, name='name',
+                          path='path', pathname='pathname')
+        self.processor.process_IN_MODIFY(event)
+        # we should be getting two different method, and open and a close
+        self.assertEqual(2, len(self.general.called_methods))
+        self.assertEqual('push_event',
+                         self.general.called_methods[0][0])
+        self.assertEqual('push_event',
+                         self.general.called_methods[1][0])
+        self.assertEqual(event.dir, self.general.called_methods[0][1].dir)
+        self.assertEqual(event.wd, self.general.called_methods[0][1].wd)
+        self.assertEqual(event.name, self.general.called_methods[0][1].name)
+        self.assertEqual(event.path, self.general.called_methods[0][1].path)
+        self.assertEqual(event.pathname,
+                         self.general.called_methods[0][1].pathname)
+        self.assertEqual(IN_OPEN,
+                         self.general.called_methods[0][1].mask)
+        self.assertEqual(event.dir, self.general.called_methods[1][1].dir)
+        self.assertEqual(event.wd, self.general.called_methods[1][1].wd)
+        self.assertEqual(event.name, self.general.called_methods[1][1].name)
+        self.assertEqual(event.path, self.general.called_methods[1][1].path)
+        self.assertEqual(event.pathname,
+                         self.general.called_methods[1][1].pathname)
+        self.assertEqual(IN_CLOSE_WRITE,
+                         self.general.called_methods[1][1].mask)
+
+    def test_process_IN_MOVED_FROM(self):
+        """Test that the in moved from works as expected."""
+        event = FakeEvent(dir=False, wd=0, name='name',
+                          path='path', pathname='pathname')
+        self.processor.process_IN_MOVED_FROM(event)
+        self.assertEqual(event, self.processor.held_event)
+
+    def test_process_IN_MOVED_TO_dir(self):
+        """Test that the in moved to works as expected."""
+        event = FakeEvent(wd=0, dir=True, name='name', path='path',
+                          pathname=os.path.join('test', 'pathname'),
+                          cookie='cookie')
+        held_event = FakeEvent(wd=0, dir=True, name='hname', path='hpath',
+                               pathname=os.path.join('test', 'hpathname'),
+                               cookie='cookie')
+        self.general.share_id = 'my_share_id'
+        self.processor.held_event = held_event
+        self.processor.process_IN_MOVED_TO(event)
+        self.assertEqual(5, len(self.general.called_methods))
+        # assert that the ignores are called
+        self.assertEqual('is_ignored', self.general.called_methods[0][0])
+        self.assertEqual(held_event.pathname,
+                         self.general.called_methods[0][1])
+        self.assertEqual('is_ignored', self.general.called_methods[1][0])
+        self.assertEqual(event.pathname, self.general.called_methods[1][1])
+        # assert that we do request the share_id
+        self.assertEqual('get_path_share_id',
+                         self.general.called_methods[2][0])
+        self.assertEqual(os.path.split(event.pathname)[0],
+                         self.general.called_methods[2][1],
+                         'Get the share_id for event')
+        self.assertEqual('get_path_share_id',
+                         self.general.called_methods[3][0])
+        self.assertEqual(os.path.split(held_event.pathname)[0],
+                         self.general.called_methods[3][1],
+                         'Get the share_id for held event.')
+
+        self.assertEqual('eq_push', self.general.called_methods[4][0])
+        self.assertEqual('FS_DIR_MOVE', self.general.called_methods[4][1])
+        self.assertEqual(event.pathname, self.general.called_methods[4][3])
+        self.assertEqual(held_event.pathname,
+                         self.general.called_methods[4][4])
+
+    def test_process_IN_MOVED_TO_file(self):
+        """Test that the in moved to works as expected."""
+        event = FakeEvent(wd=0, dir=False, name='name', path='path',
+                          pathname=os.path.join('test', 'pathname'),
+                          cookie='cookie')
+        held_event = FakeEvent(wd=0, dir=False, name='hname', path='hpath',
+                               pathname=os.path.join('test', 'hpathname'),
+                               cookie='cookie')
+        self.general.share_id = 'my_share_id'
+        self.processor.held_event = held_event
+        self.processor.process_IN_MOVED_TO(event)
+        self.assertEqual(5, len(self.general.called_methods))
+        # assert that the ignores are called
+        self.assertEqual('is_ignored', self.general.called_methods[0][0])
+        self.assertEqual(held_event.pathname,
+                         self.general.called_methods[0][1])
+        self.assertEqual('is_ignored', self.general.called_methods[1][0])
+        self.assertEqual(event.pathname, self.general.called_methods[1][1])
+        # assert that we do request the share_id
+        self.assertEqual('get_path_share_id',
+                         self.general.called_methods[2][0])
+        self.assertEqual(os.path.split(event.pathname)[0],
+                         self.general.called_methods[2][1],
+                         'Get the share_id for event')
+        self.assertEqual('get_path_share_id',
+                         self.general.called_methods[3][0])
+        self.assertEqual(os.path.split(held_event.pathname)[0],
+                         self.general.called_methods[3][1],
+                         'Get the share_id for held event.')
+
+        self.assertEqual('eq_push', self.general.called_methods[4][0])
+        self.assertEqual('FS_FILE_MOVE', self.general.called_methods[4][1])
+        self.assertEqual(event.pathname, self.general.called_methods[4][3])
+        self.assertEqual(held_event.pathname,
+                         self.general.called_methods[4][4])
+
+    def test_fake_create_event_dir(self):
+        """Test that the in moved to works as expected."""
+        event = FakeEvent(wd=0, dir=True, name='name', path='path',
+                          pathname='pathname')
+        self.processor._fake_create_event(event)
+        self.assertEqual(1, len(self.general.called_methods))
+        self.assertEqual('eq_push', self.general.called_methods[0][0])
+        self.assertEqual('FS_DIR_CREATE', self.general.called_methods[0][1])
+        self.assertEqual(event.pathname, self.general.called_methods[0][2])
+
+    def test_fake_create_event_file(self):
+        """Test that the in moved to works as expected."""
+        event = FakeEvent(wd=0, dir=False, name='name', path='path',
+                          pathname='pathname')
+        self.processor._fake_create_event(event)
+        self.assertEqual(2, len(self.general.called_methods))
+        self.assertEqual('eq_push', self.general.called_methods[0][0])
+        self.assertEqual('FS_FILE_CREATE', self.general.called_methods[0][1])
+        self.assertEqual(event.pathname, self.general.called_methods[0][2])
+        self.assertEqual('eq_push', self.general.called_methods[1][0])
+        self.assertEqual('FS_FILE_CLOSE_WRITE',
+                         self.general.called_methods[1][1])
+        self.assertEqual(event.pathname, self.general.called_methods[1][2])
+
+    def test_fake_delete_create_event_dir(self):
+        """Test that we do fake a delete and a later delete."""
+        event = FakeEvent(wd=0, dir=True, name='name', path='path',
+                          pathname='pathname')
+        held_event = FakeEvent(wd=0, dir=True, name='hname', path='hpath',
+                               pathname='hpathname')
+        self.processor.held_event = held_event
+        self.processor._fake_delete_create_event(event)
+        self.assertEqual(2, len(self.general.called_methods))
+        self.assertEqual('eq_push', self.general.called_methods[0][0])
+        self.assertEqual('FS_DIR_DELETE', self.general.called_methods[0][1])
+        self.assertEqual(held_event.pathname,
+                         self.general.called_methods[0][2])
+        self.assertEqual('eq_push', self.general.called_methods[1][0])
+        self.assertEqual('FS_DIR_CREATE', self.general.called_methods[1][1])
+        self.assertEqual(event.pathname, self.general.called_methods[1][2])
+
+    def test_fake_delete_create_event_file(self):
+        """Test that we do fake a delete and a later delete."""
+        event = FakeEvent(wd=0, dir=False, name='name', path='path',
+                          pathname='pathname')
+        held_event = FakeEvent(wd=0, dir=False, name='hname', path='hpath',
+                               pathname='hpathname')
+        self.processor.held_event = held_event
+        self.processor._fake_delete_create_event(event)
+        self.assertEqual(3, len(self.general.called_methods))
+        self.assertEqual('eq_push', self.general.called_methods[0][0])
+        self.assertEqual('FS_FILE_DELETE', self.general.called_methods[0][1])
+        self.assertEqual(held_event.pathname,
+                         self.general.called_methods[0][2])
+        self.assertEqual('eq_push', self.general.called_methods[1][0])
+        self.assertEqual('FS_FILE_CREATE', self.general.called_methods[1][1])
+        self.assertEqual(event.pathname, self.general.called_methods[1][2])
+        self.assertEqual('eq_push', self.general.called_methods[2][0])
+        self.assertEqual('FS_FILE_CLOSE_WRITE',
+                         self.general.called_methods[2][1])
+        self.assertEqual(event.pathname, self.general.called_methods[2][2])
+
+    def test_process_default_no_held(self):
+        """Test that the process default works as expected."""
+        event = 'event'
+        self.processor.process_default(event)
+        self.assertEqual(1, len(self.general.called_methods))
+        self.assertEqual('push_event',
+                         self.general.called_methods[0][0])
+        self.assertEqual(event,
+                         self.general.called_methods[0][1])
+
+    def test_process_default_with_held(self):
+        """Test that the process default works as expected."""
+        event = 'event'
+        held_event = 'held_event'
+        self.processor.held_event = held_event
+        self.processor.process_default(event)
+        self.assertEqual(2, len(self.general.called_methods))
+        self.assertEqual('push_event',
+                         self.general.called_methods[0][0])
+        self.assertEqual(held_event,
+                         self.general.called_methods[0][1])
+        self.assertEqual('push_event',
+                         self.general.called_methods[1][0])
+        self.assertEqual(event,
+                         self.general.called_methods[1][1])
+
+    def test_handle_dir_delete_files(self):
+        """Test that the handle dir delete works as expected."""
+        path = 'path'
+        present_files = 'abcde'
+        # create files and dirs to be returned from the get paths
+        for file_name in present_files:
+            self.general.paths_to_return.append((file_name, False))
+        self.processor.handle_dir_delete(path)
+        # there are calls for, rm the watch, get paths and then one per file
+        self.assertEqual(len(present_files) + 2,
+                         len(self.general.called_methods))
+        rm_call = self.general.called_methods.pop(0)
+        self.assertEqual('rm_watch', rm_call[0])
+        self.assertEqual(path, rm_call[1])
+        paths_call = self.general.called_methods.pop(0)
+        self.assertEqual('get_paths_starting_with', paths_call[0])
+        self.assertEqual(path, paths_call[1])
+        self.assertFalse(paths_call[2])
+        # we need to push the delete events in reverse order because we want
+        # to delete children before we delete parents
+        present_files = present_files[::-1]
+        for i, called_method in enumerate(self.general.called_methods):
+            self.assertEqual('eq_push', called_method[0])
+            self.assertEqual('FS_FILE_DELETE', called_method[1])
+            self.assertEqual(present_files[i], called_method[2])
+
+    def test_handle_dir_delete_dirs(self):
+        """Test that the handle dir delete works as expected."""
+        path = 'path'
+        present_files = 'abcde'
+        # create files and dirs to be returned from the get paths
+        for file_name in present_files:
+            self.general.paths_to_return.append((file_name, True))
+        self.processor.handle_dir_delete(path)
+        # there are calls for, rm the watch, get paths and then one per file
+        self.assertEqual(2 * len(present_files) + 2,
+                         len(self.general.called_methods))
+        rm_call = self.general.called_methods.pop(0)
+        self.assertEqual('rm_watch', rm_call[0])
+        self.assertEqual(path, rm_call[1])
+        paths_call = self.general.called_methods.pop(0)
+        self.assertEqual('get_paths_starting_with', paths_call[0])
+        self.assertEqual(path, paths_call[1])
+        self.assertFalse(paths_call[2])
+        # we need to push the delete events in reverse order because we want
+        # to delete children before we delete parents
+        present_files = present_files[::-1]
+
+        # from http://docs.python.org/library/itertools.html#recipes
+        def grouper(n, iterable, fillvalue=None):
+            "grouper(3, 'ABCDEFG', 'x') --> ABC DEF Gxx"
+            args = [iter(iterable)] * n
+            return itertools.izip_longest(fillvalue=fillvalue, *args)
+
+        for i, called_methods in enumerate(grouper(2,
+                                           self.general.called_methods)):
+            rm_call = called_methods[0]
+            self.assertEqual('rm_watch', rm_call[0])
+            self.assertEqual(present_files[i], rm_call[1])
+            push_call = called_methods[1]
+            self.assertEqual('eq_push', push_call[0])
+            self.assertEqual('FS_DIR_DELETE', push_call[1])
+            self.assertEqual(present_files[i], push_call[2])
+
+    def test_freeze_begin(self):
+        """Test that the freeze being works as expected."""
+        path = 'path'
+        self.processor.freeze_begin(path)
+        self.assertEqual(1, len(self.general.called_methods))
+        self.assertEqual('freeze_begin',
+                         self.general.called_methods[0][0])
+        self.assertEqual(path, self.general.called_methods[0][1])
+
+    def test_freeze_rollback(self):
+        """Test that the freeze rollback works as expected."""
+        self.processor.freeze_rollback()
+        self.assertEqual(1, len(self.general.called_methods))
+        self.assertEqual('freeze_rollback',
+                         self.general.called_methods[0][0])
+
+    def test_freeze_commit(self):
+        """Test that the freeze commit works as expected."""
+        path = 'path'
+        self.processor.freeze_commit(path)
+        self.assertEqual(1, len(self.general.called_methods))
+        self.assertEqual('freeze_commit',
+                         self.general.called_methods[0][0])
+        self.assertEqual(path, self.general.called_methods[0][1])
+
+
+class FilesystemMonitorTestCase(BaseFSMonitorTestCase):
+    """Tests for the FilesystemMonitor."""
+
+    def test_add_watch_twice(self):
+        """Check the deferred returned by a second add_watch."""
+        self.patch(Watch, "start_watching", lambda self: None)
+        self.patch(Watch, "started", lambda self: True)
+        manager = WatchManager(None)
+        # no need to stop watching because start_watching is fake
+
+        path = '/Users/username/path'
+        mask = 'fake bit mask'
+        d1 = manager.add_watch(path, mask)
+        d2 = manager.add_watch(path, mask)
+
+        self.assertTrue(d1.result, "Should not be called yet.")
+        self.assertTrue(d2, "Should not be called yet.")
 
=== modified file 'tests/platform/filesystem_notifications/test_filesystem_notifications.py'
--- tests/platform/filesystem_notifications/test_filesystem_notifications.py	2012-06-28 10:26:50 +0000
+++ tests/platform/filesystem_notifications/test_filesystem_notifications.py	2012-07-12 13:53:19 +0000
@@ -39,6 +39,7 @@
     BaseTwistedTestCase,
     FakeVolumeManager,
     skip_if_win32_missing_fs_event,
+    skip_if_darwin_missing_fs_event,
 )
 from ubuntuone.platform import (
     remove_file,
@@ -264,6 +265,7 @@
         mute_filter = self.monitor._processor.mute_filter
         return sum(len(x) for x in mute_filter._cnt.values())
 
+    @skip_if_darwin_missing_fs_event
     @skip_if_win32_missing_fs_event
     @defer.inlineCallbacks
     def test_file_open(self):
@@ -281,6 +283,7 @@
         test_result = yield self._deferred
         defer.returnValue(test_result)
 
+    @skip_if_darwin_missing_fs_event
     @skip_if_win32_missing_fs_event
     @defer.inlineCallbacks
     def test_file_close_nowrite(self):
@@ -298,6 +301,7 @@
         test_result = yield self._deferred
         defer.returnValue(test_result)
 
+    @skip_if_darwin_missing_fs_event
     @defer.inlineCallbacks
     def test_file_create_close_write(self):
         """Test receiving the create and close_write signals on files."""
@@ -423,6 +427,7 @@
         test_result = yield self._deferred
         defer.returnValue(test_result)
 
+    @skip_if_darwin_missing_fs_event
     @defer.inlineCallbacks
     def test_file_moved_from_partial(self):
         """Test the handling of the FILE_MOVE event when source is partial."""
 
=== modified file 'tests/platform/filesystem_notifications/test_windows.py'
--- tests/platform/filesystem_notifications/test_windows.py	2012-07-10 20:24:04 +0000
+++ tests/platform/filesystem_notifications/test_windows.py	2012-07-12 13:53:19 +0000
@@ -1321,7 +1321,7 @@
         for file_name in present_files:
             self.general.paths_to_return.append((file_name, False))
         self.processor.handle_dir_delete(path)
-        # there are calls for, rm the wathc, get paths and then one per file
+        # there are calls for, rm the watch, get paths and then one per file
         self.assertEqual(len(present_files) + 2,
                          len(self.general.called_methods))
         rm_call = self.general.called_methods.pop(0)
@@ -1347,7 +1347,7 @@
         for file_name in present_files:
             self.general.paths_to_return.append((file_name, True))
         self.processor.handle_dir_delete(path)
-        # there are calls for, rm the wathc, get paths and then one per file
+        # there are calls for, rm the watch, get paths and then one per file
         self.assertEqual(2 * len(present_files) + 2,
                          len(self.general.called_methods))
         rm_call = self.general.called_methods.pop(0)
 
=== modified file 'ubuntuone/platform/filesystem_notifications/darwin.py'
--- ubuntuone/platform/filesystem_notifications/darwin.py	2012-07-11 09:48:03 +0000
+++ ubuntuone/platform/filesystem_notifications/darwin.py	2012-07-12 13:53:19 +0000
@@ -61,7 +61,7 @@
 }
 
 
-def path_is_ignored(self, path):
+def path_is_ignored(path):
     """Should we ignore this path in the current platform.?"""
     # don't support links yet
     if os.path.islink(path):
@@ -135,10 +135,6 @@
         self.observer = fsevents.Observer()
         self.observer.start()
 
-    def __del__(self):
-        """Stop the observer."""
-        self.observer.stop()
-
     def stop_watch(self, watch):
         """Stop a given watch."""
         watch.stop_watching()
@@ -148,10 +144,11 @@
     def stop(self):
         """Stop the manager."""
         self.observer.stop()
+        self.observer.join()
 
     def del_watch(self, watch):
         """Delete the watch and clean resources."""
-        self.observer.unschedule(watch.stream)
+        self.observer.unschedule(watch.platform_watch.stream)
 
     def add_watch(self, watch):
         """This method perform actually the action of registering the watch."""
@@ -160,4 +157,3 @@
 
     def rm_watch(self, watch):
         """Remove the the watch with the given wd."""
-        self.observer.unschedule(watch.platform_watch.stream)