1 # copyright 2003-2014 LOGILAB S.A. (Paris, FRANCE), all rights reserved.

     2 # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr

     3 #

     4 # This file is part of CubicWeb.

     5 #

     6 # CubicWeb is free software: you can redistribute it and/or modify it under the

     7 # terms of the GNU Lesser General Public License as published by the Free

     8 # Software Foundation, either version 2.1 of the License, or (at your option)

     9 # any later version.

    10 #

    11 # CubicWeb is distributed in the hope that it will be useful, but WITHOUT

    12 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS

    13 # FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more

    14 # details.

    15 #

    16 # You should have received a copy of the GNU Lesser General Public License along

    17 # with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.

    18 

    19 from cubicweb.devtools.testlib import CubicWebTC

    20 from cubicweb.server.session import HOOKS_ALLOW_ALL, HOOKS_DENY_ALL

    21 from cubicweb.server import hook

    22 from cubicweb.predicates import is_instance

    23 

    24 class SessionTC(CubicWebTC):

    25 

    26     def test_hooks_control(self):

    27         session = self.session

    28         # this test check the "old" behavior of session with automatic connection management

    29         # close the default cnx, we do nto want it to interfer with the test

    30         self.cnx.close()

    31         # open a dedicated one

    32         session.set_cnx('Some-random-cnx-unrelated-to-the-default-one')

    33         # go test go

    34         self.assertEqual(HOOKS_ALLOW_ALL, session.hooks_mode)

    35         self.assertEqual(set(), session.disabled_hook_categories)

    36         self.assertEqual(set(), session.enabled_hook_categories)

    37         self.assertEqual(1, len(session._cnxs))

    38         with session.deny_all_hooks_but('metadata'):

    39             self.assertEqual(HOOKS_DENY_ALL, session.hooks_mode)

    40             self.assertEqual(set(), session.disabled_hook_categories)

    41             self.assertEqual(set(('metadata',)), session.enabled_hook_categories)

    42             session.commit()

    43             self.assertEqual(HOOKS_DENY_ALL, session.hooks_mode)

    44             self.assertEqual(set(), session.disabled_hook_categories)

    45             self.assertEqual(set(('metadata',)), session.enabled_hook_categories)

    46             session.rollback()

    47             self.assertEqual(HOOKS_DENY_ALL, session.hooks_mode)

    48             self.assertEqual(set(), session.disabled_hook_categories)

    49             self.assertEqual(set(('metadata',)), session.enabled_hook_categories)

    50             with session.allow_all_hooks_but('integrity'):

    51                 self.assertEqual(HOOKS_ALLOW_ALL, session.hooks_mode)

    52                 self.assertEqual(set(('integrity',)), session.disabled_hook_categories)

    53                 self.assertEqual(set(('metadata',)), session.enabled_hook_categories) # not changed in such case

    54             self.assertEqual(HOOKS_DENY_ALL, session.hooks_mode)

    55             self.assertEqual(set(), session.disabled_hook_categories)

    56             self.assertEqual(set(('metadata',)), session.enabled_hook_categories)

    57         # leaving context manager with no transaction running should reset the

    58         # transaction local storage (and associated cnxset)

    59         self.assertEqual({}, session._cnxs)

    60         self.assertEqual(None, session.cnxset)

    61         self.assertEqual(HOOKS_ALLOW_ALL, session.hooks_mode, session.HOOKS_ALLOW_ALL)

    62         self.assertEqual(set(), session.disabled_hook_categories)

    63         self.assertEqual(set(), session.enabled_hook_categories)

    64 

    65     def test_explicit_connection(self):

    66         with self.session.new_cnx() as cnx:

    67             rset = cnx.execute('Any X LIMIT 1 WHERE X is CWUser')

    68             self.assertEqual(1, len(rset))

    69             user = rset.get_entity(0, 0)

    70             user.cw_delete()

    71             cnx.rollback()

    72             new_user = cnx.entity_from_eid(user.eid)

    73             self.assertIsNotNone(new_user.login)

    74         self.assertFalse(cnx._open)

    75 

    76     def test_internal_cnx(self):

    77         with self.repo.internal_cnx() as cnx:

    78             rset = cnx.execute('Any X LIMIT 1 WHERE X is CWUser')

    79             self.assertEqual(1, len(rset))

    80             user = rset.get_entity(0, 0)

    81             user.cw_delete()

    82             cnx.rollback()

    83             new_user = cnx.entity_from_eid(user.eid)

    84             self.assertIsNotNone(new_user.login)

    85         self.assertFalse(cnx._open)

    86 

    87     def test_connection_exit(self):

    88         """exiting a connection should roll back the transaction, including any

    89         pending operations"""

    90         self.rollbacked = False

    91         class RollbackOp(hook.Operation):

    92             _test = self

    93             def rollback_event(self):

    94                 self._test.rollbacked = True

    95         class RollbackHook(hook.Hook):

    96             __regid__ = 'rollback'

    97             events = ('after_update_entity',)

    98             __select__ = hook.Hook.__select__ & is_instance('CWGroup')

    99             def __call__(self):

   100                 RollbackOp(self._cw)

   101         with self.temporary_appobjects(RollbackHook):

   102             with self.admin_access.client_cnx() as cnx:

   103                 cnx.execute('SET G name "foo" WHERE G is CWGroup, G name "managers"')

   104             self.assertTrue(self.rollbacked)

   105 

   106 if __name__ == '__main__':

   107     from logilab.common.testlib import unittest_main

   108     unittest_main()