6 :contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr |
6 :contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr |
7 :license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses |
7 :license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses |
8 """ |
8 """ |
9 |
9 |
10 from logilab.common.testlib import unittest_main |
10 from logilab.common.testlib import unittest_main |
11 from cubicweb.devtools.apptest import RepositoryBasedTC |
11 from cubicweb.devtools.testlib import CubicWebTC |
12 |
12 |
13 from cubicweb.server.pool import LateOperation, Operation, SingleLastOperation |
13 from cubicweb.server.pool import LateOperation, Operation, SingleLastOperation |
14 from cubicweb.server.hookhelper import * |
14 from cubicweb.server.hookhelper import * |
|
15 from cubicweb.server import hooks, schemahooks |
15 |
16 |
16 |
17 |
17 class HookHelpersTC(RepositoryBasedTC): |
18 def clean_session_ops(func): |
|
19 def wrapper(self, *args, **kwargs): |
|
20 try: |
|
21 return func(self, *args, **kwargs) |
|
22 finally: |
|
23 self.session.pending_operations[:] = [] |
|
24 return wrapper |
|
25 |
|
26 class HookHelpersTC(CubicWebTC): |
18 |
27 |
19 def setUp(self): |
28 def setUp(self): |
20 RepositoryBasedTC.setUp(self) |
29 CubicWebTC.setUp(self) |
21 self.hm = self.repo.hm |
30 self.hm = self.repo.hm |
22 |
31 |
|
32 @clean_session_ops |
23 def test_late_operation(self): |
33 def test_late_operation(self): |
24 session = self.session |
34 session = self.session |
25 l1 = LateOperation(session) |
35 l1 = LateOperation(session) |
26 l2 = LateOperation(session) |
36 l2 = LateOperation(session) |
27 l3 = Operation(session) |
37 l3 = Operation(session) |
28 self.assertEquals(session.pending_operations, [l3, l1, l2]) |
38 self.assertEquals(session.pending_operations, [l3, l1, l2]) |
29 |
39 |
|
40 @clean_session_ops |
30 def test_single_last_operation(self): |
41 def test_single_last_operation(self): |
31 session = self.session |
42 session = self.session |
32 l0 = SingleLastOperation(session) |
43 l0 = SingleLastOperation(session) |
33 l1 = LateOperation(session) |
44 l1 = LateOperation(session) |
34 l2 = LateOperation(session) |
45 l2 = LateOperation(session) |
35 l3 = Operation(session) |
46 l3 = Operation(session) |
36 self.assertEquals(session.pending_operations, [l3, l1, l2, l0]) |
47 self.assertEquals(session.pending_operations, [l3, l1, l2, l0]) |
37 l4 = SingleLastOperation(session) |
48 l4 = SingleLastOperation(session) |
38 self.assertEquals(session.pending_operations, [l3, l1, l2, l4]) |
49 self.assertEquals(session.pending_operations, [l3, l1, l2, l4]) |
39 |
50 |
|
51 @clean_session_ops |
40 def test_global_operation_order(self): |
52 def test_global_operation_order(self): |
41 from cubicweb.server import hooks, schemahooks |
|
42 session = self.session |
53 session = self.session |
43 op1 = hooks.DelayedDeleteOp(session) |
54 op1 = hooks.DelayedDeleteOp(session) |
44 op2 = schemahooks.MemSchemaRDefDel(session) |
55 op2 = schemahooks.MemSchemaRDefDel(session) |
45 # equivalent operation generated by op2 but replace it here by op3 so we |
56 # equivalent operation generated by op2 but replace it here by op3 so we |
46 # can check the result... |
57 # can check the result... |
78 searchedops = [op for op in self.session.pending_operations |
89 searchedops = [op for op in self.session.pending_operations |
79 if isinstance(op, SendMailOp)] |
90 if isinstance(op, SendMailOp)] |
80 self.assertEquals(len(searchedops), 1, |
91 self.assertEquals(len(searchedops), 1, |
81 self.session.pending_operations) |
92 self.session.pending_operations) |
82 self.commit() |
93 self.commit() |
83 searchedops = [op for op in self.session.pending_operations |
94 self.assertEquals([], self.session.pending_operations) |
84 if isinstance(op, SendMailOp)] |
|
85 self.assertEquals(len(searchedops), 0, |
|
86 self.session.pending_operations) |
|
87 |
95 |
88 if __name__ == '__main__': |
96 if __name__ == '__main__': |
89 unittest_main() |
97 unittest_main() |