Logo Search packages:      
Sourcecode: ubuntuone-client version File versions

test_vm.py

# -*- coding: utf-8 -*-
#
# Author: Guillermo Gonzalez <guillermo.gonzalez@canonical.com>
#
# Copyright 2009 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License version 3, as published
# by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE.  See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program.  If not, see <http://www.gnu.org/licenses/>.

""" Tests for the Volume Manager """
import logging
import os
import uuid

from ubuntuone.storageprotocol.client import ListShares
from ubuntuone.storageprotocol.sharersp import (
    NotifyShareHolder,
    ShareResponse,
)
from contrib.testing.testcase import (
    FakeMain,
    BaseTwistedTestCase,
)
from ubuntuone.syncdaemon.volume_manager import (
    Share,
    VolumeManager,
    ShareFileShelf,
)
from twisted.internet import defer, reactor


00041 class BaseVolumeManagerTests(BaseTwistedTestCase):
    """ Bas TestCase for Volume Manager tests """

00044     def setUp(self):
        """ setup the test """
        BaseTwistedTestCase.setUp(self)
        self.log = logging.getLogger("ubuntuone.SyncDaemon.TEST")
        self.log.info("starting test %s.%s", self.__class__.__name__,
                      self._testMethodName)
        self.root_dir = self.mktemp('root_dir')
        self.data_dir = self.mktemp('data_dir')
        self.shares_dir = self.mktemp('shares_dir')
        self.main = FakeMain(self.root_dir, self.shares_dir, self.data_dir)
        self.vm = self.main.vm

00056     def tearDown(self):
        """ cleanup main and remove the temp dir """
        self.main.shutdown()
        self.rmtree(self.root_dir)
        self.rmtree(self.data_dir)
        self.rmtree(self.shares_dir)
        self.log.info("finished test %s.%s", self.__class__.__name__,
                      self._testMethodName)
        return BaseTwistedTestCase.tearDown(self)


00067 class VolumeManagerTests(BaseVolumeManagerTests):
    """ Tests for Volume Manager internal API. """

00070     def test_on_server_root(self):
        """ check that list_shares is called in on_server_root """
        d = defer.Deferred()
        # helper method, pylint: disable-msg=C0111
        def list_shares():
            mdobj = self.main.fs.get_by_path(self.root_dir)
            d.callback(mdobj)
        self.main.action_q.list_shares = list_shares
        self.vm.on_server_root('root_uuid')
        return d

00081     def test_add_share(self):
        """ test the add_share method. """
        # initialize the the root
        self.vm.on_server_root('root_uuid')
        share_path = os.path.join(self.shares_dir, 'fake_share')
        share = Share(share_path, share_id='share_id')
        self.vm.add_share(share)
        self.assertIn(share.id, self.vm.shares)

00090     def test_share_deleted(self):
        """ Check that a share is deleted from the share mapping. """
        # initialize the the root
        self.vm.on_server_root('root_uuid')
        share_path = os.path.join(self.shares_dir, 'fake_share')
        share = Share(share_path, share_id='share_id')
        self.vm.add_share(share)
        self.assertIn(share.id, self.vm.shares)
        self.vm.share_deleted(share.id)
        self.assertNotIn(share.id, self.vm.shares)

00101     def test_share_changed(self):
        """ check that VM.share_changed updates the access_level. """
        share_holder = NotifyShareHolder.from_params('share_id', None,
                                                     'fake_share',
                                                     'test_username',
                                                     'visible_name', 'Modify')
        # initialize the the root
        self.vm.on_server_root('root_uuid')
        share_path = os.path.join(self.shares_dir, share_holder.share_name)
        share = Share(share_path, share_id=share_holder.share_id,
                      access_level='View')
        self.vm.add_share(share)
        self.vm.share_changed(share_holder)
        self.assertEquals('Modify', self.vm.shares[share.id].access_level)

00116     def test_handle_AQ_SHARES_LIST(self):
        """ test the handling of the AQ_SHARE_LIST event. """
        share_response = ShareResponse.from_params('share_id', 'to_me',
                                                   'fake_share_uuid',
                                                   'fake_share', 'username',
                                                   'visible_username', 'yes',
                                                   'View')
        # initialize the the root
        self.vm.on_server_root('root_uuid')
        response = ListShares(None)
        response.shares = [share_response]
        self.vm.handle_AQ_SHARES_LIST(response)
        self.assertEquals(2, len(self.vm.shares)) # the new shares and root
        # check that the share is in the shares dict
        self.assertTrue('share_id' in self.vm.shares)
        share = self.vm.shares['share_id']
        self.assertEquals('fake_share', share.name)
        self.assertEquals('fake_share_uuid', share.subtree)

00135     def test_handle_SV_SHARE_CHANGED(self):
        """ test the handling of the AQ_SHARE_LIST event. """
        share_holder = NotifyShareHolder.from_params('share_id', None,
                                                     'fake_share',
                                                     'test_username',
                                                     'visible_name', 'Modify')
        # initialize the the root
        self.vm.on_server_root('root_uuid')
        # create a share
        share_path = os.path.join(self.shares_dir, share_holder.share_name)
        share = Share(share_path, share_id=share_holder.share_id,
                      access_level='View')
        self.vm.add_share(share)
        self.vm.handle_SV_SHARE_CHANGED('changed', share_holder)
        self.assertEquals('Modify', self.vm.shares['share_id'].access_level)
        self.vm.handle_SV_SHARE_CHANGED('deleted', share_holder)
        self.assertNotIn('share_id', self.vm.shares)

00153     def test_persistence(self):
        """ Test that the persistence of shares works as expected. """
        share_path = os.path.join(self.shares_dir, 'my_share')
        share = Share(share_path, share_id='a_share_id', access_level='View')
        self.vm.add_share(share)
        other_vm = VolumeManager(self.main)
        for key in self.vm.shares:
            self.assertEquals(self.vm.shares[key].__dict__,
                              other_vm.shares[key].__dict__)

00163     def test_handle_AQ_SHARES_LIST_shared(self):
        """test the handling of the AQ_SHARE_LIST event, with a shared dir."""
        share_response = ShareResponse.from_params('share_id', 'to_me',
                                                   'fake_share_uuid',
                                                   'fake_share', 'username',
                                                   'visible_username', 'yes',
                                                   'View')
        shared_response = ShareResponse.from_params('shared_id', 'from_me',
                                                   'shared_uuid',
                                                   'fake_shared', 'myname',
                                                   'my_visible_name', 'yes',
                                                   'Modify')
        # initialize the the root
        self.vm.on_server_root('root_uuid')
        shared_dir = os.path.join(self.root_dir, 'shared_dir')
        self.main.fs.create(path=shared_dir, share_id="", is_dir=True)
        self.main.fs.set_node_id(shared_dir, shared_response.subtree)
        response = ListShares(None)
        response.shares = [share_response, shared_response]
        self.vm.handle_AQ_SHARES_LIST(response)
        self.assertEquals(2, len(self.vm.shares)) # the new shares and root
        self.assertEquals(1, len(self.vm.shared)) # the new shares and root
        shared = self.vm.shared['shared_id']
        self.assertEquals('fake_shared', shared.name)
        # check that the uuid is stored in fs
        mdobj = self.main.fs.get_by_path(shared.path)
        self.assertEquals(shared.subtree, mdobj.node_id)

00191     def test_add_shared(self):
        """ Test VolumeManager.add_shared """
        path = os.path.join(self.vm.root.path, 'shared_path')
        self.main.fs.create(path, "")
        self.main.fs.set_node_id(path, 'node_id')
        mdid = self.main.fs.get_by_node_id("", 'node_id')

        # helper method, pylint: disable-msg=C0111
        def fake_create_share(node_id, user, name, access_level, marker):
            self.assertTrue(marker in self.vm.marker_share_map)
            self.vm.handle_AQ_CREATE_SHARE_OK(share_id='share_id',
                                              marker=marker)
        self.main.action_q.create_share = fake_create_share
        self.vm.create_share(path, 'fake_user', 'shared_name', 'View')

        self.assertTrue(self.vm.shared.get('share_id') is not None)
        share = self.vm.shared.get('share_id')
        self.assertEquals('fake_user', share.other_username)
        self.assertEquals('shared_name', share.name)
        self.assertEquals('View', share.access_level)
        self.assertEquals('node_id', share.subtree)
        self.assertEquals('share_id', share.id)

00214     def test_create_share(self):
        """ Test VolumeManager.create_share """
        path = os.path.join(self.vm.root.path, 'shared_path')
        self.main.fs.create(path, "")
        self.main.fs.set_node_id(path, 'node_id')
        mdid = self.main.fs.get_by_node_id("", 'node_id')

        # helper method, pylint: disable-msg=C0111
        def fake_create_share(node_id, user, name, access_level, marker):
            self.assertEquals('node_id', node_id)
            self.assertEquals('fake_user', user)
            self.assertEquals('shared_name', name)
            self.assertEquals('View', access_level)
            self.assertTrue(marker is not None)
            share = self.vm.marker_share_map[marker]
            self.assertEquals(path, share.path)
            self.assertEquals('View', share.access_level)
            self.assertEquals(marker, share.id)
            self.assertEquals('fake_user', share.other_username)
            self.assertEquals('node_id', share.subtree)

        self.main.action_q.create_share = fake_create_share
        self.vm.create_share(path, 'fake_user', 'shared_name', 'View')

00238     def test_create_share_error(self):
        """ Test VolumeManager.create_share """
        path = os.path.join(self.vm.root.path, 'shared_path')
        self.main.fs.create(path, "")
        self.main.fs.set_node_id(path, 'node_id')
        mdid = self.main.fs.get_by_node_id("", 'node_id')

        # helper method, pylint: disable-msg=C0111
        def fake_create_share(node_id, user, name, access_level, marker):
            self.vm.handle_AQ_CREATE_SHARE_ERROR(marker, 'a fake error')

        self.main.action_q.create_share = fake_create_share
        self.vm.create_share(path, 'fake_user', 'shared_name', 'View')

00252     def test_accept_share(self):
        """ Test the accept_share method. """
        d = defer.Deferred()
        self.vm.on_server_root('root_uuid')
        share_path = os.path.join(self.shares_dir, 'fake_share')
        share = Share(share_path, share_id='share_id', subtree="node_id")
        self.vm.add_share(share)
        self.assertIn(share.id, self.vm.shares)
        self.assertEquals(False, share.accepted)
        # helper method, pylint: disable-msg=C0111
        def answer_share(share_id, answer):
            reactor.callLater(0.2, d.callback, (share_id, answer))
            return d
        self.main.action_q.answer_share = answer_share
        def callback(result):
            share_id, answer = result
            self.assertEquals(share.id, share_id)
            self.assertEquals('Yes', answer)
        d.addCallback(callback)
        self.vm.accept_share(share.id, True)
        return d

00274     def test_handle_AQ_SHARES_LIST_shared_missing_md(self):
        """test the handling of the AQ_SHARE_LIST event, when the md
        isn't there yet.
        """
        shared_response = ShareResponse.from_params('shared_id', 'from_me',
                                                   'shared_uuid',
                                                   'fake_shared', 'myname',
                                                   'my_visible_name', 'yes',
                                                   'Modify')
        # initialize the the root
        self.vm.on_server_root('root_uuid')
        shared_dir = os.path.join(self.root_dir, 'shared_dir')
        response = ListShares(None)
        response.shares = [shared_response]
        self.vm.handle_AQ_SHARES_LIST(response)
        self.assertEquals(1, len(self.vm.shared)) # the new shares and root
        shared = self.vm.shared['shared_id']
        self.assertEquals('fake_shared', shared.name)
        # check that the uuid is stored in fs
        self.assertEquals(shared_response.subtree, shared.subtree)
        self.assertEquals(None, shared.path)

00296     def test_handle_SV_SHARE_ANSWERED(self):
        """ test the handling of the AQ_SHARE_ANSWERED. """
        path = os.path.join(self.vm.root.path, 'shared_path')
        self.main.fs.create(path, "")
        self.main.fs.set_node_id(path, 'node_id')
        mdid = self.main.fs.get_by_node_id("", 'node_id')
        # initialize the the root
        self.vm.on_server_root('root_uuid')
        # add the shared folder
        share = Share(path, share_id='share_id', access_level='View')
        self.vm.add_shared(share)
        self.assertEquals(False, self.vm.shared['share_id'].accepted)
        # check that a answer notify of a missing share don't blowup
        self.vm.handle_SV_SHARE_ANSWERED('share_id', 'Yes')
        self.assertEquals(True, self.vm.shared['share_id'].accepted)

00312     def test_handle_SV_SHARE_ANSWERED_missing(self):
        """ test the handling of the AQ_SHARE_ANSWERED. """
        path = os.path.join(self.vm.root.path, 'shared_path')
        self.main.fs.create(path, "")
        self.main.fs.set_node_id(path, 'node_id')
        mdid = self.main.fs.get_by_node_id("", 'node_id')
        # initialize the the root
        self.vm.on_server_root('root_uuid')
        self.assertNotIn('share_id', self.vm.shared)
        # check that a answer notify of a missing share don't blowup
        self.vm.handle_SV_SHARE_ANSWERED('share_id', 'Yes')
        self.assertNotIn('share_id', self.vm.shared)

    def test_delete_share(self):
        share_response = ShareResponse.from_params('share_id', 'to_me',
                                                   'fake_share_uuid',
                                                   'fake_share', 'username',
                                                   'visible_username', 'yes',
                                                   'View')
        share_response_1 = ShareResponse.from_params('share_id_1', 'to_me',
                                                   'fake_share_uuid_1',
                                                   'fake_share_1', 'username',
                                                   'visible_username', 'yes',
                                                   'View')
        # initialize the the root
        self.vm.on_server_root('root_uuid')
        response = ListShares(None)
        response.shares = [share_response, share_response_1]
        self.vm.handle_AQ_SHARES_LIST(response)
        self.assertEquals(3, len(self.vm.shares)) # the new shares and root
        # check that the share is in the shares dict
        self.assertTrue('share_id' in self.vm.shares)
        self.assertTrue('share_id_1' in self.vm.shares)
        share = self.vm.shares['share_id']
        self.assertEquals('fake_share', share.name)
        self.assertEquals('fake_share_uuid', share.subtree)
        share = self.vm.shares['share_id_1']
        self.assertEquals('fake_share_1', share.name)
        self.assertEquals('fake_share_uuid_1', share.subtree)
        # inject a new ListShares response
        new_response = ListShares(None)
        new_response.shares = [share_response]
        self.vm.handle_AQ_SHARES_LIST(new_response)
        # check that the missing share was removed
        self.assertTrue('share_id' in self.vm.shares)
        self.assertFalse('share_id_1' in self.vm.shares)
        share = self.vm.shares['share_id']
        self.assertEquals('fake_share', share.name)
        self.assertEquals('fake_share_uuid', share.subtree)
        self.assertEquals(None, self.vm.shares.get('share_id_1'))


00364 class VolumeManagerUnicodeTests(BaseVolumeManagerTests):
    """Tests for Volume Manager unicode capabilities."""

00367     def test_handle_SHARES_sharename(self):
        """test the handling of AQ_SHARE_LIST with non-ascii share name."""
        share_response = ShareResponse.from_params('share_id', 'to_me',
                                                   'fake_share_uuid',
                                                   u'montón', 'username',
                                                   'visible', 'yes',
                                                   'View')
        # initialize the the root
        self.vm.on_server_root('root_uuid')
        response = ListShares(None)
        response.shares = [share_response]
        self.vm.handle_AQ_SHARES_LIST(response)

        # check
        share = self.vm.shares['share_id']
        shouldbe_dir = os.path.join(self.shares_dir,
                                    u"montón".encode("utf8") + " from visible")
        self.assertEquals(shouldbe_dir, share.path)

00386     def test_handle_SHARES_visible_username(self):
        """test the handling of AQ_SHARE_LIST with non-ascii visible uname."""
        share_response = ShareResponse.from_params('share_id', 'to_me',
                                                   'fake_share_uuid',
                                                   'sharename', 'username',
                                                   u'Darío Toño', 'yes',
                                                   'View')
        # initialize the the root
        self.vm.on_server_root('root_uuid')
        response = ListShares(None)
        response.shares = [share_response]
        self.vm.handle_AQ_SHARES_LIST(response)

        # check
        share = self.vm.shares['share_id']
        shouldbe_dir = os.path.join(self.shares_dir,
                            "sharename from " + u"Darío Toño".encode("utf8"))
        self.assertEquals(shouldbe_dir, share.path)

00405     def test_handle_SV_SHARE_CHANGED_sharename(self):
        """test the handling of SV_SHARE_CHANGED for non-ascii share name."""
        share_holder = NotifyShareHolder.from_params('share_id', None,
                                                     u'año',
                                                     'test_username',
                                                     'visible', 'Modify')
        self.vm.on_server_root('root_uuid')
        self.vm.handle_SV_SHARE_CHANGED('changed', share_holder)
        shouldbe_dir = os.path.join(self.shares_dir,
                                    u"año".encode("utf8") + " from visible")
        self.assertEquals(shouldbe_dir, self.vm.shares['share_id'].path)

00417     def test_handle_SV_SHARE_CHANGED_visible(self):
        """test the handling of SV_SHARE_CHANGED for non-ascii visible name."""
        share_holder = NotifyShareHolder.from_params('share_id', None,
                                                     'share',
                                                     'test_username',
                                                     u'Ramón', 'Modify')
        self.vm.on_server_root('root_uuid')
        self.vm.handle_SV_SHARE_CHANGED('changed', share_holder)
        shouldbe_dir = os.path.join(self.shares_dir,
                                    "share from " + u"Ramón".encode("utf8"))
        self.assertEquals(shouldbe_dir, self.vm.shares['share_id'].path)



00431 class ShareShelfUpgradeTests(BaseTwistedTestCase):
    """ Tests for shares shelf upgrades"""

00434     def setUp(self):
        """ setup the test """
        BaseTwistedTestCase.setUp(self)
        self.root_dir = self.mktemp('root_dir')
        self.data_dir = self.mktemp('data_dir')
        self.shares_dir = self.mktemp('shares_dir')

00441     def tearDown(self):
        """ cleanup main and remove the temp dir """
        main = getattr(self, 'main', None)
        if main:
            main.shutdown()
        self.rmtree(self.root_dir)
        self.rmtree(self.data_dir)
        self.rmtree(self.shares_dir)
        return BaseTwistedTestCase.tearDown(self)

00451     def test_0_to_1(self):
        """ Test the upgrade from the first shelf layout version to v. 1"""
        # ensure a clean data_dir
        self.rmtree(self.data_dir)
        vm_data_dir = os.path.join(self.data_dir, 'vm')
        old_shelf = ShareFileShelf(vm_data_dir)
        # add the root_uuid key
        root_share = Share(self.root_dir)
        root_share.access_level = 'Modify'
        old_shelf[''] = root_share
        for idx in range(1, 10):
            old_shelf[str(uuid.uuid4())] = idx
        # ShareFileShelf.keys returns a generator
        old_keys = [key for key in old_shelf.keys()]
        self.assertEquals(10, len(old_keys))
        # we want to keep a refernece to main in order to shutdown
        # pylint: disable-msg=W0201
        self.main = FakeMain(self.root_dir, self.shares_dir, self.data_dir)
        new_keys = [new_key for new_key in self.main.vm.shares.keys()]
        self.assertEquals(10, len(new_keys))
        self.assertEquals(old_keys, new_keys)
        # check the old data is still there (in the backup)
        backup_shelf = ShareFileShelf(os.path.join(vm_data_dir, '0.bkp'))
        backup_keys = [key for key in backup_shelf.keys()]
        self.assertEquals(old_keys, backup_keys)
        self.assertEquals(new_keys, backup_keys)


Generated by  Doxygen 1.6.0   Back to index