45 "SELECT * from tx_entity_actions WHERE tx_uuid='%s'" % txuuid) |
45 "SELECT * from tx_entity_actions WHERE tx_uuid='%s'" % txuuid) |
46 self.assertFalse(cu.fetchall()) |
46 self.assertFalse(cu.fetchall()) |
47 cu = self.session.system_sql( |
47 cu = self.session.system_sql( |
48 "SELECT * from tx_relation_actions WHERE tx_uuid='%s'" % txuuid) |
48 "SELECT * from tx_relation_actions WHERE tx_uuid='%s'" % txuuid) |
49 self.assertFalse(cu.fetchall()) |
49 self.assertFalse(cu.fetchall()) |
|
50 |
|
51 def assertUndoTransaction(self, txuuid, expected_errors=None): |
|
52 if expected_errors is None : |
|
53 expected_errors = [] |
|
54 try: |
|
55 self.cnx.undo_transaction(txuuid) |
|
56 except UndoTransactionException, exn: |
|
57 errors = exn.errors |
|
58 else: |
|
59 errors = [] |
|
60 self.assertEqual(errors, expected_errors) |
50 |
61 |
51 def test_undo_api(self): |
62 def test_undo_api(self): |
52 self.assertTrue(self.txuuid) |
63 self.assertTrue(self.txuuid) |
53 # test transaction api |
64 # test transaction api |
54 self.assertRaises(NoSuchTransaction, |
65 self.assertRaises(NoSuchTransaction, |
153 txuuid = self.commit() |
164 txuuid = self.commit() |
154 actions = self.cnx.transaction_info(txuuid).actions_list() |
165 actions = self.cnx.transaction_info(txuuid).actions_list() |
155 self.assertEqual(len(actions), 1) |
166 self.assertEqual(len(actions), 1) |
156 toto.cw_clear_all_caches() |
167 toto.cw_clear_all_caches() |
157 e.cw_clear_all_caches() |
168 e.cw_clear_all_caches() |
158 errors = self.cnx.undo_transaction(txuuid) |
169 self.assertUndoTransaction(txuuid) |
159 undotxuuid = self.commit() |
170 undotxuuid = self.commit() |
160 self.assertEqual(undotxuuid, None) # undo not undoable |
171 self.assertEqual(undotxuuid, None) # undo not undoable |
161 self.assertEqual(errors, []) |
|
162 self.assertTrue(self.execute('Any X WHERE X eid %(x)s', {'x': toto.eid})) |
172 self.assertTrue(self.execute('Any X WHERE X eid %(x)s', {'x': toto.eid})) |
163 self.assertTrue(self.execute('Any X WHERE X eid %(x)s', {'x': e.eid})) |
173 self.assertTrue(self.execute('Any X WHERE X eid %(x)s', {'x': e.eid})) |
164 self.assertTrue(self.execute('Any X WHERE X has_text "toto@logilab"')) |
174 self.assertTrue(self.execute('Any X WHERE X has_text "toto@logilab"')) |
165 self.assertEqual(toto.cw_adapt_to('IWorkflowable').state, 'activated') |
175 self.assertEqual(toto.cw_adapt_to('IWorkflowable').state, 'activated') |
166 self.assertEqual(toto.cw_adapt_to('IEmailable').get_email(), 'toto@logilab.org') |
176 self.assertEqual(toto.cw_adapt_to('IEmailable').get_email(), 'toto@logilab.org') |
191 c.cw_delete() |
201 c.cw_delete() |
192 txuuid = self.commit() |
202 txuuid = self.commit() |
193 c2 = session.create_entity('Card', title=u'hip', content=u'hip') |
203 c2 = session.create_entity('Card', title=u'hip', content=u'hip') |
194 p.set_relations(fiche=c2) |
204 p.set_relations(fiche=c2) |
195 self.commit() |
205 self.commit() |
196 errors = self.cnx.undo_transaction(txuuid) |
206 self.assertUndoTransaction(txuuid, [ |
|
207 "Can't restore object relation fiche to entity " |
|
208 "%s which is already linked using this relation." % p.eid]) |
197 self.commit() |
209 self.commit() |
198 p.cw_clear_all_caches() |
210 p.cw_clear_all_caches() |
199 self.assertEqual(p.fiche[0].eid, c2.eid) |
211 self.assertEqual(p.fiche[0].eid, c2.eid) |
200 self.assertEqual(len(errors), 1) |
|
201 self.assertEqual(errors[0], |
|
202 "Can't restore object relation fiche to entity " |
|
203 "%s which is already linked using this relation." % p.eid) |
|
204 |
212 |
205 def test_undo_deletion_integrity_2(self): |
213 def test_undo_deletion_integrity_2(self): |
206 # test validation error raised if we can't restore a required relation |
214 # test validation error raised if we can't restore a required relation |
207 session = self.session |
215 session = self.session |
208 g = session.create_entity('CWGroup', name=u'staff') |
216 g = session.create_entity('CWGroup', name=u'staff') |
227 def test_undo_creation_1(self): |
234 def test_undo_creation_1(self): |
228 session = self.session |
235 session = self.session |
229 c = session.create_entity('Card', title=u'hop', content=u'hop') |
236 c = session.create_entity('Card', title=u'hop', content=u'hop') |
230 p = session.create_entity('Personne', nom=u'louis', fiche=c) |
237 p = session.create_entity('Personne', nom=u'louis', fiche=c) |
231 txuuid = self.commit() |
238 txuuid = self.commit() |
232 errors = self.cnx.undo_transaction(txuuid) |
239 self.assertUndoTransaction(txuuid) |
233 self.commit() |
240 self.commit() |
234 self.assertFalse(errors) |
|
235 self.assertFalse(self.execute('Any X WHERE X eid %(x)s', {'x': c.eid})) |
241 self.assertFalse(self.execute('Any X WHERE X eid %(x)s', {'x': c.eid})) |
236 self.assertFalse(self.execute('Any X WHERE X eid %(x)s', {'x': p.eid})) |
242 self.assertFalse(self.execute('Any X WHERE X eid %(x)s', {'x': p.eid})) |
237 self.assertFalse(self.execute('Any X,Y WHERE X fiche Y')) |
243 self.assertFalse(self.execute('Any X,Y WHERE X fiche Y')) |
238 self.session.set_cnxset() |
244 self.session.set_cnxset() |
239 for eid in (p.eid, c.eid): |
245 for eid in (p.eid, c.eid): |
286 # {'in_group-subject': u'at least one relation in_group is ' |
292 # {'in_group-subject': u'at least one relation in_group is ' |
287 # 'required on CWUser (%s)' % self.toto.eid}) |
293 # 'required on CWUser (%s)' % self.toto.eid}) |
288 |
294 |
289 # test implicit 'replacement' of an inlined relation |
295 # test implicit 'replacement' of an inlined relation |
290 |
296 |
|
297 def test_undo_inline_rel_remove_ok(self): |
|
298 """Undo remove relation Personne (?) fiche (?) Card |
|
299 |
|
300 NB: processed by `_undo_r` as expected""" |
|
301 session = self.session |
|
302 c = session.create_entity('Card', title=u'hop', content=u'hop') |
|
303 p = session.create_entity('Personne', nom=u'louis', fiche=c) |
|
304 self.commit() |
|
305 p.set_relations(fiche=None) |
|
306 txuuid = self.commit() |
|
307 self.assertUndoTransaction(txuuid) |
|
308 self.commit() |
|
309 p.cw_clear_all_caches() |
|
310 self.assertEqual(p.fiche[0].eid, c.eid) |
|
311 |
|
312 def test_undo_inline_rel_remove_ko(self): |
|
313 """Restore an inlined relation to a deleted entity, with an error. |
|
314 |
|
315 NB: processed by `_undo_r` as expected""" |
|
316 session = self.session |
|
317 c = session.create_entity('Card', title=u'hop', content=u'hop') |
|
318 p = session.create_entity('Personne', nom=u'louis', fiche=c) |
|
319 self.commit() |
|
320 p.set_relations(fiche=None) |
|
321 txuuid = self.commit() |
|
322 c.cw_delete() |
|
323 self.commit() |
|
324 self.assertUndoTransaction(txuuid, [ |
|
325 "Can't restore relation fiche, object entity %d doesn't exist anymore." % c.eid]) |
|
326 self.commit() |
|
327 p.cw_clear_all_caches() |
|
328 self.assertFalse(p.fiche) |
|
329 self.assertIsNone(session.system_sql( |
|
330 'SELECT cw_fiche FROM cw_Personne WHERE cw_eid=%s' % p.eid).fetchall()[0][0]) |
|
331 |
|
332 def test_undo_inline_rel_add_ok(self): |
|
333 """Undo add relation Personne (?) fiche (?) Card |
|
334 |
|
335 Caution processed by `_undo_u`, not `_undo_a` !""" |
|
336 session = self.session |
|
337 c = session.create_entity('Card', title=u'hop', content=u'hop') |
|
338 p = session.create_entity('Personne', nom=u'louis') |
|
339 self.commit() |
|
340 p.set_relations(fiche=c) |
|
341 txuuid = self.commit() |
|
342 self.assertUndoTransaction(txuuid) |
|
343 self.commit() |
|
344 p.cw_clear_all_caches() |
|
345 self.assertFalse(p.fiche) |
|
346 |
|
347 def test_undo_inline_rel_add_ko(self): |
|
348 """Undo add relation Personne (?) fiche (?) Card |
|
349 |
|
350 Caution processed by `_undo_u`, not `_undo_a` !""" |
|
351 session = self.session |
|
352 c = session.create_entity('Card', title=u'hop', content=u'hop') |
|
353 p = session.create_entity('Personne', nom=u'louis') |
|
354 self.commit() |
|
355 p.set_relations(fiche=c) |
|
356 txuuid = self.commit() |
|
357 c.cw_delete() |
|
358 self.commit() |
|
359 self.assertUndoTransaction(txuuid) |
|
360 |
|
361 def test_undo_inline_rel_replace_ok(self): |
|
362 """Undo changing relation Personne (?) fiche (?) Card |
|
363 |
|
364 Caution processed by `_undo_u` """ |
|
365 session = self.session |
|
366 c1 = session.create_entity('Card', title=u'hop', content=u'hop') |
|
367 c2 = session.create_entity('Card', title=u'hip', content=u'hip') |
|
368 p = session.create_entity('Personne', nom=u'louis', fiche=c1) |
|
369 self.commit() |
|
370 p.set_relations(fiche=c2) |
|
371 txuuid = self.commit() |
|
372 self.assertUndoTransaction(txuuid) |
|
373 self.commit() |
|
374 p.cw_clear_all_caches() |
|
375 self.assertEqual(p.fiche[0].eid, c1.eid) |
|
376 |
|
377 def test_undo_inline_rel_replace_ko(self): |
|
378 """Undo changing relation Personne (?) fiche (?) Card, with an error |
|
379 |
|
380 Caution processed by `_undo_u` """ |
|
381 session = self.session |
|
382 c1 = session.create_entity('Card', title=u'hop', content=u'hop') |
|
383 c2 = session.create_entity('Card', title=u'hip', content=u'hip') |
|
384 p = session.create_entity('Personne', nom=u'louis', fiche=c1) |
|
385 self.commit() |
|
386 p.set_relations(fiche=c2) |
|
387 txuuid = self.commit() |
|
388 c1.cw_delete() |
|
389 self.commit() |
|
390 self.assertUndoTransaction(txuuid, [ |
|
391 "can't restore entity %s of type Personne, target of fiche (eid %s)" |
|
392 " does not exist any longer" % (p.eid, c1.eid)]) |
|
393 self.commit() |
|
394 p.cw_clear_all_caches() |
|
395 self.assertFalse(p.fiche) |
|
396 |
|
397 def test_undo_attr_update_ok(self): |
|
398 session = self.session |
|
399 p = session.create_entity('Personne', nom=u'toto') |
|
400 session.commit() |
|
401 self.session.set_cnxset() |
|
402 p.set_attributes(nom=u'titi') |
|
403 txuuid = self.commit() |
|
404 self.assertUndoTransaction(txuuid) |
|
405 p.cw_clear_all_caches() |
|
406 self.assertEqual(p.nom, u'toto') |
|
407 |
|
408 def test_undo_attr_update_ko(self): |
|
409 session = self.session |
|
410 p = session.create_entity('Personne', nom=u'toto') |
|
411 session.commit() |
|
412 self.session.set_cnxset() |
|
413 p.set_attributes(nom=u'titi') |
|
414 txuuid = self.commit() |
|
415 p.cw_delete() |
|
416 self.commit() |
|
417 self.assertUndoTransaction(txuuid, [ |
|
418 u"can't restore state of entity %s, it has been deleted inbetween" % p.eid]) |
|
419 |
291 |
420 |
292 class UndoExceptionInUnicode(CubicWebTC): |
421 class UndoExceptionInUnicode(CubicWebTC): |
293 |
422 |
294 # problem occurs in string manipulation for python < 2.6 |
423 # problem occurs in string manipulation for python < 2.6 |
295 def test___unicode__method(self): |
424 def test___unicode__method(self): |
296 u = UndoException(u"voilà") |
425 u = _UndoException(u"voilà") |
297 self.assertIsInstance(unicode(u), unicode) |
426 self.assertIsInstance(unicode(u), unicode) |
298 |
427 |
299 |
428 |
300 if __name__ == '__main__': |
429 if __name__ == '__main__': |
301 from logilab.common.testlib import unittest_main |
430 from logilab.common.testlib import unittest_main |