1 """unit tests for the hooks manager

     2 """

     3 

     4 from logilab.common.testlib import TestCase, unittest_main

     5 

     6 from cubicweb.server.hooksmanager import HooksManager, Hook

     7 from cubicweb.devtools import TestServerConfiguration

     8 from cubicweb.devtools.apptest import RepositoryBasedTC

     9 

    10 class HookCalled(Exception): pass

    11 

    12 config = TestServerConfiguration('data')

    13 config.bootstrap_cubes()

    14 schema = config.load_schema()

    15 

    16 class HooksManagerTC(TestCase):

    17     args = (None,)

    18     kwargs = {'a': 1}

    19     

    20     def setUp(self):

    21         """ called before each test from this class """

    22         self.o = HooksManager(schema)

    23 

    24     def test_register_hook_raise_keyerror(self):

    25         self.assertRaises(AssertionError,

    26                           self.o.register_hook, self._hook, 'before_add_entiti')

    27         self.assertRaises(AssertionError,

    28                           self.o.register_hook, self._hook, 'session_login', 'EEType')

    29         self.assertRaises(AssertionError,

    30                           self.o.register_hook, self._hook, 'session_logout', 'EEType')

    31         self.assertRaises(AssertionError,

    32                           self.o.register_hook, self._hook, 'server_startup', 'EEType')

    33         self.assertRaises(AssertionError,

    34                           self.o.register_hook, self._hook, 'server_shutdown', 'EEType')

    35         

    36     def test_register_hook1(self):

    37         self.o.register_hook(self._hook, 'before_add_entity')

    38         self.o.register_hook(self._hook, 'before_delete_entity', 'Personne')

    39         self._test_called_hooks()

    40         

    41     def test_register_hook2(self):

    42         self.o.register_hook(self._hook, 'before_add_entity', '')

    43         self.o.register_hook(self._hook, 'before_delete_entity', 'Personne')

    44         self._test_called_hooks()

    45         

    46     def test_register_hook3(self):

    47         self.o.register_hook(self._hook, 'before_add_entity', None)

    48         self.o.register_hook(self._hook, 'before_delete_entity', 'Personne')

    49         self._test_called_hooks()

    50         

    51     def test_register_hooks(self):

    52         self.o.register_hooks({'before_add_entity' : {'': [self._hook]},

    53                                'before_delete_entity' : {'Personne': [self._hook]},

    54                                })

    55         self._test_called_hooks()

    56 

    57     def test_unregister_hook(self):

    58         self.o.register_hook(self._hook, 'after_delete_entity', 'Personne')

    59         self.assertRaises(HookCalled,

    60                           self.o.call_hooks, 'after_delete_entity', 'Personne',

    61                           *self.args, **self.kwargs)

    62         self.o.unregister_hook(self._hook, 'after_delete_entity', 'Personne')

    63         # no hook should be called there

    64         self.o.call_hooks('after_delete_entity', 'Personne')

    65         

    66 

    67     def _test_called_hooks(self):

    68         self.assertRaises(HookCalled,

    69                           self.o.call_hooks, 'before_add_entity', '',

    70                           *self.args, **self.kwargs)

    71         self.assertRaises(HookCalled,

    72                           self.o.call_hooks, 'before_add_entity', None,

    73                           *self.args, **self.kwargs)

    74         self.assertRaises(HookCalled,

    75                           self.o.call_hooks, 'before_add_entity', 'Personne',

    76                           *self.args, **self.kwargs)

    77         self.assertRaises(HookCalled,

    78                           self.o.call_hooks, 'before_delete_entity', 'Personne',

    79                           *self.args, **self.kwargs)

    80         # no hook should be called there

    81         self.o.call_hooks('before_delete_entity', None)

    82         self.o.call_hooks('before_delete_entity', 'Societe')

    83 

    84 

    85     def _hook(self, *args, **kwargs):

    86         # check arguments

    87         self.assertEqual(args, self.args)

    88         self.assertEqual(kwargs, self.kwargs)

    89         raise HookCalled()

    90 

    91 

    92 class RelationHookTC(TestCase):

    93     """testcase for relation hooks grouping"""

    94     def setUp(self):

    95         """ called before each test from this class """

    96         self.o = HooksManager(schema)

    97         self.called = []

    98 

    99     def test_before_add_relation(self):

   100         """make sure before_xxx_relation hooks are called directly"""

   101         self.o.register_hook(self._before_relation_hook,

   102                              'before_add_relation', 'concerne')

   103         self.assertEquals(self.called, [])

   104         self.o.call_hooks('before_add_relation', 'concerne', 'USER',

   105                           1, 'concerne', 2)        

   106         self.assertEquals(self.called, [(1, 'concerne', 2)])

   107         

   108     def test_after_add_relation(self):

   109         """make sure after_xxx_relation hooks are deferred"""

   110         self.o.register_hook(self._after_relation_hook,

   111                              'after_add_relation', 'concerne')

   112         self.assertEquals(self.called, [])

   113         self.o.call_hooks('after_add_relation', 'concerne', 'USER',

   114                           1, 'concerne', 2)

   115         self.o.call_hooks('after_add_relation', 'concerne', 'USER',

   116                           3, 'concerne', 4)

   117         self.assertEquals(self.called, [(1, 'concerne', 2), (3, 'concerne', 4)])

   118     

   119     def test_before_delete_relation(self):

   120         """make sure before_xxx_relation hooks are called directly"""

   121         self.o.register_hook(self._before_relation_hook,

   122                              'before_delete_relation', 'concerne')

   123         self.assertEquals(self.called, [])

   124         self.o.call_hooks('before_delete_relation', 'concerne', 'USER',

   125                           1, 'concerne', 2)        

   126         self.assertEquals(self.called, [(1, 'concerne', 2)])

   127 

   128     def test_after_delete_relation(self):

   129         """make sure after_xxx_relation hooks are deferred"""

   130         self.o.register_hook(self._after_relation_hook,

   131                              'after_delete_relation', 'concerne')

   132         self.o.call_hooks('after_delete_relation', 'concerne', 'USER',

   133                           1, 'concerne', 2)

   134         self.o.call_hooks('after_delete_relation', 'concerne', 'USER',

   135                           3, 'concerne', 4)

   136         self.assertEquals(self.called, [(1, 'concerne', 2), (3, 'concerne', 4)])

   137 

   138 

   139     def _before_relation_hook(self, pool, subject, r_type, object):

   140         self.called.append((subject, r_type, object))

   141 

   142     def _after_relation_hook(self, pool, subject, r_type, object):

   143         self.called.append((subject, r_type, object))

   144 

   145 

   146 class SystemHooksTC(RepositoryBasedTC):

   147 

   148     def test_startup_shutdown(self):

   149         import hooks # cubicweb/server/test/data/hooks.py

   150         self.assertEquals(hooks.CALLED_EVENTS['server_startup'], True)

   151         # don't actually call repository.shutdown !

   152         self.repo.hm.call_hooks('server_shutdown', repo=None)

   153         self.assertEquals(hooks.CALLED_EVENTS['server_shutdown'], True)

   154 

   155     def test_session_open_close(self):

   156         import hooks # cubicweb/server/test/data/hooks.py

   157         cnx = self.login('anon')

   158         self.assertEquals(hooks.CALLED_EVENTS['session_open'], 'anon')

   159         cnx.close()

   160         self.assertEquals(hooks.CALLED_EVENTS['session_close'], 'anon')

   161 

   162 

   163 from itertools import repeat

   164 

   165 class MyHook(Hook):

   166     schema = schema # set for actual hooks at registration time

   167     events = ('whatever', 'another')

   168     accepts = ('Societe', 'Division')

   169     

   170 class HookTC(RepositoryBasedTC):

   171     def test_inheritance(self):

   172         self.assertEquals(list(MyHook.register_to()),

   173                           zip(repeat('whatever'), ('Societe', 'Division', 'SubDivision'))

   174                           + zip(repeat('another'), ('Societe', 'Division', 'SubDivision')))

   175 

   176 

   177 if __name__ == '__main__':

   178     unittest_main()