|
1 # -*- coding: utf-8 -*- |
|
2 """unit/functional tests for cubicweb.server.hookhelper""" |
|
3 |
|
4 from logilab.common.testlib import unittest_main |
|
5 from cubicweb.devtools.apptest import RepositoryBasedTC |
|
6 |
|
7 from cubicweb.server.pool import LateOperation |
|
8 from cubicweb.server.hookhelper import * |
|
9 |
|
10 |
|
11 class HookHelpersTC(RepositoryBasedTC): |
|
12 |
|
13 def setUp(self): |
|
14 RepositoryBasedTC.setUp(self) |
|
15 self.hm = self.repo.hm |
|
16 |
|
17 def test_late_operation(self): |
|
18 session = self.session |
|
19 l1 = LateOperation(session) |
|
20 l2 = LateOperation(session) |
|
21 l3 = Operation(session) |
|
22 self.assertEquals(session.pending_operations, [l3, l1, l2]) |
|
23 |
|
24 def test_single_last_operation(self): |
|
25 session = self.session |
|
26 l0 = SingleLastOperation(session) |
|
27 l1 = LateOperation(session) |
|
28 l2 = LateOperation(session) |
|
29 l3 = Operation(session) |
|
30 self.assertEquals(session.pending_operations, [l3, l1, l2, l0]) |
|
31 l4 = SingleLastOperation(session) |
|
32 self.assertEquals(session.pending_operations, [l3, l1, l2, l4]) |
|
33 |
|
34 def test_global_operation_order(self): |
|
35 from cubicweb.server import hooks, schemahooks |
|
36 session = self.session |
|
37 op1 = hooks.DelayedDeleteOp(session) |
|
38 op2 = schemahooks.DelErdefOp(session) |
|
39 # equivalent operation generated by op2 but replace it here by op3 so we |
|
40 # can check the result... |
|
41 op3 = schemahooks.UpdateSchemaOp(session) |
|
42 op4 = hooks.DelayedDeleteOp(session) |
|
43 op5 = hooks.CheckORelationOp(session) |
|
44 self.assertEquals(session.pending_operations, [op1, op2, op4, op5, op3]) |
|
45 |
|
46 |
|
47 def test_in_state_notification(self): |
|
48 result = [] |
|
49 # test both email notification and transition_information |
|
50 # whatever if we can connect to the default stmp server, transaction |
|
51 # should not fail |
|
52 def in_state_changed(session, eidfrom, rtype, eidto): |
|
53 tr = previous_state(session, eidfrom) |
|
54 if tr is None: |
|
55 result.append(tr) |
|
56 return |
|
57 content = u'trÀnsition from %s to %s' % (tr.name, entity_name(session, eidto)) |
|
58 result.append(content) |
|
59 SendMailOp(session, msg=content, recipients=['test@logilab.fr']) |
|
60 self.hm.register_hook(in_state_changed, |
|
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"') |
|
63 self.assertEquals(result, [None]) |
|
64 searchedops = [op for op in self.session.pending_operations |
|
65 if isinstance(op, SendMailOp)] |
|
66 self.assertEquals(len(searchedops), 0, |
|
67 self.session.pending_operations) |
|
68 self.commit() |
|
69 self.execute('SET X in_state S WHERE X login "paf", S name "deactivated"') |
|
70 self.assertEquals(result, [None, u'trÀnsition from activated to deactivated']) |
|
71 # one to send the mail, one to close the smtp connection |
|
72 searchedops = [op for op in self.session.pending_operations |
|
73 if isinstance(op, SendMailOp)] |
|
74 self.assertEquals(len(searchedops), 1, |
|
75 self.session.pending_operations) |
|
76 self.commit() |
|
77 searchedops = [op for op in self.session.pending_operations |
|
78 if isinstance(op, SendMailOp)] |
|
79 self.assertEquals(len(searchedops), 0, |
|
80 self.session.pending_operations) |
|
81 |
|
82 if __name__ == '__main__': |
|
83 unittest_main() |