server/test/unittest_hook.py
branchstable
changeset 6340 470d8e828fda
parent 5921 b3fcb004d0e1
child 6781 5062d86d6ffe
equal deleted inserted replaced
6339:bdc3dc94d744 6340:470d8e828fda
    45     def test_late_operation(self):
    45     def test_late_operation(self):
    46         session = self.session
    46         session = self.session
    47         l1 = hook.LateOperation(session)
    47         l1 = hook.LateOperation(session)
    48         l2 = hook.LateOperation(session)
    48         l2 = hook.LateOperation(session)
    49         l3 = hook.Operation(session)
    49         l3 = hook.Operation(session)
    50         self.assertEquals(session.pending_operations, [l3, l1, l2])
    50         self.assertEqual(session.pending_operations, [l3, l1, l2])
    51 
    51 
    52     @clean_session_ops
    52     @clean_session_ops
    53     def test_single_last_operation(self):
    53     def test_single_last_operation(self):
    54         session = self.session
    54         session = self.session
    55         l0 = hook.SingleLastOperation(session)
    55         l0 = hook.SingleLastOperation(session)
    56         l1 = hook.LateOperation(session)
    56         l1 = hook.LateOperation(session)
    57         l2 = hook.LateOperation(session)
    57         l2 = hook.LateOperation(session)
    58         l3 = hook.Operation(session)
    58         l3 = hook.Operation(session)
    59         self.assertEquals(session.pending_operations, [l3, l1, l2, l0])
    59         self.assertEqual(session.pending_operations, [l3, l1, l2, l0])
    60         l4 = hook.SingleLastOperation(session)
    60         l4 = hook.SingleLastOperation(session)
    61         self.assertEquals(session.pending_operations, [l3, l1, l2, l4])
    61         self.assertEqual(session.pending_operations, [l3, l1, l2, l4])
    62 
    62 
    63     @clean_session_ops
    63     @clean_session_ops
    64     def test_global_operation_order(self):
    64     def test_global_operation_order(self):
    65         session = self.session
    65         session = self.session
    66         op1 = integrity._DelayedDeleteOp(session)
    66         op1 = integrity._DelayedDeleteOp(session)
    68         # equivalent operation generated by op2 but replace it here by op3 so we
    68         # equivalent operation generated by op2 but replace it here by op3 so we
    69         # can check the result...
    69         # can check the result...
    70         op3 = syncschema.MemSchemaNotifyChanges(session)
    70         op3 = syncschema.MemSchemaNotifyChanges(session)
    71         op4 = integrity._DelayedDeleteOp(session)
    71         op4 = integrity._DelayedDeleteOp(session)
    72         op5 = integrity._CheckORelationOp(session)
    72         op5 = integrity._CheckORelationOp(session)
    73         self.assertEquals(session.pending_operations, [op1, op2, op4, op5, op3])
    73         self.assertEqual(session.pending_operations, [op1, op2, op4, op5, op3])
    74 
    74 
    75 
    75 
    76 class HookCalled(Exception): pass
    76 class HookCalled(Exception): pass
    77 
    77 
    78 config = TestServerConfiguration('data')
    78 config = TestServerConfiguration('data')
   100 
   100 
   101     def test_register_bad_hook1(self):
   101     def test_register_bad_hook1(self):
   102         class _Hook(hook.Hook):
   102         class _Hook(hook.Hook):
   103             events = ('before_add_entiti',)
   103             events = ('before_add_entiti',)
   104         ex = self.assertRaises(Exception, self.o.register, _Hook)
   104         ex = self.assertRaises(Exception, self.o.register, _Hook)
   105         self.assertEquals(str(ex), 'bad event before_add_entiti on %s._Hook' % __name__)
   105         self.assertEqual(str(ex), 'bad event before_add_entiti on %s._Hook' % __name__)
   106 
   106 
   107     def test_register_bad_hook2(self):
   107     def test_register_bad_hook2(self):
   108         class _Hook(hook.Hook):
   108         class _Hook(hook.Hook):
   109             events = None
   109             events = None
   110         ex = self.assertRaises(Exception, self.o.register, _Hook)
   110         ex = self.assertRaises(Exception, self.o.register, _Hook)
   111         self.assertEquals(str(ex), 'bad .events attribute None on %s._Hook' % __name__)
   111         self.assertEqual(str(ex), 'bad .events attribute None on %s._Hook' % __name__)
   112 
   112 
   113     def test_register_bad_hook3(self):
   113     def test_register_bad_hook3(self):
   114         class _Hook(hook.Hook):
   114         class _Hook(hook.Hook):
   115             events = 'before_add_entity'
   115             events = 'before_add_entity'
   116         ex = self.assertRaises(Exception, self.o.register, _Hook)
   116         ex = self.assertRaises(Exception, self.o.register, _Hook)
   117         self.assertEquals(str(ex), 'bad event b on %s._Hook' % __name__)
   117         self.assertEqual(str(ex), 'bad event b on %s._Hook' % __name__)
   118 
   118 
   119     def test_call_hook(self):
   119     def test_call_hook(self):
   120         self.o.register(AddAnyHook)
   120         self.o.register(AddAnyHook)
   121         dis = set()
   121         dis = set()
   122         cw = mock_object(vreg=self.vreg,
   122         cw = mock_object(vreg=self.vreg,
   136 
   136 
   137 class SystemHooksTC(CubicWebTC):
   137 class SystemHooksTC(CubicWebTC):
   138 
   138 
   139     def test_startup_shutdown(self):
   139     def test_startup_shutdown(self):
   140         import hooks # cubicweb/server/test/data/hooks.py
   140         import hooks # cubicweb/server/test/data/hooks.py
   141         self.assertEquals(hooks.CALLED_EVENTS['server_startup'], True)
   141         self.assertEqual(hooks.CALLED_EVENTS['server_startup'], True)
   142         # don't actually call repository.shutdown !
   142         # don't actually call repository.shutdown !
   143         self.repo.hm.call_hooks('server_shutdown', repo=self.repo)
   143         self.repo.hm.call_hooks('server_shutdown', repo=self.repo)
   144         self.assertEquals(hooks.CALLED_EVENTS['server_shutdown'], True)
   144         self.assertEqual(hooks.CALLED_EVENTS['server_shutdown'], True)
   145 
   145 
   146     def test_session_open_close(self):
   146     def test_session_open_close(self):
   147         import hooks # cubicweb/server/test/data/hooks.py
   147         import hooks # cubicweb/server/test/data/hooks.py
   148         cnx = self.login('anon')
   148         cnx = self.login('anon')
   149         self.assertEquals(hooks.CALLED_EVENTS['session_open'], 'anon')
   149         self.assertEqual(hooks.CALLED_EVENTS['session_open'], 'anon')
   150         cnx.close()
   150         cnx.close()
   151         self.assertEquals(hooks.CALLED_EVENTS['session_close'], 'anon')
   151         self.assertEqual(hooks.CALLED_EVENTS['session_close'], 'anon')
   152 
   152 
   153 
   153 
   154 # class RelationHookTC(TestCase):
   154 # class RelationHookTC(TestCase):
   155 #     """testcase for relation hooks grouping"""
   155 #     """testcase for relation hooks grouping"""
   156 #     def setUp(self):
   156 #     def setUp(self):
   160 
   160 
   161 #     def test_before_add_relation(self):
   161 #     def test_before_add_relation(self):
   162 #         """make sure before_xxx_relation hooks are called directly"""
   162 #         """make sure before_xxx_relation hooks are called directly"""
   163 #         self.o.register(self._before_relation_hook,
   163 #         self.o.register(self._before_relation_hook,
   164 #                              'before_add_relation', 'concerne')
   164 #                              'before_add_relation', 'concerne')
   165 #         self.assertEquals(self.called, [])
   165 #         self.assertEqual(self.called, [])
   166 #         self.o.call_hooks('before_add_relation', 'concerne', 'USER',
   166 #         self.o.call_hooks('before_add_relation', 'concerne', 'USER',
   167 #                           1, 'concerne', 2)
   167 #                           1, 'concerne', 2)
   168 #         self.assertEquals(self.called, [(1, 'concerne', 2)])
   168 #         self.assertEqual(self.called, [(1, 'concerne', 2)])
   169 
   169 
   170 #     def test_after_add_relation(self):
   170 #     def test_after_add_relation(self):
   171 #         """make sure after_xxx_relation hooks are deferred"""
   171 #         """make sure after_xxx_relation hooks are deferred"""
   172 #         self.o.register(self._after_relation_hook,
   172 #         self.o.register(self._after_relation_hook,
   173 #                              'after_add_relation', 'concerne')
   173 #                              'after_add_relation', 'concerne')
   174 #         self.assertEquals(self.called, [])
   174 #         self.assertEqual(self.called, [])
   175 #         self.o.call_hooks('after_add_relation', 'concerne', 'USER',
   175 #         self.o.call_hooks('after_add_relation', 'concerne', 'USER',
   176 #                           1, 'concerne', 2)
   176 #                           1, 'concerne', 2)
   177 #         self.o.call_hooks('after_add_relation', 'concerne', 'USER',
   177 #         self.o.call_hooks('after_add_relation', 'concerne', 'USER',
   178 #                           3, 'concerne', 4)
   178 #                           3, 'concerne', 4)
   179 #         self.assertEquals(self.called, [(1, 'concerne', 2), (3, 'concerne', 4)])
   179 #         self.assertEqual(self.called, [(1, 'concerne', 2), (3, 'concerne', 4)])
   180 
   180 
   181 #     def test_before_delete_relation(self):
   181 #     def test_before_delete_relation(self):
   182 #         """make sure before_xxx_relation hooks are called directly"""
   182 #         """make sure before_xxx_relation hooks are called directly"""
   183 #         self.o.register(self._before_relation_hook,
   183 #         self.o.register(self._before_relation_hook,
   184 #                              'before_delete_relation', 'concerne')
   184 #                              'before_delete_relation', 'concerne')
   185 #         self.assertEquals(self.called, [])
   185 #         self.assertEqual(self.called, [])
   186 #         self.o.call_hooks('before_delete_relation', 'concerne', 'USER',
   186 #         self.o.call_hooks('before_delete_relation', 'concerne', 'USER',
   187 #                           1, 'concerne', 2)
   187 #                           1, 'concerne', 2)
   188 #         self.assertEquals(self.called, [(1, 'concerne', 2)])
   188 #         self.assertEqual(self.called, [(1, 'concerne', 2)])
   189 
   189 
   190 #     def test_after_delete_relation(self):
   190 #     def test_after_delete_relation(self):
   191 #         """make sure after_xxx_relation hooks are deferred"""
   191 #         """make sure after_xxx_relation hooks are deferred"""
   192 #         self.o.register(self._after_relation_hook,
   192 #         self.o.register(self._after_relation_hook,
   193 #                         'after_delete_relation', 'concerne')
   193 #                         'after_delete_relation', 'concerne')
   194 #         self.o.call_hooks('after_delete_relation', 'concerne', 'USER',
   194 #         self.o.call_hooks('after_delete_relation', 'concerne', 'USER',
   195 #                           1, 'concerne', 2)
   195 #                           1, 'concerne', 2)
   196 #         self.o.call_hooks('after_delete_relation', 'concerne', 'USER',
   196 #         self.o.call_hooks('after_delete_relation', 'concerne', 'USER',
   197 #                           3, 'concerne', 4)
   197 #                           3, 'concerne', 4)
   198 #         self.assertEquals(self.called, [(1, 'concerne', 2), (3, 'concerne', 4)])
   198 #         self.assertEqual(self.called, [(1, 'concerne', 2), (3, 'concerne', 4)])
   199 
   199 
   200 
   200 
   201 #     def _before_relation_hook(self, pool, subject, r_type, object):
   201 #     def _before_relation_hook(self, pool, subject, r_type, object):
   202 #         self.called.append((subject, r_type, object))
   202 #         self.called.append((subject, r_type, object))
   203 
   203