Merge lp:~alecu/ubuntuone-client/watch-finished-deferred into lp:ubuntuone-client

[Alejandro J. Cura]

A deferred that is fired when watches are completely removed. (LP: #824819)

A deferred that is fired when watches are completely removed. (LP: #824819)

[Natalia Bidart (nataliabidart)]

Looks great!

review: Approve

[Guillermo Gonzalez (verterok)]

looks good.

review: Approve

=== modified file 'run-tests.bat'
--- run-tests.bat	2011-07-12 14:57:17 +0000
+++ run-tests.bat	2011-08-17 21:29:24 +0000
@@ -52,7 +52,7 @@
 ECHO Python found, executing the tests...
 COPY windows\clientdefs.py ubuntuone\clientdefs.py
 :: execute the tests with a number of ignored linux only modules
-"%PYTHONEXEPATH%\python.exe" "%PYTHONEXEPATH%\Scripts\u1trial" --reactor=twisted -c tests -p tests\platform\linux %1
+"%PYTHONEXEPATH%\python.exe" "%PYTHONEXEPATH%\Scripts\u1trial" --reactor=twisted -c tests -p tests\platform\linux %*
 "%PYTHONEXEPATH%\python.exe" "%PYTHONEXEPATH%\Scripts\u1lint"
 :: test for style if we can, if pep8 is not present, move to the end
 IF EXIST "%PYTHONEXEPATH%\Scripts\pep8.exe"

=== modified file 'tests/platform/test_filesystem_notifications.py'
--- tests/platform/test_filesystem_notifications.py	2011-08-09 12:56:06 +0000
+++ tests/platform/test_filesystem_notifications.py	2011-08-17 21:29:24 +0000
@@ -86,6 +86,7 @@
         self.vm = testcase.FakeVolumeManager(self.root_dir)
         self.tritcask_dir = self.mktemp("tritcask_dir")
         self.db = Tritcask(self.tritcask_dir)
+        self.addCleanup(self.db.shutdown)
         self.fs = filesystem_manager.FileSystemManager(fsmdir, partials_dir,
                                                        self.vm, self.db)
         self.fs.create(path=self.root_dir, share_id='', is_dir=True)
@@ -102,17 +103,11 @@
 
         eq.subscribe(HitMe())
         self.monitor = eq.monitor
+        self.addCleanup(self.monitor.shutdown)
         self.log_handler = MementoHandler()
         self.log_handler.setLevel(logging.DEBUG)
         self.monitor.log.addHandler(self.log_handler)
-
-    @defer.inlineCallbacks
-    def tearDown(self):
-        """Clean up the tests."""
-        self.monitor.shutdown()
-        self.db.shutdown()
-        self.monitor.log.removeHandler(self.log_handler)
-        yield super(BaseFSMonitorTestCase, self).tearDown()
+        self.addCleanup(self.monitor.log.removeHandler, self.log_handler)
 
 
 class DynamicHitMe(object):

=== modified file 'tests/platform/test_os_helper.py'
--- tests/platform/test_os_helper.py	2011-08-15 20:29:25 +0000
+++ tests/platform/test_os_helper.py	2011-08-17 21:29:24 +0000
@@ -313,7 +313,8 @@
         make_link(self.testfile, destination)
 
         self.assertTrue(is_link(destination))
-        self.assertEqual(self.testfile, read_link(destination))
+        self.assertEqual(os.path.normcase(self.testfile),
+                         os.path.normcase(read_link(destination)))
 
     def test_movetotrash_file_ok(self):
         """Move a file to trash ok.

=== modified file 'tests/platform/windows/test_filesystem_notifications.py'
--- tests/platform/windows/test_filesystem_notifications.py	2011-08-11 21:49:15 +0000
+++ tests/platform/windows/test_filesystem_notifications.py	2011-08-17 21:29:24 +0000
@@ -618,6 +618,12 @@
         watch = Watch(1, path, None, True, None)
         self.assertEqual(watch.started, watch._watch_started_deferred)
 
+    def test_stopped_property(self):
+        """Test that the stopped property returns the stopped deferred."""
+        path = u'\\\\?\\C:\\path'  # a valid windows path
+        watch = Watch(1, path, None, True, None)
+        self.assertEqual(watch.stopped, watch._watch_stopped_deferred)
+
     def random_error(self, *args):
         """Throw a fake exception."""
         raise FakeException()
@@ -646,14 +652,28 @@
         """CloseHandle is called when there's an error in the watch thread."""
         test_path = self.mktemp("test_directory")
         close_called = []
+        self.patch(filesystem_notifications, "CreateFileW", lambda *_: None)
         self.patch(filesystem_notifications, "CloseHandle",
                    close_called.append)
         self.patch(filesystem_notifications, "ReadDirectoryChangesW",
                    self.random_error)
-        watch = Watch(1, test_path, None, True, None)
+        watch = Watch(1, test_path, self.mask, True, None)
         d = watch.start_watching()
         yield self.assertFailure(d, FakeException)
         self.assertEqual(len(close_called), 1)
+        yield watch.stop_watching()
+
+    @defer.inlineCallbacks
+    def test_stop_watching_fired_when_watch_thread_finishes(self):
+        """The deferred returned is fired when the watch thread finishes."""
+        events = []
+        test_path = self.mktemp("another_test_directory")
+        fake_processor = events.append
+        watch = Watch(1, test_path, self.mask, True, fake_processor)
+        yield watch.start_watching()
+        self.assertNotEqual(watch._watch_handle, None)
+        yield watch.stop_watching()
+        self.assertEqual(watch._watch_handle, None)
 
 
 class TestWatchManager(BaseTwistedTestCase):
@@ -668,17 +688,39 @@
         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 stop_watching():
+        def fake_stop_watching(watch):
             """Fake stop watch."""
             self.was_called = True
-
-        self.watch.stop_watching = stop_watching
-        self.manager.stop()
-        self.assertTrue(self.was_called, 'The watch stop was not called.')
+            return defer.succeed(True)
+
+        self.patch(Watch, "stop_watching", fake_stop_watching)
+        yield self.manager.stop()
+        self.assertTrue(self.was_called, 'The watch stop should be called.')
+
+    @defer.inlineCallbacks
+    def test_stop_multiple(self):
+        """Test that stop is fired when *all* watches have stopped."""
+
+        def fake_stop_watching(watch):
+            """Another fake stop watch."""
+            return watch.stopped
+
+        self.patch(Watch, "stop_watching", fake_stop_watching)
+        second_path = self.parent_path + u"second_path"
+        second_watch = Watch(2, second_path, None, True, None)
+        self.manager._wdm[2] = second_watch
+        d = self.manager.stop()
+        self.assertFalse(d.called, "Not fired before all watches end")
+        self.watch.stopped.callback(None)
+        self.assertFalse(d.called, "Not fired before all watches end")
+        second_watch.stopped.callback(None)
+        yield d
+        self.assertTrue(d.called, "Fired after the watches ended")
 
     def test_get_present_watch(self):
         """Test that we can get a Watch using is wd."""
@@ -688,6 +730,7 @@
         """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
@@ -695,20 +738,21 @@
         def stop_watching():
             """Fake stop watch."""
             self.was_called = True
+            return defer.succeed(True)
 
         self.watch.stop_watching = stop_watching
-        self.manager.del_watch(1)
+        yield self.manager.del_watch(1)
         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 start_watching(*args):
-            """Fake stop watch."""
+        def fake_start_watching(*args):
+            """Fake start watch."""
             self.was_called = True
 
-        Watch.start_watching = start_watching
+        self.patch(Watch, "start_watching", fake_start_watching)
         self.manager._wdm = {}
 
         mask = 'bit_mask'
@@ -748,9 +792,16 @@
         """A watch on an unwatched path returns None."""
         self.assertEqual(None, self.manager.get_wd(self.parent_path))
 
+    @defer.inlineCallbacks
     def test_rm_present_wd(self):
         """Test the removal of a present watch."""
-        self.manager.rm_watch(1)
+
+        def fake_stop_watching():
+            """Fake stop watch."""
+            return defer.succeed(True)
+
+        self.patch(self.watch, "stop_watching", fake_stop_watching)
+        yield self.manager.rm_watch(1)
         self.assertEqual(None, self.manager._wdm.get(1))
 
     def test_rm_root_path(self):
@@ -791,12 +842,13 @@
 
 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: self.started)
         manager = WatchManager(None)
-        self.addCleanup(manager.stop)
+        # no need to stop watching because start_watching is fake
 
         path = u'\\\\?\\C:\\test'  # a valid windows path
         mask = 'fake bit mask'
@@ -1243,12 +1295,13 @@
 
 class FilesystemMonitorTestCase(BaseTwistedTestCase):
     """Tests for the FilesystemMonitor."""
+    timeout = 5
 
     def test_add_watch_twice(self):
         """Check the deferred returned by a second add_watch."""
         self.patch(Watch, "start_watching", lambda self: self.started)
         monitor = FilesystemMonitor(None, None)
-        self.addCleanup(monitor.shutdown)
+        # no need to stop watching because start_watching is fake
 
         parent_path = 'C:\\test'  # a valid windows path in utf-8 bytes
         child_path = parent_path + "\\child"

=== modified file 'tests/syncdaemon/test_eq_inotify.py'
--- tests/syncdaemon/test_eq_inotify.py	2011-08-10 18:27:24 +0000
+++ tests/syncdaemon/test_eq_inotify.py	2011-08-17 21:29:24 +0000
@@ -577,6 +577,7 @@
         self.partials_dir = self.mktemp('partials_dir')
         self.main = FakeMain(self.root_dir, self.shares_dir,
                              self.data_dir, self.partials_dir)
+        self.addCleanup(self.main.shutdown)
         self.eq = self.main.event_q
         assert self.eq is self.eq.fs.vm.m.event_q
 
@@ -621,7 +622,6 @@
     def tearDown(self):
         """Cleanup."""
         self.eq.unsubscribe(self.listener)
-        self.main.shutdown()
 
         if self.old_value is None:
             os.environ.pop(self.env_var)
@@ -649,6 +649,7 @@
         return first
     assertEqual = failUnlessEqual
 
+    @skipIfOS('win32', "Event watches are not set on ancestors on windows.")
     @defer.inlineCallbacks
     def test_file_events_are_ignored_on_udf_ancestor(self):
         """Events on UDF ancestors are ignored."""

=== modified file 'tests/syncdaemon/test_eventqueue.py'
--- tests/syncdaemon/test_eventqueue.py	2011-07-27 14:26:36 +0000
+++ tests/syncdaemon/test_eventqueue.py	2011-08-17 21:29:24 +0000
@@ -21,13 +21,15 @@
 
 import logging
 
+from twisted.internet import defer
+from twisted.trial.unittest import TestCase
+
+from contrib.testing.testcase import BaseTwistedTestCase, FakeVolumeManager
 from ubuntuone.syncdaemon import (
     event_queue,
     filesystem_manager,
     tritcask,
 )
-from contrib.testing.testcase import BaseTwistedTestCase, FakeVolumeManager
-from twisted.internet import defer
 from ubuntuone.devtools.handlers import MementoHandler
 
 
@@ -379,3 +381,32 @@
 
         d.addCallback(callback)
         return d
+
+
+class FakeMonitor(object):
+    """A fake FilesystemMonitor."""
+
+    def __init__(self, *args):
+        """Initialize this fake."""
+        self.shutdown_d = defer.Deferred()
+
+    def shutdown(self):
+        """Get the shutdown deferred."""
+        return self.shutdown_d
+
+
+class EventQueueShutdownTestCase(TestCase):
+    """Test the shutdown method in EQ."""
+
+    timeout = 2
+
+    @defer.inlineCallbacks
+    def test_shutdown_defers(self):
+        """The shutdown method in eq defers on the shutdown of the monitor."""
+        self.patch(event_queue, "FilesystemMonitor", FakeMonitor)
+        eq = event_queue.EventQueue(None)
+        d = eq.shutdown()
+        self.assertFalse(d.called, "shutdown is fired after the monitor.")
+        eq.monitor.shutdown_d.callback(True)
+        self.assertTrue(d.called, "shutdown is fired after the monitor.")
+        yield d

=== modified file 'tests/syncdaemon/test_main.py'
--- tests/syncdaemon/test_main.py	2011-08-10 18:27:24 +0000
+++ tests/syncdaemon/test_main.py	2011-08-17 21:29:24 +0000
@@ -123,6 +123,7 @@
         main.event_q.push('SYS_UNKNOWN_ERROR')
         self.assertTrue(self.restarted)
 
+    @defer.inlineCallbacks
     def test_shutdown_pushes_sys_quit(self):
         """When shutting down, the SYS_QUIT event is pushed."""
         params = get_main_params(self, _get_main_common_params(self))
@@ -131,7 +132,7 @@
         self.patch(main.event_q, 'push',
                    lambda *a, **kw: events.append((a, kw)))
 
-        main.shutdown()
+        yield main.shutdown()
         expected = [(('SYS_USER_DISCONNECT',), {}), (('SYS_QUIT',), {})]
         self.assertEqual(expected, events)
 

=== modified file 'tests/syncdaemon/test_tritcask.py'
--- tests/syncdaemon/test_tritcask.py	2011-08-10 12:14:19 +0000
+++ tests/syncdaemon/test_tritcask.py	2011-08-17 21:29:24 +0000
@@ -726,6 +726,7 @@
         self.db.live_file.make_immutable()
         self.db.shutdown()
         self.db = Tritcask(self.base_dir)
+        self.addCleanup(self.db.shutdown)
         self.db.put(1, 'foo1', 'bar1')
         self.assertEqual(1, len(self.db._immutable))
         # read from the old file

=== modified file 'tests/syncdaemon/test_vm_helper.py'
--- tests/syncdaemon/test_vm_helper.py	2011-08-03 20:44:10 +0000
+++ tests/syncdaemon/test_vm_helper.py	2011-08-17 21:29:24 +0000
@@ -25,6 +25,9 @@
 
 from tests.syncdaemon.test_vm import BaseVolumeManagerTests
 
+from contrib.testing.testcase import BaseTwistedTestCase
+from ubuntuone.platform import os_helper
+from ubuntuone.syncdaemon import vm_helper
 from ubuntuone.syncdaemon.vm_helper import (
     create_shares_link,
     get_share_dir_name,
@@ -78,16 +81,35 @@
         self.assertRaises(ValueError, get_udf_suggested_path, None)
         self.assertRaises(ValueError, get_udf_suggested_path, relative_home)
 
-    def _test_create_shares_link(self):
-        """Test for create_shares_link."""
-        source = 'source'
-        dest = 'dest'
-        make_link = self.mocker.replace(
-                            'ubuntuone.platform.windows.os_helper.make_link')
-        make_link(source, dest)
-        self.mocker.replay()
-        create_shares_link(source, dest)
-        self.mocker.verify()
+
+class VMHelperLinkTestCase(BaseTwistedTestCase):
+    """Tests for the VM Helper symlinks."""
+
+    def test_create_shares_link_exists(self):
+        """create_shares_link is noop when there's something with that name."""
+        base = self.mktemp("test_create_shares_link_exists")
+        source_path = os.path.join(base, "source")
+        dest_path = os.path.join(base, "dest")
+        os_helper.make_dir(dest_path)
+        self.assertFalse(create_shares_link(source_path, dest_path))
+
+    def test_create_shares_link_makes_the_link(self):
+        """create_shares_link makes the link as expected."""
+        base = self.mktemp("test_create_shares_link_makes_the_link")
+        source_path = os.path.join(base, "source")
+        dest_path = os.path.join(base, "dest")
+        os_helper.make_dir(source_path)
+        self.assertTrue(create_shares_link(source_path, dest_path))
+        self.assertTrue(vm_helper.is_link(dest_path))
+
+    def test_create_shares_link_existing(self):
+        """create_shares_link on an existing path does nothing."""
+        base = self.mktemp("test_create_shares_link_makes_the_link")
+        source_path = os.path.join(base, "source")
+        dest_path = os.path.join(base, "dest")
+        os_helper.make_dir(source_path)
+        self.assertTrue(create_shares_link(source_path, dest_path))
+        self.assertFalse(create_shares_link(source_path, dest_path))
 
 
 class GetShareDirNameTests(BaseVolumeManagerTests):

=== modified file 'ubuntuone/platform/linux/os_helper.py'
--- ubuntuone/platform/linux/os_helper.py	2011-08-15 18:54:45 +0000
+++ ubuntuone/platform/linux/os_helper.py	2011-08-17 21:29:24 +0000
@@ -131,7 +131,7 @@
 
 def remove_link(path):
     """Removes a link."""
-    if os.path.exists(path):
+    if is_link(path):
         os.unlink(path)
 
 

=== modified file 'ubuntuone/platform/windows/filesystem_notifications.py'
--- ubuntuone/platform/windows/filesystem_notifications.py	2011-08-11 21:27:07 +0000
+++ ubuntuone/platform/windows/filesystem_notifications.py	2011-08-17 21:29:24 +0000
@@ -1,5 +1,7 @@
+#
 # Authors: Manuel de la Pena <[e-mail]>
 #          Natalia B. Bidart <[e-mail]>
+#          Alejandro J. Cura <[e-mail]>
 #
 # Copyright 2011 Canonical Ltd.
 #
@@ -23,6 +25,7 @@
 from uuid import uuid4
 
 from twisted.internet import defer, reactor
+from twisted.python.failure import Failure
 from pywintypes import OVERLAPPED
 from win32api import CloseHandle
 from win32con import (
@@ -137,6 +140,7 @@
         self._cookie = None
         self._source_pathname = None
         self._process_thread = None
+        self._watch_handle = None
         # remember the subdirs we have so that when we have a delete we can
         # check if it was a remove
         self._subdirs = []
@@ -149,6 +153,8 @@
         # this deferred is fired when the watch has started monitoring
         # a directory from a thread
         self._watch_started_deferred = defer.Deferred()
+        # and this one is fired when the watch has stopped
+        self._watch_stopped_deferred = defer.Deferred()
 
     @is_valid_windows_path(path_indexes=[1])
     def _path_is_dir(self, path):
@@ -170,7 +176,7 @@
         return is_dir
 
     def _process_events(self, events):
-        """Process the events form the queue."""
+        """Process the events from the queue."""
         # do not do it if we stop watching and the events are empty
         if not self._watching:
             return
@@ -227,9 +233,9 @@
         """Wrap _watch, and errback on any unhandled error."""
         try:
             self._watch()
-        except Exception as error:
+        except Exception:
             reactor.callFromThread(self._call_deferred,
-                self._watch_started_deferred.errback, error)
+                self._watch_started_deferred.errback, Failure())
 
     def _watch(self):
         """Watch a path that is a directory."""
@@ -238,7 +244,7 @@
                        os.path.exists(self._path), os.path.isdir(self._path))
         # we are going to be using the ReadDirectoryChangesW whihc requires
         # a directory handle and the mask to be used.
-        handle = CreateFileW(
+        self._watch_handle = CreateFileW(
             self._path,
             FILE_LIST_DIRECTORY,
             FILE_SHARE_READ | FILE_SHARE_WRITE,
@@ -248,9 +254,11 @@
             None)
 
         try:
-            self._watch_loop(handle)
+            self._watch_loop(self._watch_handle)
         finally:
-            CloseHandle(handle)
+            CloseHandle(self._watch_handle)
+            self._watch_handle = None
+            reactor.callFromThread(self.stopped.callback, True)
 
     def _watch_loop(self, handle):
         """The loop where we watch the directory."""
@@ -325,6 +333,7 @@
         SetEvent(self._wait_stop)
         self._watching = False
         self._subdirs = []
+        return self.stopped
 
     def update(self, mask, auto_add=False):
         """Update the info used by the watcher."""
@@ -346,6 +355,11 @@
         """A deferred that will be called when the watch is running."""
         return self._watch_started_deferred
 
+    @property
+    def stopped(self):
+        """A deferred fired when the watch thread has finished."""
+        return self._watch_stopped_deferred
+
 
 class WatchManager(object):
     """Implement the same functions as pyinotify.WatchManager.
@@ -368,19 +382,22 @@
     def stop(self):
         """Close the manager and stop all watches."""
         self.log.debug('Stopping watches.')
+        wait_list = []
         for current_wd in self._wdm:
-            self._wdm[current_wd].stop_watching()
-            self.log.debug('Watch for %s stopped.', self._wdm[current_wd].path)
+            wait_list.append(self._wdm[current_wd].stop_watching())
+            self.log.debug('Stopping Watch on %r.', self._wdm[current_wd].path)
+        return defer.DeferredList(wait_list)
 
     def get_watch(self, wd):
         """Return the watch with the given descriptor."""
         return self._wdm[wd]
 
+    @defer.inlineCallbacks
     def del_watch(self, wd):
         """Delete the watch with the given descriptor."""
         try:
             watch = self._wdm[wd]
-            watch.stop_watching()
+            yield watch.stop_watching()
             del self._wdm[wd]
             self.log.debug('Watch %s removed.', wd)
         except KeyError, e:
@@ -445,11 +462,12 @@
         if watch:
             return watch.path
 
+    @defer.inlineCallbacks
     def rm_watch(self, wd, rec=False, quiet=True):
         """Remove the the watch with the given wd."""
         try:
             watch = self._wdm[wd]
-            watch.stop_watching()
+            yield watch.stop_watching()
             del self._wdm[wd]
         except KeyError, err:
             self.log.error(str(err))
@@ -700,7 +718,7 @@
 
     def shutdown(self):
         """Prepares the EQ to be closed."""
-        self._watch_manager.stop()
+        return self._watch_manager.stop()
 
     @windowspath(path_indexes=[1])
     def rm_watch(self, dirpath):

=== modified file 'ubuntuone/syncdaemon/event_queue.py'
--- ubuntuone/syncdaemon/event_queue.py	2011-08-03 22:01:22 +0000
+++ ubuntuone/syncdaemon/event_queue.py	2011-08-17 21:29:24 +0000
@@ -21,6 +21,8 @@
 import functools
 import logging
 
+from twisted.internet import defer
+
 from ubuntuone.platform import FilesystemMonitor
 
 # these are our internal events, what is inserted into the whole system
@@ -202,9 +204,10 @@
         if not self.empty_event_queue_callbacks:
             self._have_empty_eq_cback = False
 
+    @defer.inlineCallbacks
     def shutdown(self):
         """Make the monitor shutdown."""
-        self.monitor.shutdown()
+        yield self.monitor.shutdown()
         # clean up all registered listeners
         if len(self.listener_map.items()) > 0:
             for k, v in self.listener_map.items():

=== modified file 'ubuntuone/syncdaemon/main.py'
--- ubuntuone/syncdaemon/main.py	2011-08-10 14:34:36 +0000
+++ ubuntuone/syncdaemon/main.py	2011-08-17 21:29:24 +0000
@@ -199,6 +199,7 @@
         # hook the event queue to the root dir
         self.event_q.push('SYS_INIT_DONE')
 
+    @defer.inlineCallbacks
     def shutdown(self, with_restart=False):
         """Shutdown the daemon; optionally restart it."""
         self.event_q.push('SYS_USER_DISCONNECT')
@@ -215,7 +216,7 @@
         self.hash_q.shutdown()
         self.external.shutdown(with_restart)
 
-        self.event_q.shutdown()
+        yield self.event_q.shutdown()
         self.db.shutdown()
         self.mark.stop()
 

=== modified file 'ubuntuone/syncdaemon/vm_helper.py'
--- ubuntuone/syncdaemon/vm_helper.py	2011-08-09 12:56:06 +0000
+++ ubuntuone/syncdaemon/vm_helper.py	2011-08-17 21:29:24 +0000
@@ -27,7 +27,7 @@
     make_link,
     path_exists,
     read_link,
-    remove_file,
+    remove_link,
 )
 
 
@@ -71,7 +71,7 @@
     if not path_exists(dest):
         # remove the symlink if it's broken
         if is_link(dest) and read_link(dest) != source:
-            remove_file(dest)
+            remove_link(dest)
         make_link(source, dest)
         return True
     else: