server/test/unittest_hooksmanager.py
author Sandrine Ribeau <sandrine.ribeau@logilab.fr>
Thu, 06 Nov 2008 16:21:57 -0800
changeset 11 db9c539e0b1b
parent 0 b97547f5f1fa
child 1398 5fe84a5f7035
permissions -rw-r--r--
Add module wfobjs to enable workflow in gae. Add module vcard to enable usage of cube person in gae. Add init file creation for cubes gae directory.

"""unit tests for the hooks manager
"""

from logilab.common.testlib import TestCase, unittest_main

from cubicweb.server.hooksmanager import HooksManager, Hook
from cubicweb.devtools import TestServerConfiguration
from cubicweb.devtools.apptest import RepositoryBasedTC

class HookCalled(Exception): pass

config = TestServerConfiguration('data')
config.bootstrap_cubes()
schema = config.load_schema()

class HooksManagerTC(TestCase):
    args = (None,)
    kwargs = {'a': 1}
    
    def setUp(self):
        """ called before each test from this class """
        self.o = HooksManager(schema)

    def test_register_hook_raise_keyerror(self):
        self.assertRaises(AssertionError,
                          self.o.register_hook, self._hook, 'before_add_entiti')
        self.assertRaises(AssertionError,
                          self.o.register_hook, self._hook, 'session_login', 'EEType')
        self.assertRaises(AssertionError,
                          self.o.register_hook, self._hook, 'session_logout', 'EEType')
        self.assertRaises(AssertionError,
                          self.o.register_hook, self._hook, 'server_startup', 'EEType')
        self.assertRaises(AssertionError,
                          self.o.register_hook, self._hook, 'server_shutdown', 'EEType')
        
    def test_register_hook1(self):
        self.o.register_hook(self._hook, 'before_add_entity')
        self.o.register_hook(self._hook, 'before_delete_entity', 'Personne')
        self._test_called_hooks()
        
    def test_register_hook2(self):
        self.o.register_hook(self._hook, 'before_add_entity', '')
        self.o.register_hook(self._hook, 'before_delete_entity', 'Personne')
        self._test_called_hooks()
        
    def test_register_hook3(self):
        self.o.register_hook(self._hook, 'before_add_entity', None)
        self.o.register_hook(self._hook, 'before_delete_entity', 'Personne')
        self._test_called_hooks()
        
    def test_register_hooks(self):
        self.o.register_hooks({'before_add_entity' : {'': [self._hook]},
                               'before_delete_entity' : {'Personne': [self._hook]},
                               })
        self._test_called_hooks()

    def test_unregister_hook(self):
        self.o.register_hook(self._hook, 'after_delete_entity', 'Personne')
        self.assertRaises(HookCalled,
                          self.o.call_hooks, 'after_delete_entity', 'Personne',
                          *self.args, **self.kwargs)
        self.o.unregister_hook(self._hook, 'after_delete_entity', 'Personne')
        # no hook should be called there
        self.o.call_hooks('after_delete_entity', 'Personne')
        

    def _test_called_hooks(self):
        self.assertRaises(HookCalled,
                          self.o.call_hooks, 'before_add_entity', '',
                          *self.args, **self.kwargs)
        self.assertRaises(HookCalled,
                          self.o.call_hooks, 'before_add_entity', None,
                          *self.args, **self.kwargs)
        self.assertRaises(HookCalled,
                          self.o.call_hooks, 'before_add_entity', 'Personne',
                          *self.args, **self.kwargs)
        self.assertRaises(HookCalled,
                          self.o.call_hooks, 'before_delete_entity', 'Personne',
                          *self.args, **self.kwargs)
        # no hook should be called there
        self.o.call_hooks('before_delete_entity', None)
        self.o.call_hooks('before_delete_entity', 'Societe')


    def _hook(self, *args, **kwargs):
        # check arguments
        self.assertEqual(args, self.args)
        self.assertEqual(kwargs, self.kwargs)
        raise HookCalled()


class RelationHookTC(TestCase):
    """testcase for relation hooks grouping"""
    def setUp(self):
        """ called before each test from this class """
        self.o = HooksManager(schema)
        self.called = []

    def test_before_add_relation(self):
        """make sure before_xxx_relation hooks are called directly"""
        self.o.register_hook(self._before_relation_hook,
                             'before_add_relation', 'concerne')
        self.assertEquals(self.called, [])
        self.o.call_hooks('before_add_relation', 'concerne', 'USER',
                          1, 'concerne', 2)        
        self.assertEquals(self.called, [(1, 'concerne', 2)])
        
    def test_after_add_relation(self):
        """make sure after_xxx_relation hooks are deferred"""
        self.o.register_hook(self._after_relation_hook,
                             'after_add_relation', 'concerne')
        self.assertEquals(self.called, [])
        self.o.call_hooks('after_add_relation', 'concerne', 'USER',
                          1, 'concerne', 2)
        self.o.call_hooks('after_add_relation', 'concerne', 'USER',
                          3, 'concerne', 4)
        self.assertEquals(self.called, [(1, 'concerne', 2), (3, 'concerne', 4)])
    
    def test_before_delete_relation(self):
        """make sure before_xxx_relation hooks are called directly"""
        self.o.register_hook(self._before_relation_hook,
                             'before_delete_relation', 'concerne')
        self.assertEquals(self.called, [])
        self.o.call_hooks('before_delete_relation', 'concerne', 'USER',
                          1, 'concerne', 2)        
        self.assertEquals(self.called, [(1, 'concerne', 2)])

    def test_after_delete_relation(self):
        """make sure after_xxx_relation hooks are deferred"""
        self.o.register_hook(self._after_relation_hook,
                             'after_delete_relation', 'concerne')
        self.o.call_hooks('after_delete_relation', 'concerne', 'USER',
                          1, 'concerne', 2)
        self.o.call_hooks('after_delete_relation', 'concerne', 'USER',
                          3, 'concerne', 4)
        self.assertEquals(self.called, [(1, 'concerne', 2), (3, 'concerne', 4)])


    def _before_relation_hook(self, pool, subject, r_type, object):
        self.called.append((subject, r_type, object))

    def _after_relation_hook(self, pool, subject, r_type, object):
        self.called.append((subject, r_type, object))


class SystemHooksTC(RepositoryBasedTC):

    def test_startup_shutdown(self):
        import hooks # cubicweb/server/test/data/hooks.py
        self.assertEquals(hooks.CALLED_EVENTS['server_startup'], True)
        # don't actually call repository.shutdown !
        self.repo.hm.call_hooks('server_shutdown', repo=None)
        self.assertEquals(hooks.CALLED_EVENTS['server_shutdown'], True)

    def test_session_open_close(self):
        import hooks # cubicweb/server/test/data/hooks.py
        cnx = self.login('anon')
        self.assertEquals(hooks.CALLED_EVENTS['session_open'], 'anon')
        cnx.close()
        self.assertEquals(hooks.CALLED_EVENTS['session_close'], 'anon')


from itertools import repeat

class MyHook(Hook):
    schema = schema # set for actual hooks at registration time
    events = ('whatever', 'another')
    accepts = ('Societe', 'Division')
    
class HookTC(RepositoryBasedTC):
    def test_inheritance(self):
        self.assertEquals(list(MyHook.register_to()),
                          zip(repeat('whatever'), ('Societe', 'Division', 'SubDivision'))
                          + zip(repeat('another'), ('Societe', 'Division', 'SubDivision')))


if __name__ == '__main__':
    unittest_main()