diff -r 058bb3dc685f -r 0b59724cb3f2 cubicweb/server/test/unittest_hook.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/cubicweb/server/test/unittest_hook.py Sat Jan 16 13:48:51 2016 +0100 @@ -0,0 +1,193 @@ +# -*- coding: utf-8 -*- +# copyright 2003-2014 LOGILAB S.A. (Paris, FRANCE), all rights reserved. +# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr +# +# This file is part of CubicWeb. +# +# CubicWeb is free software: you can redistribute it and/or modify it under the +# terms of the GNU Lesser General Public License as published by the Free +# Software Foundation, either version 2.1 of the License, or (at your option) +# any later version. +# +# CubicWeb is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more +# details. +# +# You should have received a copy of the GNU Lesser General Public License along +# with CubicWeb. If not, see . +"""unit/functional tests for cubicweb.server.hook""" + +from logilab.common.testlib import TestCase, unittest_main, mock_object + +from cubicweb.devtools import TestServerConfiguration, fake +from cubicweb.devtools.testlib import CubicWebTC +from cubicweb.server import hook +from cubicweb.hooks import integrity, syncschema + +class OperationsTC(CubicWebTC): + + def setUp(self): + CubicWebTC.setUp(self) + self.hm = self.repo.hm + + def test_late_operation(self): + with self.admin_access.repo_cnx() as cnx: + l1 = hook.LateOperation(cnx) + l2 = hook.LateOperation(cnx) + l3 = hook.Operation(cnx) + self.assertEqual(cnx.pending_operations, [l3, l1, l2]) + + def test_single_last_operation(self): + with self.admin_access.repo_cnx() as cnx: + l0 = hook.SingleLastOperation(cnx) + l1 = hook.LateOperation(cnx) + l2 = hook.LateOperation(cnx) + l3 = hook.Operation(cnx) + self.assertEqual(cnx.pending_operations, [l3, l1, l2, l0]) + l4 = hook.SingleLastOperation(cnx) + self.assertEqual(cnx.pending_operations, [l3, l1, l2, l4]) + + def test_global_operation_order(self): + with self.admin_access.repo_cnx() as cnx: + op1 = syncschema.RDefDelOp(cnx) + op2 = integrity._CheckORelationOp(cnx) + op3 = syncschema.MemSchemaNotifyChanges(cnx) + self.assertEqual([op1, op2, op3], cnx.pending_operations) + +class HookCalled(Exception): pass + +config = TestServerConfiguration('data') +config.bootstrap_cubes() +schema = config.load_schema() + +def tearDownModule(*args): + global config, schema + del config, schema + +class AddAnyHook(hook.Hook): + __regid__ = 'addany' + category = 'cat1' + events = ('before_add_entity',) + def __call__(self): + raise HookCalled() + + +class HooksRegistryTC(TestCase): + + def setUp(self): + """ called before each test from this class """ + self.vreg = mock_object(config=config, schema=schema) + self.o = hook.HooksRegistry(self.vreg) + + def test_register_bad_hook1(self): + class _Hook(hook.Hook): + events = ('before_add_entiti',) + with self.assertRaises(Exception) as cm: + self.o.register(_Hook) + self.assertEqual(str(cm.exception), 'bad event before_add_entiti on %s._Hook' % __name__) + + def test_register_bad_hook2(self): + class _Hook(hook.Hook): + events = None + with self.assertRaises(Exception) as cm: + self.o.register(_Hook) + self.assertEqual(str(cm.exception), 'bad .events attribute None on %s._Hook' % __name__) + + def test_register_bad_hook3(self): + class _Hook(hook.Hook): + events = 'before_add_entity' + with self.assertRaises(Exception) as cm: + self.o.register(_Hook) + self.assertEqual(str(cm.exception), 'bad event b on %s._Hook' % __name__) + + def test_call_hook(self): + self.o.register(AddAnyHook) + dis = set() + cw = fake.FakeSession() + cw.is_hook_activated = lambda cls: cls.category not in dis + self.assertRaises(HookCalled, + self.o.call_hooks, 'before_add_entity', cw) + dis.add('cat1') + self.o.call_hooks('before_add_entity', cw) # disabled hooks category, not called + dis.remove('cat1') + self.assertRaises(HookCalled, + self.o.call_hooks, 'before_add_entity', cw) + self.o.unregister(AddAnyHook) + self.o.call_hooks('before_add_entity', cw) # nothing to call + + +class SystemHooksTC(CubicWebTC): + + def test_startup_shutdown(self): + import hooks # cubicweb/server/test/data/hooks.py + self.assertEqual(hooks.CALLED_EVENTS['server_startup'], True) + # don't actually call repository.shutdown ! + self.repo.hm.call_hooks('server_shutdown', repo=self.repo) + self.assertEqual(hooks.CALLED_EVENTS['server_shutdown'], True) + + def test_session_open_close(self): + import hooks # cubicweb/server/test/data/hooks.py + anonaccess = self.new_access('anon') + with anonaccess.repo_cnx() as cnx: + self.assertEqual(hooks.CALLED_EVENTS['session_open'], 'anon') + anonaccess.close() + self.assertEqual(hooks.CALLED_EVENTS['session_close'], 'anon') + + +# 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(self._before_relation_hook, +# 'before_add_relation', 'concerne') +# self.assertEqual(self.called, []) +# self.o.call_hooks('before_add_relation', 'concerne', 'USER', +# 1, 'concerne', 2) +# self.assertEqual(self.called, [(1, 'concerne', 2)]) + +# def test_after_add_relation(self): +# """make sure after_xxx_relation hooks are deferred""" +# self.o.register(self._after_relation_hook, +# 'after_add_relation', 'concerne') +# self.assertEqual(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.assertEqual(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(self._before_relation_hook, +# 'before_delete_relation', 'concerne') +# self.assertEqual(self.called, []) +# self.o.call_hooks('before_delete_relation', 'concerne', 'USER', +# 1, 'concerne', 2) +# self.assertEqual(self.called, [(1, 'concerne', 2)]) + +# def test_after_delete_relation(self): +# """make sure after_xxx_relation hooks are deferred""" +# self.o.register(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.assertEqual(self.called, [(1, 'concerne', 2), (3, 'concerne', 4)]) + + +# def _before_relation_hook(self, cnxset, subject, r_type, object): +# self.called.append((subject, r_type, object)) + +# def _after_relation_hook(self, cnxset, subject, r_type, object): +# self.called.append((subject, r_type, object)) + + +if __name__ == '__main__': + unittest_main()