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

test_fsm.py

#
# Author: Facundo Batista <facundo@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 File System Manager.'''

from __future__ import with_statement

import os
import shutil
import unittest
import time

from contrib.testing.testcase import (
    FakeVolumeManager,
)

from ubuntuone.syncdaemon.filesystem_manager import (
                            FileSystemManager, InconsistencyError)
from ubuntuone.syncdaemon.volume_manager import Share

TESTS_DIR = os.path.join(os.getcwd(), "tmp")

00037 class FSMTestCase(unittest.TestCase):
    """ Base test case for FSM """

00040     def setUp(self):
        """ Setup the test """
        unittest.TestCase.setUp(self)
        try:
            os.mkdir(TESTS_DIR)
        except OSError:
            # already there, remove it to clean and create again
            shutil.rmtree(TESTS_DIR)
            os.mkdir(TESTS_DIR)
        self.shares_dir = os.path.join(TESTS_DIR, 'shares')
        os.makedirs(self.shares_dir)
        self.root_dir = os.path.join(TESTS_DIR, 'root')
        os.makedirs(self.root_dir)
        self.fsmdir = os.path.join(TESTS_DIR, "fsmdir")
        self.fsm = FileSystemManager(self.fsmdir,
                                     FakeVolumeManager(self.root_dir))
        self.share = self.create_share('share', 'share_name',
                                            self.fsm, self.shares_dir)
        self.share_path = self.share.path

00060     def tearDown(self):
        """ Clean up the tests. """
        shutil.rmtree(TESTS_DIR)

    @staticmethod
00065     def create_share(share_id, share_name, fsm, shares_dir,
                     access_level='Modify'):
        """ creates a share """
        share_path = os.path.join(shares_dir, share_name)
        os.makedirs(share_path)
        share = Share(share_path, share_id=share_id,
                      access_level=access_level)
        fsm.vm.add_share(share)
        return share


00076 class StartupTests(unittest.TestCase):
    '''Test the basic startup behaviour.'''

00079     def setUp(self):
        """ Sets up a test. """
        try:
            os.mkdir(TESTS_DIR)
        except OSError:
            # already there, remove it to clean and create again
            shutil.rmtree(TESTS_DIR)
            os.mkdir(TESTS_DIR)

00088     def tearDown(self):
        '''Clean up the tests.'''
        shutil.rmtree(TESTS_DIR)

00092     def test_basic_startup(self):
        '''Test the init interface.'''
        # only one arg
        self.assertRaises(TypeError, FileSystemManager)
        self.assertRaises(TypeError, FileSystemManager, 1, 2)

        # that creates the dir
        fsmdir = os.path.join(TESTS_DIR, "a_fsmdir")
        fsm = FileSystemManager(fsmdir, FakeVolumeManager(fsmdir))
        self.assertTrue(os.path.exists(fsmdir))

00103     def test_complex_startup(self):
        '''Test startup after having data.'''
        # open an empty one
        fsmdir = os.path.join(TESTS_DIR, "fsmdir")
        fsm = FileSystemManager(fsmdir, FakeVolumeManager(fsmdir))
        share = FSMTestCase.create_share('share', 'share_name',
                                              fsm, fsmdir)
        self.assertEqual(fsm._idx_path, {})
        self.assertEqual(fsm._idx_node_id, {})

        # write some data, one with node_id
        path1 = os.path.join(share.path, 'path1')
        fsm.create(path1, "share")
        created_mdid1 = fsm._idx_path[path1]
        self.assertEqual(fsm._idx_path, {path1:created_mdid1})
        fsm.set_node_id(path1, "uuid1")
        self.assertEqual(fsm._idx_node_id, {("share","uuid1"):created_mdid1})

        # ...and one without
        path2 = os.path.join(share.path, 'path2')
        fsm.create(path2, "share")
        created_mdid2 = fsm._idx_path[path2]
        self.assertEqual(fsm._idx_path,
                         {path1:created_mdid1, path2:created_mdid2})

        # open a second one to see if everything is ok
        fsm = FileSystemManager(fsmdir, fsm.vm)
        self.assertEqual(fsm._idx_path,
                         {path1:created_mdid1, path2:created_mdid2})
        self.assertEqual(fsm._idx_node_id, {("share","uuid1"):created_mdid1})
        self.assertTrue(fsm.get_by_mdid(created_mdid1))
        self.assertTrue(fsm.get_by_mdid(created_mdid2))


00137 class CreationTests(FSMTestCase):
    '''Test the creation behaviour.'''

00140     def test_simple(self):
        '''Test simple creation.'''
        # create, but not twice
        path = os.path.join(self.share.path, 'path')
        self.fsm.create(path, "share")
        self.assertRaises(ValueError, self.fsm.create, path, "share")
        self.assertRaises(ValueError, self.fsm.create, path, "other")
        mdobj = self.fsm.get_by_path(path)
        self.assertEqual(mdobj.path, "path")
        self.assertEqual(mdobj.share_id, "share")
        when = mdobj.info.created
        now = time.time()
        self.assertTrue(now-3 <= when <= now) # 3 seconds test range

        # set uuid using valid path, but not twice
        self.fsm.set_node_id(path, "uuid")
        self.assertRaises(ValueError, self.fsm.set_node_id, path, "whatever")
        mdobj = self.fsm.get_by_path(path)
        when = mdobj.info.node_id_assigned
        now = time.time()
        self.assertTrue(now-3 <= when <= now) # 3 seconds test range

00162     def test_invalid_args(self):
        '''Test using invalid args in set_node_id.'''
        path = os.path.join(self.share.path, 'path')
        self.fsm.create(path, "share")

        # set uuid using an invalid path
        self.assertRaises(KeyError, self.fsm.set_node_id, "no-path", "whatevr")

        # set uuid using an invalid node_id
        self.assertRaises(ValueError, self.fsm.set_node_id, path, None)

00173     def test_no_twice(self):
        '''Test that assignments must be done once, even in different FSMs.'''
        # using the first FSM
        path = os.path.join(self.share.path, 'path')
        self.fsm.create(path, "share")
        self.fsm.set_node_id(path, "uuid")

        # opening another FSM
        fsm = FileSystemManager(self.fsmdir, self.fsm.vm)
        self.assertRaises(ValueError, fsm.create, path, "share")
        self.assertRaises(ValueError, fsm.set_node_id, path, "whatever")

00185     def test_fresh_metadata(self):
        '''Initing with nothing in the metadata, it should leave it in v1.'''
        fsm = FileSystemManager(self.fsmdir, self.fsm.vm)
        md_version = open(os.path.join(self.fsmdir, "metadata_version")).read()
        self.assertEqual(md_version, "2")

00191     def test_old_metadata_None(self):
        '''Initing with old metadata, it should leave it in v2.'''
        # create some stuff
        path = os.path.join(self.share.path, 'path')
        open(path, "w").close()
        mdid = self.fsm.create(path, "share")
        self.fsm.set_node_id(path, "uuid")

        # break the node on purpose
        real_mdobj = self.fsm.fs[mdid]
        del real_mdobj["stat"]
        real_mdobj["path"] = unicode(real_mdobj["path"])
        self.fsm.fs[mdid] = real_mdobj

        # delete the version that should have left the previous fsm
        version_file = os.path.join(self.fsmdir, "metadata_version")
        os.remove(version_file)

        # start up again, and check
        newfsm = FileSystemManager(self.fsmdir, self.fsm.vm)
        md_version = open(version_file).read()
        self.assertEqual(md_version, "2")
        newmdobj = newfsm.get_by_path(path)
        self.assertEqual(newmdobj.mdid, mdid)
        self.assertEqual(newmdobj.stat, os.stat(path))
        self.assertTrue(isinstance(newmdobj.path, str))

00218     def test_old_metadata_1(self):
        '''Initing with old metadata, it should leave it in v2.'''
        # create some stuff
        path1 = os.path.join(self.share.path, 'path1')
        path2 = os.path.join(self.share.path, 'path2')
        mdid1 = self.fsm.create(path1, "share")
        self.fsm.set_node_id(path1, "uuid1")
        mdid2 = self.fsm.create(path2, "share")
        self.fsm.set_node_id(path2, "uuid2")

        # break the node on purpose, with unicode valid and not
        real_mdobj = self.fsm.fs[mdid1]
        real_mdobj["path"] = unicode(real_mdobj["path"])
        self.fsm.fs[mdid1] = real_mdobj
        real_mdobj = self.fsm.fs[mdid2]
        real_mdobj["path"] = "asdas\x00\xff\xffasd"
        self.fsm.fs[mdid2] = real_mdobj

        # put the version file in 1
        version_file = os.path.join(self.fsmdir, "metadata_version")
        with open(version_file, "w") as fh:
            fh.write("1")

        # start up again, and check
        newfsm = FileSystemManager(self.fsmdir, self.fsm.vm)
        md_version = open(version_file).read()
        self.assertEqual(md_version, "2")
        newmdobj = newfsm.get_by_path(path1)
        self.assertEqual(newmdobj.mdid, mdid1)
        self.assertTrue(isinstance(newmdobj.path, str))
        self.assertEqual(1, len(newfsm._idx_node_id))



00252 class GetSetTests(FSMTestCase):
    '''Test the get/set interface.'''

00255     def test_bad_data(self):
        '''No such info is allowed as path.'''
        self.assertRaises(ValueError, self.fsm.create, "", "share")
        self.assertRaises(ValueError, self.fsm.create, " ", "share")

00260     def test_basic(self):
        '''Test basic retrieval.'''
        # write some data
        path1 = os.path.join(self.share_path, "path1")
        newmdid = self.fsm.create(path1, "share")
        mdid1 = self.fsm._idx_path[path1]
        self.fsm.set_node_id(path1, "uuid1")
        self.assertEqual(newmdid, mdid1)

        mdobj = self.fsm.get_by_mdid(mdid1)
        self.assertEqual(mdobj.node_id, "uuid1")
        self.assertEqual(mdobj.path, "path1")
        self.assertEqual(mdobj.share_id, "share")
        self.assertEqual(mdobj.local_hash, None)
        self.assertEqual(mdobj.server_hash, None)
        self.assertEqual(mdobj.info.is_partial, False)
        self.assertEqual(mdobj.is_dir, False)
        self.assertEqual(mdobj.mdid, mdid1)
        self.assertEqual(self.fsm.get_by_node_id("share", "uuid1"), mdobj)
        self.assertEqual(self.fsm.get_by_path(path1), mdobj)

        # write more data
        path2 = os.path.join(self.share_path, "path2")
        newmdid = self.fsm.create(path2, "share", is_dir=True)
        mdid2 = self.fsm._idx_path[path2]
        self.fsm.set_node_id(path2, "uuid2")
        self.assertEqual(newmdid, mdid2)

        # check that is not mixed
        mdobj = self.fsm.get_by_mdid(mdid1)
        self.assertEqual(mdobj.node_id, "uuid1")
        self.assertEqual(mdobj.path, "path1")
        self.assertEqual(mdobj.share_id, "share")
        self.assertEqual(mdobj.mdid, mdid1)
        self.assertEqual(mdobj.is_dir, False)
        self.assertEqual(self.fsm.get_by_node_id("share", "uuid1"), mdobj)
        self.assertEqual(self.fsm.get_by_path(path1), mdobj)
        mdobj = self.fsm.get_by_mdid(mdid2)
        self.assertEqual(mdobj.node_id, "uuid2")
        self.assertEqual(mdobj.path, "path2")
        self.assertEqual(mdobj.share_id, "share")
        self.assertEqual(mdobj.mdid, mdid2)
        self.assertEqual(mdobj.is_dir, True)
        self.assertEqual(self.fsm.get_by_node_id("share", "uuid2"), mdobj)
        self.assertEqual(self.fsm.get_by_path(path2), mdobj)

00306     def test_iteration(self):
        '''Test basic retrieval.'''
        # create a few objects
        mdids = []
        path_names = "path1 path2 path3".split()
        paths = []
        for path in path_names:
            path = os.path.join(self.share.path, path)
            paths.append(path)
            mdid = self.fsm.create(path, "share")
            mdids.append(mdid)
        mdids.sort()

        # get them
        retrieved_mdids = []
        retrieved_paths = []
        for mdobj in self.fsm.get_mdobjs_by_share_id("share"):
            retrieved_mdids.append(mdobj.mdid)
            retrieved_paths.append(mdobj.path)
        retrieved_mdids.sort()
        retrieved_paths.sort()

        # check them
        self.assertEqual(mdids, retrieved_mdids)
        self.assertEqual(path_names, retrieved_paths)

00332     def test_getacopy(self):
        '''Test that we receive only a copy.'''
        # write some data
        path = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(path, "share")
        self.fsm.set_node_id(path, "uuid")
        self.fsm.set_by_mdid(mdid, newarg="foo")

        # test getting a copy with mdid
        d = self.fsm.get_by_mdid(mdid)
        d.newarg = "bar"
        d = self.fsm.get_by_mdid(mdid)
        self.assertEqual(d.newarg, "foo")

        # test getting a copy with uuid
        d = self.fsm.get_by_node_id("share", "uuid")
        d.newarg = "bar"
        d = self.fsm.get_by_node_id("share", "uuid")
        self.assertEqual(d.newarg, "foo")

        # test getting a copy with path
        d = self.fsm.get_by_path(path)
        d.newarg = "bar"
        d = self.fsm.get_by_path(path)
        self.assertEqual(d.newarg, "foo")

00358     def test_get_raises(self):
        '''Test that we get an exception if the object is not there.'''
        # write some data
        path = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(path, "share")
        self.fsm.set_node_id(path, "uuid")

        # with mdid, ok and bad
        self.fsm.get_by_mdid(mdid)
        self.assertRaises(KeyError, self.fsm.get_by_mdid, "no-such-key")

        # with uuid, ok and bad
        self.fsm.get_by_node_id("share", "uuid")
        self.assertRaises(KeyError, self.fsm.get_by_node_id,
                          "share", "no-such-key")
        self.assertRaises(ValueError, self.fsm.get_by_node_id,
                          "share", None)

        # with path, ok and bad
        self.fsm.get_by_path(path)
        self.assertRaises(KeyError, self.fsm.get_by_path, "no-such-key")

00380     def test_setters_simple(self):
        '''Test that setters work.'''
        # create some data
        path = os.path.join(self.share.path, 'path')
        mdid = self.fsm.create(path, "share")
        self.fsm.set_node_id(path, "uuid")

        # test using mdid
        self.fsm.set_by_mdid(mdid, foo="foo1")
        self.fsm.set_by_mdid(mdid, bar="bar1", baz="baz1")
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertEqual(mdobj.foo, "foo1")
        self.assertEqual(mdobj.bar, "bar1")
        self.assertEqual(mdobj.baz, "baz1")

        # test using uuid
        self.assertRaises(ValueError, self.fsm.set_by_node_id, None, "sh", f=3)
        self.fsm.set_by_node_id("uuid", "share", foo="foo2")
        self.fsm.set_by_node_id("uuid", "share", bar="bar2", baz="baz2")
        mdobj = self.fsm.get_by_node_id("share", "uuid")
        self.assertEqual(mdobj.foo, "foo2")
        self.assertEqual(mdobj.bar, "bar2")
        self.assertEqual(mdobj.baz, "baz2")

        # test using path
        self.fsm.set_by_path(path, foo="foo3")
        self.fsm.set_by_path(path, bar="bar3", baz="baz3")
        mdobj = self.fsm.get_by_path(path)
        self.assertEqual(mdobj.foo, "foo3")
        self.assertEqual(mdobj.bar, "bar3")
        self.assertEqual(mdobj.baz, "baz3")

00412     def test_setters_mixed(self):
        '''Test the setters using different combinations.'''
        # create some data
        path = os.path.join(self.share.path, 'path')
        mdid = self.fsm.create(path, "share")
        self.fsm.set_node_id(path, "uuid")

        # set with mdid, get with uuid and path
        self.fsm.set_by_mdid(mdid, foo="foo1")
        self.assertEqual(self.fsm.get_by_node_id("share", "uuid").foo, "foo1")
        self.assertEqual(self.fsm.get_by_path(path).foo, "foo1")

        # set with uuid, get with mdid and path
        self.fsm.set_by_node_id("uuid", "share", foo="foo2")
        self.assertEqual(self.fsm.get_by_mdid(mdid).foo, "foo2")
        self.assertEqual(self.fsm.get_by_path(path).foo, "foo2")

        # set with path, get with uuid and mdid
        self.fsm.set_by_path(path, foo="foo3")
        self.assertEqual(self.fsm.get_by_node_id("share", "uuid").foo, "foo3")
        self.assertEqual(self.fsm.get_by_mdid(mdid).foo, "foo3")

00434     def test_setters_raises(self):
        '''Test that setters raise ok.'''
        # create some data
        path = os.path.join(self.share.path, 'path')
        mdid = self.fsm.create(path, "share")
        self.fsm.set_node_id(path, "uuid")

        # test with bad values
        self.assertRaises(KeyError, self.fsm.get_by_node_id, "share", "bad")
        self.assertRaises(KeyError, self.fsm.get_by_mdid, "bad-value")
        self.assertRaises(KeyError, self.fsm.get_by_path, "bad-value")

00446     def test_setting_forbidden_values(self):
        '''Test trying to set forbidden values.'''
        # create some data
        path = os.path.join(self.share.path, 'path')
        mdid = self.fsm.create(path, "share")
        self.fsm.set_node_id(path, "uuid")

        # test with forbidden uuid
        self.assertRaises(ValueError, self.fsm.set_by_mdid, mdid, node_id="-")
        self.assertRaises(ValueError, self.fsm.set_by_path, path, node_id="")

        # test with forbidden path
        self.assertRaises(ValueError, self.fsm.set_by_node_id, "uuid", "share",
                          path="-")
        self.assertRaises(ValueError, self.fsm.set_by_mdid, mdid, path="-")

        # test with forbidden info
        self.assertRaises(ValueError, self.fsm.set_by_node_id, "uuid", "share",
                          info="-")
        self.assertRaises(ValueError, self.fsm.set_by_mdid, mdid, info="-")
        self.assertRaises(ValueError, self.fsm.set_by_path, path, info="-")

        # test with forbidden stat
        self.assertRaises(ValueError, self.fsm.set_by_node_id, "uuid", "share",
                          stat="-")
        self.assertRaises(ValueError, self.fsm.set_by_mdid, mdid, stat="-")
        self.assertRaises(ValueError, self.fsm.set_by_path, path, stat="-")

        # test with forbidden share
        self.assertRaises(ValueError, self.fsm.set_by_mdid, mdid, share_id="-")
        self.assertRaises(ValueError, self.fsm.set_by_path, path, share_id="-")

        # test with forbidden mdid
        self.assertRaises(ValueError, self.fsm.set_by_node_id, "uuid", "share",
                          mdid="-")
        self.assertRaises(ValueError, self.fsm.set_by_path, path, mdid="-")

        # test with forbidden is_dir
        self.assertRaises(ValueError, self.fsm.set_by_mdid, mdid, is_dir="-")
        self.assertRaises(ValueError, self.fsm.set_by_path, path, is_dir="-")
        self.assertRaises(ValueError,
                          self.fsm.set_by_node_id, "uuid", "share", is_dir="-")

00489     def test_setting_forbidden_mixed(self):
        '''Test that when trying with forbidden, nothing happens at all.'''
        # create some data
        path = os.path.join(self.share.path, 'path')
        mdid = self.fsm.create(path, "share")
        self.fsm.set_node_id(path, "uuid")

        # test with mixed stuff
        self.assertRaises(TypeError, self.fsm.set_by_node_id, info="n", foo="")
        self.assertRaises(TypeError, self.fsm.set_by_mdid, path="nop", bar="?")
        self.assertRaises(TypeError, self.fsm.set_by_path, node_id="n", baz="")

        # see that it still is unchanged
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertEqual(mdobj.path, "path")
        self.assertEqual(mdobj.node_id, "uuid")

00506     def test_get_root(self):
        """ Test that the root of a share is stored properly. """
        # write some data
        mdid = self.fsm.create(self.share_path, "share")
        self.fsm.set_node_id(self.share_path, "uuid")

        # with path, ok and bad
        self.fsm.get_by_path(self.share_path)
        self.assertRaises(KeyError, self.fsm.get_by_path, "no-such-key")

        # check the path stored in the mdobj
        mdobj = self.fsm.get_by_node_id("share", "uuid")
        self.assertEquals(self.share_path, mdobj.path)

00520     def test_get_all_by_share(self):
        """ Test that it returns all the mdids in a share. """
        # create the shares
        share1 = self.create_share('share_id1', 'share_name1',  self.fsm,
                                       self.shares_dir, access_level='View')
        share2 = self.create_share('share_id2', 'share_name2',  self.fsm,
                                       self.shares_dir, access_level='View')
        mdid1 = self.fsm.create(share1.path, "share_id1", is_dir=True)
        self.fsm.set_node_id(share1.path, "uuid1")
        mdid2 = self.fsm.create(share2.path, "share_id2", is_dir=True)
        self.fsm.set_node_id(share2.path, "uuid2")

        # create some nodes in share 1
        path3 = os.path.join(share1.path, "a")
        mdid3 = self.fsm.create(path3, "share_id1", is_dir=True)
        self.fsm.set_node_id(path3, "uuid3")
        path4 = os.path.join(share1.path, "a", "b")
        mdid4 = self.fsm.create(path4, "share_id1")
        self.fsm.set_node_id(path4, "uuid4")
        path5 = os.path.join(share1.path, "c")
        mdid5 = self.fsm.create(path5, "share_id1")
        self.fsm.set_node_id(path5, "uuid5")

        # create some nodes in share 2
        path6 = os.path.join(share2.path, "a")
        mdid6 = self.fsm.create(path6, "share_id2", is_dir=True)
        self.fsm.set_node_id(path6, "uuid6")
        path7 = os.path.join(share2.path, "c")
        mdid7 = self.fsm.create(path7, "share_id2")
        self.fsm.set_node_id(path7, "uuid7")

        # tricky: node without node_id yet
        path8 = os.path.join(share2.path, "d")
        mdid8 = self.fsm.create(path8, "share_id2")


        # get them
        all = set()
        for mdobj in self.fsm.get_mdobjs_by_share_id("share_id1"):
            all.add(mdobj.mdid)
        self.assertTrue(mdid3 in all)
        self.assertTrue(mdid4 in all)
        self.assertTrue(mdid5 in all)
        self.assertTrue(mdid6 not in all)
        self.assertTrue(mdid7 not in all)
        self.assertTrue(mdid8 not in all)

        all = set()
        for mdobj in self.fsm.get_mdobjs_by_share_id("share_id2"):
            all.add(mdobj.mdid)
        self.assertTrue(mdid3 not in all)
        self.assertTrue(mdid4 not in all)
        self.assertTrue(mdid5 not in all)
        self.assertTrue(mdid6 in all)
        self.assertTrue(mdid7 in all)
        self.assertTrue(mdid8 in all)

00577     def test_get_all_by_share_mixed(self):
        """Test that it returns all the mdids in a share with mixed nodes."""
        # create the shares
        share = self.create_share('share_id', 'sharetest',  self.fsm,
                                       self.shares_dir, access_level='View')
        self.fsm.create(share.path, "share_id", is_dir=True)
        self.fsm.set_node_id(share.path, "uuid")

        # create one real node...
        path1 = os.path.join(share.path, "a")
        mdid1 = self.fsm.create(path1, "share_id", is_dir=True)
        self.fsm.set_node_id(path1, "uuid1")

        # ...and two without node_id's
        path2 = os.path.join(share.path, "b")
        mdid2 = self.fsm.create(path2, "share_id")
        path3 = os.path.join(share.path, "c")
        mdid3 = self.fsm.create(path3, "share_id")

        # get them
        all = set()
        for mdobj in self.fsm.get_mdobjs_by_share_id("share_id"):
            all.add(mdobj.mdid)
        self.assertTrue(mdid1 in all)
        self.assertTrue(mdid2 in all)
        self.assertTrue(mdid3 in all)


00605 class StatTests(FSMTestCase):
    '''Test all the behaviour regarding the stats.'''

00608     def create_node(self, name):
        '''Creates a node.'''
        path = os.path.join(self.share.path, name)
        mdid = self.fsm.create(path, self.share.id)
        self.fsm.set_node_id(path, "uuid1")
        mdobj = self.fsm.get_by_mdid(mdid)
        return mdobj

00616     def test_create_nofile(self):
        '''Test creation when there's no file.'''
        mdobj = self.create_node("foo")
        self.assertEqual(mdobj.stat, None)

00621     def test_create_file(self):
        '''Test creation when there's a file.'''
        # file
        path = os.path.join(self.share.path, "thisfile")
        open(path, "w").close()
        mdobj = self.create_node("thisfile")
        self.assertEqual(mdobj.stat, os.stat(path))

        # dir
        path = os.path.join(self.share.path, "thisdir")
        os.mkdir(path)
        mdobj = self.create_node("thisdir")
        self.assertEqual(mdobj.stat, os.stat(path))

00635     def test_refresh(self):
        '''Test that refresh_stat works.'''
        path = os.path.join(self.share.path, "thisfile")
        open(path, "w").close()
        mdobj = self.create_node("thisfile")
        mdid = mdobj.mdid
        oldstat = os.stat(path)
        self.assertEqual(mdobj.stat, oldstat)

        # touch the file, it's not automagically updated
        open(path, "w").close()
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertEqual(mdobj.stat, oldstat)

        # it's updated when asked
        self.fsm.refresh_stat(path)
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertEqual(mdobj.stat, os.stat(path))

00654     def test_commit_partial(self):
        '''Test that it's updated in the commit.'''
        path = os.path.join(self.share.path, "thisfile")
        open(path, "w").close()
        mdobj = self.create_node("thisfile")
        mdid = mdobj.mdid
        oldstat = os.stat(path)
        self.assertEqual(mdobj.stat, oldstat)

        # create a partial
        self.fsm.create_partial(mdobj.node_id, mdobj.share_id)
        fh = self.fsm.get_partial_for_writing(mdobj.node_id, mdobj.share_id)
        fh.write("foobar")
        fh.close()
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertEqual(mdobj.stat, oldstat)

        # commit the partial
        self.fsm.commit_partial(mdobj.node_id, mdobj.share_id, "localhash")
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertEqual(mdobj.stat, os.stat(path))

00676     def test_move(self):
        '''Test that move refreshes stat.'''
        path1 = os.path.join(self.share.path, "thisfile1")
        path2 = os.path.join(self.share.path, "thisfile2")
        open(path1, "w").close()
        mdobj = self.create_node(path1)
        self.assertEqual(mdobj.stat, os.stat(path1))

        # move
        self.fsm.move_file("share", path1, path2)

        # check
        mdobj = self.fsm.get_by_path(path2)
        self.assertEqual(mdobj.stat, os.stat(path2))

00691     def test_move_overwriting(self):
        '''Test that move refreshes stat when overwrites other file.'''
        path1 = os.path.join(self.share.path, "thisfile1")
        path2 = os.path.join(self.share.path, "thisfile2")
        open(path1, "w").close()
        open(path2, "w").close()
        mdobj1 = self.create_node(path1)
        mdobj2 = self.create_node(path2)
        self.assertEqual(mdobj1.stat, os.stat(path1))
        self.assertEqual(mdobj2.stat, os.stat(path2))

        # move
        self.fsm.move_file("share", path1, path2)

        # check
        self.assertRaises(KeyError, self.fsm.get_by_path, path1)
        mdobj2 = self.fsm.get_by_path(path2)
        self.assertEqual(mdobj2.stat, os.stat(path2))

00710     def test_update_stat(self):
        '''Test that update_stat works.'''
        path = os.path.join(self.share.path, "thisfile")
        open(path, "w").close()
        mdobj = self.create_node("thisfile")
        mdid = mdobj.mdid
        oldstat = os.stat(path)
        self.assertEqual(mdobj.stat, oldstat)

        # touch the file, it's not automagically updated
        open(path, "w").close()
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertEqual(mdobj.stat, oldstat)

        # it's updated when asked, even if it's an old stat
        self.fsm.update_stat(mdid, oldstat)
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertEqual(mdobj.stat, oldstat)


00730 class PartialTests(FSMTestCase):
    '''Test all the .partial nitty gritty.'''

00733     def test_create_file(self):
        '''Test create .partial for a file.'''
        testfile = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(testfile, "share")
        self.fsm.set_node_id(testfile, "uuid")

        # create partial ok
        self.fsm.create_partial("uuid", "share")
        self.assertTrue(self.fsm.get_by_mdid(mdid).info.is_partial)
        self.assertTrue(os.path.exists(testfile + ".partial"))
        mdobj = self.fsm.get_by_mdid(mdid)
        when = mdobj.info.last_partial_created
        now = time.time()
        self.assertTrue(now-3 <= when <= now) # 3 seconds test range

        # invalid uuid
        self.assertRaises(KeyError, self.fsm.create_partial, "foo", "share")
        self.assertRaises(ValueError, self.fsm.create_partial, None, "share")

        # already has a partial!
        self.assertRaises(ValueError, self.fsm.create_partial, "uuid", "share")

00755     def test_commit_file(self):
        '''Test commit the .partial for a file, after a successful download.'''
        testfile = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(testfile, "share")
        self.fsm.set_node_id(testfile, "uuid")

        # create partial
        self.fsm.create_partial("uuid", "share")
        self.assertTrue(self.fsm.get_by_mdid(mdid).info.is_partial)
        with open(testfile + ".partial", "w") as fh:
            fh.write("test info!")

        # commit partial, and check that the file is moved, and metadata is ok
        self.fsm.commit_partial("uuid", "share", local_hash=9876)
        self.assertFalse(os.path.exists(testfile + ".partial"))
        with open(testfile) as fh:
            in_file = fh.read()
        self.assertEqual(in_file, "test info!")
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertFalse(mdobj.info.is_partial)
        self.assertEqual(mdobj.local_hash, 9876)
        when = mdobj.info.last_downloaded
        now = time.time()
        self.assertTrue(now-3 <= when <= now) # 3 seconds test range

        # invalid uuid
        self.assertRaises(KeyError, self.fsm.commit_partial,
                          "foo", "share", 123)
        self.assertRaises(ValueError, self.fsm.commit_partial,
                          None, "share", 123)
        # it has no partial!
        self.assertRaises(ValueError, self.fsm.commit_partial,
                          "uuid", "share", 1)

00789     def test_remove_file(self):
        '''Test removing the .partial for a file, because a bad download.'''
        testfile = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(testfile, "share")
        self.fsm.set_node_id(testfile, "uuid")

        # create partial
        self.fsm.create_partial("uuid", "share")
        self.assertTrue(self.fsm.get_by_mdid(mdid).info.is_partial)
        with open(testfile + ".partial", "w") as fh:
            fh.write("test info!")
        with open(testfile, "w") as fh:
            fh.write("previous stuff!")

        # remove partial, and check that the file is gone, and metadata is ok
        self.fsm.remove_partial("uuid", "share")
        self.assertFalse(os.path.exists(testfile + ".partial"))
        with open(testfile) as fh:
            in_file = fh.read()
        self.assertEqual(in_file, "previous stuff!")
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertFalse(mdobj.info.is_partial)
        when = mdobj.info.last_partial_removed
        now = time.time()
        self.assertTrue(now-3 <= when <= now) # 3 seconds test range

        # invalid uuid
        self.assertRaises(KeyError, self.fsm.remove_partial, "foo", "share")
        self.assertRaises(ValueError, self.fsm.remove_partial, None, "share")

        # it has no partial!
        self.fsm.remove_partial("uuid", "share")


00823     def test_create_dir_previous(self):
        '''Test create .partial for a dir when the dir existed.'''
        testdir = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(testdir, "share", is_dir=True)
        self.fsm.set_node_id(testdir, "uuid")
        os.mkdir(testdir)

        # create partial ok
        self.fsm.create_partial("uuid", "share")
        self.assertTrue(self.fsm.get_by_mdid(mdid).info.is_partial)
        self.assertTrue(os.path.exists(os.path.join(testdir, ".partial")))
        mdobj = self.fsm.get_by_mdid(mdid)
        when = mdobj.info.last_partial_created
        now = time.time()
        self.assertTrue(now-3 <= when <= now) # 3 seconds test range

        # invalid uuid
        self.assertRaises(KeyError, self.fsm.create_partial, "foo", "share")

        # already has a partial!
        self.assertRaises(ValueError, self.fsm.create_partial, "uuid", "share")

00845     def test_create_dir_notprevious(self):
        '''Test create .partial for a dir when the dir didn't exist.'''
        testdir = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(testdir, "share", is_dir=True)
        self.fsm.set_node_id(testdir, "uuid")

        # create partial ok
        self.fsm.create_partial("uuid", "share")
        self.assertTrue(self.fsm.get_by_mdid(mdid).info.is_partial)
        self.assertTrue(os.path.exists(testdir))
        self.assertTrue(os.path.exists(os.path.join(testdir, ".partial")))
        mdobj = self.fsm.get_by_mdid(mdid)
        when = mdobj.info.last_partial_created
        now = time.time()
        self.assertTrue(now-3 <= when <= now) # 3 seconds test range

        # invalid uuid
        self.assertRaises(KeyError, self.fsm.create_partial, "foo", "share")

        # already has a partial!
        self.assertRaises(ValueError, self.fsm.create_partial, "uuid", "share")

00867     def test_commit_dir(self):
        '''Test commit the .partial for a dir, after a successful download.'''
        testdir = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(testdir, "share", is_dir=True)
        self.fsm.set_node_id(testdir, "uuid")

        # create partial
        self.fsm.create_partial("uuid", "share")
        self.assertTrue(self.fsm.get_by_mdid(mdid).info.is_partial)

        # commit is forbidden for directories
        self.assertRaises(ValueError, self.fsm.commit_partial, "uuid", "share",
                          local_hash=9876)

00881     def test_remove_dir(self):
        '''Test removing the .partial for a dir, because a bad download.'''
        testdir = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(testdir, "share", is_dir=True)
        self.fsm.set_node_id(testdir, "uuid")

        # create partial
        self.fsm.create_partial("uuid", "share")
        self.assertTrue(self.fsm.get_by_mdid(mdid).info.is_partial)

        # remove partial, and check that the file is gone, and metadata is ok
        self.fsm.remove_partial("uuid", "share")
        self.assertFalse(os.path.exists(os.path.join(testdir, ".partial")))
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertFalse(mdobj.info.is_partial)
        when = mdobj.info.last_partial_removed
        now = time.time()
        self.assertTrue(now-3 <= when <= now) # 3 seconds test range

        # invalid uuid
        self.assertRaises(KeyError, self.fsm.remove_partial, "foo", "share")

        # it has no partial!
        self.fsm.remove_partial("uuid", "share")


00907 class FileHandlingTests(FSMTestCase):
    '''Test the file handling services.'''

00910     def test_move_to_conflict(self):
        '''Test that the conflict stuff works.'''
        testfile = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(testfile, "share")
        self.fsm.set_node_id(testfile, "uuid")
        with open(testfile, "w") as fh:
            fh.write("test!")

        # move first time
        self.fsm.move_to_conflict(mdid)
        self.assertFalse(os.path.exists(testfile))
        with open(testfile + ".conflict") as fh:
            in_file = fh.read()
        self.assertEqual(in_file, "test!")
        mdobj = self.fsm.get_by_mdid(mdid)
        when = mdobj.info.last_conflicted
        now = time.time()
        self.assertTrue(now-3 <= when <= now) # 3 seconds test range

        # move second time, start the .N serie
        with open(testfile, "w") as fh:
            fh.write("test 1!")
        self.fsm.move_to_conflict(mdid)
        self.assertFalse(os.path.exists(testfile))
        with open(testfile + ".conflict.1") as fh:
            in_file = fh.read()
        self.assertEqual(in_file, "test 1!")

        # create a few more, test a higher one
        open(testfile + ".conflict.2", "w").close()
        open(testfile + ".conflict.3", "w").close()
        open(testfile + ".conflict.4", "w").close()
        open(testfile + ".conflict.5", "w").close()
        with open(testfile, "w") as fh:
            fh.write("test 6!")
        self.fsm.move_to_conflict(mdid)
        self.assertFalse(os.path.exists(testfile))
        with open(testfile + ".conflict.6") as fh:
            in_file = fh.read()
        self.assertEqual(in_file, "test 6!")

        # invalid uuid
        self.assertRaises(KeyError, self.fsm.move_to_conflict, "no-such-mdid")

00954     def test_upload_finished(self):
        '''Test upload finished.'''
        path = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(path, "share")
        self.fsm.set_node_id(path, "uuid")

        # finish the upload!
        self.fsm.upload_finished(mdid, server_hash=1234567890)
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertEqual(mdobj.server_hash, 1234567890)
        when = mdobj.info.last_uploaded
        now = time.time()
        self.assertTrue(now-3 <= when <= now) # 3 seconds test range

        # invalid mdid
        self.assertRaises(KeyError, self.fsm.upload_finished,
                          "no-such-mdid", 123)

        # bad arguments
        self.assertRaises(TypeError, self.fsm.upload_finished, mdid)

00975     def test_move_file_withfile(self):
        '''Test that a file is moved from one point to the other.'''
        testfile = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(testfile, "share")
        self.fsm.set_node_id(testfile, "uuid")
        with open(testfile, "w") as fh:
            fh.write("test!")

        # move the file
        to_path = os.path.join(self.share_path, "path2")
        self.fsm.move_file("share", testfile, to_path)
        self.assertFalse(os.path.exists(testfile))
        with open(to_path) as fh:
            in_file = fh.read()
        self.assertEqual(in_file, "test!")
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertEqual(mdobj.info.last_moved_from, testfile)
        when = mdobj.info.last_moved_time
        now = time.time()
        self.assertTrue(now-3 <= when <= now) # 3 seconds test range

        # move again, to a directory
        from_path = to_path
        os.mkdir(os.path.join(self.share_path, "testdir"))
        to_path = os.path.join(self.share_path, "testdir", "path3")
        self.fsm.move_file("share", from_path, to_path)
        self.assertFalse(os.path.exists(from_path))
        with open(to_path) as fh:
            in_file = fh.read()
        self.assertEqual(in_file, "test!")

        # invalid path
        self.assertRaises(KeyError, self.fsm.move_file,
                          "share", "no-path", "dest")

        # other share
        self.assertRaises(KeyError, self.fsm.move_file,
                                            "othershare", testfile, to_path)

        # invalid args
        self.assertRaises(TypeError, self.fsm.move_file, "one-path")

01017     def test_move_file_overwrite(self):
        '''Test that a file is moved over other one.'''
        testfile1 = os.path.join(self.share_path, "path1")
        mdid1 = self.fsm.create(testfile1, "share")
        self.fsm.set_node_id(testfile1, "uuid1")
        with open(testfile1, "w") as fh:
            fh.write("test 1")

        testfile2 = os.path.join(self.share_path, "path2")
        mdid2 = self.fsm.create(testfile2, "share")
        self.fsm.set_node_id(testfile2, "uuid2")
        with open(testfile2, "w") as fh:
            fh.write("test 2")

        # move the file
        self.fsm.move_file("share", testfile1, testfile2)
        self.assertFalse(os.path.exists(testfile1))
        with open(testfile2) as fh:
            in_file = fh.read()
        self.assertEqual(in_file, "test 1")
        mdobj = self.fsm.get_by_mdid(mdid1)
        self.assertEqual(mdobj.path, "path2")
        mdobj = self.fsm.get_by_path(testfile2)
        self.assertEqual(mdobj.mdid, mdid1)

        # check that the info for the overwritten one is gone
        self.assertRaises(KeyError, self.fsm.get_by_mdid, mdid2)
        self.assertRaises(KeyError, self.fsm.get_by_node_id, "share", "uuid2")

01046     def test_move_file_withdir(self):
        '''Test that a dir is moved from one point to the other.'''
        from_path = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(from_path, "share", is_dir=True)
        self.fsm.set_node_id(from_path, "uuid")

        # move the file
        os.mkdir(from_path)
        to_path = os.path.join(self.share_path, "path2")
        self.fsm.move_file("share", from_path, to_path)
        self.assertFalse(os.path.exists(from_path))
        self.assertTrue(os.path.exists(to_path))
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertEqual(mdobj.info.last_moved_from, from_path)
        when = mdobj.info.last_moved_time
        now = time.time()
        self.assertTrue(now-3 <= when <= now) # 3 seconds test range

        # move again, to a directory
        from_path = to_path
        os.mkdir(os.path.join(self.share_path, "testdir"))
        to_path = os.path.join(self.share_path, "testdir", "path3")
        self.fsm.move_file("share", from_path, to_path)
        self.assertFalse(os.path.exists(from_path))
        self.assertTrue(os.path.exists(to_path))
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertEqual(mdobj.info.last_moved_from, from_path)
        when = mdobj.info.last_moved_time
        now = time.time()
        self.assertTrue(now-3 <= when <= now) # 3 seconds test range

01077     def test_move_file_withfulldir(self):
        '''Test that a dir is moved from even having a file inside.'''
        # the containing dir
        from_path = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(from_path, "share", is_dir=True)
        self.fsm.set_node_id(from_path, "uuid")
        os.mkdir(from_path)

        # the file outside, with a similar name just to confuse
        otherfile = os.path.join(self.share_path, "pa")
        self.fsm.create(otherfile, "share", is_dir=False)
        self.fsm.set_node_id(otherfile, "otheruuid")
        open(otherfile, "w").close()

        # the file inside
        filepath = os.path.join(from_path, "file.txt")
        fileid = self.fsm.create(filepath, "share", is_dir=False)
        self.fsm.set_node_id(filepath, "fileuuid")
        open(filepath, "w").close()

        # move the dir
        to_path = os.path.join(self.share_path, "path2")
        self.fsm.move_file("share", from_path, to_path)
        self.assertFalse(os.path.exists(from_path))
        self.assertTrue(os.path.exists(to_path))
        mdobj = self.fsm.get_by_mdid(mdid)
        self.assertEqual(mdobj.info.last_moved_from, from_path)
        when = mdobj.info.last_moved_time
        now = time.time()
        self.assertTrue(now-3 <= when <= now) # 3 seconds test range

        # check that file inside is ok
        newfilepath = os.path.join(to_path, "file.txt")
        self.assertFalse(os.path.exists(filepath))
        self.assertTrue(os.path.exists(newfilepath))
        mdobj = self.fsm.get_by_path(newfilepath)
        self.assertEqual(mdobj.mdid, fileid)
        self.assertEqual(mdobj.path, "path2/file.txt")

        # check the outer file
        self.assertTrue(os.path.exists(otherfile))
        mdobj = self.fsm.get_by_path(otherfile)
        self.assertEqual(mdobj.path, "pa")

01121     def test_delete_file(self):
        '''Test that a file is deleted.'''
        testfile = os.path.join(self.share_path, "path")
        open(testfile, "w").close()
        mdid = self.fsm.create(testfile, "share")
        self.fsm.set_node_id(testfile, "uuid")

        # delete the file
        self.fsm.delete_file(testfile)
        self.assertFalse(os.path.exists(testfile))
        self.assertRaises(KeyError, self.fsm.get_by_mdid, mdid)
        self.assertRaises(KeyError, self.fsm.get_by_path, testfile)
        self.assertRaises(KeyError, self.fsm.get_by_node_id, "share", "uuid")

01135     def test_delete_dir(self):
        '''Test that a dir is deleted.'''
        testdir = os.path.join(self.share.path, "path")
        os.mkdir(testdir)
        mdid = self.fsm.create(testdir, "share", is_dir=True)
        self.fsm.set_node_id(testdir, "uuid")

        # try to delete the dir, but has files on it
        open(os.path.join(testdir, "foo"), "w").close()
        self.assertRaises(OSError, self.fsm.delete_file, testdir)
        self.assertEqual(self.fsm.get_by_mdid(mdid).path, "path")
        self.assertEqual(self.fsm.get_by_path(testdir).path, "path")
        self.assertEqual(self.fsm.get_by_node_id("share", "uuid").path, "path")
        os.remove(os.path.join(testdir, "foo"))

        # really delete the dir
        self.fsm.delete_file(testdir)
        self.assertFalse(os.path.exists(testdir))
        self.assertRaises(KeyError, self.fsm.get_by_mdid, mdid)
        self.assertRaises(KeyError, self.fsm.get_by_path, testdir)
        self.assertRaises(KeyError, self.fsm.get_by_node_id, "share", "uuid")


01158 class SyntheticInfoTests(FSMTestCase):
    '''Test the methods that generates attributes.'''

01161     def test_has_metadata(self):
        '''Test the has_metadata option.'''
        # not yet
        self.assertFalse(self.fsm.has_metadata(path="path"))
        self.assertFalse(self.fsm.has_metadata(node_id="uuid",
                                                            share_id="share"))
        self.assertFalse(self.fsm.has_metadata(mdid="garbage"))

        # path created
        path = os.path.join(self.share.path, 'path')
        mdid = self.fsm.create(path, "share")
        self.assertTrue(self.fsm.has_metadata(path=path))
        self.assertFalse(self.fsm.has_metadata(node_id="uuid",
                                               share_id="share"))
        self.assertTrue(self.fsm.has_metadata(mdid=mdid))

        # uuid set
        self.fsm.set_node_id(path, "uuid")
        self.assertTrue(self.fsm.has_metadata(path=path))
        self.assertTrue(self.fsm.has_metadata(node_id="uuid",
                                              share_id="share"))
        self.assertTrue(self.fsm.has_metadata(mdid=mdid))
        self.assertRaises(ValueError, self.fsm.has_metadata, node_id=None,
                                                             share_id="share")

01186     def test_is_dir(self):
        '''Test the is_directory option.'''
        # standard file
        testfiledir = os.path.join(self.share_path, "path1")
        mdid = self.fsm.create(testfiledir, "share", is_dir=False)
        self.fsm.set_node_id(testfiledir, "uuid1")
        self.assertFalse(self.fsm.is_dir(path=testfiledir))
        self.assertFalse(self.fsm.is_dir(node_id="uuid1", share_id="share"))
        self.assertFalse(self.fsm.is_dir(mdid=mdid))

        # directory
        testfiledir = os.path.join(self.share_path, "path2")
        mdid = self.fsm.create(testfiledir, "share", is_dir=True)
        self.fsm.set_node_id(testfiledir, "uuid2")
        self.assertTrue(self.fsm.is_dir(path=testfiledir))
        self.assertTrue(self.fsm.is_dir(node_id="uuid2", share_id="share"))
        self.assertTrue(self.fsm.is_dir(mdid=mdid))

01204     def test_changed_server(self):
        '''Test the changed option when in SERVER state.'''
        # SERVER means: local_hash != server_hash and is_partial == False
        testfile = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(testfile, "share")
        self.fsm.set_node_id(testfile, "uuid")

        # set conditions and test
        self.fsm.set_by_mdid(mdid, server_hash=98765)
        # local_hash is None so far
        self.assertTrue(self.fsm.changed(mdid=mdid), "SERVER")
        self.assertTrue(self.fsm.changed(node_id="uuid", share_id="share"),
                                                                    "SERVER")
        self.assertTrue(self.fsm.changed(path=testfile), "SERVER")

        # put a .partial by hand, to see it crash
        open(testfile + ".partial", "w").close()
        self.assertRaises(InconsistencyError,
                          self.fsm._check_partial, mdid=mdid)

01224     def test_changed_none(self):
        '''Test the changed option when in NONE state.'''
        # NONE means: local_hash == server_hash and is_partial == False
        testfile = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(testfile, "share")
        self.fsm.set_node_id(testfile, "uuid")

        # all conditions are set: by default, local_hash and server_hash
        # are both None
        self.assertTrue(self.fsm.changed(mdid=mdid), "NONE")
        self.assertTrue(self.fsm.changed(node_id="uuid", share_id="share"),
                                                                        "NONE")
        self.assertTrue(self.fsm.changed(path=testfile), "NONE")

        # put a .partial by hand, to see it crash
        open(testfile + ".partial", "w").close()
        self.assertRaises(InconsistencyError,
                          self.fsm._check_partial, mdid=mdid)

01243     def test_changed_local(self):
        '''Test the changed option when in LOCAL state.'''
        # LOCAL means: local_hash != server_hash and is_partial == True
        testfile = os.path.join(self.share_path, "path")
        mdid = self.fsm.create(testfile, "share")
        self.fsm.set_node_id(testfile, "uuid")

        # set conditions and test
        self.fsm.set_by_mdid(mdid, server_hash=98765)
        # local_hash is None so far
        self.fsm.create_partial("uuid", "share")
        self.assertTrue(self.fsm.changed(mdid=mdid), "LOCAL")
        self.assertTrue(self.fsm.changed(node_id="uuid", share_id="share"),
                                                                    "LOCAL")
        self.assertTrue(self.fsm.changed(path=testfile), "LOCAL")

        # remove the .partial by hand, to see it crash
        os.remove(testfile + ".partial")
        self.assertRaises(InconsistencyError,
                          self.fsm._check_partial, mdid=mdid)

01264     def test_dir_content(self):
        '''Test the dir_content method.'''
        # create a structure in md
        to_create = []
        dir1 = os.path.join(self.share_path, "foo")
        to_create.append((dir1, True))
        to_create.append((os.path.join(dir1, "file2"), False))
        to_create.append((os.path.join(dir1, "file1"), False))

        dir2 = os.path.join(dir1, "bar")
        to_create.append((dir2, True))
        to_create.append((os.path.join(dir2, "file3"), False))
        to_create.append((os.path.join(dir2, "file5"), False))
        to_create.append((os.path.join(dir2, "file4"), False))
        to_create.append((os.path.join(dir2, "file6"), False))

        dir3 = os.path.join(dir2, "baz")
        to_create.append((dir3, True))
        to_create.append((os.path.join(dir3, "file7"), False))
        to_create.append((os.path.join(dir3, "file9"), False))
        to_create.append((os.path.join(dir3, "file8"), False))

        dir4 = os.path.join(dir2, "other")
        to_create.append((dir4, True))

        for i, (path, is_dir) in enumerate(to_create):
            self.fsm.create(path, "share", is_dir=is_dir)
            self.fsm.set_node_id(path, "uuid" + str(i))

        # ask for the info for dir1
        should_be = [
            ("bar", True, "uuid3"),
            ("file1", False, "uuid2"),
            ("file2", False, "uuid1"),
        ]
        content = self.fsm.dir_content(dir1)
        self.assertEqual(should_be, content)

        # ask for the info for dir2
        should_be = [
            ("baz", True, "uuid8"),
            ("file3", False, "uuid4"),
            ("file4", False, "uuid6"),
            ("file5", False, "uuid5"),
            ("file6", False, "uuid7"),
            ("other", True, "uuid12"),
        ]
        content = self.fsm.dir_content(dir2)
        self.assertEqual(should_be, content)

        # ask for the info for dir3
        should_be = [
            ("file7", False, "uuid9"),
            ("file8", False, "uuid11"),
            ("file9", False, "uuid10"),
        ]
        content = self.fsm.dir_content(dir3)
        self.assertEqual(should_be, content)

        # ask for the info for an empty dir
        content = self.fsm.dir_content(dir4)
        self.assertEqual([], content)

        # ask for the info for an inexistant dir
        self.assertRaises(KeyError, self.fsm.dir_content, "no-such-dir")

        # ask for the info for file
        just_a_file = os.path.join(dir3, "file9")
        self.assertRaises(ValueError, self.fsm.dir_content, just_a_file)


01335 class SharesTests(FSMTestCase):
    """ Tests fsm with ro and rw shares. """

01338     def tearDown(self):
        """ cleanup the mess """
        for dirpath, dirs, files in os.walk(TESTS_DIR):
            for dir in dirs:
                os.chmod(os.path.join(dirpath, dir), 0777)
            for file in files:
                os.chmod(os.path.join(dirpath, file), 0666)
        shutil.rmtree(TESTS_DIR)

01347     def test_file_ro_share_fail(self):
        """ Test that manual creation of a file, fails on a ro-share. """
        share = self.create_share('ro_share', 'ro_share_name',  self.fsm,
                                       self.shares_dir, access_level='View')
        testfile = os.path.join(share.path, "a_file")
        self.assertRaises(IOError, open, testfile, 'w')

01354     def test_dir_ro_share(self):
        """ Test that the creation of a file using fsm, works on a ro-share."""
        share = self.create_share('ro_share', 'ro_share_name',  self.fsm,
                                       self.shares_dir, access_level='View')
        testdir = os.path.join(share.path, "path2")
        mdid = self.fsm.create(testdir, share.id, is_dir=True)
        self.fsm.set_node_id(testdir, "uuid2")
        self.fsm.create_partial('uuid2', share.id)
        fd = self.fsm.get_partial_for_writing('uuid2', share.id)
        fd.flush()
        fd.close()
        self.assertTrue(os.path.exists(testdir))

01367     def test_file_ro_share(self):
        """ Test that the creation of a file using fsm, works on a ro-share."""
        self.share = self.create_share('ro_share', 'ro_share_name',  self.fsm,
                                       self.shares_dir, access_level='View')
        testfile = os.path.join(self.share.path, "a_file")
        file_mdid = self.fsm.create(testfile, self.share.id, is_dir=False)
        self.fsm.set_node_id(testfile, "uuid3")
        self.fsm.create_partial('uuid3', self.share.id)
        fd = self.fsm.get_partial_for_writing('uuid3', self.share.id)
        fd.flush()
        fd.close()
        self.fsm.commit_partial('uuid3', self.share.id, None)
        self.assertTrue(os.path.exists(testfile))

01381     def test_delete_dir_ro_share(self):
        """ Test that fsm is able to delete a dir in a ro.share. """
        share = self.create_share('ro_share', 'ro_share_name',  self.fsm,
                                       self.shares_dir, access_level='View')
        testdir = os.path.join(share.path, "path2")
        mdid = self.fsm.create(testdir, share.id, is_dir=True)
        self.fsm.set_node_id(testdir, "uuid2")
        self.fsm.create_partial('uuid2', share.id)
        fd = self.fsm.get_partial_for_writing('uuid2', share.id)
        fd.flush()
        fd.close()
        self.fsm.remove_partial('uuid2', share.id)
        self.assertTrue(os.path.exists(testdir))
        self.fsm.delete_file(testdir)
        self.assertFalse(os.path.exists(testdir))

01397     def test_delete_file_ro_share(self):
        """ Test that fsm is able to delete a file in a ro-share. """
        self.share = self.create_share('ro_share', 'ro_share_name',  self.fsm,
                                       self.shares_dir, access_level='View')
        testfile = os.path.join(self.share.path, "a_file")
        file_mdid = self.fsm.create(testfile, self.share.id, is_dir=False)
        self.fsm.set_node_id(testfile, "uuid3")
        self.fsm.create_partial('uuid3', self.share.id)
        fd = self.fsm.get_partial_for_writing('uuid3', self.share.id)
        fd.flush()
        fd.close()
        self.fsm.commit_partial('uuid3', self.share.id, None)
        self.assertTrue(os.path.exists(testfile))
        self.fsm.delete_file(testfile)
        self.assertFalse(os.path.exists(testfile))

01413     def test_move_to_conflict_ro_share(self):
        """ Test that fsm is able to handle move_to_conflict in a ro-share. """
        self.share = self.create_share('ro_share', 'ro_share_name',  self.fsm,
                                       self.shares_dir, access_level='View')
        testfile = os.path.join(self.share.path, "a_file")
        file_mdid = self.fsm.create(testfile, self.share.id, is_dir=False)
        self.fsm.set_node_id(testfile, "uuid3")
        self.fsm.create_partial('uuid3', self.share.id)
        fd = self.fsm.get_partial_for_writing('uuid3', self.share.id)
        fd.flush()
        fd.close()
        self.fsm.commit_partial('uuid3', self.share.id, None)
        self.assertTrue(os.path.exists(testfile))
        self.fsm.move_to_conflict(file_mdid)
        self.assertTrue(os.path.exists(testfile + ".conflict"))


01430     def test_file_rw_share_no_fail(self):
        """ Test that manual creation of a file, ona  rw-share. """
        share = self.create_share('ro_share', 'ro_share_name',  self.fsm,
                                       self.shares_dir)
        testfile = os.path.join(share.path, "a_file")
        open(testfile, 'w').close()
        self.assertTrue(os.path.exists(testfile))

01438     def test_dir_rw_share(self):
        """ Test that the creation of a file using fsm, works on a rw-share."""
        share = self.create_share('ro_share', 'ro_share_name',  self.fsm,
                                       self.shares_dir)
        testdir = os.path.join(share.path, "path2")
        mdid = self.fsm.create(testdir, share.id, is_dir=True)
        self.fsm.set_node_id(testdir, "uuid2")
        self.fsm.create_partial('uuid2', share.id)
        fd = self.fsm.get_partial_for_writing('uuid2', share.id)
        fd.flush()
        fd.close()
        self.assertTrue(os.path.exists(testdir))

01451     def test_file_rw_share(self):
        """ Test that the creation of a file using fsm, works on a rw-share."""
        self.share = self.create_share('ro_share', 'ro_share_name',  self.fsm,
                                       self.shares_dir)
        testfile = os.path.join(self.share.path, "a_file")
        file_mdid = self.fsm.create(testfile, self.share.id, is_dir=False)
        self.fsm.set_node_id(testfile, "uuid3")
        self.fsm.create_partial('uuid3', self.share.id)
        fd = self.fsm.get_partial_for_writing('uuid3', self.share.id)
        fd.flush()
        fd.close()
        self.fsm.commit_partial('uuid3', self.share.id, None)
        self.assertTrue(os.path.exists(testfile))


01466     def test_share_and_root(self):
        """ Test the creation of a file with the same relative path in a share
        and in the root.
        """
        a_dir_root = os.path.join(self.root_dir, "a_dir")
        a_dir_share = os.path.join(self.share.path, "a_dir")
        root_mdid = self.fsm.create(a_dir_root, "", is_dir=True)
        self.fsm.set_node_id(a_dir_root, "uuid1")
        share_mdid = self.fsm.create(a_dir_share, self.share.id, is_dir=True)
        self.fsm.set_node_id(a_dir_share, "uuid2")
        self.fsm.create_partial('uuid1', "")
        fd = self.fsm.get_partial_for_writing('uuid1', "")
        fd.flush()
        fd.close()
        self.fsm.create_partial('uuid2', self.share.id)
        fd = self.fsm.get_partial_for_writing('uuid2', self.share.id)
        fd.flush()
        fd.close()
        self.assertTrue(os.path.exists(self.fsm.get_abspath("", a_dir_root)))
        self.assertTrue(os.path.exists(a_dir_share))

def test_suite():
    # pylint: disable-msg=C0111
    return unittest.TestLoader().loadTestsFromName(__name__)

if __name__ == "__main__":
    unittest.main()

Generated by  Doxygen 1.6.0   Back to index