~didrocks/ubuntuone-client/dont-suffer-zg-crash

#
# Author: Facundo Batista <facundo@canonical.com>
#
# 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 Event Queue."""

import logging
import os
import unittest

from twisted.internet import defer, reactor

from ubuntuone.syncdaemon import (
    event_queue,
    filesystem_manager,
)
from contrib.testing import testcase
from ubuntuone.devtools.handlers import MementoHandler
from ubuntuone.platform.linux.filesystem_notifications import (
    _GeneralINotifyProcessor
)
from ubuntuone.syncdaemon import volume_manager
from ubuntuone.syncdaemon.tritcask import Tritcask


class BaseFSMonitorTestCase(testcase.BaseTwistedTestCase):
    """Test the structures where we have the path/watch."""

    timeout = 3

    def setUp(self):
        """Set up."""
        testcase.BaseTwistedTestCase.setUp(self)
        fsmdir = self.mktemp('fsmdir')
        partials_dir = self.mktemp('partials_dir')
        self.root_dir = self.mktemp('root_dir')
        self.home_dir = self.mktemp('home_dir')
        self.vm = testcase.FakeVolumeManager(self.root_dir)
        self.tritcask_dir = self.mktemp("tritcask_dir")
        self.db = Tritcask(self.tritcask_dir)
        self.fs = filesystem_manager.FileSystemManager(fsmdir, partials_dir,
                                                       self.vm, self.db)
        self.fs.create(path=self.root_dir, share_id='', is_dir=True)
        self.fs.set_by_path(path=self.root_dir,
                            local_hash=None, server_hash=None)
        eq = event_queue.EventQueue(self.fs)

        self.deferred = deferred = defer.Deferred()

        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_default(innerself, event, **args):
                deferred.callback(True)

        eq.subscribe(HitMe())
        self.monitor = eq.monitor
        self.log_handler = MementoHandler()
        self.log_handler.setLevel(logging.DEBUG)
        self.monitor.log.addHandler(self.log_handler)

    def tearDown(self):
        """Clean up the tests."""
        self.monitor.shutdown()
        self.rmtree(self.tmpdir)
        self.monitor.log.removeHandler(self.log_handler)
        testcase.BaseTwistedTestCase.tearDown(self)


class WatchManagerTests(BaseFSMonitorTestCase):
    """Test the structures where we have the path/watch."""

    def test_watch_updated_when_deleting_dir(self):
        """Internal data structures are fixed when deleting the dir."""
        path = os.path.join(self.root_dir, "path")
        os.mkdir(path)
        self.monitor.add_watch(self.root_dir)
        self.monitor.add_watch(path)

        # we have the watch, remove the dir, the watch should be gone
        self.assertTrue(self.monitor.has_watch(path))
        os.rmdir(path)

        def check(_):
            """Check state after the event."""
            self.assertFalse(self.monitor.has_watch(path))

        self.deferred.addCallback(check)
        return self.deferred

    def test_watch_updated_when_renaming_dir(self):
        """Internal data structures are fixed when renaming the dir."""
        path1 = os.path.join(self.root_dir, "path1")
        path2 = os.path.join(self.root_dir, "path2")
        os.mkdir(path1)
        self.monitor.add_watch(self.root_dir)
        self.monitor.add_watch(path1)

        # we have the watch, rename the dir, new name should have the watch,
        # old name should not
        self.assertTrue(self.monitor.has_watch(path1))
        os.rename(path1, path2)

        def check(_):
            """Check state after the event."""
            self.assertTrue(self.monitor.has_watch(path2))
            self.assertFalse(self.monitor.has_watch(path1))

        self.deferred.addCallback(check)
        return self.deferred

    def test_watch_updated_when_movingout_dir(self):
        """Internal data structures are fixed when moving out the dir."""
        notu1 = os.path.join(self.root_dir, "notu1")
        path1 = os.path.join(self.root_dir, "path1")
        path2 = os.path.join(notu1, "path2")
        os.mkdir(notu1)
        os.mkdir(path1)
        self.monitor.add_watch(self.root_dir)
        self.monitor.add_watch(path1)

        # we have the watch, move it outside watched structure, no more watches
        self.assertTrue(self.monitor.has_watch(path1))
        os.rename(path1, path2)

        def check(_):
            """Check state after the event."""
            self.assertFalse(self.monitor.has_watch(path1))
            self.assertFalse(self.monitor.has_watch(path2))

        self.deferred.addCallback(check)
        return self.deferred

    def test_fix_path_not_there(self):
        """Try to fix path but it's not there."""
        self.monitor._general_watchs = {}
        self.monitor._ancestors_watchs = {}
        self.monitor.inotify_watch_fix("not-there", "new-one")
        self.assertTrue(self.log_handler.check_warning("Tried to fix",
                                                       "not-there"))

    def test_fix_path_general(self):
        """Fixing path in general watches."""
        self.monitor._general_watchs = {'/path1/foo': 1, '/other': 2}
        self.monitor._ancestors_watchs = {'/foo': 3}
        self.monitor.inotify_watch_fix('/path1/foo', '/path1/new')
        self.assertEqual(self.monitor._general_watchs, {'/path1/new': 1, '/other': 2})
        self.assertEqual(self.monitor._ancestors_watchs, {'/foo': 3})

    def test_fix_path_ancestors(self):
        """Fixing path in ancestors watches."""
        self.monitor._general_watchs = {'/bar': 3}
        self.monitor._ancestors_watchs = {'/oth': 1, '/other': 2}
        self.monitor.inotify_watch_fix('/oth', '/baz')
        self.assertEqual(self.monitor._general_watchs, {'/bar': 3})
        self.assertEqual(self.monitor._ancestors_watchs, {'/baz': 1, '/other': 2})


class DynamicHitMe(object):
    """Helper class to test a sequence of signals."""

    def __init__(self, should_events, test_machinery):
        self.should_events = []
        for i, info in enumerate(should_events):
            self.should_events.append((i,) + info)
        if self.should_events:
            self.final_step = self.should_events[-1][0]
            self.should_events.reverse()
        self.test_machinery = test_machinery

    def __getattr__(self, name):
        """typical method faker"""
        if not name.startswith("handle_"):
            return

        asked_event = name[7:]

        # to what we should match
        test_info = self.should_events.pop()
        step = test_info[0]
        should_evtname = test_info[1]
        should_args = test_info[2:]

        def to_check(*asked_args):
            """the function that actually be called"""
            if asked_args != should_args:
                self.test_machinery.finished_error(
                    "In step %d received wrong args (%r)" % (step, asked_args))
            else:
                if step == self.final_step:
                    self.test_machinery.finished_ok()

        if should_evtname != asked_event:
            msg = "Event %r asked in bad order (%d)" % (asked_event, step)
            self.test_machinery.finished_error(msg)
        else:
            return to_check


class BaseTwisted(BaseFSMonitorTestCase):
    """Base class for twisted tests."""

    # this timeout must be bigger than the one used in event_queue
    timeout = 2

    def setUp(self):
        """Setup the test."""
        BaseFSMonitorTestCase.setUp(self)

        # create the deferred for the tests
        self._deferred = defer.Deferred()

    def finished_ok(self):
        """Called to indicate that the tests finished ok."""
        self._deferred.callback(True)

    def finished_error(self, msg):
        """Called to indicate that the tests finished badly."""
        self._deferred.errback(Exception(msg))

    def failUnlessEqual(self, first, second, msg=''):
        """Fail the test if C{first} and C{second} are not equal.

        @param msg: A string describing the failure that's included in the
            exception.

        """
        if not first == second:
            if msg is None:
                msg = ''
            if len(msg) > 0:
                msg += '\n'
            exception = self.failureException(
                '%snot equal:\na = %s\nb = %s\n'
                % (msg, repr(first), repr(second)))
            self.finished_error(exception)
            raise exception
        return first
    assertEqual = assertEquals = failUnlessEquals = failUnlessEqual


class IgnoreFileTests(unittest.TestCase):
    """Tests the ignore files behaviour."""

    def test_filter_none(self):
        """Still works ok even if not receiving a regex to ignore."""
        p = _GeneralINotifyProcessor(None)
        self.assertFalse(p.is_ignored("froo.pyc"))

    def test_filter_one(self):
        """Filters stuff that matches (or not) this one regex."""
        p = _GeneralINotifyProcessor(None, ['\A.*\\.pyc\Z'])
        self.assertTrue(p.is_ignored("froo.pyc"))
        self.assertFalse(p.is_ignored("froo.pyc.real"))
        self.assertFalse(p.is_ignored("otherstuff"))

    def test_filter_two_simple(self):
        """Filters stuff that matches (or not) these simple regexes."""
        p = _GeneralINotifyProcessor(None, ['\A.*foo\Z', '\A.*bar\Z'])
        self.assertTrue(p.is_ignored("blah_foo"))
        self.assertTrue(p.is_ignored("blah_bar"))
        self.assertFalse(p.is_ignored("bar_xxx"))
        self.assertFalse(p.is_ignored("--foo--"))
        self.assertFalse(p.is_ignored("otherstuff"))

    def test_filter_two_complex(self):
        """Filters stuff that matches (or not) these complex regexes."""
        p = _GeneralINotifyProcessor(None,
                                     ['\A.*foo\Z|\Afoo.*\Z', '\A.*bar\Z'])
        self.assertTrue(p.is_ignored("blah_foo"))
        self.assertTrue(p.is_ignored("blah_bar"))
        self.assertTrue(p.is_ignored("foo_xxx"))
        self.assertFalse(p.is_ignored("--foo--"))
        self.assertFalse(p.is_ignored("otherstuff"))


class WatchTests(BaseFSMonitorTestCase):
    """Test the EQ API to add and remove watchs."""

    def _create_udf(self, path):
        """Create an UDF and returns it and the volume"""
        os.makedirs(path)
        udf = volume_manager.UDF("vol_id", "node_id", path.decode('utf-8'),
                                 path, True)
        self.vm.add_udf(udf)

    def test_add_general_watch(self):
        """Test that general watchs can be added."""
        # we should have what we asked for
        self.monitor.add_watch(self.root_dir)

        # check only added dir in watchs
        # pylint: disable-msg=W0212
        self.assertTrue(self.root_dir in self.monitor._general_watchs)
        self.assertTrue("not-added-dir" not in self.monitor._general_watchs)

        # nothing in the udf ancestors watch
        self.assertEqual(self.monitor._ancestors_watchs, {})

    def test_add_watch_on_udf_ancestor(self):
        """Test that ancestors watchs can be added."""
        # create the udf and add the watch
        path_udf = os.path.join(self.home_dir, "path/to/UDF")
        self._create_udf(path_udf)
        path_ancestor = os.path.join(self.home_dir, "path")
        self.monitor.add_watch(path_ancestor)

        # check only added dir in watchs
        # pylint: disable-msg=W0212
        self.assertTrue(path_ancestor in self.monitor._ancestors_watchs)
        self.assertTrue("not-added-dir" not in self.monitor._ancestors_watchs)

        # nothing in the general watch
        self.assertEqual(self.monitor._general_watchs, {})

    def test_add_watch_on_udf_exact(self):
        """Test adding the watch exactly on UDF."""
        # create the udf and add the watch
        path_udf = os.path.join(self.home_dir, "path/to/UDF")
        self._create_udf(path_udf)
        self.monitor.add_watch(path_udf)

        # pylint: disable-msg=W0212
        self.assertTrue(path_udf in self.monitor._general_watchs)
        self.assertEqual(self.monitor._ancestors_watchs, {})

    def test_add_watch_on_udf_child(self):
        """Test adding the watch inside UDF."""
        # create the udf and add the watch
        path_udf = os.path.join(self.home_dir, "path/to/UDF")
        self._create_udf(path_udf)
        path_ancestor = os.path.join(self.home_dir, "path/to/UDF/inside")
        os.mkdir(path_ancestor)
        self.monitor.add_watch(path_ancestor)

        # pylint: disable-msg=W0212
        self.assertTrue(path_ancestor in self.monitor._general_watchs)
        self.assertEqual(self.monitor._ancestors_watchs, {})

    def test_rm_watch_not_dir_anymore(self):
        """Test that a watch can be removed even not having the dir anymore.

        This is the case where the directory is deleted from the filesystem,
        the watch is automatically removed in pyinotify but we need to take
        care of it from our own data structures.
        """
        not_existing_dir = "not-added-dir"
        self.monitor.add_watch(not_existing_dir)
        self.assertTrue(self.monitor.has_watch(not_existing_dir))
        self.monitor.rm_watch(not_existing_dir)
        self.assertFalse(self.monitor.has_watch(not_existing_dir))

    def test_rm_watch_wrong(self):
        """Test that general watchs can be removed."""
        # add two types of watchs
        self.monitor.add_watch(self.root_dir)
        path_udf = os.path.join(self.home_dir, "path/to/UDF")
        self._create_udf(path_udf)
        path_ancestor = os.path.join(self.home_dir, "path")
        self.monitor.add_watch(path_ancestor)

        # remove different stuff
        self.monitor.rm_watch("not-added-dir")
        self.assertTrue(self.log_handler.check_warning('remove', 'watch',
                                                       'not-added-dir'))

    def test_rm_watch_general(self):
        """Test that general watchs can be removed."""
        # remove what we added
        self.monitor.add_watch(self.root_dir)
        self.monitor.rm_watch(self.root_dir)

        # pylint: disable-msg=W0212
        self.assertEqual(self.monitor._general_watchs, {})
        self.assertEqual(self.monitor._ancestors_watchs, {})

    def test_rm_watch_ancestor(self):
        """Test that ancestor watchs can be removed."""
        # create the udf and add the watch
        path_udf = os.path.join(self.home_dir, "path/to/UDF")
        self._create_udf(path_udf)
        path_ancestor = os.path.join(self.home_dir, "path")
        self.monitor.add_watch(path_ancestor)

        # remove what we added
        self.monitor.rm_watch(path_ancestor)
        # pylint: disable-msg=W0212
        self.assertEqual(self.monitor._general_watchs, {})
        self.assertEqual(self.monitor._ancestors_watchs, {})

    def test_has_watch_general(self):
        """Test that a general path is watched."""
        self.assertFalse(self.monitor.has_watch(self.root_dir))

        # add
        self.monitor.add_watch(self.root_dir)
        self.assertTrue(self.monitor.has_watch(self.root_dir))

        # remove
        self.monitor.rm_watch(self.root_dir)
        self.assertFalse(self.monitor.has_watch(self.root_dir))

    def test_has_watch_ancestor(self):
        """Test that an ancestor path is watched."""
        path_udf = os.path.join(self.home_dir, "path/to/UDF")
        self._create_udf(path_udf)
        path_ancestor = os.path.join(self.home_dir, "path")

        self.assertFalse(self.monitor.has_watch(path_ancestor))

        # add
        # create the udf and add the watch
        self.monitor.add_watch(path_ancestor)
        self.assertTrue(self.monitor.has_watch(path_ancestor))

        # remove
        self.monitor.rm_watch(path_ancestor)
        self.assertFalse(self.monitor.has_watch(path_ancestor))


class MutedSignalsTests(BaseTwisted):
    """Test that EQ filter some signals on demand."""

    def check_filter(self, _=None):
        self.assertFalse(self.monitor._processor._to_mute._cnt)
        self.finished_ok()

    def test_mute_and_remove(self):
        """Test add and remove the mute."""
        # add
        self.monitor.add_to_mute_filter('FS_FILE_OPEN', path='somepath')
        self.assertEqual(self.monitor._processor._to_mute._cnt,
                         {'FS_FILE_OPEN': [{'path': 'somepath'}]})
        self.monitor.add_to_mute_filter('FS_FILE_OPEN', path='somepath')
        self.assertEqual(self.monitor._processor._to_mute._cnt,
                         {'FS_FILE_OPEN': [{'path': 'somepath'},
                                           {'path': 'somepath'}]})
        self.monitor.add_to_mute_filter('FS_FILE_OPEN', path='otherpath')
        self.assertEqual(self.monitor._processor._to_mute._cnt,
                         {'FS_FILE_OPEN': [{'path': 'somepath'},
                                           {'path': 'somepath'},
                                           {'path': 'otherpath'}]})

        # remove
        self.monitor.rm_from_mute_filter('FS_FILE_OPEN', path='somepath')
        self.assertEqual(self.monitor._processor._to_mute._cnt,
                         {'FS_FILE_OPEN': [{'path': 'somepath'},
                                           {'path': 'otherpath'}]})
        self.monitor.rm_from_mute_filter('FS_FILE_OPEN', path='otherpath')
        self.assertEqual(self.monitor._processor._to_mute._cnt,
                         {'FS_FILE_OPEN': [{'path': 'somepath'}]})
        self.monitor.rm_from_mute_filter('FS_FILE_OPEN', path='somepath')
        self.assertEqual(self.monitor._processor._to_mute._cnt, {})

    def _how_many_muted(self):
        """Return how many events are muted."""
        mute_filter = self.monitor._processor._to_mute
        return sum(len(x) for x in mute_filter._cnt.values())

    def test_file_open(self):
        """Test receiving the open signal on files."""
        testfile = os.path.join(self.root_dir, "foo")
        open(testfile, "w").close()
        self.monitor.add_to_mute_filter("FS_FILE_OPEN", path=testfile)
        self.monitor.add_to_mute_filter("FS_FILE_CLOSE_NOWRITE", path=testfile)
        self.assertEqual(self._how_many_muted(), 2)
        self.monitor.add_watch(self.root_dir)

        # generate the event
        open(testfile)
        reactor.callLater(.1, self.check_filter)
        return self._deferred

    def test_file_close_nowrite(self):
        """Test receiving the close_nowrite signal on files."""
        testfile = os.path.join(self.root_dir, "foo")
        open(testfile, "w").close()
        fh = open(testfile)
        self.monitor.add_to_mute_filter("FS_FILE_CLOSE_NOWRITE", path=testfile)
        self.assertEqual(self._how_many_muted(), 1)
        self.monitor.add_watch(self.root_dir)

        # generate the event
        fh.close()
        reactor.callLater(.1, self.check_filter)
        return self._deferred

    def test_file_create_close_write(self):
        """Test receiving the create and close_write signals on files."""
        testfile = os.path.join(self.root_dir, "foo")
        self.monitor.add_to_mute_filter("FS_FILE_CREATE", path=testfile)
        self.monitor.add_to_mute_filter("FS_FILE_OPEN", path=testfile)
        self.monitor.add_to_mute_filter("FS_FILE_CLOSE_WRITE", path=testfile)
        self.assertEqual(self._how_many_muted(), 3)
        self.monitor.add_watch(self.root_dir)

        # generate the event
        open(testfile, "w").close()
        reactor.callLater(.1, self.check_filter)
        return self._deferred

    def test_dir_create(self):
        """Test receiving the create signal on dirs."""
        testdir = os.path.join(self.root_dir, "foo")
        self.monitor.add_to_mute_filter("FS_DIR_CREATE", path=testdir)
        self.assertEqual(self._how_many_muted(), 1)
        self.monitor.add_watch(self.root_dir)

        # generate the event
        os.mkdir(testdir)
        reactor.callLater(.1, self.check_filter)
        return self._deferred

    def test_file_delete(self):
        """Test the delete signal on a file."""
        testfile = os.path.join(self.root_dir, "foo")
        open(testfile, "w").close()
        self.monitor.add_to_mute_filter("FS_FILE_DELETE", path=testfile)
        self.assertEqual(self._how_many_muted(), 1)
        self.monitor.add_watch(self.root_dir)

        # generate the event
        os.remove(testfile)
        reactor.callLater(.1, self.check_filter)
        return self._deferred

    def test_dir_delete(self):
        """Test the delete signal on a dir."""
        testdir = os.path.join(self.root_dir, "foo")
        os.mkdir(testdir)
        self.monitor.add_to_mute_filter("FS_DIR_DELETE", path=testdir)
        self.assertEqual(self._how_many_muted(), 1)
        self.monitor.add_watch(self.root_dir)

        # generate the event
        os.rmdir(testdir)
        reactor.callLater(.1, self.check_filter)
        return self._deferred

    def test_file_moved_inside(self):
        """Test the synthesis of the FILE_MOVE event."""
        fromfile = os.path.join(self.root_dir, "foo")
        self.fs.create(fromfile, "")
        self.fs.set_node_id(fromfile, "from_node_id")
        tofile = os.path.join(self.root_dir, "bar")
        self.fs.create(tofile, "")
        self.fs.set_node_id(tofile, "to_node_id")
        open(fromfile, "w").close()
        self.monitor.add_to_mute_filter("FS_FILE_MOVE",
                                        path_from=fromfile, path_to=tofile)
        self.assertEqual(self._how_many_muted(), 1)
        self.monitor.add_watch(self.root_dir)

        # generate the event
        os.rename(fromfile, tofile)
        reactor.callLater(.1, self.check_filter)
        return self._deferred

    def test_dir_moved_inside(self):
        """Test the synthesis of the DIR_MOVE event."""
        fromdir = os.path.join(self.root_dir, "foo")
        self.fs.create(fromdir, "")
        self.fs.set_node_id(fromdir, "from_node_id")
        todir = os.path.join(self.root_dir, "bar")
        self.fs.create(todir, "")
        self.fs.set_node_id(todir, "to_node_id")
        os.mkdir(fromdir)
        self.monitor.add_to_mute_filter("FS_DIR_MOVE",
                                        path_from=fromdir, path_to=todir)
        self.assertEqual(self._how_many_muted(), 1)
        self.monitor.add_watch(self.root_dir)

        # generate the event
        os.rename(fromdir, todir)
        reactor.callLater(.1, self.check_filter)
        return self._deferred

    def test_file_moved_from_conflict(self):
        """Test the handling of the FILE_MOVE event when source is conflict."""
        fromfile = os.path.join(self.root_dir, "foo.u1conflict")
        self.fs.create(fromfile, "")
        self.fs.set_node_id(fromfile, "from_node_id")
        tofile = os.path.join(self.root_dir, "foo")
        self.fs.create(tofile, "")
        self.fs.set_node_id(tofile, "to_node_id")
        open(fromfile, "w").close()
        self.monitor.add_to_mute_filter("FS_FILE_MOVE",
                                        path_from=fromfile, path_to=tofile)
        self.assertEqual(self._how_many_muted(), 2)
        self.monitor.add_watch(self.root_dir)

        # generate the event
        os.rename(fromfile, tofile)
        reactor.callLater(.1, self.check_filter)
        return self._deferred

    def test_file_moved_from_partial(self):
        """Test the handling of the FILE_MOVE event when source is partial."""
        fromfile = os.path.join(self.root_dir, "mdid.u1partial.foo")
        root_dir = os.path.join(self.root_dir, "my_files")
        tofile = os.path.join(root_dir, "foo")
        os.mkdir(root_dir)
        open(fromfile, "w").close()
        self.monitor.add_to_mute_filter("FS_FILE_CREATE", path=tofile)
        self.monitor.add_to_mute_filter("FS_FILE_CLOSE_WRITE", path=tofile)
        self.assertEqual(self._how_many_muted(), 2)
        self.monitor.add_watch(root_dir)

        # generate the event
        os.rename(fromfile, tofile)
        reactor.callLater(.1, self.check_filter)
        return self._deferred


def test_suite():
    """Collect the tests."""
    import sys
    if sys.platform == 'linux2':
        tests = unittest.TestLoader().loadTestsFromName(__name__)
    else:
        tests = []
    return tests

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