"""unit tests for the hooks manager:license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses"""fromlogilab.common.testlibimportTestCase,unittest_mainfromcubicweb.server.hooksmanagerimportHooksManager,Hookfromcubicweb.devtoolsimportTestServerConfigurationfromcubicweb.devtools.apptestimportRepositoryBasedTCclassHookCalled(Exception):passconfig=TestServerConfiguration('data')config.bootstrap_cubes()schema=config.load_schema()classHooksManagerTC(TestCase):args=(None,)kwargs={'a':1}defsetUp(self):""" called before each test from this class """self.o=HooksManager(schema)deftest_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','CWEType')self.assertRaises(AssertionError,self.o.register_hook,self._hook,'session_logout','CWEType')self.assertRaises(AssertionError,self.o.register_hook,self._hook,'server_startup','CWEType')self.assertRaises(AssertionError,self.o.register_hook,self._hook,'server_shutdown','CWEType')deftest_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()deftest_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()deftest_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()deftest_register_hooks(self):self.o.register_hooks({'before_add_entity':{'':[self._hook]},'before_delete_entity':{'Personne':[self._hook]},})self._test_called_hooks()deftest_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 thereself.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 thereself.o.call_hooks('before_delete_entity',None)self.o.call_hooks('before_delete_entity','Societe')def_hook(self,*args,**kwargs):# check argumentsself.assertEqual(args,self.args)self.assertEqual(kwargs,self.kwargs)raiseHookCalled()classRelationHookTC(TestCase):"""testcase for relation hooks grouping"""defsetUp(self):""" called before each test from this class """self.o=HooksManager(schema)self.called=[]deftest_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)])deftest_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)])deftest_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)])deftest_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))classSystemHooksTC(RepositoryBasedTC):deftest_startup_shutdown(self):importhooks# cubicweb/server/test/data/hooks.pyself.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)deftest_session_open_close(self):importhooks# cubicweb/server/test/data/hooks.pycnx=self.login('anon')self.assertEquals(hooks.CALLED_EVENTS['session_open'],'anon')cnx.close()self.assertEquals(hooks.CALLED_EVENTS['session_close'],'anon')fromitertoolsimportrepeatclassMyHook(Hook):schema=schema# set for actual hooks at registration timeevents=('whatever','another')accepts=('Societe','Division')classHookTC(RepositoryBasedTC):deftest_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()