Merge lp:~nataliabidart/ubuntuone-client/share-subscription-ops into lp:ubuntuone-client

[Natalia Bidart]

- Shares can be subscribed to/unsubscribed from (part of LP: #708335).

[lp:~nataliabidart/ubuntuone-client/share-subscription-ops]

833. By Natalia Bidart on 2011-02-01

Changes from dependency branch.

834. By Natalia Bidart on 2011-02-02

Merged trunk in.

835. By Natalia Bidart on 2011-02-02

Merged trunk in.

836. By Natalia Bidart on 2011-02-02

Changes from dependency branch.

[Roberto Alsina (ralsina)]

+1 looks good to me.

[Roberto Alsina (ralsina)]

review: Approve

[lp:~nataliabidart/ubuntuone-client/share-subscription-ops]

837. By Natalia Bidart on 2011-02-03

Merged trunk in.

[dobey (dobey)]

review: Approve

=== modified file 'tests/syncdaemon/test_vm.py'
--- tests/syncdaemon/test_vm.py	2011-02-01 14:24:41 +0000
+++ tests/syncdaemon/test_vm.py	2011-02-03 12:52:26 +0000
@@ -983,6 +983,232 @@
                 self.assertEqual(set(defined_args[1:]), set(evtargs))
 
 
+class ViewSharesSubscriptionTests(BaseVolumeManagerTests):
+    """Test Shares subscription operations when access_level is View."""
+
+    access_level = 'View'
+
+    @defer.inlineCallbacks
+    def test_subscribe_share(self):
+        """Test subscribe_share method."""
+        share = self._create_share(access_level=self.access_level,
+                                   subscribed=False)
+        yield self.vm.add_share(share)
+        self.assertFalse(self.vm.shares[share.volume_id].subscribed)
+        # subscribe to it
+        yield self.vm.subscribe_share(share.volume_id)
+        self.assertTrue(self.vm.shares[share.volume_id].subscribed)
+
+    @defer.inlineCallbacks
+    def test_subscribe_share_missing_path(self):
+        """Test subscribe_share with a missing path """
+        share = self._create_share(access_level=self.access_level,
+                                   subscribed=False)
+        yield self.vm.add_share(share)
+        self.assertFalse(os.path.exists(share.path))
+        self.assertFalse(self.vm.shares[share.id].subscribed)
+        # subscribe to it
+        yield self.vm.subscribe_share(share.id)
+        self.assertTrue(self.vm.shares[share.id].subscribed)
+        self.assertTrue(os.path.exists(share.path))
+
+    @defer.inlineCallbacks
+    def test_subscribe_share_missing_volume(self):
+        """Test subscribe_share with a invalid volume_id."""
+        try:
+            yield self.vm.subscribe_share('invalid_share_id')
+        except VolumeDoesNotExist, e:
+            self.assertEquals('DOES_NOT_EXIST', e.args[0])
+            self.assertEquals('invalid_share_id', e.args[1])
+        else:
+            self.fail('Must get a VolumeDoesNotExist!')
+
+    @defer.inlineCallbacks
+    def test_unsubscribe_share(self):
+        """Test unsubscribe_share method."""
+        share = self._create_share(access_level=self.access_level,
+                                   subscribed=True)
+        yield self.vm.add_share(share)
+        self.assertTrue(self.vm.shares[share.volume_id].subscribed)
+        # unsubscribe from it
+        self.vm.unsubscribe_share(share.volume_id)
+        self.assertFalse(self.vm.shares[share.volume_id].subscribed)
+
+    @defer.inlineCallbacks
+    def test_unsubscribe_share_with_content(self):
+        """Test unsubscribe_share method in a share with content."""
+        share = self._create_share(access_level=self.access_level,
+                                   subscribed=True)
+        yield self.vm.add_share(share)
+
+        self.assertTrue(self.vm.shares[share.volume_id].subscribed)
+        # create a few files and directories
+        dirs = ['dir', 'dir/subdir', 'dir/empty_dir']
+        for i, dir in enumerate(dirs):
+            path = os.path.join(share.path, dir)
+            with allow_writes(os.path.split(share.path)[0]):
+                with allow_writes(share.path):
+                    if not os.path.exists(path):
+                        os.makedirs(path)
+            self.main.fs.create(path, share.volume_id, is_dir=True)
+            self.main.fs.set_node_id(path, 'dir_node_id'+str(i))
+            # add a inotify watch to the dir
+            self.main.event_q.add_watch(path)
+        files = ['a_file', 'dir/file', 'dir/subdir/file']
+        for i, file in enumerate(files):
+            path = os.path.join(share.path, file)
+            with allow_writes(os.path.split(share.path)[0]):
+                with allow_writes(share.path):
+                    open(path, 'w').close()
+            self.main.fs.create(path, share.volume_id)
+            self.main.fs.set_node_id(path, 'file_node_id'+str(i))
+        paths = self.main.fs.get_paths_starting_with(share.path)
+        self.assertEquals(len(paths), len(dirs+files)+1)
+
+        # unsubscribe from it
+        self.vm.unsubscribe_share(share.volume_id)
+
+        self.assertEquals(2, len(self.vm.shares))  # share and root
+        self.assertFalse(self.vm.shares[share.volume_id].subscribed)
+        # check that the share is in the fsm metadata
+        self.assertTrue(self.main.fs.get_by_path(share.path))
+        # get the childs (should be an empty list)
+        paths = list(self.main.fs.get_paths_starting_with(share.path))
+        self.assertEquals(len(dirs+files)+1, len(paths))
+        # check that there isn't a watch in the share
+        self.assertFalse(self.main.event_q.has_watch(share.path))
+        # check that the childs don't have a watch
+        for path, is_dir in paths:
+            if is_dir:
+                self.assertFalse(self.main.event_q.has_watch(path))
+
+
+class ModifySharesSubscriptionTests(ViewSharesSubscriptionTests):
+    """Test Shares subscription operations when access_level is Modify."""
+
+    access_level = 'Modify'
+
+    @defer.inlineCallbacks
+    def test_subscribe_share_missing_fsm_md(self):
+        """Test subscribe_share with a missing node in fsm."""
+        share = self._create_share(access_level=self.access_level,
+                                   subscribed=False)
+        yield self.vm.add_share(share)
+        self.assertFalse(os.path.exists(share.path))
+        self.assertFalse(self.vm.shares[share.id].subscribed)
+        yield self.vm.subscribe_share(share.id)
+        yield self.vm.unsubscribe_share(share.id)
+        # delete the fsm metadata
+        self.main.fs.delete_metadata(share.path)
+        # subscribe to it and fail!
+        try:
+            yield self.vm.subscribe_share(share.id)
+        except KeyError, e:
+            self.assertIn(share.path, e.args[0])
+        else:
+            self.fail('Must get a KeyError!')
+
+    @defer.inlineCallbacks
+    def _test_subscribe_share_generations(self, share):
+        """Test subscribe_share with a generation."""
+        scratch_d = defer.Deferred()
+        def fake_rescan_from_scratch(volume_id):
+            """A fake rescan_from_scratch that check the arguments."""
+            self.assertEquals(share.volume_id, volume_id)
+            scratch_d.callback(None)
+        self.main.action_q.rescan_from_scratch = fake_rescan_from_scratch
+        # subscribe to it
+        yield self.vm.subscribe_share(share.volume_id)
+        yield scratch_d
+        self.assertTrue(self.vm.shares[share.volume_id].subscribed)
+
+    @defer.inlineCallbacks
+    def test_subscribe_share_valid_generation(self):
+        """Test subscribe_share with a valid generation."""
+        share = self._create_share(access_level=self.access_level,
+                                   subscribed=False)
+        yield self.vm.add_share(share)
+        self.assertFalse(self.vm.shares[share.volume_id].subscribed)
+        # update share generation
+        self.vm.update_generation(share.volume_id, 0)
+        yield self._test_subscribe_share_generations(share)
+
+    @defer.inlineCallbacks
+    def test_subscribe_share_without_generation(self):
+        """Test subscribe_share without a valid generation."""
+        share = self._create_share(access_level=self.access_level,
+                                   subscribed=False)
+        yield self.vm.add_share(share)
+        self.assertFalse(self.vm.shares[share.volume_id].subscribed)
+        # update share generation
+        self.vm.update_generation(share.volume_id, None)
+        yield self._test_subscribe_share_generations(share)
+
+    @defer.inlineCallbacks
+    def test_unsubscribe_subscribe_share_with_content(self):
+        """Test for re-subscribing to a share."""
+        share = self._create_share(access_level=self.access_level,
+                                   subscribed=True)
+        yield self.vm.add_share(share)
+        self.main.event_q.rm_watch(share.path)
+        self.assertTrue(self.vm.shares[share.volume_id].subscribed)
+        # create a few files and directories
+        dirs = ['dir', 'dir/subdir', 'dir/empty_dir']
+        for i, dir in enumerate(dirs):
+            path = os.path.join(share.path, dir)
+            with allow_writes(os.path.split(share.path)[0]):
+                with allow_writes(share.path):
+                    if not os.path.exists(path):
+                        os.makedirs(path)
+            self.main.fs.create(path, share.volume_id, is_dir=True)
+            self.main.fs.set_node_id(path, 'dir_node_id'+str(i))
+        files = ['a_file', 'dir/file', 'dir/subdir/file']
+        for i, path in enumerate(files):
+            path = os.path.join(share.path, path)
+            with allow_writes(os.path.split(share.path)[0]):
+                with allow_writes(share.path):
+                    open(path, 'w').close()
+            self.main.fs.create(path, share.volume_id)
+            self.main.fs.set_node_id(path, 'file_node_id'+str(i))
+        paths = list(self.main.fs.get_paths_starting_with(share.path))
+        # add a inotify watch to the dirs
+        for path, is_dir in paths:
+            if is_dir:
+                self.main.event_q.add_watch(path)
+        self.assertEquals(len(paths), len(dirs+files)+1, paths)
+
+        # unsubscribe from it
+        self.vm.unsubscribe_share(share.volume_id)
+
+        self.assertEquals(2, len(self.vm.shares))  # share and root
+        self.assertFalse(self.vm.shares[share.volume_id].subscribed)
+        # check that the share is in the fsm metadata
+        self.assertTrue(self.main.fs.get_by_path(share.path))
+        # check that there isn't a watch in the share
+        self.assertFalse(self.main.event_q.has_watch(share.path))
+        # check that the childs don't have a watch
+        for path, is_dir in paths:
+            if is_dir:
+                self.assertFalse(self.main.event_q.has_watch(path))
+        # check the childs
+        paths = self.main.fs.get_paths_starting_with(share.path)
+        self.assertEquals(len(dirs+files)+1, len(paths))
+        # resubscribe to it
+        yield self.vm.subscribe_share(share.volume_id)
+        paths = list(self.main.fs.get_paths_starting_with(share.path))
+        # we should only have the dirs, as the files metadata is
+        # delete by local rescan (both hashes are '')
+        self.assertEquals(len(dirs)+1, len(paths))
+        # check that there is a watch in the share
+        self.assertTrue(self.main.event_q.has_watch(share.path))
+        # check that the child dirs have a watch
+        for path, is_dir in paths:
+            if is_dir:
+                self.assertTrue(self.main.event_q.has_watch(path),
+                                '%s has a watch' % path)
+                self.vm._remove_watch(path)
+
+
 class VolumeManagerUnicodeTests(BaseVolumeManagerTests):
     """Tests for Volume Manager unicode capabilities."""
 

=== modified file 'ubuntuone/syncdaemon/event_queue.py'
--- ubuntuone/syncdaemon/event_queue.py	2011-01-21 15:45:16 +0000
+++ ubuntuone/syncdaemon/event_queue.py	2011-02-03 12:52:26 +0000
@@ -151,6 +151,10 @@
     'VM_UDF_UNSUBSCRIBE_ERROR': ('udf_id', 'error'),
     'VM_UDF_CREATED': ('udf',),
     'VM_UDF_CREATE_ERROR': ('path', 'error'),
+    'VM_SHARE_SUBSCRIBED': ('share',),
+    'VM_SHARE_SUBSCRIBE_ERROR': ('share_id', 'error'),
+    'VM_SHARE_UNSUBSCRIBED': ('share',),
+    'VM_SHARE_UNSUBSCRIBE_ERROR': ('share_id', 'error'),
     'VM_SHARE_CREATED': ('share_id',),
     'VM_SHARE_DELETED': ('share',),
     'VM_SHARE_DELETE_ERROR': ('share_id', 'error'),

=== modified file 'ubuntuone/syncdaemon/volume_manager.py'
--- ubuntuone/syncdaemon/volume_manager.py	2011-02-02 18:33:45 +0000
+++ ubuntuone/syncdaemon/volume_manager.py	2011-02-03 12:52:26 +0000
@@ -945,6 +945,19 @@
         if not share.can_write():
             set_dir_readonly(share.path)
 
+    def _create_udf_dir(self, udf):
+        """Create the udf dir if does not exist."""
+        if not path_exists(udf.path):
+            # the udf path isn't there, create it!
+            make_dir(udf.path, recursive=True)
+
+    def _create_volume_dir(self, volume):
+        """Create the volume dir if does not exist, set perms for shares."""
+        if isinstance(volume, (Share, Root)):
+            self._create_share_dir(volume)
+        elif isinstance(volume, UDF):
+            self._create_udf_dir(volume)
+
     def _create_fsm_object(self, path, volume_id, node_id):
         """ Creates the mdobj for this share in fs manager. """
         try:
@@ -1103,6 +1116,13 @@
             if all_volumes or volume.active:
                 yield volume
 
+    def store_volume(self, volume):
+        """Store 'volume'."""
+        if isinstance(volume, (Share, Root)):
+            self.shares[volume.volume_id] = volume
+        elif isinstance(volume, UDF):
+            self.udfs[volume.volume_id] = volume
+
     def _is_nested_udf(self, path):
         """Check if it's ok to create a UDF in 'path'.
 
@@ -1168,44 +1188,74 @@
         volume = self.get_volume(volume_id)
         self.m.action_q.delete_volume(volume.id, volume.path)
 
+    def subscribe_share(self, share_id):
+        """Mark the Share with 'share_id' as subscribed.
+
+        Also fire a local and server rescan.
+
+        """
+        push_error = functools.partial(self.m.event_q.push,
+            'VM_SHARE_SUBSCRIBE_ERROR', share_id=share_id)
+        push_success = lambda volume: \
+            self.m.event_q.push('VM_SHARE_SUBSCRIBED', share=volume)
+        self.log.info('subscribe_share: %r', share_id)
+        d = self._subscribe_volume(share_id, push_success, push_error)
+        return d
+
     def subscribe_udf(self, udf_id):
-        """Mark the UDF with id as subscribed.
+        """Mark the UDF with 'udf_id' as subscribed.
 
         Also fire a local and server rescan.
 
         """
         push_error = functools.partial(self.m.event_q.push,
                                        'VM_UDF_SUBSCRIBE_ERROR', udf_id=udf_id)
+        push_success = lambda volume: self.m.event_q.push('VM_UDF_SUBSCRIBED',
+                                                          udf=volume)
         self.log.info('subscribe_udf: %r', udf_id)
+        d = self._subscribe_volume(udf_id, push_success, push_error)
+        return d
+
+    def _subscribe_volume(self, volume_id, push_success, push_error):
+        """Mark the volume with 'volume_id' as subscribed.
+
+        If can_write(), fire a local and server rescan while temporary
+        unsubscribing from it.
+
+        """
+        self.log.debug('_subscribe_volume: %r', volume_id)
         try:
-            udf = self.udfs[udf_id]
-        except KeyError:
+            volume = self.get_volume(volume_id)
+        except VolumeDoesNotExist, e:
             push_error(error="DOES_NOT_EXIST")
-            d = defer.fail(VolumeDoesNotExist(udf_id))
-        else:
-            if not path_exists(udf.path):
-                # the udf path isn't there, create it!
-                make_dir(udf.path, recursive=True)
-            def subscribe(result):
-                """Subscribe the UDF after the local rescan.
-
-                As we don't wait for server rescan to finish, the udf is
-                subscribed just after the local rescan it's done.
-
-                """
-                udf.subscribed = True
-                self.udfs[udf_id] = udf
-                return result
+            return defer.fail(e)
+
+        self._create_volume_dir(volume)
+
+        def subscribe(result):
+            """Subscribe the volume after the local rescan.
+
+            As we don't wait for server rescan to finish, the volume is
+            subscribed just after the local rescan it's done.
+
+            """
+            volume.subscribed = True
+            self.store_volume(volume)
+            return result
+
+        if volume.can_write():
             try:
-                d = self._scan_udf(udf)
+                d = self._scan_volume(volume)
             except KeyError, e:
                 push_error(error="METADATA_DOES_NOT_EXIST")
-                d = defer.fail(e)
-            else:
-                d.addCallback(subscribe)
-                d.addCallbacks(
-                    lambda _: self.m.event_q.push('VM_UDF_SUBSCRIBED', udf=udf),
-                    lambda f: push_error(error=f.getErrorMessage()))
+                return defer.fail(e)
+        else:
+            # avoid local and server rescan for read-only volumes
+            d = defer.succeed(None)
+
+        d.addCallback(subscribe)
+        d.addCallbacks(lambda _: push_success(volume),
+                       lambda f: push_error(error=f.getErrorMessage()))
         return d
 
     def _scan_udf(self, udf):
@@ -1230,21 +1280,38 @@
         d.addCallback(server_rescan)
         return d
 
+    def unsubscribe_share(self, share_id):
+        """Mark the share with share_id as unsubscribed."""
+        self.log.info('unsubscribe_share: %r', share_id)
+        push_error = functools.partial(self.m.event_q.push,
+            'VM_SHARE_UNSUBSCRIBE_ERROR', share_id=share_id)
+        push_success = lambda volume: \
+            self.m.event_q.push('VM_SHARE_UNSUBSCRIBED', share=volume)
+        self._unsubscribe_volume(share_id, push_success, push_error)
+
     def unsubscribe_udf(self, udf_id):
         """Mark the UDF with udf_id as unsubscribed."""
         self.log.info('unsubscribe_udf: %r', udf_id)
+        push_error = functools.partial(self.m.event_q.push,
+            'VM_UDF_UNSUBSCRIBE_ERROR', udf_id=udf_id)
+        push_success = lambda volume: \
+            self.m.event_q.push('VM_UDF_UNSUBSCRIBED', udf=volume)
+        self._unsubscribe_volume(udf_id, push_success, push_error)
+
+    def _unsubscribe_volume(self, volume_id, push_success, push_error):
+        """Mark the volume with volume_id as unsubscribed."""
+        self.log.debug('unsubscribe_volume: %r', volume_id)
         try:
-            udf = self.udfs[udf_id]
-        except KeyError:
-            self.m.event_q.push('VM_UDF_UNSUBSCRIBE_ERROR', udf_id=udf_id,
-                                error="DOES_NOT_EXIST")
+            volume = self.get_volume(volume_id)
+        except VolumeDoesNotExist:
+            push_error(error="DOES_NOT_EXIST")
         else:
             # remove the inotify watches, but don't delete the metadata
-            self._remove_watches(udf.path)
-            # makr the udf as unsubscribed
-            udf.subscribed = False
-            self.udfs[udf_id] = udf
-            self.m.event_q.push('VM_UDF_UNSUBSCRIBED', udf=udf)
+            self._remove_watches(volume.path)
+            # mark the volume as unsubscribed
+            volume.subscribed = False
+            self.store_volume(volume)
+            push_success(volume)
 
     def handle_AQ_CREATE_UDF_OK(self, marker, volume_id, node_id):
         """Handle AQ_CREATE_UDF_OK."""