branch | tls-sprint |
changeset 1787 | 71c143c0ada3 |
parent 1398 | 5fe84a5f7035 |
child 1949 | 9064102b4deb |
1786:eccd1885d42e | 1787:71c143c0ada3 |
---|---|
38 TYPEMAP = _GenericAdvFuncHelper.TYPE_MAPPING |
38 TYPEMAP = _GenericAdvFuncHelper.TYPE_MAPPING |
39 |
39 |
40 class MakeSchemaTC(TestCase): |
40 class MakeSchemaTC(TestCase): |
41 def test_known_values(self): |
41 def test_known_values(self): |
42 solution = {'A': 'String', 'B': 'CWUser'} |
42 solution = {'A': 'String', 'B': 'CWUser'} |
43 self.assertEquals(make_schema((Variable('A'), Variable('B')), solution, |
43 self.assertEquals(make_schema((Variable('A'), Variable('B')), solution, |
44 'table0', TYPEMAP), |
44 'table0', TYPEMAP), |
45 ('C0 text,C1 integer', {'A': 'table0.C0', 'B': 'table0.C1'})) |
45 ('C0 text,C1 integer', {'A': 'table0.C0', 'B': 'table0.C1'})) |
46 |
46 |
47 |
47 |
48 repo, cnx = init_test_database('sqlite') |
48 repo, cnx = init_test_database('sqlite') |
49 |
49 |
50 |
50 |
51 |
51 |
52 class UtilsTC(BaseQuerierTC): |
52 class UtilsTC(BaseQuerierTC): |
53 repo = repo |
53 repo = repo |
54 |
54 |
55 def get_max_eid(self): |
55 def get_max_eid(self): |
56 # no need for cleanup here |
56 # no need for cleanup here |
57 return None |
57 return None |
58 def cleanup(self): |
58 def cleanup(self): |
59 # no need for cleanup here |
59 # no need for cleanup here |
60 pass |
60 pass |
61 |
61 |
62 def test_preprocess_1(self): |
62 def test_preprocess_1(self): |
63 reid = self.execute('Any X WHERE X is CWRType, X name "owned_by"')[0][0] |
63 reid = self.execute('Any X WHERE X is CWRType, X name "owned_by"')[0][0] |
64 rqlst = self._prepare('Any COUNT(RDEF) WHERE RDEF relation_type X, X eid %(x)s', {'x': reid}) |
64 rqlst = self._prepare('Any COUNT(RDEF) WHERE RDEF relation_type X, X eid %(x)s', {'x': reid}) |
65 self.assertEquals(rqlst.solutions, [{'RDEF': 'CWAttribute'}, {'RDEF': 'CWRelation'}]) |
65 self.assertEquals(rqlst.solutions, [{'RDEF': 'CWAttribute'}, {'RDEF': 'CWRelation'}]) |
66 |
66 |
67 def test_preprocess_2(self): |
67 def test_preprocess_2(self): |
68 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
68 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
69 #geid = self.execute("CWGroup G WHERE G name 'users'")[0][0] |
69 #geid = self.execute("CWGroup G WHERE G name 'users'")[0][0] |
70 #self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s", |
70 #self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s", |
71 # {'g': geid, 't': teid}, 'g') |
71 # {'g': geid, 't': teid}, 'g') |
72 rqlst = self._prepare('Any X WHERE E eid %(x)s, E tags X', {'x': teid}) |
72 rqlst = self._prepare('Any X WHERE E eid %(x)s, E tags X', {'x': teid}) |
73 # the query may be optimized, should keep only one solution |
73 # the query may be optimized, should keep only one solution |
74 # (any one, etype will be discarded) |
74 # (any one, etype will be discarded) |
75 self.assertEquals(len(rqlst.solutions), 1) |
75 self.assertEquals(len(rqlst.solutions), 1) |
76 |
76 |
77 def test_preprocess_security(self): |
77 def test_preprocess_security(self): |
78 plan = self._prepare_plan('Any ETN,COUNT(X) GROUPBY ETN ' |
78 plan = self._prepare_plan('Any ETN,COUNT(X) GROUPBY ETN ' |
79 'WHERE X is ET, ET name ETN') |
79 'WHERE X is ET, ET name ETN') |
80 plan.session = self._user_session(('users',))[1] |
80 plan.session = self._user_session(('users',))[1] |
81 union = plan.rqlst |
81 union = plan.rqlst |
107 'J': 'Affaire', |
107 'J': 'Affaire', |
108 'X': 'Affaire', |
108 'X': 'Affaire', |
109 'ET': 'CWEType', 'ETN': 'String'}]) |
109 'ET': 'CWEType', 'ETN': 'String'}]) |
110 rql, solutions = partrqls[1] |
110 rql, solutions = partrqls[1] |
111 self.assertEquals(rql, 'Any ETN,X WHERE X is ET, ET name ETN, ET is CWEType, ' |
111 self.assertEquals(rql, 'Any ETN,X WHERE X is ET, ET name ETN, ET is CWEType, ' |
112 'X is IN(Bookmark, Card, Comment, Division, CWCache, CWConstraint, CWConstraintType, CWEType, CWAttribute, CWGroup, CWRelation, CWPermission, CWProperty, CWRType, CWUser, Email, EmailAddress, EmailPart, EmailThread, File, Folder, Image, Note, Personne, RQLExpression, Societe, State, SubDivision, Tag, TrInfo, Transition)') |
112 'X is IN(Bookmark, CWAttribute, CWCache, CWConstraint, CWConstraintType, CWEType, CWGroup, CWPermission, CWProperty, CWRType, CWRelation, CWUser, Card, Comment, Division, Email, EmailAddress, EmailPart, EmailThread, File, Folder, Image, Note, Personne, RQLExpression, Societe, State, SubDivision, Tag, TrInfo, Transition)') |
113 self.assertListEquals(sorted(solutions), |
113 self.assertListEquals(sorted(solutions), |
114 sorted([{'X': 'Bookmark', 'ETN': 'String', 'ET': 'CWEType'}, |
114 sorted([{'X': 'Bookmark', 'ETN': 'String', 'ET': 'CWEType'}, |
115 {'X': 'Card', 'ETN': 'String', 'ET': 'CWEType'}, |
115 {'X': 'Card', 'ETN': 'String', 'ET': 'CWEType'}, |
116 {'X': 'Comment', 'ETN': 'String', 'ET': 'CWEType'}, |
116 {'X': 'Comment', 'ETN': 'String', 'ET': 'CWEType'}, |
117 {'X': 'Division', 'ETN': 'String', 'ET': 'CWEType'}, |
117 {'X': 'Division', 'ETN': 'String', 'ET': 'CWEType'}, |
160 self.assertEquals(len(union.children[0].with_), 1) |
160 self.assertEquals(len(union.children[0].with_), 1) |
161 subq = union.children[0].with_[0].query |
161 subq = union.children[0].with_[0].query |
162 self.assertEquals(len(subq.children), 3) |
162 self.assertEquals(len(subq.children), 3) |
163 self.assertEquals([t.as_string() for t in union.children[0].selection], |
163 self.assertEquals([t.as_string() for t in union.children[0].selection], |
164 ['MAX(X)']) |
164 ['MAX(X)']) |
165 |
165 |
166 def test_preprocess_nonregr(self): |
166 def test_preprocess_nonregr(self): |
167 rqlst = self._prepare('Any S ORDERBY SI WHERE NOT S ecrit_par O, S para SI') |
167 rqlst = self._prepare('Any S ORDERBY SI WHERE NOT S ecrit_par O, S para SI') |
168 self.assertEquals(len(rqlst.solutions), 1) |
168 self.assertEquals(len(rqlst.solutions), 1) |
169 |
169 |
170 def test_build_description(self): |
170 def test_build_description(self): |
171 # should return an empty result set |
171 # should return an empty result set |
172 rset = self.execute('Any X WHERE X eid %(x)s', {'x': self.session.user.eid}) |
172 rset = self.execute('Any X WHERE X eid %(x)s', {'x': self.session.user.eid}) |
173 self.assertEquals(rset.description[0][0], 'CWUser') |
173 self.assertEquals(rset.description[0][0], 'CWUser') |
174 rset = self.execute('Any 1') |
174 rset = self.execute('Any 1') |
205 'Any X WHERE X is CWRType, X name "öwned_by"') |
205 'Any X WHERE X is CWRType, X name "öwned_by"') |
206 |
206 |
207 def test_unknown_eid(self): |
207 def test_unknown_eid(self): |
208 # should return an empty result set |
208 # should return an empty result set |
209 self.failIf(self.execute('Any X WHERE X eid 99999999')) |
209 self.failIf(self.execute('Any X WHERE X eid 99999999')) |
210 |
210 |
211 # selection queries tests ################################################# |
211 # selection queries tests ################################################# |
212 |
212 |
213 def test_select_1(self): |
213 def test_select_1(self): |
214 rset = self.execute('Any X ORDERBY X WHERE X is CWGroup') |
214 rset = self.execute('Any X ORDERBY X WHERE X is CWGroup') |
215 result, descr = rset.rows, rset.description |
215 result, descr = rset.rows, rset.description |
216 self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,)]) |
216 self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,)]) |
217 self.assertEquals(descr, [('CWGroup',), ('CWGroup',), ('CWGroup',), ('CWGroup',)]) |
217 self.assertEquals(descr, [('CWGroup',), ('CWGroup',), ('CWGroup',), ('CWGroup',)]) |
218 |
218 |
219 def test_select_2(self): |
219 def test_select_2(self): |
220 rset = self.execute('Any X ORDERBY N WHERE X is CWGroup, X name N') |
220 rset = self.execute('Any X ORDERBY N WHERE X is CWGroup, X name N') |
221 self.assertEquals(tuplify(rset.rows), [(3,), (1,), (4,), (2,)]) |
221 self.assertEquals(tuplify(rset.rows), [(3,), (1,), (4,), (2,)]) |
222 self.assertEquals(rset.description, [('CWGroup',), ('CWGroup',), ('CWGroup',), ('CWGroup',)]) |
222 self.assertEquals(rset.description, [('CWGroup',), ('CWGroup',), ('CWGroup',), ('CWGroup',)]) |
223 rset = self.execute('Any X ORDERBY N DESC WHERE X is CWGroup, X name N') |
223 rset = self.execute('Any X ORDERBY N DESC WHERE X is CWGroup, X name N') |
224 self.assertEquals(tuplify(rset.rows), [(2,), (4,), (1,), (3,)]) |
224 self.assertEquals(tuplify(rset.rows), [(2,), (4,), (1,), (3,)]) |
225 |
225 |
226 def test_select_3(self): |
226 def test_select_3(self): |
227 rset = self.execute('Any N GROUPBY N WHERE X is CWGroup, X name N') |
227 rset = self.execute('Any N GROUPBY N WHERE X is CWGroup, X name N') |
228 result, descr = rset.rows, rset.description |
228 result, descr = rset.rows, rset.description |
229 result.sort() |
229 result.sort() |
230 self.assertEquals(tuplify(result), [('guests',), ('managers',), ('owners',), ('users',)]) |
230 self.assertEquals(tuplify(result), [('guests',), ('managers',), ('owners',), ('users',)]) |
231 self.assertEquals(descr, [('String',), ('String',), ('String',), ('String',)]) |
231 self.assertEquals(descr, [('String',), ('String',), ('String',), ('String',)]) |
232 |
232 |
233 def test_select_is(self): |
233 def test_select_is(self): |
234 rset = self.execute('Any X, TN ORDERBY TN LIMIT 10 WHERE X is T, T name TN') |
234 rset = self.execute('Any X, TN ORDERBY TN LIMIT 10 WHERE X is T, T name TN') |
235 result, descr = rset.rows, rset.description |
235 result, descr = rset.rows, rset.description |
236 self.assertEquals(result[0][1], descr[0][0]) |
236 self.assertEquals(result[0][1], descr[0][0]) |
237 |
237 |
238 def test_select_is_aggr(self): |
238 def test_select_is_aggr(self): |
239 rset = self.execute('Any TN, COUNT(X) GROUPBY TN ORDERBY 2 DESC WHERE X is T, T name TN') |
239 rset = self.execute('Any TN, COUNT(X) GROUPBY TN ORDERBY 2 DESC WHERE X is T, T name TN') |
240 result, descr = rset.rows, rset.description |
240 result, descr = rset.rows, rset.description |
241 self.assertEquals(descr[0][0], 'String') |
241 self.assertEquals(descr[0][0], 'String') |
242 self.assertEquals(descr[0][1], 'Int') |
242 self.assertEquals(descr[0][1], 'Int') |
243 self.assertEquals(result[0][0], 'CWRelation') |
243 self.assertEquals(result[0][0], 'CWRelation') |
244 |
244 |
245 def test_select_groupby_orderby(self): |
245 def test_select_groupby_orderby(self): |
246 rset = self.execute('Any N GROUPBY N ORDERBY N WHERE X is CWGroup, X name N') |
246 rset = self.execute('Any N GROUPBY N ORDERBY N WHERE X is CWGroup, X name N') |
247 self.assertEquals(tuplify(rset.rows), [('guests',), ('managers',), ('owners',), ('users',)]) |
247 self.assertEquals(tuplify(rset.rows), [('guests',), ('managers',), ('owners',), ('users',)]) |
248 self.assertEquals(rset.description, [('String',), ('String',), ('String',), ('String',)]) |
248 self.assertEquals(rset.description, [('String',), ('String',), ('String',), ('String',)]) |
249 |
249 |
250 def test_select_complex_groupby(self): |
250 def test_select_complex_groupby(self): |
251 rset = self.execute('Any N GROUPBY N WHERE X name N') |
251 rset = self.execute('Any N GROUPBY N WHERE X name N') |
252 rset = self.execute('Any N,MAX(D) GROUPBY N LIMIT 5 WHERE X name N, X creation_date D') |
252 rset = self.execute('Any N,MAX(D) GROUPBY N LIMIT 5 WHERE X name N, X creation_date D') |
253 |
253 |
254 def test_select_inlined_groupby(self): |
254 def test_select_inlined_groupby(self): |
255 seid = self.execute('State X WHERE X name "deactivated"')[0][0] |
255 seid = self.execute('State X WHERE X name "deactivated"')[0][0] |
256 rset = self.execute('Any U,L,S GROUPBY U,L,S WHERE X in_state S, U login L, S eid %s' % seid) |
256 rset = self.execute('Any U,L,S GROUPBY U,L,S WHERE X in_state S, U login L, S eid %s' % seid) |
257 |
257 |
258 def test_select_complex_orderby(self): |
258 def test_select_complex_orderby(self): |
259 rset1 = self.execute('Any N ORDERBY N WHERE X name N') |
259 rset1 = self.execute('Any N ORDERBY N WHERE X name N') |
260 self.assertEquals(sorted(rset1.rows), rset1.rows) |
260 self.assertEquals(sorted(rset1.rows), rset1.rows) |
261 rset = self.execute('Any N ORDERBY N LIMIT 5 OFFSET 1 WHERE X name N') |
261 rset = self.execute('Any N ORDERBY N LIMIT 5 OFFSET 1 WHERE X name N') |
262 self.assertEquals(rset.rows[0][0], rset1.rows[1][0]) |
262 self.assertEquals(rset.rows[0][0], rset1.rows[1][0]) |
263 self.assertEquals(len(rset), 5) |
263 self.assertEquals(len(rset), 5) |
264 |
264 |
265 def test_select_5(self): |
265 def test_select_5(self): |
266 rset = self.execute('Any X, TMP ORDERBY TMP WHERE X name TMP, X is CWGroup') |
266 rset = self.execute('Any X, TMP ORDERBY TMP WHERE X name TMP, X is CWGroup') |
267 self.assertEquals(tuplify(rset.rows), [(3, 'guests',), (1, 'managers',), (4, 'owners',), (2, 'users',)]) |
267 self.assertEquals(tuplify(rset.rows), [(3, 'guests',), (1, 'managers',), (4, 'owners',), (2, 'users',)]) |
268 self.assertEquals(rset.description, [('CWGroup', 'String',), ('CWGroup', 'String',), ('CWGroup', 'String',), ('CWGroup', 'String',)]) |
268 self.assertEquals(rset.description, [('CWGroup', 'String',), ('CWGroup', 'String',), ('CWGroup', 'String',), ('CWGroup', 'String',)]) |
269 |
269 |
270 def test_select_6(self): |
270 def test_select_6(self): |
271 self.execute("INSERT Personne X: X nom 'bidule'")[0] |
271 self.execute("INSERT Personne X: X nom 'bidule'")[0] |
272 rset = self.execute('Any Y where X name TMP, Y nom in (TMP, "bidule")') |
272 rset = self.execute('Any Y where X name TMP, Y nom in (TMP, "bidule")') |
273 #self.assertEquals(rset.description, [('Personne',), ('Personne',)]) |
273 #self.assertEquals(rset.description, [('Personne',), ('Personne',)]) |
274 self.assert_(('Personne',) in rset.description) |
274 self.assert_(('Personne',) in rset.description) |
275 rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")') |
275 rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")') |
276 self.assert_(('Personne',) in rset.description) |
276 self.assert_(('Personne',) in rset.description) |
277 |
277 |
278 def test_select_not_attr(self): |
278 def test_select_not_attr(self): |
279 self.execute("INSERT Personne X: X nom 'bidule'") |
279 self.execute("INSERT Personne X: X nom 'bidule'") |
280 self.execute("INSERT Societe X: X nom 'chouette'") |
280 self.execute("INSERT Societe X: X nom 'chouette'") |
281 rset = self.execute('Personne X WHERE NOT X nom "bidule"') |
281 rset = self.execute('Personne X WHERE NOT X nom "bidule"') |
282 self.assertEquals(len(rset.rows), 0, rset.rows) |
282 self.assertEquals(len(rset.rows), 0, rset.rows) |
283 rset = self.execute('Personne X WHERE NOT X nom "bid"') |
283 rset = self.execute('Personne X WHERE NOT X nom "bid"') |
284 self.assertEquals(len(rset.rows), 1, rset.rows) |
284 self.assertEquals(len(rset.rows), 1, rset.rows) |
285 self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'") |
285 self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'") |
286 rset = self.execute('Personne X WHERE NOT X travaille S') |
286 rset = self.execute('Personne X WHERE NOT X travaille S') |
287 self.assertEquals(len(rset.rows), 0, rset.rows) |
287 self.assertEquals(len(rset.rows), 0, rset.rows) |
288 |
288 |
289 def test_select_is_in(self): |
289 def test_select_is_in(self): |
290 self.execute("INSERT Personne X: X nom 'bidule'") |
290 self.execute("INSERT Personne X: X nom 'bidule'") |
291 self.execute("INSERT Societe X: X nom 'chouette'") |
291 self.execute("INSERT Societe X: X nom 'chouette'") |
292 self.assertEquals(len(self.execute("Any X WHERE X is IN (Personne, Societe)")), |
292 self.assertEquals(len(self.execute("Any X WHERE X is IN (Personne, Societe)")), |
293 2) |
293 2) |
294 |
294 |
295 def test_select_not_rel(self): |
295 def test_select_not_rel(self): |
296 self.execute("INSERT Personne X: X nom 'bidule'") |
296 self.execute("INSERT Personne X: X nom 'bidule'") |
297 self.execute("INSERT Societe X: X nom 'chouette'") |
297 self.execute("INSERT Societe X: X nom 'chouette'") |
298 self.execute("INSERT Personne X: X nom 'autre'") |
298 self.execute("INSERT Personne X: X nom 'autre'") |
299 self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'") |
299 self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'") |
300 rset = self.execute('Personne X WHERE NOT X travaille S') |
300 rset = self.execute('Personne X WHERE NOT X travaille S') |
301 self.assertEquals(len(rset.rows), 1, rset.rows) |
301 self.assertEquals(len(rset.rows), 1, rset.rows) |
302 rset = self.execute('Personne X WHERE NOT X travaille S, S nom "chouette"') |
302 rset = self.execute('Personne X WHERE NOT X travaille S, S nom "chouette"') |
303 self.assertEquals(len(rset.rows), 1, rset.rows) |
303 self.assertEquals(len(rset.rows), 1, rset.rows) |
304 |
304 |
305 def test_select_nonregr_inlined(self): |
305 def test_select_nonregr_inlined(self): |
306 self.execute("INSERT Note X: X para 'bidule'") |
306 self.execute("INSERT Note X: X para 'bidule'") |
307 self.execute("INSERT Personne X: X nom 'chouette'") |
307 self.execute("INSERT Personne X: X nom 'chouette'") |
308 self.execute("INSERT Personne X: X nom 'autre'") |
308 self.execute("INSERT Personne X: X nom 'autre'") |
309 self.execute("SET X ecrit_par P WHERE X para 'bidule', P nom 'chouette'") |
309 self.execute("SET X ecrit_par P WHERE X para 'bidule', P nom 'chouette'") |
310 rset = self.execute('Any U,T ORDERBY T DESC WHERE U is CWUser, ' |
310 rset = self.execute('Any U,T ORDERBY T DESC WHERE U is CWUser, ' |
311 'N ecrit_par U, N type T')#, {'x': self.ueid}) |
311 'N ecrit_par U, N type T')#, {'x': self.ueid}) |
312 self.assertEquals(len(rset.rows), 0) |
312 self.assertEquals(len(rset.rows), 0) |
313 |
313 |
314 def test_select_nonregr_edition_not(self): |
314 def test_select_nonregr_edition_not(self): |
315 groupeids = set((1, 2, 3)) |
315 groupeids = set((1, 2, 3)) |
316 groupreadperms = set(r[0] for r in self.execute('Any Y WHERE X name "CWGroup", Y eid IN(1, 2, 3), X read_permission Y')) |
316 groupreadperms = set(r[0] for r in self.execute('Any Y WHERE X name "CWGroup", Y eid IN(1, 2, 3), X read_permission Y')) |
317 rset = self.execute('DISTINCT Any Y WHERE X is CWEType, X name "CWGroup", Y eid IN(1, 2, 3), NOT X read_permission Y') |
317 rset = self.execute('DISTINCT Any Y WHERE X is CWEType, X name "CWGroup", Y eid IN(1, 2, 3), NOT X read_permission Y') |
318 self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms)) |
318 self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms)) |
319 rset = self.execute('DISTINCT Any Y WHERE X name "CWGroup", Y eid IN(1, 2, 3), NOT X read_permission Y') |
319 rset = self.execute('DISTINCT Any Y WHERE X name "CWGroup", Y eid IN(1, 2, 3), NOT X read_permission Y') |
320 self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms)) |
320 self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms)) |
321 |
321 |
322 def test_select_outer_join(self): |
322 def test_select_outer_join(self): |
323 peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0] |
323 peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0] |
324 peid2 = self.execute("INSERT Personne X: X nom 'autre'")[0][0] |
324 peid2 = self.execute("INSERT Personne X: X nom 'autre'")[0][0] |
325 seid1 = self.execute("INSERT Societe X: X nom 'chouette'")[0][0] |
325 seid1 = self.execute("INSERT Societe X: X nom 'chouette'")[0][0] |
326 seid2 = self.execute("INSERT Societe X: X nom 'chouetos'")[0][0] |
326 seid2 = self.execute("INSERT Societe X: X nom 'chouetos'")[0][0] |
329 self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'") |
329 self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'") |
330 rset = self.execute('Any X,S ORDERBY X WHERE X travaille S?') |
330 rset = self.execute('Any X,S ORDERBY X WHERE X travaille S?') |
331 self.assertEquals(rset.rows, [[peid1, seid1], [peid2, None]]) |
331 self.assertEquals(rset.rows, [[peid1, seid1], [peid2, None]]) |
332 rset = self.execute('Any S,X ORDERBY S WHERE X? travaille S') |
332 rset = self.execute('Any S,X ORDERBY S WHERE X? travaille S') |
333 self.assertEquals(rset.rows, [[seid1, peid1], [seid2, None]]) |
333 self.assertEquals(rset.rows, [[seid1, peid1], [seid2, None]]) |
334 |
334 |
335 def test_select_outer_join_optimized(self): |
335 def test_select_outer_join_optimized(self): |
336 peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0] |
336 peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0] |
337 rset = self.execute('Any X WHERE X eid %(x)s, P? connait X', {'x':peid1}, 'x') |
337 rset = self.execute('Any X WHERE X eid %(x)s, P? connait X', {'x':peid1}, 'x') |
338 self.assertEquals(rset.rows, [[peid1]]) |
338 self.assertEquals(rset.rows, [[peid1]]) |
339 rset = self.execute('Any X WHERE X eid %(x)s, X require_permission P?', {'x':peid1}, 'x') |
339 rset = self.execute('Any X WHERE X eid %(x)s, X require_permission P?', {'x':peid1}, 'x') |
370 {'g': geid, 't': teid}, 'g') |
370 {'g': geid, 't': teid}, 'g') |
371 rset = self.execute("Any GN,TN ORDERBY GN WHERE T? tags G, T name TN, G name GN") |
371 rset = self.execute("Any GN,TN ORDERBY GN WHERE T? tags G, T name TN, G name GN") |
372 self.failUnless(['users', 'tag'] in rset.rows) |
372 self.failUnless(['users', 'tag'] in rset.rows) |
373 self.failUnless(['activated', None] in rset.rows) |
373 self.failUnless(['activated', None] in rset.rows) |
374 rset = self.execute("Any GN,TN ORDERBY GN WHERE T tags G?, T name TN, G name GN") |
374 rset = self.execute("Any GN,TN ORDERBY GN WHERE T tags G?, T name TN, G name GN") |
375 self.assertEquals(rset.rows, [[None, 'tagbis'], ['users', 'tag']]) |
375 self.assertEquals(rset.rows, [[None, 'tagbis'], ['users', 'tag']]) |
376 |
376 |
377 def test_select_not_inline_rel(self): |
377 def test_select_not_inline_rel(self): |
378 self.execute("INSERT Personne X: X nom 'bidule'") |
378 self.execute("INSERT Personne X: X nom 'bidule'") |
379 self.execute("INSERT Note X: X type 'a'") |
379 self.execute("INSERT Note X: X type 'a'") |
380 self.execute("INSERT Note X: X type 'b'") |
380 self.execute("INSERT Note X: X type 'b'") |
381 self.execute("SET X ecrit_par Y WHERE X type 'a', Y nom 'bidule'") |
381 self.execute("SET X ecrit_par Y WHERE X type 'a', Y nom 'bidule'") |
382 rset = self.execute('Note X WHERE NOT X ecrit_par P') |
382 rset = self.execute('Note X WHERE NOT X ecrit_par P') |
383 self.assertEquals(len(rset.rows), 1, rset.rows) |
383 self.assertEquals(len(rset.rows), 1, rset.rows) |
384 |
384 |
385 def test_select_not_unlinked_multiple_solutions(self): |
385 def test_select_not_unlinked_multiple_solutions(self): |
386 self.execute("INSERT Personne X: X nom 'bidule'") |
386 self.execute("INSERT Personne X: X nom 'bidule'") |
387 self.execute("INSERT Note X: X type 'a'") |
387 self.execute("INSERT Note X: X type 'a'") |
388 self.execute("INSERT Note X: X type 'b'") |
388 self.execute("INSERT Note X: X type 'b'") |
389 self.execute("SET Y evaluee X WHERE X type 'a', Y nom 'bidule'") |
389 self.execute("SET Y evaluee X WHERE X type 'a', Y nom 'bidule'") |
393 def test_select_aggregat_count(self): |
393 def test_select_aggregat_count(self): |
394 rset = self.execute('Any COUNT(X)') |
394 rset = self.execute('Any COUNT(X)') |
395 self.assertEquals(len(rset.rows), 1) |
395 self.assertEquals(len(rset.rows), 1) |
396 self.assertEquals(len(rset.rows[0]), 1) |
396 self.assertEquals(len(rset.rows[0]), 1) |
397 self.assertEquals(rset.description, [('Int',)]) |
397 self.assertEquals(rset.description, [('Int',)]) |
398 |
398 |
399 def test_select_aggregat_sum(self): |
399 def test_select_aggregat_sum(self): |
400 rset = self.execute('Any SUM(O) WHERE X ordernum O') |
400 rset = self.execute('Any SUM(O) WHERE X ordernum O') |
401 self.assertEquals(len(rset.rows), 1) |
401 self.assertEquals(len(rset.rows), 1) |
402 self.assertEquals(len(rset.rows[0]), 1) |
402 self.assertEquals(len(rset.rows[0]), 1) |
403 self.assertEquals(rset.description, [('Int',)]) |
403 self.assertEquals(rset.description, [('Int',)]) |
404 |
404 |
405 def test_select_aggregat_min(self): |
405 def test_select_aggregat_min(self): |
406 rset = self.execute('Any MIN(X) WHERE X is Personne') |
406 rset = self.execute('Any MIN(X) WHERE X is Personne') |
407 self.assertEquals(len(rset.rows), 1) |
407 self.assertEquals(len(rset.rows), 1) |
408 self.assertEquals(len(rset.rows[0]), 1) |
408 self.assertEquals(len(rset.rows[0]), 1) |
409 self.assertEquals(rset.description, [('Personne',)]) |
409 self.assertEquals(rset.description, [('Personne',)]) |
410 rset = self.execute('Any MIN(O) WHERE X ordernum O') |
410 rset = self.execute('Any MIN(O) WHERE X ordernum O') |
411 self.assertEquals(len(rset.rows), 1) |
411 self.assertEquals(len(rset.rows), 1) |
412 self.assertEquals(len(rset.rows[0]), 1) |
412 self.assertEquals(len(rset.rows[0]), 1) |
413 self.assertEquals(rset.description, [('Int',)]) |
413 self.assertEquals(rset.description, [('Int',)]) |
414 |
414 |
415 def test_select_aggregat_max(self): |
415 def test_select_aggregat_max(self): |
416 rset = self.execute('Any MAX(X) WHERE X is Personne') |
416 rset = self.execute('Any MAX(X) WHERE X is Personne') |
417 self.assertEquals(len(rset.rows), 1) |
417 self.assertEquals(len(rset.rows), 1) |
418 self.assertEquals(len(rset.rows[0]), 1) |
418 self.assertEquals(len(rset.rows[0]), 1) |
419 self.assertEquals(rset.description, [('Personne',)]) |
419 self.assertEquals(rset.description, [('Personne',)]) |
442 self.failUnlessEqual(len(rset), 1) |
442 self.failUnlessEqual(len(rset), 1) |
443 self.failUnlessEqual(rset[0][1], 'managers') |
443 self.failUnlessEqual(rset[0][1], 'managers') |
444 rset = self.execute('Any X,N ORDERBY GROUP_SORT_VALUE(N) WHERE X is CWGroup, X name N, NOT U in_group X, U login "admin"') |
444 rset = self.execute('Any X,N ORDERBY GROUP_SORT_VALUE(N) WHERE X is CWGroup, X name N, NOT U in_group X, U login "admin"') |
445 self.failUnlessEqual(len(rset), 3) |
445 self.failUnlessEqual(len(rset), 3) |
446 self.failUnlessEqual(rset[0][1], 'owners') |
446 self.failUnlessEqual(rset[0][1], 'owners') |
447 |
447 |
448 def test_select_aggregat_sort(self): |
448 def test_select_aggregat_sort(self): |
449 rset = self.execute('Any G, COUNT(U) GROUPBY G ORDERBY 2 WHERE U in_group G') |
449 rset = self.execute('Any G, COUNT(U) GROUPBY G ORDERBY 2 WHERE U in_group G') |
450 self.assertEquals(len(rset.rows), 2) |
450 self.assertEquals(len(rset.rows), 2) |
451 self.assertEquals(len(rset.rows[0]), 2) |
451 self.assertEquals(len(rset.rows[0]), 2) |
452 self.assertEquals(rset.description[0], ('CWGroup', 'Int',)) |
452 self.assertEquals(rset.description[0], ('CWGroup', 'Int',)) |
473 self.skip('retry me once http://www.sqlite.org/cvstrac/tktview?tn=3773 is fixed') |
473 self.skip('retry me once http://www.sqlite.org/cvstrac/tktview?tn=3773 is fixed') |
474 rset = self.execute('Any X ORDERBY X,D LIMIT 5 WHERE X creation_date D') |
474 rset = self.execute('Any X ORDERBY X,D LIMIT 5 WHERE X creation_date D') |
475 result = rset.rows |
475 result = rset.rows |
476 result.sort() |
476 result.sort() |
477 self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,), (5,)]) |
477 self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,), (5,)]) |
478 |
478 |
479 def test_select_upper(self): |
479 def test_select_upper(self): |
480 rset = self.execute('Any X, UPPER(L) ORDERBY L WHERE X is CWUser, X login L') |
480 rset = self.execute('Any X, UPPER(L) ORDERBY L WHERE X is CWUser, X login L') |
481 self.assertEquals(len(rset.rows), 2) |
481 self.assertEquals(len(rset.rows), 2) |
482 self.assertEquals(rset.rows[0][1], 'ADMIN') |
482 self.assertEquals(rset.rows[0][1], 'ADMIN') |
483 self.assertEquals(rset.description[0], ('CWUser', 'String',)) |
483 self.assertEquals(rset.description[0], ('CWUser', 'String',)) |
492 ## ueid = self.session.user.eid |
492 ## ueid = self.session.user.eid |
493 ## rset = self.execute('Any L WHERE %s login L'%ueid) |
493 ## rset = self.execute('Any L WHERE %s login L'%ueid) |
494 ## self.assertEquals(rset.rows[0][0], 'admin') |
494 ## self.assertEquals(rset.rows[0][0], 'admin') |
495 ## rset = self.execute('Any L WHERE %(x)s login L', {'x':ueid}) |
495 ## rset = self.execute('Any L WHERE %(x)s login L', {'x':ueid}) |
496 ## self.assertEquals(rset.rows[0][0], 'admin') |
496 ## self.assertEquals(rset.rows[0][0], 'admin') |
497 |
497 |
498 def test_select_searchable_text_1(self): |
498 def test_select_searchable_text_1(self): |
499 rset = self.execute(u"INSERT Personne X: X nom 'bidüle'") |
499 rset = self.execute(u"INSERT Personne X: X nom 'bidüle'") |
500 rset = self.execute(u"INSERT Societe X: X nom 'bidüle'") |
500 rset = self.execute(u"INSERT Societe X: X nom 'bidüle'") |
501 rset = self.execute("INSERT Societe X: X nom 'chouette'") |
501 rset = self.execute("INSERT Societe X: X nom 'chouette'") |
502 self.commit() |
502 self.commit() |
507 biduleeids = [r[0] for r in rset.rows] |
507 biduleeids = [r[0] for r in rset.rows] |
508 rset = self.execute(u'Any N where NOT N has_text "bidüle"') |
508 rset = self.execute(u'Any N where NOT N has_text "bidüle"') |
509 self.failIf([r[0] for r in rset.rows if r[0] in biduleeids]) |
509 self.failIf([r[0] for r in rset.rows if r[0] in biduleeids]) |
510 # duh? |
510 # duh? |
511 rset = self.execute('Any X WHERE X has_text %(text)s', {'text': u'ça'}) |
511 rset = self.execute('Any X WHERE X has_text %(text)s', {'text': u'ça'}) |
512 |
512 |
513 def test_select_searchable_text_2(self): |
513 def test_select_searchable_text_2(self): |
514 rset = self.execute("INSERT Personne X: X nom 'bidule'") |
514 rset = self.execute("INSERT Personne X: X nom 'bidule'") |
515 rset = self.execute("INSERT Personne X: X nom 'chouette'") |
515 rset = self.execute("INSERT Personne X: X nom 'chouette'") |
516 rset = self.execute("INSERT Societe X: X nom 'bidule'") |
516 rset = self.execute("INSERT Societe X: X nom 'bidule'") |
517 self.commit() |
517 self.commit() |
518 rset = self.execute('Personne N where N has_text "bidule"') |
518 rset = self.execute('Personne N where N has_text "bidule"') |
519 self.assertEquals(len(rset.rows), 1, rset.rows) |
519 self.assertEquals(len(rset.rows), 1, rset.rows) |
520 |
520 |
521 def test_select_searchable_text_3(self): |
521 def test_select_searchable_text_3(self): |
522 rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'M'") |
522 rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'M'") |
523 rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'F'") |
523 rset = self.execute("INSERT Personne X: X nom 'bidule', X sexe 'F'") |
524 rset = self.execute("INSERT Societe X: X nom 'bidule'") |
524 rset = self.execute("INSERT Societe X: X nom 'bidule'") |
525 self.commit() |
525 self.commit() |
526 rset = self.execute('Any X where X has_text "bidule" and X sexe "M"') |
526 rset = self.execute('Any X where X has_text "bidule" and X sexe "M"') |
527 self.assertEquals(len(rset.rows), 1, rset.rows) |
527 self.assertEquals(len(rset.rows), 1, rset.rows) |
528 |
528 |
529 def test_select_multiple_searchable_text(self): |
529 def test_select_multiple_searchable_text(self): |
530 self.execute(u"INSERT Personne X: X nom 'bidüle'") |
530 self.execute(u"INSERT Personne X: X nom 'bidüle'") |
531 self.execute("INSERT Societe X: X nom 'chouette', S travaille X") |
531 self.execute("INSERT Societe X: X nom 'chouette', S travaille X") |
532 self.execute(u"INSERT Personne X: X nom 'bidüle'") |
532 self.execute(u"INSERT Personne X: X nom 'bidüle'") |
533 self.commit() |
533 self.commit() |
534 rset = self.execute('Personne X WHERE X has_text %(text)s, X travaille S, S has_text %(text2)s', |
534 rset = self.execute('Personne X WHERE X has_text %(text)s, X travaille S, S has_text %(text2)s', |
535 {'text': u'bidüle', |
535 {'text': u'bidüle', |
536 'text2': u'chouette',} |
536 'text2': u'chouette',} |
537 ) |
537 ) |
538 self.assertEquals(len(rset.rows), 1, rset.rows) |
538 self.assertEquals(len(rset.rows), 1, rset.rows) |
539 |
539 |
540 def test_select_no_descr(self): |
540 def test_select_no_descr(self): |
541 rset = self.execute('Any X WHERE X is CWGroup', build_descr=0) |
541 rset = self.execute('Any X WHERE X is CWGroup', build_descr=0) |
542 rset.rows.sort() |
542 rset.rows.sort() |
543 self.assertEquals(tuplify(rset.rows), [(1,), (2,), (3,), (4,)]) |
543 self.assertEquals(tuplify(rset.rows), [(1,), (2,), (3,), (4,)]) |
544 self.assertEquals(rset.description, ()) |
544 self.assertEquals(rset.description, ()) |
547 rset = self.execute('CWGroup X ORDERBY N LIMIT 2 WHERE X name N') |
547 rset = self.execute('CWGroup X ORDERBY N LIMIT 2 WHERE X name N') |
548 self.assertEquals(tuplify(rset.rows), [(3,), (1,)]) |
548 self.assertEquals(tuplify(rset.rows), [(3,), (1,)]) |
549 self.assertEquals(rset.description, [('CWGroup',), ('CWGroup',)]) |
549 self.assertEquals(rset.description, [('CWGroup',), ('CWGroup',)]) |
550 rset = self.execute('CWGroup X ORDERBY N LIMIT 2 OFFSET 2 WHERE X name N') |
550 rset = self.execute('CWGroup X ORDERBY N LIMIT 2 OFFSET 2 WHERE X name N') |
551 self.assertEquals(tuplify(rset.rows), [(4,), (2,)]) |
551 self.assertEquals(tuplify(rset.rows), [(4,), (2,)]) |
552 |
552 |
553 def test_select_symetric(self): |
553 def test_select_symetric(self): |
554 self.execute("INSERT Personne X: X nom 'machin'") |
554 self.execute("INSERT Personne X: X nom 'machin'") |
555 self.execute("INSERT Personne X: X nom 'bidule'") |
555 self.execute("INSERT Personne X: X nom 'bidule'") |
556 self.execute("INSERT Personne X: X nom 'chouette'") |
556 self.execute("INSERT Personne X: X nom 'chouette'") |
557 self.execute("INSERT Personne X: X nom 'trucmuche'") |
557 self.execute("INSERT Personne X: X nom 'trucmuche'") |
567 self.assertEquals(len(rset.rows), 1, rset.rows) |
567 self.assertEquals(len(rset.rows), 1, rset.rows) |
568 rset = self.execute('Any P where P connait P2, P2 nom "chouette"') |
568 rset = self.execute('Any P where P connait P2, P2 nom "chouette"') |
569 self.assertEquals(len(rset.rows), 2, rset.rows) |
569 self.assertEquals(len(rset.rows), 2, rset.rows) |
570 rset = self.execute('Any P where P2 connait P, P2 nom "chouette"') |
570 rset = self.execute('Any P where P2 connait P, P2 nom "chouette"') |
571 self.assertEquals(len(rset.rows), 2, rset.rows) |
571 self.assertEquals(len(rset.rows), 2, rset.rows) |
572 |
572 |
573 def test_select_inline(self): |
573 def test_select_inline(self): |
574 self.execute("INSERT Personne X: X nom 'bidule'") |
574 self.execute("INSERT Personne X: X nom 'bidule'") |
575 self.execute("INSERT Note X: X type 'a'") |
575 self.execute("INSERT Note X: X type 'a'") |
576 self.execute("SET X ecrit_par Y WHERE X type 'a', Y nom 'bidule'") |
576 self.execute("SET X ecrit_par Y WHERE X type 'a', Y nom 'bidule'") |
577 rset = self.execute('Any N where N ecrit_par X, X nom "bidule"') |
577 rset = self.execute('Any N where N ecrit_par X, X nom "bidule"') |
578 self.assertEquals(len(rset.rows), 1, rset.rows) |
578 self.assertEquals(len(rset.rows), 1, rset.rows) |
579 |
579 |
580 def test_select_creation_date(self): |
580 def test_select_creation_date(self): |
581 self.execute("INSERT Personne X: X nom 'bidule'") |
581 self.execute("INSERT Personne X: X nom 'bidule'") |
582 rset = self.execute('Any D WHERE X nom "bidule", X creation_date D') |
582 rset = self.execute('Any D WHERE X nom "bidule", X creation_date D') |
583 self.assertEqual(len(rset.rows), 1) |
583 self.assertEqual(len(rset.rows), 1) |
584 |
584 |
591 rset = self.execute('DISTINCT Any P WHERE P travaille S1 OR P travaille S2, S1 nom "logilab", S2 nom "caesium"') |
591 rset = self.execute('DISTINCT Any P WHERE P travaille S1 OR P travaille S2, S1 nom "logilab", S2 nom "caesium"') |
592 self.assertEqual(len(rset.rows), 1) |
592 self.assertEqual(len(rset.rows), 1) |
593 self.execute("SET P travaille S WHERE P nom 'chouette', S nom 'caesium'") |
593 self.execute("SET P travaille S WHERE P nom 'chouette', S nom 'caesium'") |
594 rset = self.execute('DISTINCT Any P WHERE P travaille S1 OR P travaille S2, S1 nom "logilab", S2 nom "caesium"') |
594 rset = self.execute('DISTINCT Any P WHERE P travaille S1 OR P travaille S2, S1 nom "logilab", S2 nom "caesium"') |
595 self.assertEqual(len(rset.rows), 2) |
595 self.assertEqual(len(rset.rows), 2) |
596 |
596 |
597 def test_select_or_sym_relation(self): |
597 def test_select_or_sym_relation(self): |
598 self.execute("INSERT Personne X: X nom 'bidule'") |
598 self.execute("INSERT Personne X: X nom 'bidule'") |
599 self.execute("INSERT Personne X: X nom 'chouette'") |
599 self.execute("INSERT Personne X: X nom 'chouette'") |
600 self.execute("INSERT Personne X: X nom 'truc'") |
600 self.execute("INSERT Personne X: X nom 'truc'") |
601 self.execute("SET P connait S WHERE P nom 'bidule', S nom 'chouette'") |
601 self.execute("SET P connait S WHERE P nom 'bidule', S nom 'chouette'") |
606 self.execute("SET P connait S WHERE P nom 'chouette', S nom 'truc'") |
606 self.execute("SET P connait S WHERE P nom 'chouette', S nom 'truc'") |
607 rset = self.execute('DISTINCT Any P WHERE S connait P, S nom "chouette"') |
607 rset = self.execute('DISTINCT Any P WHERE S connait P, S nom "chouette"') |
608 self.assertEqual(len(rset.rows), 2, rset.rows) |
608 self.assertEqual(len(rset.rows), 2, rset.rows) |
609 rset = self.execute('DISTINCT Any P WHERE P connait S OR S connait P, S nom "chouette"') |
609 rset = self.execute('DISTINCT Any P WHERE P connait S OR S connait P, S nom "chouette"') |
610 self.assertEqual(len(rset.rows), 2, rset.rows) |
610 self.assertEqual(len(rset.rows), 2, rset.rows) |
611 |
611 |
612 def test_select_follow_relation(self): |
612 def test_select_follow_relation(self): |
613 self.execute("INSERT Affaire X: X sujet 'cool'") |
613 self.execute("INSERT Affaire X: X sujet 'cool'") |
614 self.execute("INSERT Societe X: X nom 'chouette'") |
614 self.execute("INSERT Societe X: X nom 'chouette'") |
615 self.execute("SET A concerne S WHERE A is Affaire, S is Societe") |
615 self.execute("SET A concerne S WHERE A is Affaire, S is Societe") |
616 self.execute("INSERT Note X: X para 'truc'") |
616 self.execute("INSERT Note X: X para 'truc'") |
633 self.execute("INSERT Affaire X: X sujet 'normal'") |
633 self.execute("INSERT Affaire X: X sujet 'normal'") |
634 self.execute("INSERT Affaire X: X sujet 'zou'") |
634 self.execute("INSERT Affaire X: X sujet 'zou'") |
635 self.execute("INSERT Affaire X: X sujet 'abcd'") |
635 self.execute("INSERT Affaire X: X sujet 'abcd'") |
636 rset = self.execute('DISTINCT Any S ORDERBY S WHERE A is Affaire, A sujet S') |
636 rset = self.execute('DISTINCT Any S ORDERBY S WHERE A is Affaire, A sujet S') |
637 self.assertEqual(rset.rows, [['abcd'], ['important'], ['minor'], ['normal'], ['zou']]) |
637 self.assertEqual(rset.rows, [['abcd'], ['important'], ['minor'], ['normal'], ['zou']]) |
638 |
638 |
639 def test_select_ordered_distinct_3(self): |
639 def test_select_ordered_distinct_3(self): |
640 rset = self.execute('DISTINCT Any N ORDERBY GROUP_SORT_VALUE(N) WHERE X is CWGroup, X name N') |
640 rset = self.execute('DISTINCT Any N ORDERBY GROUP_SORT_VALUE(N) WHERE X is CWGroup, X name N') |
641 self.assertEqual(rset.rows, [['owners'], ['guests'], ['users'], ['managers']]) |
641 self.assertEqual(rset.rows, [['owners'], ['guests'], ['users'], ['managers']]) |
642 |
642 |
643 def test_select_or_value(self): |
643 def test_select_or_value(self): |
648 |
648 |
649 def test_select_explicit_eid(self): |
649 def test_select_explicit_eid(self): |
650 rset = self.execute('Any X,E WHERE X owned_by U, X eid E, U eid %(u)s', {'u': self.session.user.eid}) |
650 rset = self.execute('Any X,E WHERE X owned_by U, X eid E, U eid %(u)s', {'u': self.session.user.eid}) |
651 self.failUnless(rset) |
651 self.failUnless(rset) |
652 self.assertEquals(rset.description[0][1], 'Int') |
652 self.assertEquals(rset.description[0][1], 'Int') |
653 |
653 |
654 # def test_select_rewritten_optional(self): |
654 # def test_select_rewritten_optional(self): |
655 # eid = self.execute("INSERT Affaire X: X sujet 'cool'")[0][0] |
655 # eid = self.execute("INSERT Affaire X: X sujet 'cool'")[0][0] |
656 # rset = self.execute('Any X WHERE X eid %(x)s, EXISTS(X owned_by U) OR EXISTS(X concerne S?, S owned_by U)', |
656 # rset = self.execute('Any X WHERE X eid %(x)s, EXISTS(X owned_by U) OR EXISTS(X concerne S?, S owned_by U)', |
657 # {'x': eid}, 'x') |
657 # {'x': eid}, 'x') |
658 # self.assertEquals(rset.rows, [[eid]]) |
658 # self.assertEquals(rset.rows, [[eid]]) |
659 |
659 |
660 def test_today_bug(self): |
660 def test_today_bug(self): |
661 self.execute("INSERT Tag X: X name 'bidule', X creation_date NOW") |
661 self.execute("INSERT Tag X: X name 'bidule', X creation_date NOW") |
662 self.execute("INSERT Tag Y: Y name 'toto'") |
662 self.execute("INSERT Tag Y: Y name 'toto'") |
663 rset = self.execute("Any D WHERE X name in ('bidule', 'toto') , X creation_date D") |
663 rset = self.execute("Any D WHERE X name in ('bidule', 'toto') , X creation_date D") |
664 self.assert_(isinstance(rset.rows[0][0], datetime), rset.rows) |
664 self.assert_(isinstance(rset.rows[0][0], datetime), rset.rows) |
687 'Date', 'Datetime', |
687 'Date', 'Datetime', |
688 'Decimal', 'Float', |
688 'Decimal', 'Float', |
689 'Int', 'Interval', |
689 'Int', 'Interval', |
690 'Password', 'String', |
690 'Password', 'String', |
691 'Time']) |
691 'Time']) |
692 |
692 |
693 def test_select_constant(self): |
693 def test_select_constant(self): |
694 rset = self.execute('Any X, "toto" ORDERBY X WHERE X is CWGroup') |
694 rset = self.execute('Any X, "toto" ORDERBY X WHERE X is CWGroup') |
695 self.assertEquals(rset.rows, |
695 self.assertEquals(rset.rows, |
696 map(list, zip((1,2,3,4), ('toto','toto','toto','toto',)))) |
696 map(list, zip((1,2,3,4), ('toto','toto','toto','toto',)))) |
697 self.assertIsInstance(rset[0][1], unicode) |
697 self.assertIsInstance(rset[0][1], unicode) |
716 self.assertEquals([x[1] for x in rset.rows], |
716 self.assertEquals([x[1] for x in rset.rows], |
717 ['activate', 'activated', 'deactivate', 'deactivated']) |
717 ['activate', 'activated', 'deactivate', 'deactivated']) |
718 self.assertEquals(rset.description, |
718 self.assertEquals(rset.description, |
719 [('Transition', 'String'), ('State', 'String'), |
719 [('Transition', 'String'), ('State', 'String'), |
720 ('Transition', 'String'), ('State', 'String')]) |
720 ('Transition', 'String'), ('State', 'String')]) |
721 |
721 |
722 def test_select_union_aggregat(self): |
722 def test_select_union_aggregat(self): |
723 # meaningless, the goal in to have group by done on different attribute |
723 # meaningless, the goal in to have group by done on different attribute |
724 # for each sub-query |
724 # for each sub-query |
725 self.execute('(Any N,COUNT(X) GROUPBY N WHERE X name N, X is State)' |
725 self.execute('(Any N,COUNT(X) GROUPBY N WHERE X name N, X is State)' |
726 ' UNION ' |
726 ' UNION ' |
727 '(Any N,COUNT(X) GROUPBY N ORDERBY 2 WHERE X login N)') |
727 '(Any N,COUNT(X) GROUPBY N ORDERBY 2 WHERE X login N)') |
728 |
728 |
729 def test_select_union_aggregat_independant_group(self): |
729 def test_select_union_aggregat_independant_group(self): |
730 self.execute('INSERT State X: X name "hop"') |
730 self.execute('INSERT State X: X name "hop"') |
731 self.execute('INSERT State X: X name "hop"') |
731 self.execute('INSERT State X: X name "hop"') |
732 self.execute('INSERT Transition X: X name "hop"') |
732 self.execute('INSERT Transition X: X name "hop"') |
733 self.execute('INSERT Transition X: X name "hop"') |
733 self.execute('INSERT Transition X: X name "hop"') |
734 rset = self.execute('Any N,NX ORDERBY 2 WITH N,NX BEING ' |
734 rset = self.execute('Any N,NX ORDERBY 2 WITH N,NX BEING ' |
735 '((Any N,COUNT(X) GROUPBY N WHERE X name N, X is State HAVING COUNT(X)>1)' |
735 '((Any N,COUNT(X) GROUPBY N WHERE X name N, X is State HAVING COUNT(X)>1)' |
736 ' UNION ' |
736 ' UNION ' |
737 '(Any N,COUNT(X) GROUPBY N WHERE X name N, X is Transition HAVING COUNT(X)>1))') |
737 '(Any N,COUNT(X) GROUPBY N WHERE X name N, X is Transition HAVING COUNT(X)>1))') |
738 self.assertEquals(rset.rows, [[u'hop', 2], [u'hop', 2]]) |
738 self.assertEquals(rset.rows, [[u'hop', 2], [u'hop', 2]]) |
739 |
739 |
740 def test_select_union_selection_with_diff_variables(self): |
740 def test_select_union_selection_with_diff_variables(self): |
741 rset = self.execute('(Any N WHERE X name N, X is State)' |
741 rset = self.execute('(Any N WHERE X name N, X is State)' |
742 ' UNION ' |
742 ' UNION ' |
743 '(Any NN WHERE XX name NN, XX is Transition)') |
743 '(Any NN WHERE XX name NN, XX is Transition)') |
744 self.assertEquals(sorted(r[0] for r in rset.rows), |
744 self.assertEquals(sorted(r[0] for r in rset.rows), |
745 ['abort', 'activate', 'activated', 'ben non', |
745 ['abort', 'activate', 'activated', 'ben non', |
746 'deactivate', 'deactivated', 'done', 'en cours', |
746 'deactivate', 'deactivated', 'done', 'en cours', |
747 'end', 'finie', 'markasdone', 'pitetre', 'redoit', |
747 'end', 'finie', 'markasdone', 'pitetre', 'redoit', |
748 'start', 'todo']) |
748 'start', 'todo']) |
749 |
749 |
750 def test_exists(self): |
750 def test_exists(self): |
751 geid = self.execute("INSERT CWGroup X: X name 'lulufanclub'")[0][0] |
751 geid = self.execute("INSERT CWGroup X: X name 'lulufanclub'")[0][0] |
752 self.execute("SET U in_group G WHERE G name 'lulufanclub'") |
752 self.execute("SET U in_group G WHERE G name 'lulufanclub'") |
753 peid = self.execute("INSERT Personne X: X prenom 'lulu', X nom 'petit'")[0][0] |
753 peid = self.execute("INSERT Personne X: X prenom 'lulu', X nom 'petit'")[0][0] |
754 rset = self.execute("Any X WHERE X prenom 'lulu'," |
754 rset = self.execute("Any X WHERE X prenom 'lulu'," |
782 |
782 |
783 def test_select_subquery_const(self): |
783 def test_select_subquery_const(self): |
784 rset = self.execute('Any X WITH X BEING ((Any NULL) UNION (Any "toto"))') |
784 rset = self.execute('Any X WITH X BEING ((Any NULL) UNION (Any "toto"))') |
785 self.assertEquals(rset.rows, [[None], ['toto']]) |
785 self.assertEquals(rset.rows, [[None], ['toto']]) |
786 self.assertEquals(rset.description, [(None,), ('String',)]) |
786 self.assertEquals(rset.description, [(None,), ('String',)]) |
787 |
787 |
788 # insertion queries tests ################################################# |
788 # insertion queries tests ################################################# |
789 |
789 |
790 def test_insert_is(self): |
790 def test_insert_is(self): |
791 eid, = self.execute("INSERT Personne X: X nom 'bidule'")[0] |
791 eid, = self.execute("INSERT Personne X: X nom 'bidule'")[0] |
792 etype, = self.execute("Any TN WHERE X is T, X eid %s, T name TN" % eid)[0] |
792 etype, = self.execute("Any TN WHERE X is T, X eid %s, T name TN" % eid)[0] |
793 self.assertEquals(etype, 'Personne') |
793 self.assertEquals(etype, 'Personne') |
794 self.execute("INSERT Personne X: X nom 'managers'") |
794 self.execute("INSERT Personne X: X nom 'managers'") |
795 |
795 |
796 def test_insert_1(self): |
796 def test_insert_1(self): |
797 rset = self.execute("INSERT Personne X: X nom 'bidule'") |
797 rset = self.execute("INSERT Personne X: X nom 'bidule'") |
798 self.assertEquals(len(rset.rows), 1) |
798 self.assertEquals(len(rset.rows), 1) |
799 self.assertEquals(rset.description, [('Personne',)]) |
799 self.assertEquals(rset.description, [('Personne',)]) |
800 rset = self.execute('Personne X WHERE X nom "bidule"') |
800 rset = self.execute('Personne X WHERE X nom "bidule"') |
817 |
817 |
818 def test_insert_3(self): |
818 def test_insert_3(self): |
819 self.execute("INSERT Personne X: X nom Y WHERE U login 'admin', U login Y") |
819 self.execute("INSERT Personne X: X nom Y WHERE U login 'admin', U login Y") |
820 rset = self.execute('Personne X WHERE X nom "admin"') |
820 rset = self.execute('Personne X WHERE X nom "admin"') |
821 self.assert_(rset.rows) |
821 self.assert_(rset.rows) |
822 self.assertEquals(rset.description, [('Personne',)]) |
822 self.assertEquals(rset.description, [('Personne',)]) |
823 |
823 |
824 def test_insert_4(self): |
824 def test_insert_4(self): |
825 self.execute("INSERT Societe Y: Y nom 'toto'") |
825 self.execute("INSERT Societe Y: Y nom 'toto'") |
826 self.execute("INSERT Personne X: X nom 'bidule', X travaille Y WHERE Y nom 'toto'") |
826 self.execute("INSERT Personne X: X nom 'bidule', X travaille Y WHERE Y nom 'toto'") |
827 rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto", X travaille Y') |
827 rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto", X travaille Y') |
828 self.assert_(rset.rows) |
828 self.assert_(rset.rows) |
829 self.assertEquals(rset.description, [('Personne', 'Societe',)]) |
829 self.assertEquals(rset.description, [('Personne', 'Societe',)]) |
830 |
830 |
831 def test_insert_4bis(self): |
831 def test_insert_4bis(self): |
832 peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0] |
832 peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0] |
833 seid = self.execute("INSERT Societe Y: Y nom 'toto', X travaille Y WHERE X eid %(x)s", |
833 seid = self.execute("INSERT Societe Y: Y nom 'toto', X travaille Y WHERE X eid %(x)s", |
834 {'x': str(peid)})[0][0] |
834 {'x': str(peid)})[0][0] |
835 self.assertEqual(len(self.execute('Any X, Y WHERE X travaille Y')), 1) |
835 self.assertEqual(len(self.execute('Any X, Y WHERE X travaille Y')), 1) |
836 self.execute("INSERT Personne X: X nom 'chouette', X travaille Y WHERE Y eid %(x)s", |
836 self.execute("INSERT Personne X: X nom 'chouette', X travaille Y WHERE Y eid %(x)s", |
837 {'x': str(seid)}) |
837 {'x': str(seid)}) |
838 self.assertEqual(len(self.execute('Any X, Y WHERE X travaille Y')), 2) |
838 self.assertEqual(len(self.execute('Any X, Y WHERE X travaille Y')), 2) |
839 |
839 |
840 def test_insert_4ter(self): |
840 def test_insert_4ter(self): |
841 peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0] |
841 peid = self.execute("INSERT Personne X: X nom 'bidule'")[0][0] |
842 seid = self.execute("INSERT Societe Y: Y nom 'toto', X travaille Y WHERE X eid %(x)s", |
842 seid = self.execute("INSERT Societe Y: Y nom 'toto', X travaille Y WHERE X eid %(x)s", |
843 {'x': unicode(peid)})[0][0] |
843 {'x': unicode(peid)})[0][0] |
844 self.assertEqual(len(self.execute('Any X, Y WHERE X travaille Y')), 1) |
844 self.assertEqual(len(self.execute('Any X, Y WHERE X travaille Y')), 1) |
889 def test_insertion_description_with_where(self): |
889 def test_insertion_description_with_where(self): |
890 rset = self.execute('INSERT CWUser E, EmailAddress EM: E login "X", E upassword "X", ' |
890 rset = self.execute('INSERT CWUser E, EmailAddress EM: E login "X", E upassword "X", ' |
891 'E primary_email EM, EM address "X", E in_group G ' |
891 'E primary_email EM, EM address "X", E in_group G ' |
892 'WHERE G name "managers"') |
892 'WHERE G name "managers"') |
893 self.assertEquals(list(rset.description[0]), ['CWUser', 'EmailAddress']) |
893 self.assertEquals(list(rset.description[0]), ['CWUser', 'EmailAddress']) |
894 |
894 |
895 # deletion queries tests ################################################## |
895 # deletion queries tests ################################################## |
896 |
896 |
897 def test_delete_1(self): |
897 def test_delete_1(self): |
898 self.execute("INSERT Personne Y: Y nom 'toto'") |
898 self.execute("INSERT Personne Y: Y nom 'toto'") |
899 rset = self.execute('Personne X WHERE X nom "toto"') |
899 rset = self.execute('Personne X WHERE X nom "toto"') |
900 self.assertEqual(len(rset.rows), 1) |
900 self.assertEqual(len(rset.rows), 1) |
901 self.execute("DELETE Personne Y WHERE Y nom 'toto'") |
901 self.execute("DELETE Personne Y WHERE Y nom 'toto'") |
902 rset = self.execute('Personne X WHERE X nom "toto"') |
902 rset = self.execute('Personne X WHERE X nom "toto"') |
903 self.assertEqual(len(rset.rows), 0) |
903 self.assertEqual(len(rset.rows), 0) |
904 |
904 |
905 def test_delete_2(self): |
905 def test_delete_2(self): |
906 rset = self.execute("INSERT Personne X, Personne Y, Societe Z : X nom 'syt', Y nom 'adim', Z nom 'Logilab', X travaille Z, Y travaille Z") |
906 rset = self.execute("INSERT Personne X, Personne Y, Societe Z : X nom 'syt', Y nom 'adim', Z nom 'Logilab', X travaille Z, Y travaille Z") |
907 self.assertEquals(len(rset), 1) |
907 self.assertEquals(len(rset), 1) |
908 self.assertEquals(len(rset[0]), 3) |
908 self.assertEquals(len(rset[0]), 3) |
909 self.assertEquals(rset.description[0], ('Personne', 'Personne', 'Societe')) |
909 self.assertEquals(rset.description[0], ('Personne', 'Personne', 'Societe')) |
958 sqlc = s.pool['system'] |
958 sqlc = s.pool['system'] |
959 sqlc.execute('SELECT * FROM recipients_relation') |
959 sqlc.execute('SELECT * FROM recipients_relation') |
960 self.assertEquals(len(sqlc.fetchall()), 0) |
960 self.assertEquals(len(sqlc.fetchall()), 0) |
961 sqlc.execute('SELECT * FROM owned_by_relation WHERE eid_from=%s'%eeid) |
961 sqlc.execute('SELECT * FROM owned_by_relation WHERE eid_from=%s'%eeid) |
962 self.assertEquals(len(sqlc.fetchall()), 0) |
962 self.assertEquals(len(sqlc.fetchall()), 0) |
963 |
963 |
964 def test_nonregr_delete_cache2(self): |
964 def test_nonregr_delete_cache2(self): |
965 eid = self.execute("INSERT Folder T: T name 'toto'")[0][0] |
965 eid = self.execute("INSERT Folder T: T name 'toto'")[0][0] |
966 self.commit() |
966 self.commit() |
967 # fill the cache |
967 # fill the cache |
968 self.execute("Any X WHERE X eid %(x)s", {'x': eid}, 'x') |
968 self.execute("Any X WHERE X eid %(x)s", {'x': eid}, 'x') |
977 self.assertEquals(rset.rows, []) |
977 self.assertEquals(rset.rows, []) |
978 rset = self.execute("Folder X WHERE X eid %(x)s", {'x': eid}, 'x') |
978 rset = self.execute("Folder X WHERE X eid %(x)s", {'x': eid}, 'x') |
979 self.assertEquals(rset.rows, []) |
979 self.assertEquals(rset.rows, []) |
980 rset = self.execute("Folder X WHERE X eid %s" %eid) |
980 rset = self.execute("Folder X WHERE X eid %s" %eid) |
981 self.assertEquals(rset.rows, []) |
981 self.assertEquals(rset.rows, []) |
982 |
982 |
983 # update queries tests #################################################### |
983 # update queries tests #################################################### |
984 |
984 |
985 def test_update_1(self): |
985 def test_update_1(self): |
986 self.execute("INSERT Personne Y: Y nom 'toto'") |
986 self.execute("INSERT Personne Y: Y nom 'toto'") |
987 rset = self.execute('Personne X WHERE X nom "toto"') |
987 rset = self.execute('Personne X WHERE X nom "toto"') |
988 self.assertEqual(len(rset.rows), 1) |
988 self.assertEqual(len(rset.rows), 1) |
989 self.execute("SET X nom 'tutu', X prenom 'original' WHERE X is Personne, X nom 'toto'") |
989 self.execute("SET X nom 'tutu', X prenom 'original' WHERE X is Personne, X nom 'toto'") |
990 rset = self.execute('Any Y, Z WHERE X is Personne, X nom Y, X prenom Z') |
990 rset = self.execute('Any Y, Z WHERE X is Personne, X nom Y, X prenom Z') |
991 self.assertEqual(tuplify(rset.rows), [('tutu', 'original')]) |
991 self.assertEqual(tuplify(rset.rows), [('tutu', 'original')]) |
992 |
992 |
993 def test_update_2(self): |
993 def test_update_2(self): |
994 self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
994 self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
995 #rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto"') |
995 #rset = self.execute('Any X, Y WHERE X nom "bidule", Y nom "toto"') |
996 #self.assertEqual(len(rset.rows), 1) |
996 #self.assertEqual(len(rset.rows), 1) |
997 #rset = self.execute('Any X, Y WHERE X travaille Y') |
997 #rset = self.execute('Any X, Y WHERE X travaille Y') |
998 #self.assertEqual(len(rset.rows), 0) |
998 #self.assertEqual(len(rset.rows), 0) |
999 self.execute("SET X travaille Y WHERE X nom 'bidule', Y nom 'toto'") |
999 self.execute("SET X travaille Y WHERE X nom 'bidule', Y nom 'toto'") |
1000 rset = self.execute('Any X, Y WHERE X travaille Y') |
1000 rset = self.execute('Any X, Y WHERE X travaille Y') |
1001 self.assertEqual(len(rset.rows), 1) |
1001 self.assertEqual(len(rset.rows), 1) |
1002 |
1002 |
1003 def test_update_2bis(self): |
1003 def test_update_2bis(self): |
1004 rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
1004 rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
1005 eid1, eid2 = rset[0][0], rset[0][1] |
1005 eid1, eid2 = rset[0][0], rset[0][1] |
1006 self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s", |
1006 self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s", |
1007 {'x': str(eid1), 'y': str(eid2)}) |
1007 {'x': str(eid1), 'y': str(eid2)}) |
1008 rset = self.execute('Any X, Y WHERE X travaille Y') |
1008 rset = self.execute('Any X, Y WHERE X travaille Y') |
1009 self.assertEqual(len(rset.rows), 1) |
1009 self.assertEqual(len(rset.rows), 1) |
1010 |
1010 |
1011 def test_update_2ter(self): |
1011 def test_update_2ter(self): |
1012 rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
1012 rset = self.execute("INSERT Personne X, Societe Y: X nom 'bidule', Y nom 'toto'") |
1013 eid1, eid2 = rset[0][0], rset[0][1] |
1013 eid1, eid2 = rset[0][0], rset[0][1] |
1014 self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s", |
1014 self.execute("SET X travaille Y WHERE X eid %(x)s, Y eid %(y)s", |
1015 {'x': unicode(eid1), 'y': unicode(eid2)}) |
1015 {'x': unicode(eid1), 'y': unicode(eid2)}) |
1016 rset = self.execute('Any X, Y WHERE X travaille Y') |
1016 rset = self.execute('Any X, Y WHERE X travaille Y') |
1017 self.assertEqual(len(rset.rows), 1) |
1017 self.assertEqual(len(rset.rows), 1) |
1018 |
1018 |
1019 ## def test_update_4(self): |
1019 ## def test_update_4(self): |
1020 ## self.execute("SET X know Y WHERE X ami Y") |
1020 ## self.execute("SET X know Y WHERE X ami Y") |
1021 |
1021 |
1022 def test_update_multiple1(self): |
1022 def test_update_multiple1(self): |
1023 peid1 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0] |
1023 peid1 = self.execute("INSERT Personne Y: Y nom 'tutu'")[0][0] |
1024 peid2 = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0] |
1024 peid2 = self.execute("INSERT Personne Y: Y nom 'toto'")[0][0] |
1025 self.execute("SET X nom 'tutu', Y nom 'toto' WHERE X nom 'toto', Y nom 'tutu'") |
1025 self.execute("SET X nom 'tutu', Y nom 'toto' WHERE X nom 'toto', Y nom 'tutu'") |
1026 self.assertEquals(self.execute('Any X WHERE X nom "toto"').rows, [[peid1]]) |
1026 self.assertEquals(self.execute('Any X WHERE X nom "toto"').rows, [[peid1]]) |
1052 def test_update_string_concat(self): |
1052 def test_update_string_concat(self): |
1053 beid = self.execute("INSERT Bookmark Y: Y title 'toto', Y path '/view'")[0][0] |
1053 beid = self.execute("INSERT Bookmark Y: Y title 'toto', Y path '/view'")[0][0] |
1054 self.execute('SET X title XN + %(suffix)s WHERE X is Bookmark, X title XN', {'suffix': u'-moved'}) |
1054 self.execute('SET X title XN + %(suffix)s WHERE X is Bookmark, X title XN', {'suffix': u'-moved'}) |
1055 newname = self.execute('Any XN WHERE X eid %(x)s, X title XN', {'x': beid}, 'x')[0][0] |
1055 newname = self.execute('Any XN WHERE X eid %(x)s, X title XN', {'x': beid}, 'x')[0][0] |
1056 self.assertEquals(newname, 'toto-moved') |
1056 self.assertEquals(newname, 'toto-moved') |
1057 |
1057 |
1058 def test_update_query_error(self): |
1058 def test_update_query_error(self): |
1059 self.execute("INSERT Personne Y: Y nom 'toto'") |
1059 self.execute("INSERT Personne Y: Y nom 'toto'") |
1060 self.assertRaises(Exception, self.execute, "SET X nom 'toto', X is Personne") |
1060 self.assertRaises(Exception, self.execute, "SET X nom 'toto', X is Personne") |
1061 self.assertRaises(QueryError, self.execute, "SET X nom 'toto', X has_text 'tutu' WHERE X is Personne") |
1061 self.assertRaises(QueryError, self.execute, "SET X nom 'toto', X has_text 'tutu' WHERE X is Personne") |
1062 self.assertRaises(QueryError, self.execute, "SET X login 'tutu', X eid %s" % cnx.user(self.session).eid) |
1062 self.assertRaises(QueryError, self.execute, "SET X login 'tutu', X eid %s" % cnx.user(self.session).eid) |
1063 |
1063 |
1064 |
1064 |
1065 # upassword encryption tests ################################################# |
1065 # upassword encryption tests ################################################# |
1066 |
1066 |
1067 def test_insert_upassword(self): |
1067 def test_insert_upassword(self): |
1068 rset = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'") |
1068 rset = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'") |
1069 self.assertEquals(len(rset.rows), 1) |
1069 self.assertEquals(len(rset.rows), 1) |
1070 self.assertEquals(rset.description, [('CWUser',)]) |
1070 self.assertEquals(rset.description, [('CWUser',)]) |
1071 self.assertRaises(Unauthorized, |
1071 self.assertRaises(Unauthorized, |
1072 self.execute, "Any P WHERE X is CWUser, X login 'bob', X upassword P") |
1072 self.execute, "Any P WHERE X is CWUser, X login 'bob', X upassword P") |
1073 cursor = self.pool['system'] |
1073 cursor = self.pool['system'] |
1074 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1074 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1075 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1075 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1076 passwd = cursor.fetchone()[0].getvalue() |
1076 passwd = cursor.fetchone()[0].getvalue() |
1077 self.assertEquals(passwd, crypt_password('toto', passwd[:2])) |
1077 self.assertEquals(passwd, crypt_password('toto', passwd[:2])) |
1078 rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword '%s'" % passwd) |
1078 rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword '%s'" % passwd) |
1079 self.assertEquals(len(rset.rows), 1) |
1079 self.assertEquals(len(rset.rows), 1) |
1080 self.assertEquals(rset.description, [('CWUser',)]) |
1080 self.assertEquals(rset.description, [('CWUser',)]) |
1081 |
1081 |
1082 def test_update_upassword(self): |
1082 def test_update_upassword(self): |
1083 cursor = self.pool['system'] |
1083 cursor = self.pool['system'] |
1084 rset = self.execute("INSERT CWUser X: X login 'bob', X upassword %(pwd)s", {'pwd': 'toto'}) |
1084 rset = self.execute("INSERT CWUser X: X login 'bob', X upassword %(pwd)s", {'pwd': 'toto'}) |
1085 self.assertEquals(rset.description[0][0], 'CWUser') |
1085 self.assertEquals(rset.description[0][0], 'CWUser') |
1086 rset = self.execute("SET X upassword %(pwd)s WHERE X is CWUser, X login 'bob'", |
1086 rset = self.execute("SET X upassword %(pwd)s WHERE X is CWUser, X login 'bob'", |
1087 {'pwd': 'tutu'}) |
1087 {'pwd': 'tutu'}) |
1088 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1088 cursor.execute("SELECT %supassword from %sCWUser WHERE %slogin='bob'" |
1089 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1089 % (SQL_PREFIX, SQL_PREFIX, SQL_PREFIX)) |
1090 passwd = cursor.fetchone()[0].getvalue() |
1090 passwd = cursor.fetchone()[0].getvalue() |
1091 self.assertEquals(passwd, crypt_password('tutu', passwd[:2])) |
1091 self.assertEquals(passwd, crypt_password('tutu', passwd[:2])) |
1092 rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword '%s'" % passwd) |
1092 rset = self.execute("Any X WHERE X is CWUser, X login 'bob', X upassword '%s'" % passwd) |
1093 self.assertEquals(len(rset.rows), 1) |
1093 self.assertEquals(len(rset.rows), 1) |
1094 self.assertEquals(rset.description, [('CWUser',)]) |
1094 self.assertEquals(rset.description, [('CWUser',)]) |
1095 |
1095 |
1096 # non regression tests #################################################### |
1096 # non regression tests #################################################### |
1097 |
1097 |
1098 def test_nonregr_1(self): |
1098 def test_nonregr_1(self): |
1099 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
1099 teid = self.execute("INSERT Tag X: X name 'tag'")[0][0] |
1100 self.execute("SET X tags Y WHERE X name 'tag', Y is State, Y name 'activated'") |
1100 self.execute("SET X tags Y WHERE X name 'tag', Y is State, Y name 'activated'") |
1101 rset = self.execute('Any X WHERE T tags X') |
1101 rset = self.execute('Any X WHERE T tags X') |
1102 self.assertEquals(len(rset.rows), 1, rset.rows) |
1102 self.assertEquals(len(rset.rows), 1, rset.rows) |
1111 self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s", |
1111 self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s", |
1112 {'g': geid, 't': teid}) |
1112 {'g': geid, 't': teid}) |
1113 rset = self.execute('Any X WHERE E eid %(x)s, E tags X', |
1113 rset = self.execute('Any X WHERE E eid %(x)s, E tags X', |
1114 {'x': teid}) |
1114 {'x': teid}) |
1115 self.assertEquals(rset.rows, [[geid]]) |
1115 self.assertEquals(rset.rows, [[geid]]) |
1116 |
1116 |
1117 def test_nonregr_3(self): |
1117 def test_nonregr_3(self): |
1118 """bad sql generated on the second query (destination_state is not |
1118 """bad sql generated on the second query (destination_state is not |
1119 detected as an inlined relation) |
1119 detected as an inlined relation) |
1120 """ |
1120 """ |
1121 rset = self.execute('Any S,ES,T WHERE S state_of ET, ET name "CWUser",' |
1121 rset = self.execute('Any S,ES,T WHERE S state_of ET, ET name "CWUser",' |
1149 'N filed_under T, W concerne N,' |
1149 'N filed_under T, W concerne N,' |
1150 'W filed_under A, A eid %s' % (teid1, teid2)) |
1150 'W filed_under A, A eid %s' % (teid1, teid2)) |
1151 self.assertEquals(rset1.rows, rset2.rows) |
1151 self.assertEquals(rset1.rows, rset2.rows) |
1152 self.assertEquals(rset1.rows, rset3.rows) |
1152 self.assertEquals(rset1.rows, rset3.rows) |
1153 self.assertEquals(rset1.rows, rset4.rows) |
1153 self.assertEquals(rset1.rows, rset4.rows) |
1154 |
1154 |
1155 def test_nonregr_6(self): |
1155 def test_nonregr_6(self): |
1156 self.execute('Any N,COUNT(S) GROUPBY N ORDERBY COUNT(N) WHERE S name N, S is State') |
1156 self.execute('Any N,COUNT(S) GROUPBY N ORDERBY COUNT(N) WHERE S name N, S is State') |
1157 |
1157 |
1158 def test_sqlite_encoding(self): |
1158 def test_sqlite_encoding(self): |
1159 """XXX this test was trying to show a bug on use of lower which only |
1159 """XXX this test was trying to show a bug on use of lower which only |
1160 occurs with non ascii string and misconfigured locale |
1160 occurs with non ascii string and misconfigured locale |
1161 """ |
1161 """ |
1162 self.execute("INSERT Tag X: X name %(name)s," |
1162 self.execute("INSERT Tag X: X name %(name)s," |
1213 cause: old variable ref inserted into a fresh rqlst copy |
1213 cause: old variable ref inserted into a fresh rqlst copy |
1214 (in RQLSpliter._complex_select_plan) |
1214 (in RQLSpliter._complex_select_plan) |
1215 """ |
1215 """ |
1216 self.skip('retry me once http://www.sqlite.org/cvstrac/tktview?tn=3773 is fixed') |
1216 self.skip('retry me once http://www.sqlite.org/cvstrac/tktview?tn=3773 is fixed') |
1217 self.execute('Any X ORDERBY D DESC WHERE X creation_date D') |
1217 self.execute('Any X ORDERBY D DESC WHERE X creation_date D') |
1218 |
1218 |
1219 def test_nonregr_extra_joins(self): |
1219 def test_nonregr_extra_joins(self): |
1220 ueid = self.session.user.eid |
1220 ueid = self.session.user.eid |
1221 teid1 = self.execute("INSERT Folder X: X name 'folder1'")[0][0] |
1221 teid1 = self.execute("INSERT Folder X: X name 'folder1'")[0][0] |
1222 teid2 = self.execute("INSERT Folder X: X name 'folder2'")[0][0] |
1222 teid2 = self.execute("INSERT Folder X: X name 'folder2'")[0][0] |
1223 neid1 = self.execute("INSERT Note X: X para 'note1'")[0][0] |
1223 neid1 = self.execute("INSERT Note X: X para 'note1'")[0][0] |
1249 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0] |
1249 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto'")[0][0] |
1250 self.execute("SET E in_group G, E in_state S, " |
1250 self.execute("SET E in_group G, E in_state S, " |
1251 "E firstname %(firstname)s, E surname %(surname)s " |
1251 "E firstname %(firstname)s, E surname %(surname)s " |
1252 "WHERE E eid %(x)s, G name 'users', S name 'activated'", |
1252 "WHERE E eid %(x)s, G name 'users', S name 'activated'", |
1253 {'x':ueid, 'firstname': u'jean', 'surname': u'paul'}, 'x') |
1253 {'x':ueid, 'firstname': u'jean', 'surname': u'paul'}, 'x') |
1254 |
1254 |
1255 def test_nonregr_u_owned_by_u(self): |
1255 def test_nonregr_u_owned_by_u(self): |
1256 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G " |
1256 ueid = self.execute("INSERT CWUser X: X login 'bob', X upassword 'toto', X in_group G " |
1257 "WHERE G name 'users'")[0][0] |
1257 "WHERE G name 'users'")[0][0] |
1258 rset = self.execute("CWUser U") |
1258 rset = self.execute("CWUser U") |
1259 self.assertEquals(len(rset), 3) # bob + admin + anon |
1259 self.assertEquals(len(rset), 3) # bob + admin + anon |