107 metainf = externent.metainformation() |
107 metainf = externent.metainformation() |
108 self.assertEquals(metainf['source'], {'adapter': 'pyrorql', 'base-url': 'http://extern.org/', 'uri': 'extern'}) |
108 self.assertEquals(metainf['source'], {'adapter': 'pyrorql', 'base-url': 'http://extern.org/', 'uri': 'extern'}) |
109 self.assertEquals(metainf['type'], 'Card') |
109 self.assertEquals(metainf['type'], 'Card') |
110 self.assert_(metainf['extid']) |
110 self.assert_(metainf['extid']) |
111 etype = self.sexecute('Any ETN WHERE X is ET, ET name ETN, X eid %(x)s', |
111 etype = self.sexecute('Any ETN WHERE X is ET, ET name ETN, X eid %(x)s', |
112 {'x': externent.eid}, 'x')[0][0] |
112 {'x': externent.eid})[0][0] |
113 self.assertEquals(etype, 'Card') |
113 self.assertEquals(etype, 'Card') |
114 |
114 |
115 def test_order_limit_offset(self): |
115 def test_order_limit_offset(self): |
116 rsetbase = self.sexecute('Any W,X ORDERBY W,X WHERE X wikiid W') |
116 rsetbase = self.sexecute('Any W,X ORDERBY W,X WHERE X wikiid W') |
117 self.assertEquals(len(rsetbase), 4) |
117 self.assertEquals(len(rsetbase), 4) |
127 def test_anon_has_text(self): |
127 def test_anon_has_text(self): |
128 self.repo.sources_by_uri['extern'].synchronize(MTIME) # in case fti_update has been run before |
128 self.repo.sources_by_uri['extern'].synchronize(MTIME) # in case fti_update has been run before |
129 self.sexecute('INSERT Affaire X: X ref "no readable card"')[0][0] |
129 self.sexecute('INSERT Affaire X: X ref "no readable card"')[0][0] |
130 aff1 = self.sexecute('INSERT Affaire X: X ref "card"')[0][0] |
130 aff1 = self.sexecute('INSERT Affaire X: X ref "card"')[0][0] |
131 # grant read access |
131 # grant read access |
132 self.sexecute('SET X owned_by U WHERE X eid %(x)s, U login "anon"', {'x': aff1}, 'x') |
132 self.sexecute('SET X owned_by U WHERE X eid %(x)s, U login "anon"', {'x': aff1}) |
133 self.commit() |
133 self.commit() |
134 cnx = self.login('anon') |
134 cnx = self.login('anon') |
135 cu = cnx.cursor() |
135 cu = cnx.cursor() |
136 rset = cu.execute('Any X WHERE X has_text "card"') |
136 rset = cu.execute('Any X WHERE X has_text "card"') |
137 self.assertEquals(len(rset), 5, zip(rset.rows, rset.description)) |
137 self.assertEquals(len(rset), 5, zip(rset.rows, rset.description)) |
138 Connection_close(cnx) |
138 Connection_close(cnx) |
139 |
139 |
140 def test_synchronization(self): |
140 def test_synchronization(self): |
141 cu = cnx2.cursor() |
141 cu = cnx2.cursor() |
142 assert cu.execute('Any X WHERE X eid %(x)s', {'x': self.aff1}, 'x') |
142 assert cu.execute('Any X WHERE X eid %(x)s', {'x': self.aff1}) |
143 cu.execute('SET X ref "BLAH" WHERE X eid %(x)s', {'x': self.aff1}, 'x') |
143 cu.execute('SET X ref "BLAH" WHERE X eid %(x)s', {'x': self.aff1}) |
144 aff2 = cu.execute('INSERT Affaire X: X ref "AFFREUX"')[0][0] |
144 aff2 = cu.execute('INSERT Affaire X: X ref "AFFREUX"')[0][0] |
145 cnx2.commit() |
145 cnx2.commit() |
146 try: |
146 try: |
147 # force sync |
147 # force sync |
148 self.repo.sources_by_uri['extern'].synchronize(MTIME) |
148 self.repo.sources_by_uri['extern'].synchronize(MTIME) |
153 self.repo.sources_by_uri['extern'].synchronize(MTIME) |
153 self.repo.sources_by_uri['extern'].synchronize(MTIME) |
154 rset = self.sexecute('Any X WHERE X has_text "affreux"') |
154 rset = self.sexecute('Any X WHERE X has_text "affreux"') |
155 self.failIf(rset) |
155 self.failIf(rset) |
156 finally: |
156 finally: |
157 # restore state |
157 # restore state |
158 cu.execute('SET X ref "AFFREF" WHERE X eid %(x)s', {'x': self.aff1}, 'x') |
158 cu.execute('SET X ref "AFFREF" WHERE X eid %(x)s', {'x': self.aff1}) |
159 cnx2.commit() |
159 cnx2.commit() |
160 |
160 |
161 def test_simplifiable_var(self): |
161 def test_simplifiable_var(self): |
162 affeid = self.sexecute('Affaire X WHERE X ref "AFFREF"')[0][0] |
162 affeid = self.sexecute('Affaire X WHERE X ref "AFFREF"')[0][0] |
163 rset = self.sexecute('Any X,AA,AB WHERE E eid %(x)s, E in_state X, X name AA, X modification_date AB', |
163 rset = self.sexecute('Any X,AA,AB WHERE E eid %(x)s, E in_state X, X name AA, X modification_date AB', |
164 {'x': affeid}, 'x') |
164 {'x': affeid}) |
165 self.assertEquals(len(rset), 1) |
165 self.assertEquals(len(rset), 1) |
166 self.assertEquals(rset[0][1], "pitetre") |
166 self.assertEquals(rset[0][1], "pitetre") |
167 |
167 |
168 def test_simplifiable_var_2(self): |
168 def test_simplifiable_var_2(self): |
169 affeid = self.sexecute('Affaire X WHERE X ref "AFFREF"')[0][0] |
169 affeid = self.sexecute('Affaire X WHERE X ref "AFFREF"')[0][0] |
170 rset = self.sexecute('Any E WHERE E eid %(x)s, E in_state S, NOT S name "moved"', |
170 rset = self.sexecute('Any E WHERE E eid %(x)s, E in_state S, NOT S name "moved"', |
171 {'x': affeid, 'u': self.session.user.eid}, 'x') |
171 {'x': affeid, 'u': self.session.user.eid}) |
172 self.assertEquals(len(rset), 1) |
172 self.assertEquals(len(rset), 1) |
173 |
173 |
174 def test_sort_func(self): |
174 def test_sort_func(self): |
175 self.sexecute('Affaire X ORDERBY DUMB_SORT(RF) WHERE X ref RF') |
175 self.sexecute('Affaire X ORDERBY DUMB_SORT(RF) WHERE X ref RF') |
176 |
176 |
214 try: |
214 try: |
215 c1 = self.sexecute('INSERT Card C: C title "AFFREF"')[0][0] |
215 c1 = self.sexecute('INSERT Card C: C title "AFFREF"')[0][0] |
216 rset = self.sexecute('Any X,Y WHERE X is Card, Y is Affaire, X title T, Y ref T') |
216 rset = self.sexecute('Any X,Y WHERE X is Card, Y is Affaire, X title T, Y ref T') |
217 self.assertEquals(len(rset), 2, rset.rows) |
217 self.assertEquals(len(rset), 2, rset.rows) |
218 finally: |
218 finally: |
219 cu.execute('DELETE Card X WHERE X eid %(x)s', {'x': ec2}, 'x') |
219 cu.execute('DELETE Card X WHERE X eid %(x)s', {'x': ec2}) |
220 cnx2.commit() |
220 cnx2.commit() |
221 |
221 |
222 def test_attr_unification_neq_1(self): |
222 def test_attr_unification_neq_1(self): |
223 # XXX complete |
223 # XXX complete |
224 self.sexecute('Any X,Y WHERE X is Note, Y is Affaire, X creation_date D, Y creation_date > D') |
224 self.sexecute('Any X,Y WHERE X is Note, Y is Affaire, X creation_date D, Y creation_date > D') |
256 states = set(tuple(x) for x in self.sexecute('Any S,SN WHERE S is State, S name SN')) |
256 states = set(tuple(x) for x in self.sexecute('Any S,SN WHERE S is State, S name SN')) |
257 self.session.user.clear_all_caches() |
257 self.session.user.clear_all_caches() |
258 userstate = self.session.user.in_state[0] |
258 userstate = self.session.user.in_state[0] |
259 states.remove((userstate.eid, userstate.name)) |
259 states.remove((userstate.eid, userstate.name)) |
260 notstates = set(tuple(x) for x in self.sexecute('Any S,SN WHERE S is State, S name SN, NOT X in_state S, X eid %(x)s', |
260 notstates = set(tuple(x) for x in self.sexecute('Any S,SN WHERE S is State, S name SN, NOT X in_state S, X eid %(x)s', |
261 {'x': self.session.user.eid}, 'x')) |
261 {'x': self.session.user.eid})) |
262 self.assertSetEquals(notstates, states) |
262 self.assertSetEquals(notstates, states) |
263 aff1 = self.sexecute('Any X WHERE X is Affaire, X ref "AFFREF"')[0][0] |
263 aff1 = self.sexecute('Any X WHERE X is Affaire, X ref "AFFREF"')[0][0] |
264 aff1stateeid, aff1statename = self.sexecute('Any S,SN WHERE X eid %(x)s, X in_state S, S name SN', {'x': aff1}, 'x')[0] |
264 aff1stateeid, aff1statename = self.sexecute('Any S,SN WHERE X eid %(x)s, X in_state S, S name SN', {'x': aff1})[0] |
265 self.assertEquals(aff1statename, 'pitetre') |
265 self.assertEquals(aff1statename, 'pitetre') |
266 states.add((userstate.eid, userstate.name)) |
266 states.add((userstate.eid, userstate.name)) |
267 states.remove((aff1stateeid, aff1statename)) |
267 states.remove((aff1stateeid, aff1statename)) |
268 notstates = set(tuple(x) for x in self.sexecute('Any S,SN WHERE S is State, S name SN, NOT X in_state S, X eid %(x)s', |
268 notstates = set(tuple(x) for x in self.sexecute('Any S,SN WHERE S is State, S name SN, NOT X in_state S, X eid %(x)s', |
269 {'x': aff1}, 'x')) |
269 {'x': aff1})) |
270 self.assertSetEquals(notstates, states) |
270 self.assertSetEquals(notstates, states) |
271 |
271 |
272 def test_absolute_url_base_url(self): |
272 def test_absolute_url_base_url(self): |
273 cu = cnx2.cursor() |
273 cu = cnx2.cursor() |
274 ceid = cu.execute('INSERT Card X: X title "without wikiid to get eid based url"')[0][0] |
274 ceid = cu.execute('INSERT Card X: X title "without wikiid to get eid based url"')[0][0] |