diff -r 000000000000 -r b97547f5f1fa server/test/unittest_hooksmanager.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/server/test/unittest_hooksmanager.py Wed Nov 05 15:52:50 2008 +0100 @@ -0,0 +1,178 @@ +"""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()