~rodrigo-moya/ubuntuone-client/contact-default-icon

# tests.syncdaemon.test_eq_inotify
#
# 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 Event Queue part that uses inotify."""

import functools
import logging
import os
import shutil
import unittest

from twisted.internet import defer, reactor

from contrib.testing import testcase
from ubuntuone.syncdaemon import volume_manager
from tests.syncdaemon.test_eventqueue import BaseEQTestCase


class DontHitMe(object):
    """We shouldn't be called."""

    def __init__(self, test_instance):
        self.test_instance = test_instance

    def handle_default(self, *a, **k):
        """Something here? Error!"""
        self.test_inst.finished_error("Don't hit me! Received %s %s" % (a, k))



class WatchTests(BaseEQTestCase):
    """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.eq.inotify_add_watch(self.root_dir)

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

        # nothing in the udf ancestors watch
        self.assertEqual(self.eq._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.eq.inotify_add_watch(path_ancestor)

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

        # nothing in the general watch
        self.assertEqual(self.eq._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.eq.inotify_add_watch(path_udf)

        # pylint: disable-msg=W0212
        self.assertTrue(path_udf in self.eq._general_watchs)
        self.assertEqual(self.eq._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.eq.inotify_add_watch(path_ancestor)

        # pylint: disable-msg=W0212
        self.assertTrue(path_ancestor in self.eq._general_watchs)
        self.assertEqual(self.eq._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.eq.inotify_add_watch(not_existing_dir)
        self.assertTrue(self.eq.inotify_has_watch(not_existing_dir))
        self.eq.inotify_rm_watch(not_existing_dir)
        self.assertFalse(self.eq.inotify_has_watch(not_existing_dir))

    def test_rm_watch_wrong(self):
        """Test that general watchs can be removed."""
        # add two types of watchs
        self.eq.inotify_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.eq.inotify_add_watch(path_ancestor)

        # remove different stuff
        self.eq.inotify_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.eq.inotify_add_watch(self.root_dir)
        self.eq.inotify_rm_watch(self.root_dir)

        # pylint: disable-msg=W0212
        self.assertEqual(self.eq._general_watchs, {})
        self.assertEqual(self.eq._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.eq.inotify_add_watch(path_ancestor)

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

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

        # add
        self.eq.inotify_add_watch(self.root_dir)
        self.assertTrue(self.eq.inotify_has_watch(self.root_dir))

        # remove
        self.eq.inotify_rm_watch(self.root_dir)
        self.assertFalse(self.eq.inotify_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.eq.inotify_has_watch(path_ancestor))

        # add
        # create the udf and add the watch
        self.eq.inotify_add_watch(path_ancestor)
        self.assertTrue(self.eq.inotify_has_watch(path_ancestor))

        # remove
        self.eq.inotify_rm_watch(path_ancestor)
        self.assertFalse(self.eq.inotify_has_watch(path_ancestor))


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

    timeout = 3

    def setUp(self):
        """Set up."""
        BaseEQTestCase.setUp(self)
        self.deferred = deferred = defer.Deferred()

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

        self.eq.subscribe(HitMe())

    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.eq.inotify_add_watch(self.root_dir)
        self.eq.inotify_add_watch(path)

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

        def check(_):
            """Check state after the event."""
            self.assertFalse(self.eq.inotify_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.eq.inotify_add_watch(self.root_dir)
        self.eq.inotify_add_watch(path1)

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

        def check(_):
            """Check state after the event."""
            self.assertTrue(self.eq.inotify_has_watch(path2))
            self.assertFalse(self.eq.inotify_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.eq.inotify_add_watch(self.root_dir)
        self.eq.inotify_add_watch(path1)

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

        def check(_):
            """Check state after the event."""
            self.assertFalse(self.eq.inotify_has_watch(path1))
            self.assertFalse(self.eq.inotify_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.eq._general_watchs = {}
        self.eq._ancestors_watchs = {}
        self.eq.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.eq._general_watchs = {'/path1/foo': 1, '/other': 2}
        self.eq._ancestors_watchs = {'/foo': 3}
        self.eq.inotify_watch_fix('/path1/foo', '/path1/new')
        self.assertEqual(self.eq._general_watchs, {'/path1/new': 1, '/other': 2})
        self.assertEqual(self.eq._ancestors_watchs, {'/foo': 3})

    def test_fix_path_ancestors(self):
        """Fixing path in ancestors watches."""
        self.eq._general_watchs = {'/bar': 3}
        self.eq._ancestors_watchs = {'/oth': 1, '/other': 2}
        self.eq.inotify_watch_fix('/oth', '/baz')
        self.assertEqual(self.eq._general_watchs, {'/bar': 3})
        self.assertEqual(self.eq._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(BaseEQTestCase):
    """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."""
        BaseEQTestCase.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 SignalingTests(BaseTwisted):
    """Test the whole stuff to receive signals."""

    def test_file_open(self):
        """Test receiving the open signal on files."""
        testfile = os.path.join(self.root_dir, "foo")

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_FILE_OPEN(innerself, path):
                if path != testfile:
                    self.finished_error("received a wrong path")
                else:
                    os.remove(testfile)
                    self.finished_ok()

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(HitMe())

        # generate the event
        open(testfile, "w")
        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)

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_FILE_CLOSE_NOWRITE(innerself, path):
                if path != testfile:
                    self.finished_error("received a wrong path")
                else:
                    os.remove(testfile)
                    self.finished_ok()

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(HitMe())

        # generate the event
        fh.close()
        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")

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def __init__(innerself):
                innerself.hist = []

            def handle_FS_FILE_CREATE(innerself, path):
                if path != testfile:
                    self.finished_error("received a wrong path")
                else:
                    innerself.hist.append("create")

            def handle_FS_FILE_CLOSE_WRITE(innerself, path):
                if path != testfile:
                    self.finished_error("received a wrong path")
                else:
                    if innerself.hist == ["create"]:
                        os.remove(testfile)
                        self.finished_ok()
                    else:
                        msg = "Finished in bad condition: %s" % innerself.hist
                        self.finished_error(msg)

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(HitMe())

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

    def test_dir_create(self):
        """Test receiving the create signal on dirs."""
        testdir = os.path.join(self.root_dir, "foo")

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_DIR_CREATE(innerself, path):
                if path != testdir:
                    self.finished_error("received a wrong path")
                else:
                    os.rmdir(testdir)
                    self.finished_ok()

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(HitMe())

        # generate the event
        os.mkdir(testdir)
        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()

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_FILE_DELETE(innerself, path):
                if path != testfile:
                    self.finished_error("received a wrong path")
                    return

                if not self.log_handler.check_info('FS_FILE_DELETE', testfile):
                    self.finished_error("FS_FILE_DELETE must appear in INFO.")
                    return

                self.finished_ok()

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(HitMe())

        # generate the event
        os.remove(testfile)
        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)

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_DIR_DELETE(innerself, path):
                if path != testdir:
                    self.finished_error("received a wrong path")
                    return

                if not self.log_handler.check_info('FS_DIR_DELETE', testdir):
                    self.finished_error("FS_DIR_DELETE must appear in INFO.")
                    return

                # file deletion should remove its watch
                self.assertFalse(self.eq.inotify_has_watch(testdir))

                self.finished_ok()

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(HitMe())

        # generate the event
        os.rmdir(testdir)
        return self._deferred

    def test_symlink(self):
        """Test that symlinks are ignored."""
        testdir = os.path.join(self.root_dir, "foo")
        os.mkdir(testdir)
        fromfile = os.path.join(self.root_dir, "from")
        open(fromfile, "w").close()
        symlpath = os.path.join(testdir, "syml")

        def confirm():
            """check result."""
            self.finished_ok()

        # set up everything and freeze
        self.eq.inotify_add_watch(testdir)
        self.eq.subscribe(DontHitMe(self))

        os.symlink(fromfile, symlpath)
        reactor.callLater(.1, confirm)
        return self._deferred

    def test_file_moved_from(self):
        """Test receiving the delete signal on a file when moved_from."""
        fromfile = os.path.join(self.root_dir, "foo")
        helpdir = os.path.join(self.root_dir, "dir")
        tofile = os.path.join(helpdir, "foo")
        open(fromfile, "w").close()
        os.mkdir(helpdir)

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_FILE_DELETE(innerself, path):
                if path != fromfile:
                    self.finished_error("received a wrong path")
                else:
                    os.remove(tofile)
                    os.rmdir(helpdir)
                    self.finished_ok()

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(HitMe())

        # generate the event
        os.rename(fromfile, tofile)
        return self._deferred

    def test_dir_moved_from(self):
        """Test receiving the delete signal on a dir when it's moved_from."""
        fromdir = os.path.join(self.root_dir, "foo")
        helpdir = os.path.join(self.root_dir, "dir")
        todir = os.path.join(helpdir, "foo")
        os.mkdir(fromdir)
        os.mkdir(helpdir)

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_DIR_DELETE(innerself, path):
                if path != fromdir:
                    self.finished_error("received a wrong path")
                else:
                    os.rmdir(todir)
                    os.rmdir(helpdir)
                    self.finished_ok()

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(HitMe())

        # generate the event
        os.rename(fromdir, todir)
        return self._deferred

    def test_file_moved_to(self):
        """Test receiving the create signal on a file when it's moved_to."""
        fromfile = os.path.join(self.root_dir, "dir", "foo")
        tofile = os.path.join(self.root_dir, "foo")
        helpdir = os.path.join(self.root_dir, "dir")
        os.mkdir(helpdir)
        open(fromfile, "w").close()

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_FILE_CREATE(innerself, path):
                if path != tofile:
                    self.finished_error("received a wrong path")
                else:
                    os.remove(tofile)
                    os.rmdir(helpdir)
                    self.finished_ok()

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(HitMe())

        # generate the event
        os.rename(fromfile, tofile)
        return self._deferred

    def test_dir_moved_to(self):
        """Test receiving the create signal on a file when it's moved_to."""
        fromdir = os.path.join(self.root_dir, "dir", "foo")
        todir = os.path.join(self.root_dir, "foo")
        helpdir = os.path.join(self.root_dir, "dir")
        os.mkdir(helpdir)
        os.mkdir(fromdir)

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_DIR_CREATE(innerself, path):
                if path != todir:
                    self.finished_error("received a wrong path")
                else:
                    os.rmdir(todir)
                    os.rmdir(helpdir)
                    self.finished_ok()

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(HitMe())

        # generate the event
        os.rename(fromdir, todir)
        return self._deferred

    def test_dir_moved_to_from_ignored(self):
        """Test receiving the create signal on a file when it's moved_to."""
        fromdir = os.path.join(self.root_dir, "dir", "foo.u1conflict")
        todir = os.path.join(self.root_dir, "foo")
        helpdir = os.path.join(self.root_dir, "dir")
        os.makedirs(fromdir)

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_DIR_CREATE(innerself, path):
                if path != todir:
                    self.finished_error("received a wrong path")
                else:
                    os.rmdir(todir)
                    os.rmdir(helpdir)
                    self.finished_ok()

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(HitMe())

        # generate the event
        os.rename(fromdir, todir)
        return self._deferred

    def test_lots_of_changes(self):
        """Test doing several operations on files."""
        helpdir = os.path.join(self.root_dir, "dir")
        os.mkdir(helpdir)
        mypath = functools.partial(os.path.join, self.root_dir)

        self.eq.inotify_add_watch(self.root_dir)

        should_events = [
            ("FS_FILE_CREATE", mypath("foo")),
            ("FS_FILE_OPEN", mypath("foo")),
            ("FS_FILE_CLOSE_WRITE", mypath("foo")),
            ("FS_FILE_DELETE", mypath("foo")),
            ("FS_FILE_CREATE", mypath("bar")),
            ("FS_FILE_OPEN", mypath("bar")),
            ("FS_FILE_CLOSE_WRITE", mypath("bar")),
            ("FS_FILE_CREATE", mypath("foo")),
            ("FS_FILE_CLOSE_WRITE", mypath("foo")),
            ("FS_FILE_DELETE", mypath("bar")),
            ("FS_FILE_DELETE", mypath("foo")),
            ("FS_FILE_CREATE", mypath("bar")),
            ("FS_FILE_CLOSE_WRITE", mypath("bar")),
            ("FS_FILE_DELETE", mypath("bar")),
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))

        # generate the events
        open(mypath("foo"), "w").close()
        os.rename(mypath("foo"), mypath("dir", "foo"))
        open(mypath("bar"), "w").close()
        os.rename(mypath("dir", "foo"), mypath("foo"))
        os.rename(mypath("bar"), mypath("dir", "bar"))
        os.remove(mypath("foo"))
        os.rename(mypath("dir", "bar"), mypath("bar"))
        os.remove(mypath("bar"))
        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()

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_FILE_MOVE(innerself, path_from, path_to):
                if path_from != fromfile:
                    self.finished_error("received a wrong path in from")
                elif path_to != tofile:
                    self.finished_error("received a wrong path in to")
                else:
                    os.remove(tofile)
                    self.finished_ok()

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(HitMe())

        # generate the event
        os.rename(fromfile, tofile)
        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)

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_DIR_MOVE(innerself, path_from, path_to):
                if path_from != fromdir:
                    self.finished_error("received a wrong path in from")
                elif path_to != todir:
                    self.finished_error("received a wrong path in to")
                else:
                    os.rmdir(todir)
                    self.finished_ok()

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(HitMe())

        # generate the event
        os.rename(fromdir, todir)
        return self._deferred

    def test_file_moved_inside_mixed(self):
        """Test the synthesis of the FILE_MOVE event with more events."""
        helpdir = os.path.join(self.root_dir, "dir")
        os.mkdir(helpdir)
        mypath = functools.partial(os.path.join, self.root_dir)
        self.fs.create(mypath('foo'), "")
        self.fs.set_node_id(mypath('foo'), "foo_node_id")
        self.fs.create(mypath('bar'), "")
        self.fs.set_node_id(mypath('bar'), "bar_node_id")


        self.eq.inotify_add_watch(self.root_dir)

        should_events = [
            ("FS_FILE_CREATE", mypath("foo")),
            ("FS_FILE_OPEN", mypath("foo")),
            ("FS_FILE_CLOSE_WRITE", mypath("foo")),
            ("FS_FILE_CREATE", mypath("bar")),
            ("FS_FILE_OPEN", mypath("bar")),
            ("FS_FILE_CLOSE_WRITE", mypath("bar")),
            ("FS_FILE_DELETE", mypath("foo")),
            ("FS_FILE_CREATE", mypath("foo")),
            ("FS_FILE_MOVE", mypath("bar"), mypath("baz")),
            ("FS_FILE_DELETE", mypath("foo")),
            ("FS_FILE_DELETE", mypath("baz")),
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))

        # generate the events
        open(mypath("foo"), "w").close()
        open(mypath("bar"), "w").close()
        os.rename(mypath("foo"), mypath("dir", "foo"))
        os.rename(mypath("dir", "foo"), mypath("foo"))
        os.rename(mypath("bar"), mypath("baz"))
        os.remove(mypath("foo"))
        os.remove(mypath("baz"))
        return self._deferred

    def test_dir_with_contents_moved_outside(self):
        """ test the move of a dir outside the watched diresctory."""
        root = os.path.join(self.root_dir, "watched_root")
        os.mkdir(root)
        trash = os.path.join(self.root_dir, "trash")
        os.mkdir(trash)

        testdir = os.path.join(root, "testdir")
        self.eq.fs.create(testdir, '')
        self.eq.fs.set_node_id(testdir, 'testdir_id')
        os.mkdir(testdir)
        testfile = os.path.join(testdir, "testfile")
        self.eq.fs.create(testfile, '')
        self.eq.fs.set_node_id(testfile, 'testfile_id')
        open(testfile, 'w').close()

        paths = [testdir, testfile]
        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_DIR_DELETE(innerself, path):
                expected = paths.pop()
                if path != expected:
                    self.finished_error("received a wrong path, expected:"
                                        " %s was: %s " % (expected, path))
                elif len(paths) == 0:
                    self.finished_ok()

            def handle_FS_FILE_DELETE(innerself, path):
                self.assertEqual(paths.pop(), path)

        self.eq.inotify_add_watch(root)
        self.eq.subscribe(HitMe())

        # generate the event
        os.rename(testdir, os.path.join(trash, os.path.basename(testdir)))
        return self._deferred

    def test_creation_inside_a_moved_directory(self):
        """Test that renaming a directory is supported."""
        testdir = os.path.join(self.root_dir, "testdir")
        self.eq.fs.create(testdir, '')
        self.eq.fs.set_node_id(testdir, 'testdir_id')
        os.mkdir(testdir)
        newdirname = os.path.join(self.root_dir, "newdir")

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_FILE_CREATE(innerself, path):
                if path != newfilepath:
                    self.finished_error("received a wrong path")
                else:
                    os.remove(newfilepath)
                    os.rmdir(newdirname)
                    self.finished_ok()

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.inotify_add_watch(testdir)
        self.eq.subscribe(HitMe())

        # rename the dir
        os.rename(testdir, newdirname)

        # generate the event
        newfilepath = os.path.join(newdirname, "afile")
        open(newfilepath, "w").close()
        return self._deferred

    def test_outside_file_moved_to(self):
        """Test receiving the create signal on a file when it's moved_to."""
        fromfile = os.path.join(self.root_dir, "foo")
        root_dir = os.path.join(self.root_dir, "my_files")
        tofile = os.path.join(root_dir, "foo")
        mypath = functools.partial(os.path.join, root_dir)
        os.mkdir(root_dir)
        open(fromfile, "w").close()

        should_events = [
            ("FS_FILE_CREATE", mypath("foo")),
            ("FS_FILE_CLOSE_WRITE", mypath("foo")),
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))
        self.eq.inotify_add_watch(root_dir)

        # generate the event
        os.rename(fromfile, tofile)
        return self._deferred

    def test_outside_dir_with_contents_moved_to(self):
        """Test receiving the create signal on a file when it's moved_to."""
        fromdir = os.path.join(self.root_dir, "foo_dir")
        fromfile = os.path.join(fromdir, "foo")
        root_dir = os.path.join(self.root_dir, "my_files")
        mypath = functools.partial(os.path.join, root_dir)
        todir = os.path.join(root_dir, "foo_dir")
        os.mkdir(root_dir)
        os.mkdir(fromdir)
        open(fromfile, "w").close()

        should_events = [
            ("FS_DIR_CREATE", mypath("foo_dir")),
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))
        self.eq.inotify_add_watch(root_dir)

        # generate the event
        os.rename(fromdir, todir)
        return self._deferred

    def test_delete_inside_moving_directory(self):
        """Test to assure that the DELETE signal has the correct path."""
        # basedir
        basedir = os.path.join(self.root_dir, "basedir")
        os.mkdir(basedir)
        self.fs.create(path=basedir, share_id='', is_dir=True)

        # working stuff
        dir1 = os.path.join(basedir, "inside_d")
        dir2 = os.path.join(basedir, "new_d")
        fromfile = os.path.join(dir1, "test_f")
        tofile = os.path.join(dir2, "test_f")
        os.mkdir(dir1)
        open(fromfile, "w").close()

        should_events = [
            ("FS_DIR_MOVE", dir1, dir2),
            ("FS_FILE_DELETE", tofile),
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))
        self.eq.inotify_add_watch(self.root_dir)
        self.eq.inotify_add_watch(basedir)
        self.eq.inotify_add_watch(dir1)

        # generate the event
        os.rename(dir1, dir2)
        os.remove(tofile)
        return self._deferred

    def test_move_conflict_to_new_file(self):
        """Test to assure the signal wents through as a new file."""
        testfile = os.path.join(self.root_dir, "testfile")
        destfile = os.path.join(self.root_dir, "destfile")
        mdid = self.fs.create(testfile, '')
        open(testfile, "w").close()
        self.fs.move_to_conflict(mdid)

        should_events = [
            ("FS_FILE_CREATE", destfile),
            ("FS_FILE_CLOSE_WRITE", destfile),
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))
        self.eq.inotify_add_watch(self.root_dir)

        # generate the event
        os.rename(testfile + ".u1conflict", destfile)
        return self._deferred

    def test_move_conflict_over_file(self):
        """Test to assure the signal wents through as a the file."""
        testfile = os.path.join(self.root_dir, "testfile")
        mdid = self.fs.create(testfile, '')
        open(testfile, "w").close()
        self.fs.move_to_conflict(mdid)

        should_events = [
            ("FS_FILE_CREATE", testfile),
            ("FS_FILE_CLOSE_WRITE", testfile),
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))
        self.eq.inotify_add_watch(self.root_dir)

        # generate the event
        os.rename(testfile + ".u1conflict", testfile)
        return self._deferred

    def test_move_conflict_over_dir(self):
        """Test to assure the signal wents through as a the dir."""
        testdir = os.path.join(self.root_dir, "testdir")
        mdid = self.fs.create(testdir, '', is_dir=True)
        os.makedirs(testdir)
        self.fs.move_to_conflict(mdid)

        should_events = [
            ("FS_DIR_CREATE", testdir),
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))
        self.eq.inotify_add_watch(self.root_dir)

        # generate the event
        os.rename(testdir + ".u1conflict", testdir)
        return self._deferred

    def test_move_conflict_to_new_dir(self):
        """Test to assure the signal wents through as a new dir."""
        testdir = os.path.join(self.root_dir, "testdir")
        destdir = os.path.join(self.root_dir, "destdir")
        mdid = self.fs.create(testdir, '')
        os.makedirs(testdir)
        self.fs.move_to_conflict(mdid)

        should_events = [
            ("FS_DIR_CREATE", destdir),
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))
        self.eq.inotify_add_watch(self.root_dir)

        # generate the event
        os.rename(testdir + ".u1conflict", destdir)
        return self._deferred

    def test_no_read_perms_file(self):
        """Test to assure the signal wents through as a the file."""
        testfile = os.path.join(self.root_dir, "testfile")
        self.fs.create(testfile, '')

        should_events = []
        self.eq.subscribe(DynamicHitMe(should_events, self))
        self.eq.inotify_add_watch(self.root_dir)

        d = self._deferred
        log = self.eq._processor.log
        class Handler(logging.Handler):
            """Handler that trigger the deferred callback."""
            def emit(self, record):
                """Dummy emit."""
                # cleanup, remove the handler
                log.removeHandler(self)
                d.callback(record)
        hdlr = Handler()
        hdlr.setLevel(logging.WARNING)
        log.addHandler(hdlr)

        # generate the event
        open(testfile, "w").close()
        # and change the permissions so it's ignored
        os.chmod(testfile, 0000)

        def check(record):
            self.assertIn(testfile, record.args)
            self.assertEquals(1, len(record.args))
        self._deferred.addCallback(check)
        return self._deferred

    def test_no_read_perms_dir(self):
        """Test to assure the signal wents through as a the file."""
        testdir = os.path.join(self.root_dir, "testdir")
        self.fs.create(testdir, '', is_dir=True)

        should_events = []
        self.eq.subscribe(DynamicHitMe(should_events, self))
        self.eq.inotify_add_watch(self.root_dir)

        d = self._deferred
        log = self.eq._processor.log
        class Handler(logging.Handler):
            """Handler that trigger the deferred callback."""
            def emit(self, record):
                """Dummy emit."""
                # cleanup, remove the handler
                log.removeHandler(self)
                d.callback(record)
        hdlr = Handler()
        hdlr.setLevel(logging.WARNING)
        log.addHandler(hdlr)

        # generate the event
        os.makedirs(testdir)
        # and change the permissions so it's ignored
        os.chmod(testdir, 0000)

        def check(record):
            self.assertIn(testdir, record.args)
            self.assertEquals(1, len(record.args))
        self._deferred.addCallback(check)
        return self._deferred


class FreezeTests(BaseTwisted):
    """Test the freeze mechanism."""

    def test_api(self):
        """API for freeze/freeze_commit stuff."""
        # bad args
        self.assertRaises(TypeError, self.eq.freeze_begin)
        self.assertRaises(TypeError, self.eq.freeze_begin, 1, 2)
        self.assertRaises(TypeError, self.eq.freeze_commit)
        self.assertRaises(TypeError, self.eq.freeze_commit, 1, 2)
        self.assertRaises(TypeError, self.eq.freeze_rollback, 1)
        self.assertRaises(TypeError, self.eq.is_frozen, 1)

        # nothing frozen
        self.assertRaises(ValueError, self.eq.freeze_commit, [])
        self.assertRaises(ValueError, self.eq.freeze_rollback)
        self.assertFalse(self.eq.is_frozen())

        # freeze, no-double-freeze, freeze_commit, no post-commit or rollback
        self.eq.freeze_begin(1)
        self.assertRaises(ValueError, self.eq.freeze_begin, 1)
        self.assertTrue(self.eq.is_frozen())
        self.eq.freeze_commit([])
        self.assertRaises(ValueError, self.eq.freeze_commit, [])
        self.assertRaises(ValueError, self.eq.freeze_rollback)
        self.assertFalse(self.eq.is_frozen())

        # freeze, rollback, no post-commit or rollback
        self.eq.freeze_begin(1)
        self.assertRaises(ValueError, self.eq.freeze_begin, 1)
        self.assertTrue(self.eq.is_frozen())
        self.eq.freeze_rollback()
        self.assertRaises(ValueError, self.eq.freeze_commit, [])
        self.assertRaises(ValueError, self.eq.freeze_rollback)
        self.assertFalse(self.eq.is_frozen())

    def test_commit_no_middle_events(self):
        """Commit behaviour when nothing happened in the middle."""
        testdir = os.path.join(self.root_dir, "foo")
        os.mkdir(testdir)

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_DIR_DELETE(innerself, path):
                if path != "foobar":
                    self.finished_error("received a wrong path")
                else:
                    self.finished_ok()

        def freeze_commit():
            """release with handcrafted event and check result."""
            d = self.eq.freeze_commit([("FS_DIR_DELETE", "foobar")])

            def check(dirty):
                """check dirty"""
                if dirty:
                    self.finished_error("should not be dirty here")
            d.addCallback(check)

        # set up everything and freeze
        self.eq.inotify_add_watch(testdir)
        self.eq.subscribe(HitMe())
        self.eq.freeze_begin(testdir)

        reactor.callLater(.1, freeze_commit)
        return self._deferred

    def test_commit_middle_events(self):
        """Commit behaviour when something happened in the middle."""
        testdir = os.path.join(self.root_dir, "foo")
        testfile = os.path.join(testdir, "bar")
        os.mkdir(testdir)

        def freeze_commit():
            """release and check result."""
            d = self.eq.freeze_commit([("FS_DIR_DELETE", "foobar")])
            def check(dirty):
                """check dirty"""
                if not dirty:
                    self.finished_error("it *should* be dirty here")
                else:
                    self.finished_ok()
            d.addCallback(check)

        # set up everything and freeze
        self.eq.inotify_add_watch(testdir)
        self.eq.subscribe(DontHitMe(self))
        self.eq.freeze_begin(testdir)

        open(testfile, "w").close()
        reactor.callLater(.1, freeze_commit)
        return self._deferred

    def test_rollback(self):
        """Check rollback."""
        testdir = os.path.join(self.root_dir, "foo")
        testfile = os.path.join(testdir, "bar")
        os.mkdir(testdir)

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_FS_DIR_DELETE(innerself, path):
                if path != "foobar":
                    self.finished_error("received a wrong path")
                else:
                    self.finished_ok()


        def freeze_rollback():
            """release with handcrafted event and check result."""
            self.eq.freeze_rollback()
            self.eq.freeze_begin(testdir)
            reactor.callLater(.1,
                        self.eq.freeze_commit, [("FS_DIR_DELETE", "foobar")])

        # set up everything and freeze
        self.eq.inotify_add_watch(testdir)
        self.eq.subscribe(HitMe())
        self.eq.freeze_begin(testdir)

        # don't matter if had changes, rollback cleans them
        open(testfile, "w").close()
        reactor.callLater(.1, freeze_rollback)
        return self._deferred

    def test_selective(self):
        """Check that it's frozen only for a path."""
        testdir = os.path.join(self.root_dir, "foo")
        os.mkdir(testdir)
        testfile = os.path.join(self.root_dir, "bar")

        # helper class, pylint: disable-msg=C0111
        class HitMe(object):
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def __init__(innerself):
                innerself.hist = []

            def handle_FS_FILE_CREATE(innerself, path):
                if path != testfile:
                    self.finished_error("received a wrong path")
                else:
                    innerself.hist.append("create")

            def handle_FS_FILE_CLOSE_WRITE(innerself, path):
                if path != testfile:
                    self.finished_error("received a wrong path")
                else:
                    if innerself.hist == ["create"]:
                        os.remove(testfile)
                        self.finished_ok()
                    else:
                        msg = "Finished in bad condition: %s" % innerself.hist
                        self.finished_error(msg)

        # set up everything
        self.eq.inotify_add_watch(self.root_dir)
        self.eq.inotify_add_watch(testdir)
        self.eq.subscribe(HitMe())

        # only freeze one path
        self.eq.freeze_begin(testdir)

        # generate events in the nonfrozen path
        open(testfile, "w").close()

        return self._deferred


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

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

    def test_mute_and_remove(self):
        """Test add and remove the mute."""
        # add
        self.eq.add_to_mute_filter('FS_FILE_OPEN', 'somepath')
        self.assertEqual(self.eq._processor._to_mute._cnt,
                         {('FS_FILE_OPEN', 'somepath'): 1})
        self.eq.add_to_mute_filter('FS_FILE_OPEN', 'somepath')
        self.assertEqual(self.eq._processor._to_mute._cnt,
                         {('FS_FILE_OPEN', 'somepath'): 2})
        self.eq.add_to_mute_filter('FS_FILE_OPEN', 'otherpath')
        self.assertEqual(self.eq._processor._to_mute._cnt,
                         {('FS_FILE_OPEN', 'somepath'): 2,
                          ('FS_FILE_OPEN', 'otherpath'): 1})

        # remove
        self.eq.rm_from_mute_filter('FS_FILE_OPEN', 'somepath')
        self.assertEqual(self.eq._processor._to_mute._cnt,
                         {('FS_FILE_OPEN', 'somepath'): 1,
                          ('FS_FILE_OPEN', 'otherpath'): 1})
        self.eq.rm_from_mute_filter('FS_FILE_OPEN', 'otherpath')
        self.assertEqual(self.eq._processor._to_mute._cnt,
                         {('FS_FILE_OPEN', 'somepath'): 1})
        self.eq.rm_from_mute_filter('FS_FILE_OPEN', 'somepath')
        self.assertEqual(self.eq._processor._to_mute._cnt, {})


    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.eq.add_to_mute_filter("FS_FILE_OPEN", testfile)
        self.eq.add_to_mute_filter("FS_FILE_CLOSE_NOWRITE", testfile)

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(DontHitMe(self))

        # 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.eq.add_to_mute_filter("FS_FILE_CLOSE_NOWRITE", testfile)

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(DontHitMe(self))

        # 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.eq.add_to_mute_filter("FS_FILE_CREATE", testfile)
        self.eq.add_to_mute_filter("FS_FILE_OPEN", testfile)
        self.eq.add_to_mute_filter("FS_FILE_CLOSE_WRITE", testfile)

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(DontHitMe(self))

        # 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.eq.add_to_mute_filter("FS_DIR_CREATE", testdir)

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(DontHitMe(self))

        # 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.eq.add_to_mute_filter("FS_FILE_DELETE", testfile)

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(DontHitMe(self))

        # 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.eq.add_to_mute_filter("FS_DIR_DELETE", testdir)

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(DontHitMe(self))

        # 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.eq.add_to_mute_filter("FS_FILE_MOVE", fromfile, tofile)

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(DontHitMe(self))

        # 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.eq.add_to_mute_filter("FS_DIR_MOVE", fromdir, todir)

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(DontHitMe(self))

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

    def test_file_moved_to_conflict(self):
        """Test the handling of the FILE_MOVE event when dest is conflict."""
        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, "foo.u1conflict")
        self.fs.create(tofile, "")
        self.fs.set_node_id(tofile, "to_node_id")
        open(fromfile, "w").close()
        self.eq.add_to_mute_filter("FS_FILE_MOVE", fromfile, tofile)

        self.eq.inotify_add_watch(self.root_dir)
        self.eq.subscribe(DontHitMe(self))

        # 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.eq.add_to_mute_filter("FS_FILE_CREATE", tofile)
        self.eq.add_to_mute_filter("FS_FILE_CLOSE_WRITE", tofile)

        self.eq.inotify_add_watch(root_dir)
        self.eq.subscribe(DontHitMe(self))

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


class AncestorsUDFTestCase(testcase.BaseTwistedTestCase):
    """Events over UDF's ancestor are properly handled."""

    timeout = 2

    def setUp(self):
        """Init."""
        testcase.BaseTwistedTestCase.setUp(self)
        self._deferred = defer.Deferred()
        self.home_dir = self.mktemp('ubuntuonehacker')
        self.root_dir = self.mktemp('root_dir')
        self.data_dir = self.mktemp('data_dir')
        self.shares_dir = self.mktemp('shares_dir')
        self.partials_dir = self.mktemp('partials_dir')
        self.main = testcase.FakeMain(self.root_dir, self.shares_dir,
                             self.data_dir, self.partials_dir)
        self.eq = self.main.event_q
        assert self.eq is self.eq.fs.vm.m.event_q

        self.listener = testcase.Listener()
        self.eq.subscribe(self.listener)

        self.env_var = 'HOME'
        self.old_value = os.environ.get(self.env_var, None)
        os.environ[self.env_var] = self.home_dir

        # create UDF
        suggested_path = u'~/Documents/Reading/Books/PDFs'
        udf_id, node_id = u'udf_id', u'node_id'
        path = os.path.expanduser(suggested_path).encode("utf8")
        self.udf = volume_manager.UDF(udf_id, node_id,
                                      suggested_path, path, True)
        os.makedirs(path)
        d = self.eq.fs.vm.add_udf(self.udf)

        # create a second UDF
        suggested_path = u'~/Documents/Reading/Magazines/Text'
        udf_id2, node_id2 = u'udf_id_2', u'node_id_2'
        path = os.path.expanduser(suggested_path).encode("utf8")
        self.udf2 = volume_manager.UDF(udf_id2, node_id2,
                                       suggested_path, path, True)
        os.makedirs(path)
        d.addCallback(lambda _: self.eq.fs.vm.add_udf(self.udf2))

        # every ancestor has a watch already, added by LocalRescan. Copy that.
        self.eq.inotify_add_watch(self.udf.path)
        for path in self.udf.ancestors:
            self.eq.inotify_add_watch(path)

        self.eq.inotify_add_watch(self.udf2.path)
        for path in self.udf2.ancestors:
            self.eq.inotify_add_watch(path)

        # reset events up to now
        d.addCallback(lambda _: setattr(self.listener, 'events', []))
        return d

    def tearDown(self):
        """Cleanup."""
        self.eq.unsubscribe(self.listener)
        self.main.shutdown()

        self.rmtree(self.home_dir)
        self.rmtree(self.root_dir)
        self.rmtree(self.data_dir)
        self.rmtree(self.shares_dir)
        self.rmtree(self.partials_dir)

        if self.old_value is None:
            os.environ.pop(self.env_var)
        else:
            os.environ[self.env_var] = self.old_value

        testcase.BaseTwistedTestCase.tearDown(self)

    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._deferred.errback(msg)
            raise exception
        return first
    assertEqual = assertEquals = failUnlessEquals = failUnlessEqual

    def test_file_events_are_ignored_on_udf_ancestor(self):
        """Events on UDF ancestors are ignored."""
        for path in self.udf.ancestors:
            assert self.eq.inotify_has_watch(path)

            fname = os.path.join(path, 'testit')
            # generate FS_FILE_CREATE, FS_FILE_OPEN, FS_FILE_CLOSE_WRITE
            open(fname, 'w').close()
            # generate FS_FILE_CLOSE_NOWRITE
            with open(fname) as f:
                f.read()
            # generate FS_FILE_DELETE
            os.remove(fname)

            fnamedir = os.path.join(path, 'testit.dir')
            # generate FS_DIR_CREATE
            os.mkdir(fnamedir)
            # generate FS_DIR_DELETE
            os.rmdir(fnamedir)

        def check():
            """Check."""
            self.assertEquals([], self.listener.events)
            self._deferred.callback(True)

        reactor.callLater(.1, check)
        return self._deferred

    def test_file_events_are_not_ignored_on_others(self):
        """Events in the UDF are not ignored."""
        path = self.udf.path
        assert self.eq.inotify_has_watch(path)

        fname = os.path.join(path, 'testit')
        # generate FS_FILE_CREATE, FS_FILE_OPEN, FS_FILE_CLOSE_WRITE
        open(fname, 'w').close()
        # generate FS_FILE_CLOSE_NOWRITE
        with open(fname) as f:
            f.read()
        # generate FS_FILE_DELETE
        os.remove(fname)

        fnamedir = os.path.join(path, 'testit.dir')
        # generate FS_DIR_CREATE
        os.mkdir(fnamedir)
        # generate FS_DIR_DELETE
        os.rmdir(fnamedir)

        expected = [('FS_FILE_CREATE', (fname,), {}),
                    ('FS_FILE_OPEN', (fname,), {}),
                    ('FS_FILE_CLOSE_WRITE', (fname,), {}),
                    ('FS_FILE_OPEN', (fname,), {}),
                    ('FS_FILE_CLOSE_NOWRITE', (fname,), {}),
                    ('FS_FILE_DELETE', (fname,), {}),
                    ('FS_DIR_CREATE', (fnamedir,), {}),
                    ('FS_DIR_DELETE', (fnamedir,), {})]

        def check():
            """Check."""
            self.assertEquals(expected, self.listener.events)
            self._deferred.callback(True)

        reactor.callLater(.1, check)
        return self._deferred

    def test_file_events_are_not_ignored_on_common_prefix_name(self):
        """Events in a UDF with similar name to ancestor are not ignored."""
        fname = os.path.join(self.udf2.path, 'testit')
        expected = [('FS_FILE_CREATE', (fname,), {}),
                    ('FS_FILE_OPEN', (fname,), {}),
                    ('FS_FILE_CLOSE_WRITE', (fname,), {}),
                    ('FS_FILE_OPEN', (fname,), {}),
                    ('FS_FILE_CLOSE_NOWRITE', (fname,), {})]

        def check():
            """Check."""
            self.assertEquals(expected, self.listener.events)
            self._deferred.callback(True)

        # generate FS_FILE_CREATE, FS_FILE_OPEN, FS_FILE_CLOSE_WRITE
        open(fname, 'w').close()

        # generate FS_FILE_CLOSE_NOWRITE
        with open(fname) as f:
            f.read()

        reactor.callLater(.1, check)
        return self._deferred

    def test_move_udf_ancestor(self):
        """UDF is unsubscribed on ancestor move."""
        original = self.eq.fs.vm.unsubscribe_udf
        path = self.udf.ancestors[-2] # an ancestor common to both UDFs
        # generate IN_MOVED_FROM and IN_MOVED_TO
        newpath = path + u'.old'
        os.rename(path, newpath)
        assert os.path.exists(newpath)

        unsubscribed = []
        def check():
            """Check."""
            for uid in unsubscribed:
                if uid == self.udf.id:
                    udf = self.udf
                elif uid == self.udf2.id:
                    udf = self.udf2
                else:
                    self._deferred.errback(ValueError("uid %s is wrong" % uid))
                self.assertEquals(False, self.eq.inotify_has_watch(udf.path),
                                  'watch must be removed')
                self.assertEquals(False, self.eq.fs.vm.udfs[udf.id].subscribed)
            self._deferred.callback(True)

        def unsubsc(uid):
            original(uid)
            unsubscribed.append(uid)
        self.eq.fs.vm.unsubscribe_udf = unsubsc

        def restore():
            """Restore original unsubscribe_udf."""
            self.eq.fs.vm.unsubscribe_udf = original
            shutil.rmtree(newpath)
        self.addCleanup(restore)

        reactor.callLater(.1, check)
        return self._deferred

    def test_move_udf_itself(self):
        """UDF is unsubscribed if renamed."""
        original = self.eq.fs.vm.unsubscribe_udf
        newpath = self.udf.path + u'.old'
        os.rename(self.udf.path, newpath)
        assert os.path.exists(newpath)

        unsubscribed = []
        def check():
            """Check."""
            self.assertEquals(len(unsubscribed), 1)
            uid = unsubscribed[0]
            self.assertEquals(uid, self.udf.id, "wrong UDF removed!")
            self.assertEquals(False, self.eq.inotify_has_watch(self.udf.path),
                              'watch must be removed')
            self.assertEquals(False,
                              self.eq.fs.vm.udfs[self.udf.id].subscribed)
            self._deferred.callback(True)

        def unsubsc(uid):
            original(uid)
            unsubscribed.append(uid)
        self.eq.fs.vm.unsubscribe_udf = unsubsc

        def restore():
            """Restore original unsubscribe_udf."""
            self.eq.fs.vm.unsubscribe_udf = original
            shutil.rmtree(newpath)
        self.addCleanup(restore)

        reactor.callLater(.1, check)
        return self._deferred

    def test_remove_udf_per_se_subsc(self):
        """Removing an UDF should generate VOLUME_DELETED, if subscribed."""
        original = self.eq.fs.vm.delete_volume
        expected = []
        # pylint: disable-msg=W0108
        self.eq.fs.vm.delete_volume = lambda uid: expected.append(uid)

        def restore():
            """Restore original delete_volume."""
            self.eq.fs.vm.delete_volume = original
        self.addCleanup(restore)

        uid = self.udf.volume_id
        fnamedir = self.udf.path
        os.rmdir(fnamedir)

        def check():
            """Check."""
            self.assertEquals([], self.listener.events)
            self.assertEquals(expected, [uid], 'udf deleted')
            self._deferred.callback(True)

        reactor.callLater(.1, check)
        return self._deferred

    def test_remove_udf_per_se_unsubsc(self):
        """If not subscribed, removing an UDF should do nothing."""
        original = self.eq.fs.vm.delete_volume
        expected = []
        # pylint: disable-msg=W0108
        self.eq.fs.vm.delete_volume = lambda uid: expected.append(uid)

        def restore():
            """Restore original delete_volume."""
            self.eq.fs.vm.delete_volume = original
        self.addCleanup(restore)

        # unsubscribe and reset listener events
        self.eq.fs.vm.unsubscribe_udf(self.udf.volume_id)
        self.listener.events = []

        fnamedir = self.udf.path
        os.rmdir(fnamedir)

        def check():
            """Check."""
            self.assertEquals([], self.listener.events)
            self.assertEquals(expected, [], 'udf should not be deleted')
            self._deferred.callback(True)

        reactor.callLater(.1, check)
        return self._deferred

    def test_removeudf_removeancestors(self):
        """If an UDF is removed, also the watches in its ancestors."""
        removed_watches = []
        def remove_watch(path):
            """Store the path."""
            removed_watches.append(path)

        original = self.eq.inotify_rm_watch
        self.eq.inotify_rm_watch = remove_watch

        def restore():
            """Restore original rm_watch."""
            self.eq.inotify_rm_watch = original
        self.addCleanup(restore)

        # only the parent, as the other ancestors are shared with other UDFs
        # and should not be removed
        expected = os.path.dirname(self.udf.path)
        def check():
            self.assertEquals([expected], removed_watches,
                              "Removed watches don't match the expected")
            self._deferred.callback(True)

        os.rmdir(self.udf.path)
        reactor.callLater(.1, check)
        return self._deferred

    def test_unsubscribe_removeancestors(self):
        """Remove the watches of the ancestors in an unsubscription."""
        removed_watches = []
        original = self.eq.inotify_rm_watch
        def remove_watch(path):
            """Store the path."""
            original(path)
            removed_watches.append(path)

        self.eq.inotify_rm_watch = remove_watch

        def restore():
            """Restore original rm_watch."""
            self.eq.inotify_rm_watch = original
        self.addCleanup(restore)

        # only the parent, as the other ancestors are shared with other UDFs
        # and should not be removed, and the path of the udf itself
        expected = [os.path.dirname(self.udf.path), self.udf.path]
        def check():
            self.assertEquals(sorted(expected), sorted(removed_watches),
                              "Removed watches don't match the expected")
            self._deferred.callback(True)

        os.rename(self.udf.path, self.udf.path + ".old")
        reactor.callLater(.1, check)
        return self._deferred

    def test_unsubscribe_rename_removeancestors(self):
        """Mix of unsubscription and further renaming."""
        removed_watches = []
        original = self.eq.inotify_rm_watch
        def remove_watch(path):
            """Store the path."""
            original(path)
            removed_watches.append(path)

        self.eq.inotify_rm_watch = remove_watch

        def restore():
            """Restore original rm_watch."""
            self.eq.inotify_rm_watch = original
        self.addCleanup(restore)

        # all should be removed
        expected = list(set(self.udf.ancestors) | set(self.udf2.ancestors)) + [
                            self.udf.path, self.udf2.path]
        def check():
            self.assertEquals(sorted(expected), sorted(removed_watches),
                              "Removed watches don't match the expected")
            self._deferred.callback(True)

        path = self.udf.ancestors[-2] # an ancestor common to both UDFs

        os.rename(self.udf.path, self.udf.path + ".old")
        os.rename(path, path + ".old")
        reactor.callLater(.1, check)
        return self._deferred


class NonUTF8NamesTests(BaseTwisted):
    """Test the non-utf8 name handling."""

    invalid_name = "invalid \xff\xff name"

    def test_file_open(self):
        """Test invalid_filename after open a file."""
        testfile = os.path.join(self.root_dir, self.invalid_name)
        open(testfile, "w").close()

        self.eq.inotify_add_watch(self.root_dir)
        should_events = [
            ('FS_INVALID_NAME', self.root_dir, self.invalid_name), # open
            ('FS_INVALID_NAME', self.root_dir, self.invalid_name), # close no w
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))

        # generate the event
        open(testfile)
        return self._deferred

    def test_file_close_nowrite(self):
        """Test invalid_filename after a close no write."""
        testfile = os.path.join(self.root_dir, self.invalid_name)
        open(testfile, "w").close()
        fh = open(testfile)

        self.eq.inotify_add_watch(self.root_dir)
        should_events = [
            ('FS_INVALID_NAME', self.root_dir, self.invalid_name), # close no w
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))

        # generate the event
        fh.close()
        return self._deferred

    def test_file_create_close_write(self):
        """Test invalid_filename after a create, open and close write."""
        testfile = os.path.join(self.root_dir, self.invalid_name)

        self.eq.inotify_add_watch(self.root_dir)
        should_events = [
            ('FS_INVALID_NAME', self.root_dir, self.invalid_name), # create
            ('FS_INVALID_NAME', self.root_dir, self.invalid_name), # open
            ('FS_INVALID_NAME', self.root_dir, self.invalid_name), # close w
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))

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

    def test_dir_create(self):
        """Test invalid_filename after a dir create."""
        testdir = os.path.join(self.root_dir, self.invalid_name)

        self.eq.inotify_add_watch(self.root_dir)
        should_events = [
            ('FS_INVALID_NAME', self.root_dir, self.invalid_name), # create
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))

        # generate the event
        os.mkdir(testdir)
        return self._deferred

    def test_file_delete(self):
        """Test invalid_filename after a file delete."""
        testfile = os.path.join(self.root_dir, self.invalid_name)
        open(testfile, "w").close()

        self.eq.inotify_add_watch(self.root_dir)
        should_events = [
            ('FS_INVALID_NAME', self.root_dir, self.invalid_name), # delete
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))

        # generate the event
        os.remove(testfile)
        return self._deferred

    def test_dir_delete(self):
        """Test invalid_filename after a dir delete."""
        testdir = os.path.join(self.root_dir, self.invalid_name)
        os.mkdir(testdir)

        self.eq.inotify_add_watch(self.root_dir)
        should_events = [
            ('FS_INVALID_NAME', self.root_dir, self.invalid_name), # delete
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))

        # generate the event
        os.rmdir(testdir)
        return self._deferred

    def test_file_move_to(self):
        """Test invalid_filename after moving a file into a watched dir."""
        fromfile = os.path.join(self.root_dir, self.invalid_name)
        open(fromfile, "w").close()
        destdir = os.path.join(self.root_dir, "watched_dir")
        os.mkdir(destdir)
        destfile = os.path.join(destdir, self.invalid_name)

        self.eq.inotify_add_watch(destdir)
        should_events = [
            ('FS_INVALID_NAME', destdir, self.invalid_name), # move to
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))

        # generate the event
        os.rename(fromfile, destfile)
        return self._deferred

    def test_file_move_from(self):
        """Test invalid_filename after moving a file from a watched dir."""
        fromdir = os.path.join(self.root_dir, "watched_dir")
        os.mkdir(fromdir)
        fromfile = os.path.join(fromdir, self.invalid_name)
        open(fromfile, "w").close()
        destfile = os.path.join(self.root_dir, self.invalid_name)

        self.eq.inotify_add_watch(fromdir)
        should_events = [
            ('FS_INVALID_NAME', fromdir, self.invalid_name), # move from
        ]
        self.eq.subscribe(DynamicHitMe(should_events, self))

        # generate the event
        os.rename(fromfile, destfile)
        return self._deferred


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

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