28 |
28 |
29 repo2, cnx2 = init_test_database('sqlite', config=ExternalSource1Configuration('data')) |
29 repo2, cnx2 = init_test_database('sqlite', config=ExternalSource1Configuration('data')) |
30 cu = cnx2.cursor() |
30 cu = cnx2.cursor() |
31 ec1 = cu.execute('INSERT Card X: X title "C3: An external card", X wikiid "aaa"')[0][0] |
31 ec1 = cu.execute('INSERT Card X: X title "C3: An external card", X wikiid "aaa"')[0][0] |
32 cu.execute('INSERT Card X: X title "C4: Ze external card", X wikiid "zzz"') |
32 cu.execute('INSERT Card X: X title "C4: Ze external card", X wikiid "zzz"') |
33 aff1 = cu.execute('INSERT Affaire X: X ref "AFFREF", X in_state S WHERE S name "pitetre"')[0][0] |
33 aff1 = cu.execute('INSERT Affaire X: X ref "AFFREF"')[0][0] |
34 cnx2.commit() |
34 cnx2.commit() |
35 |
35 |
36 MTIME = datetime.now() - timedelta(0, 10) |
36 MTIME = datetime.now() - timedelta(0, 10) |
37 |
37 |
38 repo3, cnx3 = init_test_database('sqlite', config=ExternalSource2Configuration('data')) |
38 repo3, cnx3 = init_test_database('sqlite', config=ExternalSource2Configuration('data')) |
120 |
120 |
121 def test_synchronization(self): |
121 def test_synchronization(self): |
122 cu = cnx2.cursor() |
122 cu = cnx2.cursor() |
123 assert cu.execute('Any X WHERE X eid %(x)s', {'x': aff1}, 'x') |
123 assert cu.execute('Any X WHERE X eid %(x)s', {'x': aff1}, 'x') |
124 cu.execute('SET X ref "BLAH" WHERE X eid %(x)s', {'x': aff1}, 'x') |
124 cu.execute('SET X ref "BLAH" WHERE X eid %(x)s', {'x': aff1}, 'x') |
125 aff2 = cu.execute('INSERT Affaire X: X ref "AFFREUX", X in_state S WHERE S name "pitetre"')[0][0] |
125 aff2 = cu.execute('INSERT Affaire X: X ref "AFFREUX"')[0][0] |
126 cnx2.commit() |
126 cnx2.commit() |
127 try: |
127 try: |
128 # force sync |
128 # force sync |
129 self.repo.sources_by_uri['extern'].synchronize(MTIME) |
129 self.repo.sources_by_uri['extern'].synchronize(MTIME) |
130 self.failUnless(self.execute('Any X WHERE X has_text "blah"')) |
130 self.failUnless(self.execute('Any X WHERE X has_text "blah"')) |
231 self.assertEquals(len(rset), 1) |
231 self.assertEquals(len(rset), 1) |
232 self.assertEquals(rset[0][1], "pitetre") |
232 self.assertEquals(rset[0][1], "pitetre") |
233 |
233 |
234 def test_not_relation(self): |
234 def test_not_relation(self): |
235 states = set(tuple(x) for x in self.execute('Any S,SN WHERE S is State, S name SN')) |
235 states = set(tuple(x) for x in self.execute('Any S,SN WHERE S is State, S name SN')) |
|
236 self.session.user.clear_all_caches() |
236 userstate = self.session.user.in_state[0] |
237 userstate = self.session.user.in_state[0] |
237 states.remove((userstate.eid, userstate.name)) |
238 states.remove((userstate.eid, userstate.name)) |
238 notstates = set(tuple(x) for x in self.execute('Any S,SN WHERE S is State, S name SN, NOT X in_state S, X eid %(x)s', |
239 notstates = set(tuple(x) for x in self.execute('Any S,SN WHERE S is State, S name SN, NOT X in_state S, X eid %(x)s', |
239 {'x': self.session.user.eid}, 'x')) |
240 {'x': self.session.user.eid}, 'x')) |
240 self.assertEquals(notstates, states) |
241 self.assertSetEquals(notstates, states) |
241 aff1 = self.execute('Any X WHERE X is Affaire, X ref "AFFREF"')[0][0] |
242 aff1 = self.execute('Any X WHERE X is Affaire, X ref "AFFREF"')[0][0] |
242 aff1stateeid, aff1statename = self.execute('Any S,SN WHERE X eid %(x)s, X in_state S, S name SN', {'x': aff1}, 'x')[0] |
243 aff1stateeid, aff1statename = self.execute('Any S,SN WHERE X eid %(x)s, X in_state S, S name SN', {'x': aff1}, 'x')[0] |
243 self.assertEquals(aff1statename, 'pitetre') |
244 self.assertEquals(aff1statename, 'pitetre') |
244 states.add((userstate.eid, userstate.name)) |
245 states.add((userstate.eid, userstate.name)) |
245 states.remove((aff1stateeid, aff1statename)) |
246 states.remove((aff1stateeid, aff1statename)) |
265 affaire = self.execute('Affaire X WHERE X ref "AFFREF"').get_entity(0, 0) |
266 affaire = self.execute('Affaire X WHERE X ref "AFFREF"').get_entity(0, 0) |
266 self.execute('Any U WHERE U in_group G, (G name IN ("managers", "logilab") OR (X require_permission P?, P name "bla", P require_group G)), X eid %(x)s, U eid %(u)s', |
267 self.execute('Any U WHERE U in_group G, (G name IN ("managers", "logilab") OR (X require_permission P?, P name "bla", P require_group G)), X eid %(x)s, U eid %(u)s', |
267 {'x': affaire.eid, 'u': ueid}) |
268 {'x': affaire.eid, 'u': ueid}) |
268 |
269 |
269 def test_nonregr2(self): |
270 def test_nonregr2(self): |
|
271 self.session.user.fire_transition('deactivate') |
270 treid = self.session.user.latest_trinfo().eid |
272 treid = self.session.user.latest_trinfo().eid |
271 rset = self.execute('Any X ORDERBY D DESC WHERE E eid %(x)s, E wf_info_for X, X modification_date D', |
273 rset = self.execute('Any X ORDERBY D DESC WHERE E eid %(x)s, E wf_info_for X, X modification_date D', |
272 {'x': treid}) |
274 {'x': treid}) |
273 self.assertEquals(len(rset), 1) |
275 self.assertEquals(len(rset), 1) |
274 self.assertEquals(rset.rows[0], [self.session.user.eid]) |
276 self.assertEquals(rset.rows[0], [self.session.user.eid]) |