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

test_eq_inotify.py

# 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 os
import unittest
import functools

from twisted.internet import defer, reactor
from tests.syncdaemon.test_eventqueue import BaseEQTestCase


00029 class WatchTests(BaseEQTestCase):
    '''Test the EQ API to add and remove watchs.'''

00032     def test_add_watch(self):
        '''Test that watchs can be added.'''
        # we should have what we asked for
        self.eq.inotify_add_watch(self.root_dir)
        self.assertTrue(self.root_dir in self.eq._watchs)

        # we shouldn't have other stuff
        self.assertTrue("not-added-dir" not in self.eq._watchs)

00041     def test_rm_watch(self):
        '''Test that watchs can be removed.'''
        # remove what we added
        self.eq.inotify_add_watch(self.root_dir)
        self.eq.inotify_rm_watch(self.root_dir)
        self.assertTrue(self.root_dir not in self.eq._watchs)

        # remove different stuff
        self.eq.inotify_add_watch(self.root_dir)
        self.assertRaises(ValueError,
                          self.eq.inotify_rm_watch, "not-added-dir")


00054 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)
        self.final_step = self.should_events[-1][0]
        self.should_events.reverse()
        self.test_machinery = test_machinery

00064     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

00092 class BaseTwisted(BaseEQTestCase):
    '''Base class for twisted tests.'''

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

00098     def setUp(self):
        '''Setup the test.'''
        BaseEQTestCase.setUp(self)
        # create the deferred for the tests
        self._deferred = defer.Deferred()

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

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


00113 class SignalingTests(BaseTwisted):
    '''Test the whole stuff to receive signals.'''

00116     def test_file_create_close(self):
        '''Test receiving the create and close 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

00150     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

00171     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")
                else:
                    self.finished_ok()

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

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

00192     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")
                else:
                    self.finished_ok()

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

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

00213     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")

        class DontHitMe(object):
            '''we shouldn't be called'''
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_default(innerself, *a):
                '''Something here? Error!'''
                self.finished_error("don't hit me! received %s" % (a,))

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

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

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

00240     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

00266     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

00292     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

00318     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

00344     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_CLOSE_WRITE", mypath("foo")),
            ("FS_FILE_DELETE", mypath("foo")),
            ("FS_FILE_CREATE", 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

00379     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

00408     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

00437     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_CLOSE_WRITE", mypath("foo")),
            ("FS_FILE_CREATE", 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

00473     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

00511     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

00542     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

00562     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

00583 class FreezeTests(BaseTwisted):
    '''Test the freeze mechanism.'''

00586     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)

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

        # 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.eq.freeze_commit([])
        self.assertRaises(ValueError, self.eq.freeze_commit, [])
        self.assertRaises(ValueError, self.eq.freeze_rollback)

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

00613     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

00645     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)

        class DontHitMe(object):
            '''we shouldn't be called'''
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def handle_default(innerself, *a):
                '''Something here? Error!'''
                self.finished_error("don't hit me! received %s" % (a,))

        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.eq.freeze_begin(testdir)

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

00678     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

00711     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


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

def Xtest_suite():
    # pylint: disable-msg=C0111
    loader = unittest.TestLoader()
    suite = unittest.TestSuite()
    suite.addTests(loader.loadTestsFromTestCase(SignalingTests))
    return suite

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

Generated by  Doxygen 1.6.0   Back to index