diff -r 058bb3dc685f -r 0b59724cb3f2 server/test/unittest_hook.py --- a/server/test/unittest_hook.py Mon Jan 04 18:40:30 2016 +0100 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,193 +0,0 @@ -# -*- 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()