21 |
21 |
22 def tearDown(self): |
22 def tearDown(self): |
23 self.restore_connection() |
23 self.restore_connection() |
24 self.session.undo_support = set() |
24 self.session.undo_support = set() |
25 super(UndoableTransactionTC, self).tearDown() |
25 super(UndoableTransactionTC, self).tearDown() |
|
26 |
|
27 def check_transaction_deleted(self, txuuid): |
|
28 # also check transaction actions have been properly deleted |
|
29 cu = self.session.system_sql( |
|
30 "SELECT * from tx_entity_actions WHERE tx_uuid='%s'" % txuuid) |
|
31 self.failIf(cu.fetchall()) |
|
32 cu = self.session.system_sql( |
|
33 "SELECT * from tx_relation_actions WHERE tx_uuid='%s'" % txuuid) |
|
34 self.failIf(cu.fetchall()) |
26 |
35 |
27 def test_undo_api(self): |
36 def test_undo_api(self): |
28 self.failUnless(self.txuuid) |
37 self.failUnless(self.txuuid) |
29 # test transaction api |
38 # test transaction api |
30 self.assertRaises(NoSuchTransaction, |
39 self.assertRaises(NoSuchTransaction, |
152 # transaction should be removed |
161 # transaction should be removed |
153 txs = self.cnx.undoable_transactions() |
162 txs = self.cnx.undoable_transactions() |
154 self.assertEquals(len(txs), 2) |
163 self.assertEquals(len(txs), 2) |
155 self.assertRaises(NoSuchTransaction, |
164 self.assertRaises(NoSuchTransaction, |
156 self.cnx.transaction_info, txuuid) |
165 self.cnx.transaction_info, txuuid) |
157 # also check transaction actions have been properly deleted |
166 self.check_transaction_deleted(txuuid) |
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 |
167 # the final test: check we can login with the previously deleted user |
165 self.login('toto') |
168 self.login('toto') |
166 |
169 |
167 def test_undo_deletion_integrity_1(self): |
170 def test_undo_deletion_integrity_1(self): |
168 session = self.session |
171 session = self.session |
194 self.toto.delete() |
197 self.toto.delete() |
195 txuuid = self.commit() |
198 txuuid = self.commit() |
196 g.delete() |
199 g.delete() |
197 self.commit() |
200 self.commit() |
198 errors = self.cnx.undo_transaction(txuuid) |
201 errors = self.cnx.undo_transaction(txuuid) |
199 self.assertRaises(ValidationError, self.commit) |
202 self.assertEquals(errors, |
200 |
203 [u"Can't restore relation in_group, object entity " |
201 def test_undo_creation(self): |
204 "%s doesn't exist anymore." % g.eid]) |
202 # XXX what about relation / composite entities which have been created |
205 ex = self.assertRaises(ValidationError, self.commit) |
203 # afterwhile and linked to the undoed addition ? |
206 self.assertEquals(ex.entity, self.toto.eid) |
204 self.skip('not implemented') |
207 self.assertEquals(ex.errors, |
|
208 {'in_group-subject': u'at least one relation in_group is ' |
|
209 'required on CWUser (%s)' % self.toto.eid}) |
|
210 |
|
211 def test_undo_creation_1(self): |
|
212 session = self.session |
|
213 c = session.create_entity('Card', title=u'hop', content=u'hop') |
|
214 p = session.create_entity('Personne', nom=u'louis', fiche=c) |
|
215 txuuid = self.commit() |
|
216 errors = self.cnx.undo_transaction(txuuid) |
|
217 self.commit() |
|
218 self.failIf(errors) |
|
219 self.failIf(self.execute('Any X WHERE X eid %(x)s', {'x': c.eid}, 'x')) |
|
220 self.failIf(self.execute('Any X WHERE X eid %(x)s', {'x': p.eid}, 'x')) |
|
221 self.failIf(self.execute('Any X,Y WHERE X fiche Y')) |
|
222 self.session.set_pool() |
|
223 for eid in (p.eid, c.eid): |
|
224 self.failIf(session.system_sql( |
|
225 'SELECT * FROM entities WHERE eid=%s' % eid).fetchall()) |
|
226 self.failIf(session.system_sql( |
|
227 'SELECT 1 FROM owned_by_relation WHERE eid_from=%s' % eid).fetchall()) |
|
228 # added by sql in hooks (except when using dataimport) |
|
229 self.failIf(session.system_sql( |
|
230 'SELECT 1 FROM is_relation WHERE eid_from=%s' % eid).fetchall()) |
|
231 self.failIf(session.system_sql( |
|
232 'SELECT 1 FROM is_instance_of_relation WHERE eid_from=%s' % eid).fetchall()) |
|
233 self.check_transaction_deleted(txuuid) |
|
234 |
|
235 |
|
236 def test_undo_creation_integrity_1(self): |
|
237 session = self.session |
|
238 tutu = self.create_user('tutu', commit=False) |
|
239 txuuid = self.commit() |
|
240 email = self.request().create_entity('EmailAddress', address=u'tutu@cubicweb.org') |
|
241 prop = self.request().create_entity('CWProperty', pkey=u'ui.default-text-format', |
|
242 value=u'text/html') |
|
243 tutu.set_relations(use_email=email, reverse_for_user=prop) |
|
244 self.commit() |
|
245 ex = self.assertRaises(ValidationError, |
|
246 self.cnx.undo_transaction, txuuid) |
|
247 self.assertEquals(ex.entity, tutu.eid) |
|
248 self.assertEquals(ex.errors, |
|
249 {None: 'some later transaction(s) touch entity, undo them first'}) |
|
250 |
|
251 def test_undo_creation_integrity_2(self): |
|
252 session = self.session |
|
253 g = session.create_entity('CWGroup', name=u'staff') |
|
254 txuuid = self.commit() |
|
255 session.execute('DELETE U in_group G WHERE U eid %(x)s', {'x': self.toto.eid}) |
|
256 self.toto.set_relations(in_group=g) |
|
257 self.commit() |
|
258 ex = self.assertRaises(ValidationError, |
|
259 self.cnx.undo_transaction, txuuid) |
|
260 self.assertEquals(ex.entity, g.eid) |
|
261 self.assertEquals(ex.errors, |
|
262 {None: 'some later transaction(s) touch entity, undo them first'}) |
|
263 # self.assertEquals(errors, |
|
264 # [u"Can't restore relation in_group, object entity " |
|
265 # "%s doesn't exist anymore." % g.eid]) |
|
266 # ex = self.assertRaises(ValidationError, self.commit) |
|
267 # self.assertEquals(ex.entity, self.toto.eid) |
|
268 # self.assertEquals(ex.errors, |
|
269 # {'in_group-subject': u'at least one relation in_group is ' |
|
270 # 'required on CWUser (%s)' % self.toto.eid}) |
205 |
271 |
206 # test implicit 'replacement' of an inlined relation |
272 # test implicit 'replacement' of an inlined relation |