42 super(UndoableTransactionTC, self).setUp() |
47 super(UndoableTransactionTC, self).setUp() |
43 |
48 |
44 def tearDown(self): |
49 def tearDown(self): |
45 cubicweb.server.session.Connection = OldConnection |
50 cubicweb.server.session.Connection = OldConnection |
46 self.restore_connection() |
51 self.restore_connection() |
47 self.session.undo_support = set() |
|
48 super(UndoableTransactionTC, self).tearDown() |
52 super(UndoableTransactionTC, self).tearDown() |
49 |
53 |
50 def check_transaction_deleted(self, txuuid): |
54 def check_transaction_deleted(self, cnx, txuuid): |
51 # also check transaction actions have been properly deleted |
55 # also check transaction actions have been properly deleted |
52 cu = self.session.system_sql( |
56 cu = cnx.system_sql( |
53 "SELECT * from tx_entity_actions WHERE tx_uuid='%s'" % txuuid) |
57 "SELECT * from tx_entity_actions WHERE tx_uuid='%s'" % txuuid) |
54 self.assertFalse(cu.fetchall()) |
58 self.assertFalse(cu.fetchall()) |
55 cu = self.session.system_sql( |
59 cu = cnx.system_sql( |
56 "SELECT * from tx_relation_actions WHERE tx_uuid='%s'" % txuuid) |
60 "SELECT * from tx_relation_actions WHERE tx_uuid='%s'" % txuuid) |
57 self.assertFalse(cu.fetchall()) |
61 self.assertFalse(cu.fetchall()) |
58 |
62 |
59 def assertUndoTransaction(self, txuuid, expected_errors=None): |
63 def assertUndoTransaction(self, cnx, txuuid, expected_errors=None): |
60 if expected_errors is None : |
64 if expected_errors is None : |
61 expected_errors = [] |
65 expected_errors = [] |
62 try: |
66 try: |
63 self.cnx.undo_transaction(txuuid) |
67 cnx.undo_transaction(txuuid) |
64 except UndoTransactionException as exn: |
68 except UndoTransactionException as exn: |
65 errors = exn.errors |
69 errors = exn.errors |
66 else: |
70 else: |
67 errors = [] |
71 errors = [] |
68 self.assertEqual(errors, expected_errors) |
72 self.assertEqual(errors, expected_errors) |
69 |
73 |
70 def test_undo_api(self): |
74 def test_undo_api(self): |
71 self.assertTrue(self.txuuid) |
75 self.assertTrue(self.txuuid) |
72 # test transaction api |
76 # test transaction api |
73 self.assertRaises(NoSuchTransaction, |
77 with self.admin_access.client_cnx() as cnx: |
74 self.cnx.transaction_info, 'hop') |
78 self.assertRaises(NoSuchTransaction, |
75 self.assertRaises(NoSuchTransaction, |
79 cnx.transaction_info, 'hop') |
76 self.cnx.transaction_actions, 'hop') |
80 self.assertRaises(NoSuchTransaction, |
77 self.assertRaises(NoSuchTransaction, |
81 cnx.transaction_actions, 'hop') |
78 self.cnx.undo_transaction, 'hop') |
82 self.assertRaises(NoSuchTransaction, |
79 txinfo = self.cnx.transaction_info(self.txuuid) |
83 cnx.undo_transaction, 'hop') |
80 self.assertTrue(txinfo.datetime) |
84 txinfo = cnx.transaction_info(self.txuuid) |
81 self.assertEqual(txinfo.user_eid, self.session.user.eid) |
85 self.assertTrue(txinfo.datetime) |
82 self.assertEqual(txinfo.user().login, 'admin') |
86 self.assertEqual(txinfo.user_eid, cnx.user.eid) |
83 actions = txinfo.actions_list() |
87 self.assertEqual(txinfo.user().login, 'admin') |
84 self.assertEqual(len(actions), 2) |
88 actions = txinfo.actions_list() |
85 actions = txinfo.actions_list(public=False) |
89 self.assertEqual(len(actions), 2) |
86 self.assertEqual(len(actions), 6) |
90 actions = txinfo.actions_list(public=False) |
87 a1 = actions[0] |
91 self.assertEqual(len(actions), 6) |
88 self.assertEqual(a1.action, 'C') |
92 a1 = actions[0] |
89 self.assertEqual(a1.eid, self.toto.eid) |
93 self.assertEqual(a1.action, 'C') |
90 self.assertEqual(a1.etype,'CWUser') |
94 self.assertEqual(a1.eid, self.totoeid) |
91 self.assertEqual(a1.ertype, 'CWUser') |
95 self.assertEqual(a1.etype,'CWUser') |
92 self.assertEqual(a1.changes, None) |
96 self.assertEqual(a1.ertype, 'CWUser') |
93 self.assertEqual(a1.public, True) |
97 self.assertEqual(a1.changes, None) |
94 self.assertEqual(a1.order, 1) |
98 self.assertEqual(a1.public, True) |
95 a4 = actions[3] |
99 self.assertEqual(a1.order, 1) |
96 self.assertEqual(a4.action, 'A') |
100 a4 = actions[3] |
97 self.assertEqual(a4.rtype, 'in_group') |
101 self.assertEqual(a4.action, 'A') |
98 self.assertEqual(a4.ertype, 'in_group') |
102 self.assertEqual(a4.rtype, 'in_group') |
99 self.assertEqual(a4.eid_from, self.toto.eid) |
103 self.assertEqual(a4.ertype, 'in_group') |
100 self.assertEqual(a4.eid_to, self.toto.in_group[0].eid) |
104 self.assertEqual(a4.eid_from, self.totoeid) |
101 self.assertEqual(a4.order, 4) |
105 self.assertEqual(a4.eid_to, self.toto(cnx).in_group[0].eid) |
102 for i, rtype in ((1, 'owned_by'), (2, 'owned_by'), |
106 self.assertEqual(a4.order, 4) |
103 (4, 'in_state'), (5, 'created_by')): |
107 for i, rtype in ((1, 'owned_by'), (2, 'owned_by'), |
104 a = actions[i] |
108 (4, 'in_state'), (5, 'created_by')): |
105 self.assertEqual(a.action, 'A') |
109 a = actions[i] |
106 self.assertEqual(a.eid_from, self.toto.eid) |
110 self.assertEqual(a.action, 'A') |
107 self.assertEqual(a.rtype, rtype) |
111 self.assertEqual(a.eid_from, self.totoeid) |
108 self.assertEqual(a.order, i+1) |
112 self.assertEqual(a.rtype, rtype) |
109 # test undoable_transactions |
113 self.assertEqual(a.order, i+1) |
110 txs = self.cnx.undoable_transactions() |
114 # test undoable_transactions |
111 self.assertEqual(len(txs), 1) |
115 txs = cnx.undoable_transactions() |
112 self.assertEqual(txs[0].uuid, self.txuuid) |
116 self.assertEqual(len(txs), 1) |
113 # test transaction_info / undoable_transactions security |
117 self.assertEqual(txs[0].uuid, self.txuuid) |
114 cnx = self.login('anon') |
118 # test transaction_info / undoable_transactions security |
115 self.assertRaises(NoSuchTransaction, |
119 with self.new_access('anon').client_cnx() as cnx: |
116 cnx.transaction_info, self.txuuid) |
120 self.assertRaises(NoSuchTransaction, |
117 self.assertRaises(NoSuchTransaction, |
121 cnx.transaction_info, self.txuuid) |
118 cnx.transaction_actions, self.txuuid) |
122 self.assertRaises(NoSuchTransaction, |
119 self.assertRaises(NoSuchTransaction, |
123 cnx.transaction_actions, self.txuuid) |
120 cnx.undo_transaction, self.txuuid) |
124 self.assertRaises(NoSuchTransaction, |
121 txs = cnx.undoable_transactions() |
125 cnx.undo_transaction, self.txuuid) |
122 self.assertEqual(len(txs), 0) |
126 txs = cnx.undoable_transactions() |
|
127 self.assertEqual(len(txs), 0) |
123 |
128 |
124 def test_undoable_transactions(self): |
129 def test_undoable_transactions(self): |
125 toto = self.toto |
130 with self.admin_access.client_cnx() as cnx: |
126 e = self.session.create_entity('EmailAddress', |
131 toto = self.toto(cnx) |
127 address=u'toto@logilab.org', |
132 e = cnx.create_entity('EmailAddress', |
128 reverse_use_email=toto) |
133 address=u'toto@logilab.org', |
129 txuuid1 = self.commit() |
134 reverse_use_email=toto) |
130 toto.cw_delete() |
135 txuuid1 = cnx.commit() |
131 txuuid2 = self.commit() |
136 toto.cw_delete() |
132 undoable_transactions = self.cnx.undoable_transactions |
137 txuuid2 = cnx.commit() |
133 txs = undoable_transactions(action='D') |
138 undoable_transactions = cnx.undoable_transactions |
134 self.assertEqual(len(txs), 1, txs) |
139 txs = undoable_transactions(action='D') |
135 self.assertEqual(txs[0].uuid, txuuid2) |
140 self.assertEqual(len(txs), 1, txs) |
136 txs = undoable_transactions(action='C') |
141 self.assertEqual(txs[0].uuid, txuuid2) |
137 self.assertEqual(len(txs), 2, txs) |
142 txs = undoable_transactions(action='C') |
138 self.assertEqual(txs[0].uuid, txuuid1) |
143 self.assertEqual(len(txs), 2, txs) |
139 self.assertEqual(txs[1].uuid, self.txuuid) |
144 self.assertEqual(txs[0].uuid, txuuid1) |
140 txs = undoable_transactions(eid=toto.eid) |
145 self.assertEqual(txs[1].uuid, self.txuuid) |
141 self.assertEqual(len(txs), 3) |
146 txs = undoable_transactions(eid=toto.eid) |
142 self.assertEqual(txs[0].uuid, txuuid2) |
147 self.assertEqual(len(txs), 3) |
143 self.assertEqual(txs[1].uuid, txuuid1) |
148 self.assertEqual(txs[0].uuid, txuuid2) |
144 self.assertEqual(txs[2].uuid, self.txuuid) |
149 self.assertEqual(txs[1].uuid, txuuid1) |
145 txs = undoable_transactions(etype='CWUser') |
150 self.assertEqual(txs[2].uuid, self.txuuid) |
146 self.assertEqual(len(txs), 2) |
151 txs = undoable_transactions(etype='CWUser') |
147 txs = undoable_transactions(etype='CWUser', action='C') |
152 self.assertEqual(len(txs), 2) |
148 self.assertEqual(len(txs), 1) |
153 txs = undoable_transactions(etype='CWUser', action='C') |
149 self.assertEqual(txs[0].uuid, self.txuuid) |
154 self.assertEqual(len(txs), 1) |
150 txs = undoable_transactions(etype='EmailAddress', action='D') |
155 self.assertEqual(txs[0].uuid, self.txuuid) |
151 self.assertEqual(len(txs), 0) |
156 txs = undoable_transactions(etype='EmailAddress', action='D') |
152 txs = undoable_transactions(etype='EmailAddress', action='D', |
157 self.assertEqual(len(txs), 0) |
153 public=False) |
158 txs = undoable_transactions(etype='EmailAddress', action='D', |
154 self.assertEqual(len(txs), 1) |
159 public=False) |
155 self.assertEqual(txs[0].uuid, txuuid2) |
160 self.assertEqual(len(txs), 1) |
156 txs = undoable_transactions(eid=toto.eid, action='R', public=False) |
161 self.assertEqual(txs[0].uuid, txuuid2) |
157 self.assertEqual(len(txs), 1) |
162 txs = undoable_transactions(eid=toto.eid, action='R', public=False) |
158 self.assertEqual(txs[0].uuid, txuuid2) |
163 self.assertEqual(len(txs), 1) |
|
164 self.assertEqual(txs[0].uuid, txuuid2) |
159 |
165 |
160 def test_undo_deletion_base(self): |
166 def test_undo_deletion_base(self): |
161 toto = self.toto |
167 with self.admin_access.client_cnx() as cnx: |
162 e = self.session.create_entity('EmailAddress', |
168 toto = self.toto(cnx) |
163 address=u'toto@logilab.org', |
169 e = cnx.create_entity('EmailAddress', |
164 reverse_use_email=toto) |
170 address=u'toto@logilab.org', |
165 # entity with inlined relation |
171 reverse_use_email=toto) |
166 p = self.session.create_entity('CWProperty', |
172 # entity with inlined relation |
167 pkey=u'ui.default-text-format', |
173 p = cnx.create_entity('CWProperty', |
168 value=u'text/rest', |
174 pkey=u'ui.default-text-format', |
169 for_user=toto) |
175 value=u'text/rest', |
170 self.commit() |
176 for_user=toto) |
171 txs = self.cnx.undoable_transactions() |
177 cnx.commit() |
172 self.assertEqual(len(txs), 2) |
178 txs = cnx.undoable_transactions() |
173 toto.cw_delete() |
179 self.assertEqual(len(txs), 2) |
174 txuuid = self.commit() |
180 toto.cw_delete() |
175 actions = self.cnx.transaction_info(txuuid).actions_list() |
181 txuuid = cnx.commit() |
176 self.assertEqual(len(actions), 1) |
182 actions = cnx.transaction_info(txuuid).actions_list() |
177 toto.cw_clear_all_caches() |
183 self.assertEqual(len(actions), 1) |
178 e.cw_clear_all_caches() |
184 toto.cw_clear_all_caches() |
179 self.assertUndoTransaction(txuuid) |
185 e.cw_clear_all_caches() |
180 undotxuuid = self.commit() |
186 self.assertUndoTransaction(cnx, txuuid) |
181 self.assertEqual(undotxuuid, None) # undo not undoable |
187 undotxuuid = cnx.commit() |
182 self.assertTrue(self.execute('Any X WHERE X eid %(x)s', {'x': toto.eid})) |
188 self.assertEqual(undotxuuid, None) # undo not undoable |
183 self.assertTrue(self.execute('Any X WHERE X eid %(x)s', {'x': e.eid})) |
189 self.assertTrue(cnx.execute('Any X WHERE X eid %(x)s', {'x': toto.eid})) |
184 self.assertTrue(self.execute('Any X WHERE X has_text "toto@logilab"')) |
190 self.assertTrue(cnx.execute('Any X WHERE X eid %(x)s', {'x': e.eid})) |
185 self.assertEqual(toto.cw_adapt_to('IWorkflowable').state, 'activated') |
191 self.assertTrue(cnx.execute('Any X WHERE X has_text "toto@logilab"')) |
186 self.assertEqual(toto.cw_adapt_to('IEmailable').get_email(), 'toto@logilab.org') |
192 self.assertEqual(toto.cw_adapt_to('IWorkflowable').state, 'activated') |
187 self.assertEqual([(p.pkey, p.value) for p in toto.reverse_for_user], |
193 self.assertEqual(toto.cw_adapt_to('IEmailable').get_email(), 'toto@logilab.org') |
188 [('ui.default-text-format', 'text/rest')]) |
194 self.assertEqual([(p.pkey, p.value) for p in toto.reverse_for_user], |
189 self.assertEqual([g.name for g in toto.in_group], |
195 [('ui.default-text-format', 'text/rest')]) |
190 ['users']) |
196 self.assertEqual([g.name for g in toto.in_group], |
191 self.assertEqual([et.name for et in toto.related('is', entities=True)], |
197 ['users']) |
192 ['CWUser']) |
198 self.assertEqual([et.name for et in toto.related('is', entities=True)], |
193 self.assertEqual([et.name for et in toto.is_instance_of], |
199 ['CWUser']) |
194 ['CWUser']) |
200 self.assertEqual([et.name for et in toto.is_instance_of], |
195 # undoing shouldn't be visble in undoable transaction, and the undone |
201 ['CWUser']) |
196 # transaction should be removed |
202 # undoing shouldn't be visble in undoable transaction, and the undone |
197 txs = self.cnx.undoable_transactions() |
203 # transaction should be removed |
198 self.assertEqual(len(txs), 2) |
204 txs = self.cnx.undoable_transactions() |
199 self.assertRaises(NoSuchTransaction, |
205 self.assertEqual(len(txs), 2) |
200 self.cnx.transaction_info, txuuid) |
206 self.assertRaises(NoSuchTransaction, |
201 self.check_transaction_deleted(txuuid) |
207 self.cnx.transaction_info, txuuid) |
202 # the final test: check we can login with the previously deleted user |
208 with self.admin_access.repo_cnx() as cnx: |
203 self.login('toto') |
209 with cnx.ensure_cnx_set: |
|
210 self.check_transaction_deleted(cnx, txuuid) |
|
211 # the final test: check we can login with the previously deleted user |
|
212 with self.new_access('toto').client_cnx(): |
|
213 pass |
204 |
214 |
205 def test_undo_deletion_integrity_1(self): |
215 def test_undo_deletion_integrity_1(self): |
206 session = self.session |
216 with self.admin_access.client_cnx() as cnx: |
207 # 'Personne fiche Card with' '??' cardinality |
217 # 'Personne fiche Card with' '??' cardinality |
208 c = session.create_entity('Card', title=u'hop', content=u'hop') |
218 c = cnx.create_entity('Card', title=u'hop', content=u'hop') |
209 p = session.create_entity('Personne', nom=u'louis', fiche=c) |
219 p = cnx.create_entity('Personne', nom=u'louis', fiche=c) |
210 self.commit() |
220 cnx.commit() |
211 c.cw_delete() |
221 c.cw_delete() |
212 txuuid = self.commit() |
222 txuuid = cnx.commit() |
213 c2 = session.create_entity('Card', title=u'hip', content=u'hip') |
223 c2 = cnx.create_entity('Card', title=u'hip', content=u'hip') |
214 p.cw_set(fiche=c2) |
224 p.cw_set(fiche=c2) |
215 self.commit() |
225 cnx.commit() |
216 self.assertUndoTransaction(txuuid, [ |
226 self.assertUndoTransaction(cnx, txuuid, [ |
217 "Can't restore object relation fiche to entity " |
227 "Can't restore object relation fiche to entity " |
218 "%s which is already linked using this relation." % p.eid]) |
228 "%s which is already linked using this relation." % p.eid]) |
219 self.commit() |
229 cnx.commit() |
220 p.cw_clear_all_caches() |
230 p.cw_clear_all_caches() |
221 self.assertEqual(p.fiche[0].eid, c2.eid) |
231 self.assertEqual(p.fiche[0].eid, c2.eid) |
222 |
232 |
223 def test_undo_deletion_integrity_2(self): |
233 def test_undo_deletion_integrity_2(self): |
224 # test validation error raised if we can't restore a required relation |
234 with self.admin_access.client_cnx() as cnx: |
225 session = self.session |
235 # test validation error raised if we can't restore a required relation |
226 g = session.create_entity('CWGroup', name=u'staff') |
236 g = cnx.create_entity('CWGroup', name=u'staff') |
227 session.execute('DELETE U in_group G WHERE U eid %(x)s', {'x': self.toto.eid}) |
237 cnx.execute('DELETE U in_group G WHERE U eid %(x)s', {'x': self.totoeid}) |
228 self.toto.cw_set(in_group=g) |
238 self.toto(cnx).cw_set(in_group=g) |
229 self.commit() |
239 cnx.commit() |
230 self.toto.cw_delete() |
240 self.toto(cnx).cw_delete() |
231 txuuid = self.commit() |
241 txuuid = cnx.commit() |
232 g.cw_delete() |
242 g.cw_delete() |
233 self.commit() |
243 cnx.commit() |
234 self.assertUndoTransaction(txuuid, [ |
244 self.assertUndoTransaction(cnx, txuuid, [ |
235 u"Can't restore relation in_group, object entity " |
245 u"Can't restore relation in_group, object entity " |
236 "%s doesn't exist anymore." % g.eid]) |
246 "%s doesn't exist anymore." % g.eid]) |
237 with self.assertRaises(ValidationError) as cm: |
247 with self.assertRaises(ValidationError) as cm: |
238 self.commit() |
248 cnx.commit() |
239 cm.exception.translate(unicode) |
249 cm.exception.translate(unicode) |
240 self.assertEqual(cm.exception.entity, self.toto.eid) |
250 self.assertEqual(cm.exception.entity, self.totoeid) |
241 self.assertEqual(cm.exception.errors, |
251 self.assertEqual(cm.exception.errors, |
242 {'in_group-subject': u'at least one relation in_group is ' |
252 {'in_group-subject': u'at least one relation in_group is ' |
243 'required on CWUser (%s)' % self.toto.eid}) |
253 'required on CWUser (%s)' % self.totoeid}) |
244 |
254 |
245 def test_undo_creation_1(self): |
255 def test_undo_creation_1(self): |
246 session = self.session |
256 with self.admin_access.client_cnx() as cnx: |
247 c = session.create_entity('Card', title=u'hop', content=u'hop') |
257 c = cnx.create_entity('Card', title=u'hop', content=u'hop') |
248 p = session.create_entity('Personne', nom=u'louis', fiche=c) |
258 p = cnx.create_entity('Personne', nom=u'louis', fiche=c) |
249 txuuid = self.commit() |
259 txuuid = cnx.commit() |
250 self.assertUndoTransaction(txuuid) |
260 self.assertUndoTransaction(cnx, txuuid) |
251 self.commit() |
261 cnx.commit() |
252 self.assertFalse(self.execute('Any X WHERE X eid %(x)s', {'x': c.eid})) |
262 self.assertFalse(cnx.execute('Any X WHERE X eid %(x)s', {'x': c.eid})) |
253 self.assertFalse(self.execute('Any X WHERE X eid %(x)s', {'x': p.eid})) |
263 self.assertFalse(cnx.execute('Any X WHERE X eid %(x)s', {'x': p.eid})) |
254 self.assertFalse(self.execute('Any X,Y WHERE X fiche Y')) |
264 self.assertFalse(cnx.execute('Any X,Y WHERE X fiche Y')) |
255 self.session.set_cnxset() |
265 with self.admin_access.repo_cnx() as cnx: |
256 for eid in (p.eid, c.eid): |
266 with cnx.ensure_cnx_set: |
257 self.assertFalse(session.system_sql( |
267 for eid in (p.eid, c.eid): |
258 'SELECT * FROM entities WHERE eid=%s' % eid).fetchall()) |
268 self.assertFalse(cnx.system_sql( |
259 self.assertFalse(session.system_sql( |
269 'SELECT * FROM entities WHERE eid=%s' % eid).fetchall()) |
260 'SELECT 1 FROM owned_by_relation WHERE eid_from=%s' % eid).fetchall()) |
270 self.assertFalse(cnx.system_sql( |
261 # added by sql in hooks (except when using dataimport) |
271 'SELECT 1 FROM owned_by_relation WHERE eid_from=%s' % eid).fetchall()) |
262 self.assertFalse(session.system_sql( |
272 # added by sql in hooks (except when using dataimport) |
263 'SELECT 1 FROM is_relation WHERE eid_from=%s' % eid).fetchall()) |
273 self.assertFalse(cnx.system_sql( |
264 self.assertFalse(session.system_sql( |
274 'SELECT 1 FROM is_relation WHERE eid_from=%s' % eid).fetchall()) |
265 'SELECT 1 FROM is_instance_of_relation WHERE eid_from=%s' % eid).fetchall()) |
275 self.assertFalse(cnx.system_sql( |
266 self.check_transaction_deleted(txuuid) |
276 'SELECT 1 FROM is_instance_of_relation WHERE eid_from=%s' % eid).fetchall()) |
267 |
277 self.check_transaction_deleted(cnx, txuuid) |
268 |
278 |
269 def test_undo_creation_integrity_1(self): |
279 def test_undo_creation_integrity_1(self): |
270 session = self.session |
280 with self.admin_access.client_cnx() as cnx: |
271 req = self.request() |
281 tutu = self.create_user(cnx, 'tutu', commit=False) |
272 tutu = self.create_user(req, 'tutu', commit=False) |
282 txuuid = cnx.commit() |
273 txuuid = self.commit() |
283 email = cnx.create_entity('EmailAddress', address=u'tutu@cubicweb.org') |
274 email = self.request().create_entity('EmailAddress', address=u'tutu@cubicweb.org') |
284 prop = cnx.create_entity('CWProperty', pkey=u'ui.default-text-format', |
275 prop = self.request().create_entity('CWProperty', pkey=u'ui.default-text-format', |
285 value=u'text/html') |
276 value=u'text/html') |
286 tutu.cw_set(use_email=email, reverse_for_user=prop) |
277 tutu.cw_set(use_email=email, reverse_for_user=prop) |
287 cnx.commit() |
278 self.commit() |
288 with self.assertRaises(ValidationError) as cm: |
279 with self.assertRaises(ValidationError) as cm: |
289 cnx.undo_transaction(txuuid) |
280 self.cnx.undo_transaction(txuuid) |
290 self.assertEqual(cm.exception.entity, tutu.eid) |
281 self.assertEqual(cm.exception.entity, tutu.eid) |
291 self.assertEqual(cm.exception.errors, |
282 self.assertEqual(cm.exception.errors, |
292 {None: 'some later transaction(s) touch entity, undo them first'}) |
283 {None: 'some later transaction(s) touch entity, undo them first'}) |
|
284 |
293 |
285 def test_undo_creation_integrity_2(self): |
294 def test_undo_creation_integrity_2(self): |
286 session = self.session |
295 with self.admin_access.client_cnx() as cnx: |
287 g = session.create_entity('CWGroup', name=u'staff') |
296 g = cnx.create_entity('CWGroup', name=u'staff') |
288 txuuid = self.commit() |
297 txuuid = cnx.commit() |
289 session.execute('DELETE U in_group G WHERE U eid %(x)s', {'x': self.toto.eid}) |
298 cnx.execute('DELETE U in_group G WHERE U eid %(x)s', {'x': self.totoeid}) |
290 self.toto.cw_set(in_group=g) |
299 self.toto(cnx).cw_set(in_group=g) |
291 self.commit() |
300 cnx.commit() |
292 with self.assertRaises(ValidationError) as cm: |
301 with self.assertRaises(ValidationError) as cm: |
293 self.cnx.undo_transaction(txuuid) |
302 cnx.undo_transaction(txuuid) |
294 self.assertEqual(cm.exception.entity, g.eid) |
303 self.assertEqual(cm.exception.entity, g.eid) |
295 self.assertEqual(cm.exception.errors, |
304 self.assertEqual(cm.exception.errors, |
296 {None: 'some later transaction(s) touch entity, undo them first'}) |
305 {None: 'some later transaction(s) touch entity, undo them first'}) |
297 # self.assertEqual(errors, |
306 # self.assertEqual(errors, |
298 # [u"Can't restore relation in_group, object entity " |
307 # [u"Can't restore relation in_group, object entity " |
299 # "%s doesn't exist anymore." % g.eid]) |
308 # "%s doesn't exist anymore." % g.eid]) |
300 # with self.assertRaises(ValidationError) as cm: self.commit() |
309 # with self.assertRaises(ValidationError) as cm: cnx.commit() |
301 # self.assertEqual(cm.exception.entity, self.toto.eid) |
310 # self.assertEqual(cm.exception.entity, self.totoeid) |
302 # self.assertEqual(cm.exception.errors, |
311 # self.assertEqual(cm.exception.errors, |
303 # {'in_group-subject': u'at least one relation in_group is ' |
312 # {'in_group-subject': u'at least one relation in_group is ' |
304 # 'required on CWUser (%s)' % self.toto.eid}) |
313 # 'required on CWUser (%s)' % self.totoeid}) |
305 |
314 |
306 # test implicit 'replacement' of an inlined relation |
315 # test implicit 'replacement' of an inlined relation |
307 |
316 |
308 def test_undo_inline_rel_remove_ok(self): |
317 def test_undo_inline_rel_remove_ok(self): |
309 """Undo remove relation Personne (?) fiche (?) Card |
318 """Undo remove relation Personne (?) fiche (?) Card |
310 |
319 |
311 NB: processed by `_undo_r` as expected""" |
320 NB: processed by `_undo_r` as expected""" |
312 session = self.session |
321 with self.admin_access.client_cnx() as cnx: |
313 c = session.create_entity('Card', title=u'hop', content=u'hop') |
322 c = cnx.create_entity('Card', title=u'hop', content=u'hop') |
314 p = session.create_entity('Personne', nom=u'louis', fiche=c) |
323 p = cnx.create_entity('Personne', nom=u'louis', fiche=c) |
315 self.commit() |
324 cnx.commit() |
316 p.cw_set(fiche=None) |
325 p.cw_set(fiche=None) |
317 txuuid = self.commit() |
326 txuuid = cnx.commit() |
318 self.assertUndoTransaction(txuuid) |
327 self.assertUndoTransaction(cnx, txuuid) |
319 self.commit() |
328 cnx.commit() |
320 p.cw_clear_all_caches() |
329 p.cw_clear_all_caches() |
321 self.assertEqual(p.fiche[0].eid, c.eid) |
330 self.assertEqual(p.fiche[0].eid, c.eid) |
322 |
331 |
323 def test_undo_inline_rel_remove_ko(self): |
332 def test_undo_inline_rel_remove_ko(self): |
324 """Restore an inlined relation to a deleted entity, with an error. |
333 """Restore an inlined relation to a deleted entity, with an error. |
325 |
334 |
326 NB: processed by `_undo_r` as expected""" |
335 NB: processed by `_undo_r` as expected""" |
327 session = self.session |
336 with self.admin_access.client_cnx() as cnx: |
328 c = session.create_entity('Card', title=u'hop', content=u'hop') |
337 c = cnx.create_entity('Card', title=u'hop', content=u'hop') |
329 p = session.create_entity('Personne', nom=u'louis', fiche=c) |
338 p = cnx.create_entity('Personne', nom=u'louis', fiche=c) |
330 self.commit() |
339 cnx.commit() |
331 p.cw_set(fiche=None) |
340 p.cw_set(fiche=None) |
332 txuuid = self.commit() |
341 txuuid = cnx.commit() |
333 c.cw_delete() |
342 c.cw_delete() |
334 self.commit() |
343 cnx.commit() |
335 self.assertUndoTransaction(txuuid, [ |
344 self.assertUndoTransaction(cnx, txuuid, [ |
336 "Can't restore relation fiche, object entity %d doesn't exist anymore." % c.eid]) |
345 "Can't restore relation fiche, object entity %d doesn't exist anymore." % c.eid]) |
337 self.commit() |
346 cnx.commit() |
338 p.cw_clear_all_caches() |
347 p.cw_clear_all_caches() |
339 self.assertFalse(p.fiche) |
348 self.assertFalse(p.fiche) |
340 self.assertIsNone(session.system_sql( |
349 with self.admin_access.repo_cnx() as cnx: |
341 'SELECT cw_fiche FROM cw_Personne WHERE cw_eid=%s' % p.eid).fetchall()[0][0]) |
350 with cnx.ensure_cnx_set: |
|
351 self.assertIsNone(cnx.system_sql( |
|
352 'SELECT cw_fiche FROM cw_Personne WHERE cw_eid=%s' % p.eid).fetchall()[0][0]) |
342 |
353 |
343 def test_undo_inline_rel_add_ok(self): |
354 def test_undo_inline_rel_add_ok(self): |
344 """Undo add relation Personne (?) fiche (?) Card |
355 """Undo add relation Personne (?) fiche (?) Card |
345 |
356 |
346 Caution processed by `_undo_u`, not `_undo_a` !""" |
357 Caution processed by `_undo_u`, not `_undo_a` !""" |
347 session = self.session |
358 with self.admin_access.client_cnx() as cnx: |
348 c = session.create_entity('Card', title=u'hop', content=u'hop') |
359 c = cnx.create_entity('Card', title=u'hop', content=u'hop') |
349 p = session.create_entity('Personne', nom=u'louis') |
360 p = cnx.create_entity('Personne', nom=u'louis') |
350 self.commit() |
361 cnx.commit() |
351 p.cw_set(fiche=c) |
362 p.cw_set(fiche=c) |
352 txuuid = self.commit() |
363 txuuid = cnx.commit() |
353 self.assertUndoTransaction(txuuid) |
364 self.assertUndoTransaction(cnx, txuuid) |
354 self.commit() |
365 cnx.commit() |
355 p.cw_clear_all_caches() |
366 p.cw_clear_all_caches() |
356 self.assertFalse(p.fiche) |
367 self.assertFalse(p.fiche) |
357 |
368 |
358 def test_undo_inline_rel_add_ko(self): |
369 def test_undo_inline_rel_add_ko(self): |
359 """Undo add relation Personne (?) fiche (?) Card |
370 """Undo add relation Personne (?) fiche (?) Card |
360 |
371 |
361 Caution processed by `_undo_u`, not `_undo_a` !""" |
372 Caution processed by `_undo_u`, not `_undo_a` !""" |
362 session = self.session |
373 with self.admin_access.client_cnx() as cnx: |
363 c = session.create_entity('Card', title=u'hop', content=u'hop') |
374 c = cnx.create_entity('Card', title=u'hop', content=u'hop') |
364 p = session.create_entity('Personne', nom=u'louis') |
375 p = cnx.create_entity('Personne', nom=u'louis') |
365 self.commit() |
376 cnx.commit() |
366 p.cw_set(fiche=c) |
377 p.cw_set(fiche=c) |
367 txuuid = self.commit() |
378 txuuid = cnx.commit() |
368 c.cw_delete() |
379 c.cw_delete() |
369 self.commit() |
380 cnx.commit() |
370 self.assertUndoTransaction(txuuid) |
381 self.assertUndoTransaction(cnx, txuuid) |
371 |
382 |
372 def test_undo_inline_rel_replace_ok(self): |
383 def test_undo_inline_rel_replace_ok(self): |
373 """Undo changing relation Personne (?) fiche (?) Card |
384 """Undo changing relation Personne (?) fiche (?) Card |
374 |
385 |
375 Caution processed by `_undo_u` """ |
386 Caution processed by `_undo_u` """ |
376 session = self.session |
387 with self.admin_access.client_cnx() as cnx: |
377 c1 = session.create_entity('Card', title=u'hop', content=u'hop') |
388 c1 = cnx.create_entity('Card', title=u'hop', content=u'hop') |
378 c2 = session.create_entity('Card', title=u'hip', content=u'hip') |
389 c2 = cnx.create_entity('Card', title=u'hip', content=u'hip') |
379 p = session.create_entity('Personne', nom=u'louis', fiche=c1) |
390 p = cnx.create_entity('Personne', nom=u'louis', fiche=c1) |
380 self.commit() |
391 cnx.commit() |
381 p.cw_set(fiche=c2) |
392 p.cw_set(fiche=c2) |
382 txuuid = self.commit() |
393 txuuid = cnx.commit() |
383 self.assertUndoTransaction(txuuid) |
394 self.assertUndoTransaction(cnx, txuuid) |
384 self.commit() |
395 cnx.commit() |
385 p.cw_clear_all_caches() |
396 p.cw_clear_all_caches() |
386 self.assertEqual(p.fiche[0].eid, c1.eid) |
397 self.assertEqual(p.fiche[0].eid, c1.eid) |
387 |
398 |
388 def test_undo_inline_rel_replace_ko(self): |
399 def test_undo_inline_rel_replace_ko(self): |
389 """Undo changing relation Personne (?) fiche (?) Card, with an error |
400 """Undo changing relation Personne (?) fiche (?) Card, with an error |
390 |
401 |
391 Caution processed by `_undo_u` """ |
402 Caution processed by `_undo_u` """ |
392 session = self.session |
403 with self.admin_access.client_cnx() as cnx: |
393 c1 = session.create_entity('Card', title=u'hop', content=u'hop') |
404 c1 = cnx.create_entity('Card', title=u'hop', content=u'hop') |
394 c2 = session.create_entity('Card', title=u'hip', content=u'hip') |
405 c2 = cnx.create_entity('Card', title=u'hip', content=u'hip') |
395 p = session.create_entity('Personne', nom=u'louis', fiche=c1) |
406 p = cnx.create_entity('Personne', nom=u'louis', fiche=c1) |
396 self.commit() |
407 cnx.commit() |
397 p.cw_set(fiche=c2) |
408 p.cw_set(fiche=c2) |
398 txuuid = self.commit() |
409 txuuid = cnx.commit() |
399 c1.cw_delete() |
410 c1.cw_delete() |
400 self.commit() |
411 cnx.commit() |
401 self.assertUndoTransaction(txuuid, [ |
412 self.assertUndoTransaction(cnx, txuuid, [ |
402 "can't restore entity %s of type Personne, target of fiche (eid %s)" |
413 "can't restore entity %s of type Personne, target of fiche (eid %s)" |
403 " does not exist any longer" % (p.eid, c1.eid)]) |
414 " does not exist any longer" % (p.eid, c1.eid)]) |
404 self.commit() |
415 cnx.commit() |
405 p.cw_clear_all_caches() |
416 p.cw_clear_all_caches() |
406 self.assertFalse(p.fiche) |
417 self.assertFalse(p.fiche) |
407 |
418 |
408 def test_undo_attr_update_ok(self): |
419 def test_undo_attr_update_ok(self): |
409 session = self.session |
420 with self.admin_access.client_cnx() as cnx: |
410 p = session.create_entity('Personne', nom=u'toto') |
421 p = cnx.create_entity('Personne', nom=u'toto') |
411 session.commit() |
422 cnx.commit() |
412 self.session.set_cnxset() |
423 p.cw_set(nom=u'titi') |
413 p.cw_set(nom=u'titi') |
424 txuuid = cnx.commit() |
414 txuuid = self.commit() |
425 self.assertUndoTransaction(cnx, txuuid) |
415 self.assertUndoTransaction(txuuid) |
426 p.cw_clear_all_caches() |
416 p.cw_clear_all_caches() |
427 self.assertEqual(p.nom, u'toto') |
417 self.assertEqual(p.nom, u'toto') |
|
418 |
428 |
419 def test_undo_attr_update_ko(self): |
429 def test_undo_attr_update_ko(self): |
420 session = self.session |
430 with self.admin_access.client_cnx() as cnx: |
421 p = session.create_entity('Personne', nom=u'toto') |
431 p = cnx.create_entity('Personne', nom=u'toto') |
422 session.commit() |
432 cnx.commit() |
423 self.session.set_cnxset() |
433 p.cw_set(nom=u'titi') |
424 p.cw_set(nom=u'titi') |
434 txuuid = cnx.commit() |
425 txuuid = self.commit() |
435 p.cw_delete() |
426 p.cw_delete() |
436 cnx.commit() |
427 self.commit() |
437 self.assertUndoTransaction(cnx, txuuid, [ |
428 self.assertUndoTransaction(txuuid, [ |
438 u"can't restore state of entity %s, it has been deleted inbetween" % p.eid]) |
429 u"can't restore state of entity %s, it has been deleted inbetween" % p.eid]) |
|
430 |
439 |
431 |
440 |
432 class UndoExceptionInUnicode(CubicWebTC): |
441 class UndoExceptionInUnicode(CubicWebTC): |
433 |
442 |
434 # problem occurs in string manipulation for python < 2.6 |
443 # problem occurs in string manipulation for python < 2.6 |