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

test_eventqueue.py

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

'''Tests for the Event Queue.'''

import logging
import unittest

from ubuntuone.syncdaemon import (
    event_queue,
    filesystem_manager,
)
from contrib.testing import testcase
from twisted.internet import defer


00033 class BaseEQTestCase(testcase.BaseTwistedTestCase):
    ''' Setup an EQ for test. '''

00036     def setUp(self):
        '''Setup the test.'''
        testcase.BaseTwistedTestCase.setUp(self)
        self.fsmdir = self.mktemp('fsmdir')
        self.root_dir = self.mktemp('root_dir')
        self.fs = filesystem_manager.FileSystemManager(self.fsmdir,
                                   testcase.FakeVolumeManager(self.root_dir))
        mdobj = self.fs.create(path=self.root_dir,
                                 share_id='', is_dir=True)
        self.fs.set_by_path(path=self.root_dir,
                              local_hash=None, server_hash=None)
        self.eq = event_queue.EventQueue(self.fs)

00049     def tearDown(self):
        '''Clean up the tests.'''
        testcase.BaseTwistedTestCase.tearDown(self)
        self.eq.shutdown()


00055 class SubscriptionTests(BaseEQTestCase):
    '''Test to subscribe and unsubscribe to the EQ.'''

00058     def test_subscription(self):
        '''Test that subscription works.'''
        # simple one
        o = object()
        self.eq.subscribe(o)
        self.assertTrue(o in self.eq._listeners)

        # not duplicates
        o = object()
        self.eq.subscribe(o)
        self.eq.subscribe(o)
        self.assertTrue(self.eq._listeners.count(o) == 1)

00071     def test_unsubscription(self):
        '''Test that unsubscription works.'''
        # simple one
        o = object()
        self.eq.subscribe(o)
        self.eq.unsubscribe(o)
        self.assertFalse(o in self.eq._listeners)

        # duplicate ok (two suscs)
        o = object()
        self.eq.subscribe(o)
        self.eq.subscribe(o)
        self.eq.unsubscribe(o)
        self.assertFalse(o in self.eq._listeners)

        # duplicate error (two unsuscs)
        o = object()
        self.eq.subscribe(o)
        self.eq.subscribe(o)
        self.eq.unsubscribe(o)
        self.assertRaises(ValueError, self.eq.unsubscribe, o)

        # indecisive
        o = object()
        self.eq.subscribe(o)
        self.eq.unsubscribe(o)
        self.eq.subscribe(o)
        self.eq.unsubscribe(o)
        self.assertFalse(o in self.eq._listeners)


00102 class PushTests(BaseEQTestCase):
    '''Test the event distribution machinery.'''

00105     def test_push_simple(self):
        '''Test that events can be pushed (not listening yet).'''
        # not even an event
        self.assertRaises(TypeError, self.eq.push)

        # bad event
        self.assertRaises(event_queue.InvalidEventError,
                          self.eq.push, "no-such-event")

        # not enough or incorrect args for that event
        self.assertRaises(TypeError, self.eq.push, "FS_FILE_MOVE")
        self.assertRaises(TypeError, self.eq.push, "FS_FILE_MOVE", 1, 2, 3)
        self.assertRaises(TypeError,
                      self.eq.push, "FS_FILE_MOVE", 1, 2, path_to=3)
        self.assertRaises(TypeError,
                      self.eq.push, "FS_FILE_MOVE", 1, path_from=2, path_to=3)

        # all ok... args, kwargs, and mixed
        self.eq.push("FS_FILE_MOVE", 1, 2)
        self.eq.push("FS_FILE_MOVE", path_from=1, path_to=2)
        self.eq.push("FS_FILE_MOVE", 1, path_to=2)

00127     def test_listened_pushs(self):
        '''Push events and listem them.'''

        # helper class, pylint: disable-msg=C0111
        class C(object):
            def __init__(self):
                self.a = None
            def handle_FS_FILE_CREATE(self, a):
                self.a = a

        # it get passed!
        c = C()
        self.eq.subscribe(c)
        self.eq.push("FS_FILE_CREATE", 1)
        self.assertEqual(c.a, 1)
        self.eq.unsubscribe(c)

        # don't get what don't listen
        c = C()
        self.eq.subscribe(c)
        self.eq.push("FS_FILE_DELETE", 1)
        self.assertEqual(c.a, None)
        self.eq.unsubscribe(c)

00151     def test_signatures(self):
        '''Check that the handle signatures are forced when passing.'''

        # helper class, pylint: disable-msg=C0111
        class C(object):
            def handle_FS_FILE_CREATE(self, a, b): # it should be only 'a' here
                pass

        # it get passed!
        c = C()
        self.eq.subscribe(c)

        # caught by EQ
        self.assertRaises(TypeError, self.eq.push, "FS_FILE_CREATE")
        self.assertRaises(TypeError, self.eq.push, "FS_FILE_CREATE", 1, 2)

        # approved by EQ, but the listener has a wrong signature
        # this is logged as an error/exception, so we hook to the logger
        handler = testcase.MementoHandler()
        handler.setLevel(logging.ERROR)
        self.eq.log.addHandler(handler)

        self.eq.push("FS_FILE_CREATE", 1)
        self.assertEquals(1, len(handler.records))
        self.assertEquals('FS_FILE_CREATE', handler.records[0].args[0])
        self.assertEquals(C, type(handler.records[0].args[1]))

        self.eq.log.removeHandler(handler)
        self.eq.unsubscribe(c)


00182 class PushTestsWithCallback(BaseEQTestCase):
    '''Test the error handling in the event distribution machinery.'''

00185     def test_keep_going(self):
        ''' Check that if a listener raises an Exception or have a
        wrong signature, the next listeners are called.
        '''
        d = defer.Deferred()
        # helper class, pylint: disable-msg=C0111
        class BadListener(object):
            def handle_FS_FILE_CREATE(self, a, b): # it should be only 'a' here
                d.callback(False)

        class GoodListener(object):
            def handle_FS_FILE_CREATE(self, a):
                d.callback(a)

        bl = BadListener()
        gl = GoodListener()
        self.eq.subscribe(bl)
        self.eq.subscribe(gl)

        def cleanup():
            """ unsubscribe the listeners """
            self.eq.unsubscribe(bl)
            self.eq.unsubscribe(gl)
        self.addCleanup(cleanup)

        # caught by EQ
        self.assertRaises(TypeError, self.eq.push, "FS_FILE_CREATE")
        self.assertRaises(TypeError, self.eq.push, "FS_FILE_CREATE", 1, 2)

        # approved by EQ, but one listener has a wrong signature
        self.eq.push("FS_FILE_CREATE", 1)
        def callback(result):
            """ asserts that GoodListener was called. """
            self.assertTrue(result)
            self.assertEquals(1, result)

        d.addCallback(callback)
        return d

00224     def test_default_handler(self):
        ''' Check that handler_default is called. '''
        d = defer.Deferred()
        # helper class, pylint: disable-msg=C0111
        class Listener(object):
            def handle_default(self, *args, **kwargs):
                d.callback(args)

        l = Listener()
        self.eq.subscribe(l)

        def cleanup():
            """ unsubscribe the listeners """
            self.eq.unsubscribe(l)
        self.addCleanup(cleanup)

        # push some event and expect it'll be handled by handle_default
        self.eq.push("FS_FILE_CREATE", 1)
        def callback(result):
            """ asserts that GoodListener was called. """
            self.assertEquals(2, len(result))
            self.assertEquals('FS_FILE_CREATE', result[0])
            self.assertEquals(1, result[1])

        d.addCallback(callback)
        return d

00251     def test_ordered_dispatch(self):
        """ Check that the events are pushed to all listeners in order. """
        d = defer.Deferred()
        # helper class, pylint: disable-msg=C0111
        class Listener(object):
            def __init__(self, eq):
                self.eq = eq
                self.events = []

            def handle_FS_FILE_CREATE(self, a):
                self.events.append('FS_FILE_CREATE')
                self.eq.push('FS_FILE_MOVE', a, 2)

            def handle_FS_FILE_DELETE(self, a):
                self.events.append('FS_FILE_DELETE')

            def handle_FS_FILE_MOVE(self, *args):
                self.events.append('FS_FILE_MOVE')
                d.callback(True)

        # create 10 listeners in order to create an event madness
        listeners = []
        for i in xrange(0, 10):
            l = Listener(self.eq)
            listeners.append(l)
            self.eq.subscribe(l)

        def cleanup():
            """ unsubscribe the listeners """
            for l in listeners:
                self.eq.unsubscribe(l)
        self.addCleanup(cleanup)

        # push some events to unleash the event madness
        self.eq.push("FS_FILE_CREATE", 1)
        self.eq.push('FS_FILE_DELETE', 2)

        def callback(result):
            """ asserts that Listener was called in the right order"""
            listeners_events = [listener.events for listener in listeners]
            for l_events in listeners_events:
                for other_l_events in listeners_events:
                    self.assertEqual(l_events, other_l_events)

        d.addCallback(callback)
        return d


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

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

Generated by  Doxygen 1.6.0   Back to index