1 # -*- coding: utf-8 -*-

     2 """unit/functional tests for cubicweb.server.hook

     3 

     4 :organization: Logilab

     5 :copyright: 2001-2009 LOGILAB S.A. (Paris, FRANCE), license is LGPL v2.

     6 :contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr

     7 :license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses

     8 """

     9 

    10 from logilab.common.testlib import unittest_main

    11 

    12 from cubicweb.devtools.testlib import CubicWebTC

    13 from cubicweb.server.hook import LateOperation, Operation, SingleLastOperation

    14 from cubicweb.hooks import integrity, syncschema

    15 

    16 

    17 def clean_session_ops(func):

    18     def wrapper(self, *args, **kwargs):

    19         try:

    20             return func(self, *args, **kwargs)

    21         finally:

    22             self.session.pending_operations[:] = []

    23     return wrapper

    24 

    25 class HookHelpersTC(CubicWebTC):

    26 

    27     def setUp(self):

    28         CubicWebTC.setUp(self)

    29         self.hm = self.repo.hm

    30 

    31     @clean_session_ops

    32     def test_late_operation(self):

    33         session = self.session

    34         l1 = LateOperation(session)

    35         l2 = LateOperation(session)

    36         l3 = Operation(session)

    37         self.assertEquals(session.pending_operations, [l3, l1, l2])

    38 

    39     @clean_session_ops

    40     def test_single_last_operation(self):

    41         session = self.session

    42         l0 = SingleLastOperation(session)

    43         l1 = LateOperation(session)

    44         l2 = LateOperation(session)

    45         l3 = Operation(session)

    46         self.assertEquals(session.pending_operations, [l3, l1, l2, l0])

    47         l4 = SingleLastOperation(session)

    48         self.assertEquals(session.pending_operations, [l3, l1, l2, l4])

    49 

    50     @clean_session_ops

    51     def test_global_operation_order(self):

    52         session = self.session

    53         op1 = integrity._DelayedDeleteOp(session)

    54         op2 = syncschema.MemSchemaRDefDel(session)

    55         # equivalent operation generated by op2 but replace it here by op3 so we

    56         # can check the result...

    57         op3 = syncschema.MemSchemaNotifyChanges(session)

    58         op4 = integrity._DelayedDeleteOp(session)

    59         op5 = integrity._CheckORelationOp(session)

    60         self.assertEquals(session.pending_operations, [op1, op2, op4, op5, op3])

    61 

    62 if __name__ == '__main__':

    63     unittest_main()