2 """unit/functional tests for cubicweb.server.hookhelper""" |
2 """unit/functional tests for cubicweb.server.hookhelper""" |
3 |
3 |
4 from logilab.common.testlib import unittest_main |
4 from logilab.common.testlib import unittest_main |
5 from cubicweb.devtools.apptest import RepositoryBasedTC |
5 from cubicweb.devtools.apptest import RepositoryBasedTC |
6 |
6 |
7 from cubicweb.server.pool import LateOperation |
7 from cubicweb.server.pool import LateOperation, Operation, SingleLastOperation |
8 from cubicweb.server.hookhelper import * |
8 from cubicweb.server.hookhelper import * |
9 |
9 |
10 |
10 |
11 class HookHelpersTC(RepositoryBasedTC): |
11 class HookHelpersTC(RepositoryBasedTC): |
12 |
12 |
13 def setUp(self): |
13 def setUp(self): |
14 RepositoryBasedTC.setUp(self) |
14 RepositoryBasedTC.setUp(self) |
15 self.hm = self.repo.hm |
15 self.hm = self.repo.hm |
16 |
16 |
17 def test_late_operation(self): |
17 def test_late_operation(self): |
18 session = self.session |
18 session = self.session |
19 l1 = LateOperation(session) |
19 l1 = LateOperation(session) |
20 l2 = LateOperation(session) |
20 l2 = LateOperation(session) |
21 l3 = Operation(session) |
21 l3 = Operation(session) |
22 self.assertEquals(session.pending_operations, [l3, l1, l2]) |
22 self.assertEquals(session.pending_operations, [l3, l1, l2]) |
23 |
23 |
24 def test_single_last_operation(self): |
24 def test_single_last_operation(self): |
25 session = self.session |
25 session = self.session |
26 l0 = SingleLastOperation(session) |
26 l0 = SingleLastOperation(session) |
27 l1 = LateOperation(session) |
27 l1 = LateOperation(session) |
28 l2 = LateOperation(session) |
28 l2 = LateOperation(session) |
29 l3 = Operation(session) |
29 l3 = Operation(session) |
30 self.assertEquals(session.pending_operations, [l3, l1, l2, l0]) |
30 self.assertEquals(session.pending_operations, [l3, l1, l2, l0]) |
31 l4 = SingleLastOperation(session) |
31 l4 = SingleLastOperation(session) |
32 self.assertEquals(session.pending_operations, [l3, l1, l2, l4]) |
32 self.assertEquals(session.pending_operations, [l3, l1, l2, l4]) |
33 |
33 |
34 def test_global_operation_order(self): |
34 def test_global_operation_order(self): |
35 from cubicweb.server import hooks, schemahooks |
35 from cubicweb.server import hooks, schemahooks |
36 session = self.session |
36 session = self.session |
37 op1 = hooks.DelayedDeleteOp(session) |
37 op1 = hooks.DelayedDeleteOp(session) |
38 op2 = schemahooks.DelErdefOp(session) |
38 op2 = schemahooks.DelErdefOp(session) |
40 # can check the result... |
40 # can check the result... |
41 op3 = schemahooks.UpdateSchemaOp(session) |
41 op3 = schemahooks.UpdateSchemaOp(session) |
42 op4 = hooks.DelayedDeleteOp(session) |
42 op4 = hooks.DelayedDeleteOp(session) |
43 op5 = hooks.CheckORelationOp(session) |
43 op5 = hooks.CheckORelationOp(session) |
44 self.assertEquals(session.pending_operations, [op1, op2, op4, op5, op3]) |
44 self.assertEquals(session.pending_operations, [op1, op2, op4, op5, op3]) |
45 |
45 |
46 |
46 |
47 def test_in_state_notification(self): |
47 def test_in_state_notification(self): |
48 result = [] |
48 result = [] |
49 # test both email notification and transition_information |
49 # test both email notification and transition_information |
50 # whatever if we can connect to the default stmp server, transaction |
50 # whatever if we can connect to the default stmp server, transaction |
51 # should not fail |
51 # should not fail |
57 content = u'trÀnsition from %s to %s' % (tr.name, entity_name(session, eidto)) |
57 content = u'trÀnsition from %s to %s' % (tr.name, entity_name(session, eidto)) |
58 result.append(content) |
58 result.append(content) |
59 SendMailOp(session, msg=content, recipients=['test@logilab.fr']) |
59 SendMailOp(session, msg=content, recipients=['test@logilab.fr']) |
60 self.hm.register_hook(in_state_changed, |
60 self.hm.register_hook(in_state_changed, |
61 'before_add_relation', 'in_state') |
61 'before_add_relation', 'in_state') |
62 self.execute('INSERT EUser X: X login "paf", X upassword "wouf", X in_state S, X in_group G WHERE S name "activated", G name "users"') |
62 self.execute('INSERT CWUser X: X login "paf", X upassword "wouf", X in_state S, X in_group G WHERE S name "activated", G name "users"') |
63 self.assertEquals(result, [None]) |
63 self.assertEquals(result, [None]) |
64 searchedops = [op for op in self.session.pending_operations |
64 searchedops = [op for op in self.session.pending_operations |
65 if isinstance(op, SendMailOp)] |
65 if isinstance(op, SendMailOp)] |
66 self.assertEquals(len(searchedops), 0, |
66 self.assertEquals(len(searchedops), 0, |
67 self.session.pending_operations) |
67 self.session.pending_operations) |