|
1 """ |
|
2 |
|
3 :organization: Logilab |
|
4 :copyright: 2001-2010 LOGILAB S.A. (Paris, FRANCE), license is LGPL v2. |
|
5 :contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr |
|
6 :license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses |
|
7 """ |
|
8 from __future__ import with_statement |
|
9 |
|
10 from cubicweb import ValidationError |
|
11 from cubicweb.devtools.testlib import CubicWebTC |
|
12 from cubicweb.transaction import * |
|
13 |
|
14 class UndoableTransactionTC(CubicWebTC): |
|
15 |
|
16 def setup_database(self): |
|
17 self.session.undo_actions = set('CUDAR') |
|
18 self.toto = self.create_user('toto', password='toto', groups=('users',), |
|
19 commit=False) |
|
20 self.txuuid = self.commit() |
|
21 |
|
22 def tearDown(self): |
|
23 self.restore_connection() |
|
24 self.session.undo_support = set() |
|
25 super(UndoableTransactionTC, self).tearDown() |
|
26 |
|
27 def test_undo_api(self): |
|
28 self.failUnless(self.txuuid) |
|
29 # test transaction api |
|
30 self.assertRaises(NoSuchTransaction, |
|
31 self.cnx.transaction_info, 'hop') |
|
32 self.assertRaises(NoSuchTransaction, |
|
33 self.cnx.transaction_actions, 'hop') |
|
34 self.assertRaises(NoSuchTransaction, |
|
35 self.cnx.undo_transaction, 'hop') |
|
36 txinfo = self.cnx.transaction_info(self.txuuid) |
|
37 self.failUnless(txinfo.datetime) |
|
38 self.assertEquals(txinfo.user_eid, self.session.user.eid) |
|
39 self.assertEquals(txinfo.user().login, 'admin') |
|
40 actions = txinfo.actions_list() |
|
41 self.assertEquals(len(actions), 2) |
|
42 actions = txinfo.actions_list(public=False) |
|
43 self.assertEquals(len(actions), 6) |
|
44 a1 = actions[0] |
|
45 self.assertEquals(a1.action, 'C') |
|
46 self.assertEquals(a1.eid, self.toto.eid) |
|
47 self.assertEquals(a1.etype,'CWUser') |
|
48 self.assertEquals(a1.changes, None) |
|
49 self.assertEquals(a1.public, True) |
|
50 self.assertEquals(a1.order, 1) |
|
51 a4 = actions[3] |
|
52 self.assertEquals(a4.action, 'A') |
|
53 self.assertEquals(a4.rtype, 'in_group') |
|
54 self.assertEquals(a4.eid_from, self.toto.eid) |
|
55 self.assertEquals(a4.eid_to, self.toto.in_group[0].eid) |
|
56 self.assertEquals(a4.order, 4) |
|
57 for i, rtype in ((1, 'owned_by'), (2, 'owned_by'), |
|
58 (4, 'created_by'), (5, 'in_state')): |
|
59 a = actions[i] |
|
60 self.assertEquals(a.action, 'A') |
|
61 self.assertEquals(a.eid_from, self.toto.eid) |
|
62 self.assertEquals(a.rtype, rtype) |
|
63 self.assertEquals(a.order, i+1) |
|
64 # test undoable_transactions |
|
65 txs = self.cnx.undoable_transactions() |
|
66 self.assertEquals(len(txs), 1) |
|
67 self.assertEquals(txs[0].uuid, self.txuuid) |
|
68 # test transaction_info / undoable_transactions security |
|
69 cnx = self.login('anon') |
|
70 self.assertRaises(NoSuchTransaction, |
|
71 cnx.transaction_info, self.txuuid) |
|
72 self.assertRaises(NoSuchTransaction, |
|
73 cnx.transaction_actions, self.txuuid) |
|
74 self.assertRaises(NoSuchTransaction, |
|
75 cnx.undo_transaction, self.txuuid) |
|
76 txs = cnx.undoable_transactions() |
|
77 self.assertEquals(len(txs), 0) |
|
78 |
|
79 def test_undoable_transactions(self): |
|
80 toto = self.toto |
|
81 e = self.session.create_entity('EmailAddress', |
|
82 address=u'toto@logilab.org', |
|
83 reverse_use_email=toto) |
|
84 txuuid1 = self.commit() |
|
85 toto.delete() |
|
86 txuuid2 = self.commit() |
|
87 undoable_transactions = self.cnx.undoable_transactions |
|
88 txs = undoable_transactions(action='D') |
|
89 self.assertEquals(len(txs), 1, txs) |
|
90 self.assertEquals(txs[0].uuid, txuuid2) |
|
91 txs = undoable_transactions(action='C') |
|
92 self.assertEquals(len(txs), 2, txs) |
|
93 self.assertEquals(txs[0].uuid, txuuid1) |
|
94 self.assertEquals(txs[1].uuid, self.txuuid) |
|
95 txs = undoable_transactions(eid=toto.eid) |
|
96 self.assertEquals(len(txs), 3) |
|
97 self.assertEquals(txs[0].uuid, txuuid2) |
|
98 self.assertEquals(txs[1].uuid, txuuid1) |
|
99 self.assertEquals(txs[2].uuid, self.txuuid) |
|
100 txs = undoable_transactions(etype='CWUser') |
|
101 self.assertEquals(len(txs), 2) |
|
102 txs = undoable_transactions(etype='CWUser', action='C') |
|
103 self.assertEquals(len(txs), 1) |
|
104 self.assertEquals(txs[0].uuid, self.txuuid) |
|
105 txs = undoable_transactions(etype='EmailAddress', action='D') |
|
106 self.assertEquals(len(txs), 0) |
|
107 txs = undoable_transactions(etype='EmailAddress', action='D', |
|
108 public=False) |
|
109 self.assertEquals(len(txs), 1) |
|
110 self.assertEquals(txs[0].uuid, txuuid2) |
|
111 txs = undoable_transactions(eid=toto.eid, action='R', public=False) |
|
112 self.assertEquals(len(txs), 1) |
|
113 self.assertEquals(txs[0].uuid, txuuid2) |
|
114 |
|
115 def test_undo_deletion_base(self): |
|
116 toto = self.toto |
|
117 e = self.session.create_entity('EmailAddress', |
|
118 address=u'toto@logilab.org', |
|
119 reverse_use_email=toto) |
|
120 # entity with inlined relation |
|
121 p = self.session.create_entity('CWProperty', |
|
122 pkey=u'ui.default-text-format', |
|
123 value=u'text/rest', |
|
124 for_user=toto) |
|
125 self.commit() |
|
126 txs = self.cnx.undoable_transactions() |
|
127 self.assertEquals(len(txs), 2) |
|
128 toto.delete() |
|
129 txuuid = self.commit() |
|
130 actions = self.cnx.transaction_info(txuuid).actions_list() |
|
131 self.assertEquals(len(actions), 1) |
|
132 toto.clear_all_caches() |
|
133 e.clear_all_caches() |
|
134 errors = self.cnx.undo_transaction(txuuid) |
|
135 undotxuuid = self.commit() |
|
136 self.assertEquals(undotxuuid, None) # undo not undoable |
|
137 self.assertEquals(errors, []) |
|
138 self.failUnless(self.execute('Any X WHERE X eid %(x)s', {'x': toto.eid}, 'x')) |
|
139 self.failUnless(self.execute('Any X WHERE X eid %(x)s', {'x': e.eid}, 'x')) |
|
140 self.failUnless(self.execute('Any X WHERE X has_text "toto@logilab"')) |
|
141 self.assertEquals(toto.state, 'activated') |
|
142 self.assertEquals(toto.get_email(), 'toto@logilab.org') |
|
143 self.assertEquals([(p.pkey, p.value) for p in toto.reverse_for_user], |
|
144 [('ui.default-text-format', 'text/rest')]) |
|
145 self.assertEquals([g.name for g in toto.in_group], |
|
146 ['users']) |
|
147 self.assertEquals([et.name for et in toto.related('is', entities=True)], |
|
148 ['CWUser']) |
|
149 self.assertEquals([et.name for et in toto.is_instance_of], |
|
150 ['CWUser']) |
|
151 # undoing shouldn't be visble in undoable transaction, and the undoed |
|
152 # transaction should be removed |
|
153 txs = self.cnx.undoable_transactions() |
|
154 self.assertEquals(len(txs), 2) |
|
155 self.assertRaises(NoSuchTransaction, |
|
156 self.cnx.transaction_info, txuuid) |
|
157 # also check transaction actions have been properly deleted |
|
158 cu = self.session.system_sql( |
|
159 "SELECT * from tx_entity_actions WHERE tx_uuid='%s'" % txuuid) |
|
160 self.failIf(cu.fetchall()) |
|
161 cu = self.session.system_sql( |
|
162 "SELECT * from tx_relation_actions WHERE tx_uuid='%s'" % txuuid) |
|
163 self.failIf(cu.fetchall()) |
|
164 # the final test: check we can login with the previously deleted user |
|
165 self.login('toto') |
|
166 |
|
167 def test_undo_deletion_integrity_1(self): |
|
168 session = self.session |
|
169 # 'Personne fiche Card with' '??' cardinality |
|
170 c = session.create_entity('Card', title=u'hop', content=u'hop') |
|
171 p = session.create_entity('Personne', nom=u'louis', fiche=c) |
|
172 self.commit() |
|
173 c.delete() |
|
174 txuuid = self.commit() |
|
175 c2 = session.create_entity('Card', title=u'hip', content=u'hip') |
|
176 p.set_relations(fiche=c2) |
|
177 self.commit() |
|
178 errors = self.cnx.undo_transaction(txuuid) |
|
179 self.commit() |
|
180 p.clear_all_caches() |
|
181 self.assertEquals(p.fiche[0].eid, c2.eid) |
|
182 self.assertEquals(len(errors), 1) |
|
183 self.assertEquals(errors[0], |
|
184 "Can't restore object relation fiche to entity " |
|
185 "%s which is already linked using this relation." % p.eid) |
|
186 |
|
187 def test_undo_deletion_integrity_2(self): |
|
188 # test validation error raised if we can't restore a required relation |
|
189 session = self.session |
|
190 g = session.create_entity('CWGroup', name=u'staff') |
|
191 session.execute('DELETE U in_group G WHERE U eid %(x)s', {'x': self.toto.eid}) |
|
192 self.toto.set_relations(in_group=g) |
|
193 self.commit() |
|
194 self.toto.delete() |
|
195 txuuid = self.commit() |
|
196 g.delete() |
|
197 self.commit() |
|
198 errors = self.cnx.undo_transaction(txuuid) |
|
199 self.assertRaises(ValidationError, self.commit) |
|
200 |
|
201 def test_undo_creation(self): |
|
202 # XXX what about relation / composite entities which have been created |
|
203 # afterwhile and linked to the undoed addition ? |
|
204 self.skip('not implemented') |
|
205 |
|
206 # test implicit 'replacement' of an inlined relation |